[MCP Server] Support Streamable HTTP and deprecate SSE in MCP server #4162

rithin-pullela-aws · 2025-09-10T16:53:20Z

Description

This PR deprecates SSE server and now provides the Streamable HTTP server.

The new endpoint: http://localhost:9200/_plugins/_ml/mcp/stream

Updated the mcp java sdk to 0.12.1 version

NOTE: The MCP Client (connectors) with the new protocol is not implemented yet. Will do it in a different PR to not clutter the current code.

Testing:

Enable MCP Server:

PUT /_cluster/settings/
{
    "persistent": {
        "plugins.ml_commons.mcp_server_enabled":"true"
    }
}

Register a Tool:

POST /_plugins/_ml/mcp/tools/_register
{
    "tools": [
        {
            "type": "ListIndexTool",
            "name": "ListIndexTool"
        }
    ]
}

Mock Client behavior via Rest Calls:

1. Initialize Connection:

POST /_plugins/_ml/mcp

{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "initialize",
    "params": {
        "protocolVersion": "2025-03-26",
        "capabilities": {
            "roots": {
                "listChanged": true
            },
            "sampling": {}
        },
        "clientInfo": {
            "name": "test-client",
            "version": "1.0.0"
        }
    }
}

Expected response:

200 OK
{
    "jsonrpc": "2.0",
    "id": 1,
    "result": {
        "protocolVersion": "2025-03-26",
        "capabilities": {
            "logging": {},
            "prompts": {
                "listChanged": false
            },
            "resources": {
                "subscribe": false,
                "listChanged": false
            },
            "tools": {
                "listChanged": true
            }
        },
        "serverInfo": {
            "name": "OpenSearch-MCP-Stateless-Server",
            "version": "0.1.0"
        },
        "instructions": "OpenSearch MCP Stateless Server - provides access to ML tools without sessions"
    }
}

2. Initialization Complete Notification

POST /_plugins/_ml/mcp
{
    "jsonrpc": "2.0",
    "method": "notifications/initialized",
    "params": {}
}

Expected response:

202 OK

3. List Available Tools

POST /_plugins/_ml/mcp
{
    "jsonrpc": "2.0",
    "id": 2,
    "method": "tools/list",
    "params": {}
}

Expected response:

200 OK
{
    "jsonrpc": "2.0",
    "id": 2,
    "result": {
        "tools": [
            {
                "name": "ListIndexTool",
                "description": "This tool returns information about indices in the OpenSearch cluster along with the index `health`, `status`, `index`, `uuid`, `pri`, `rep`, `docs.count`, `docs.deleted`, `store.size`, `pri.store. size `, `pri.store.size`, `pri.store`. Optional arguments: 1. `indices`, a comma-delimited list of one or more indices to get information from (default is an empty list meaning all indices). Use only valid index names. 2. `local`, whether to return information from the local node only instead of the cluster manager node (Default is false)",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "indices": {
                            "type": "array",
                            "items": {
                                "type": "string"
                            },
                            "description": "OpenSearch index name list, separated by comma. for example: [\"index1\", \"index2\"], use empty array [] to list all indices in the cluster"
                        }
                    },
                    "additionalProperties": false
                }
            },
            {
                "name": "ListIndexTool",
                "description": "This tool returns information about indices in the OpenSearch cluster along with the index `health`, `status`, `index`, `uuid`, `pri`, `rep`, `docs.count`, `docs.deleted`, `store.size`, `pri.store. size `, `pri.store.size`, `pri.store`. Optional arguments: 1. `indices`, a comma-delimited list of one or more indices to get information from (default is an empty list meaning all indices). Use only valid index names. 2. `local`, whether to return information from the local node only instead of the cluster manager node (Default is false)",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "indices": {
                            "type": "array",
                            "items": {
                                "type": "string"
                            },
                            "description": "OpenSearch index name list, separated by comma. for example: [\"index1\", \"index2\"], use empty array [] to list all indices in the cluster"
                        }
                    },
                    "additionalProperties": false
                }
            }
        ]
    }
}

4. Call a Tool

POST /_plugins/_ml/mcp
{
    "jsonrpc": "2.0",
    "id": 3,
    "method": "tools/call",
    "params": {
        "name": "ListIndexTool",
        "arguments": {
            "indices": []
        }
    }
}

Expected Response:

{
    "jsonrpc": "2.0",
    "id": 3,
    "result": {
        "content": [
            {
                "type": "text",
                "text": "row,health,status,index,uuid,pri(number of primary shards),rep(number of replica shards),docs.count(number of available documents),docs.deleted(number of deleted documents),store.size(store size of primary and replica shards),pri.store.size(store size of primary shards)\n1,green,open,.plugins-ml-config,nKyzDAupTGCwuybs9S_iBA,1,0,1,0,3.9kb,3.9kb\n2,green,open,.plugins-ml-mcp-tools,k1QwQKmXSeqRexmB2JDJiw,1,0,1,0,5kb,5kb\n"
            }
        ],
        "isError": false
    }
}

Other ways of Testing:

You can also test with various MCP clients

1. Fast MCP client:

import asyncio, logging
from fastmcp import Client

async def main():
    async with Client("http://localhost:9200/_plugins/_ml/mcp") as client:
        for t in await client.list_tools():
            print(t.name)
        r = await client.call_tool("ListIndexTool", {})
        print("result ->", r)

asyncio.run(main())

2. PostMan MCP client:
In the latest versions of postman we can directly test an MCP server by creating an MCP kind of request

Related Issues

Resolves #3813 since we don't need the extra variable in streamable_http protocol.

Check List

New functionality includes testing.
New functionality has been documented.
API changes companion pull request created.
Commits are signed per the DCO using --signoff.
Public documentation issue/PR created.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

plugin/src/main/java/org/opensearch/ml/plugin/MachineLearningPlugin.java

rithin-pullela-aws · 2025-09-10T19:42:23Z

The failing ITs are because of usage of deprecated Bedrock models.

{"status":404,"error":{"type":"OpenSearchStatusException","reason":"System Error","details":"Error from remote service: {\"message\":\"This model version has reached the end of its life. Please refer to the AWS documentation for more details.\"}"}}

Will raise another PR to fix

ylwu-amzn · 2025-09-10T19:59:48Z

plugin/src/main/java/org/opensearch/ml/rest/mcpserver/RestMcpStatelessStreamingAction.java

+import lombok.extern.log4j.Log4j2;
+
+@Log4j2
+public class RestMcpStatelessStreamingAction extends BaseRestHandler {


What does stateless mean? Do we also have or plan to build a stateful action?

MCP protocol supports streamable HTTP in a stateful manner as well, but that would require streaming to be enabled.
While we do not have any plans yet, by naming this endpoint stateless we are keeping the door open for potential future use cases. Also it makes it clear to readers what underlying methodology is being used keeping it consistent with MCP JAVA SDK

For future extension , should we create a new API or extend current one ? If need to extend current one, the sateless name seems not reasonable.

From https://modelcontextprotocol.io/specification/2025-03-26/basic/transports

The server MUST provide a single HTTP endpoint path (hereafter referred to as the MCP endpoint) that supports both POST and GET methods. For example, this could be a URL like https://example.com/mcp.

Does that mean we should use one HTTP endpoint for both stateless and stateful?

Renamed the Rest and Transport Actions to remove "stateless"

ylwu-amzn · 2025-09-10T20:06:46Z

plugin/src/main/java/org/opensearch/ml/rest/mcpserver/RestMcpStatelessStreamingAction.java

+        return channel -> {
+            try {
+                if (request.content() == null) {
+                    sendErrorResponse(channel, null, -32700, "Parse error: empty body");


What does this magic number mean: -32700? Define these json RPC error code as constants ?

These are defined by the JSON RPC spec which is used by MCP Protocol. These numbers indicate different errors to the client.

Reference: https://www.jsonrpc.org/specification#:~:text=NOT%20be%20included.-,5.1%20Error%20object,-When%20a%20rpc

Used constants for JSON-RPC error codes

ylwu-amzn · 2025-09-10T20:12:59Z

plugin/src/main/java/org/opensearch/ml/action/mcpserver/McpStatelessServerHolder.java

+                .async(mcpStatelessServerTransportProvider)
+                .serverInfo("OpenSearch-MCP-Stateless-Server", "0.1.0")
+                .capabilities(serverCapabilities)
+                .instructions("OpenSearch MCP Stateless Server - provides access to ML tools without sessions")


OpenSearch MCP Stateless Server this name looks confusing, do we have another type of stateful server ?

The description here does not affect the functionality. But I believe it would not hurt to have an explicit mention of the server being stateless.

Can the client side see/get the instructions and serverInfo ?

If they can see these information, let's design carefully.

Will remove the stateless

Updated the server desc to remove "stateless"

ylwu-amzn · 2025-09-10T20:27:50Z

plugin/src/main/java/org/opensearch/ml/rest/mcpserver/RestMcpStatelessStreamingAction.java

+                OpenSearchMcpStatelessServerTransportProvider transportProvider = McpStatelessServerHolder
+                    .getMcpStatelessServerTransportProvider();
+                if (transportProvider == null || !transportProvider.isHandlerReady()) {
+                    log.error("MCP transport provider not ready - server may not be properly initialized");


Is this an error ? Change to warn level?

Yes, this is an error and we are throwing an exception in the next lines of the code.

I see McpStatelessServerHolder.java line 72 using warn level log. Let's keep consistent?

Removed the warning logging since we do not need to check isHandlerReady anymore

ylwu-amzn · 2025-09-10T21:20:05Z

plugin/src/main/java/org/opensearch/ml/rest/mcpserver/RestMcpStatelessStreamingAction.java

+                // Requests: capture id for any downstream error mapping
+                final Object id = (msg instanceof McpSchema.JSONRPCRequest) ? ((McpSchema.JSONRPCRequest) msg).id() : null;
+
+                transportProvider.handleRequest(requestBody).subscribe(response -> {


Line 86 already called deserializeJsonRpcMessage , why not pass msg to handleRequest method?

Good point, I will refactor the method to accept the McpSchema.JSONRPCMessage instead of the whole requestBody

We need a transport action to control permission on API level.

Added a transport Action

Signed-off-by: rithin-pullela-aws <[email protected]>

codecov · 2025-09-18T20:48:48Z

Codecov Report

❌ Patch coverage is 87.07865% with 46 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.17%. Comparing base (7bf4b54) to head (0a73c33).
⚠️ Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
.../ml/action/mcpserver/McpStatelessServerHolder.java	83.33%	13 Missing and 2 partials ⚠️
...opensearch/ml/action/mcpserver/McpToolsHelper.java	84.12%	8 Missing and 2 partials ⚠️
...nsearch/ml/rest/mcpserver/RestMcpServerAction.java	84.48%	8 Missing and 1 partial ⚠️
...OpenSearchMcpStatelessServerTransportProvider.java	80.95%	3 Missing and 1 partial ⚠️
...cpserver/responses/server/MLMcpServerResponse.java	88.88%	1 Missing and 2 partials ⚠️
...server/TransportMcpToolsRegisterOnNodesAction.java	62.50%	2 Missing and 1 partial ⚠️
.../ml/action/mcpserver/TransportMcpServerAction.java	97.91%	0 Missing and 1 partial ⚠️
...cpserver/TransportMcpToolsRemoveOnNodesAction.java	75.00%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #4162      +/-   ##
============================================
+ Coverage     82.02%   82.17%   +0.15%     
+ Complexity     9195     9186       -9     
============================================
  Files           789      788       -1     
  Lines         39541    39437     -104     
  Branches       4387     4400      +13     
============================================
- Hits          32432    32407      -25     
+ Misses         5219     5139      -80     
- Partials       1890     1891       +1

Flag	Coverage Δ
ml-commons	`82.17% <87.07%> (+0.15%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

Signed-off-by: Rithin Pullela <[email protected]>

Signed-off-by: rithin-pullela-aws <[email protected]>

ylwu-amzn · 2025-09-22T22:31:01Z

plugin/src/main/java/org/opensearch/ml/action/mcpserver/McpStatelessServerHolder.java

+                    .prompts(false)           // We don't support prompts
+                    .build();
+
+                log.info("Building MCP server without pre-loaded tools (dynamic loading)...");


What does "pre-loaded tools" mean?

ylwu-amzn · 2025-09-22T22:32:33Z

plugin/src/main/java/org/opensearch/ml/action/mcpserver/McpStatelessServerHolder.java

+                            error -> log.error("Initial tool loading failed", error)
+                        )
+                );
+                initialized = true;


I think we should move this line inside autoLoadAllMcpTools success branch ? If load MCP tools failed, we should not set initialized as true right?

ylwu-amzn · 2025-09-22T22:41:12Z

plugin/src/main/java/org/opensearch/ml/action/mcpserver/McpStatelessServerHolder.java

+            .wrap(r -> { log.debug("Auto reload mcp tools schedule job run successfully!"); }, e -> {
+                log.error(e.getMessage(), e);
+            });
+        threadPool


Use dedicated thread pool ?
Refer to MLExecuteTaskRunner

threadPool.executor(EXECUTE_THREAD_POOL)

ylwu-amzn · 2025-09-22T22:45:33Z

plugin/src/main/java/org/opensearch/ml/action/mcpserver/McpToolsHelper.java

+        toolNames.forEach(toolName -> queryBuilder.should(QueryBuilders.matchQuery("name", toolName)));
+        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
+        searchSourceBuilder.query(queryBuilder);
+        searchRequest.source(searchSourceBuilder);


Add size ?

searchRequest.source().size(MAX_TOOL_NUMBER);

ylwu-amzn · 2025-09-22T22:48:48Z

plugin/src/main/java/org/opensearch/ml/action/mcpserver/McpToolsHelper.java

+                    McpToolRegisterInput mcpTool = parseMcpTool(x.getSourceAsString());
+                    mcpTools.add(mcpTool);
+                } catch (IOException e) {
+                    listener.onFailure(e);


If have multiple MCP tools parsing failed, will call this listener.onFailure(e); multiple times. Is that ok ? I remember that will trigger some exception. Have you done some test?

Should we use List to track all exceptions, then send back all exceptions with listener.onFailure(...) ?

ylwu-amzn · 2025-09-22T22:50:49Z

plugin/src/main/java/org/opensearch/ml/action/mcpserver/McpToolsHelper.java

+
+    public void searchToolsWithPrimaryTermAndSeqNo(List<String> toolNames, ActionListener<SearchResponse> listener) {
+        SearchRequest searchRequest = buildSearchRequest(toolNames);
+        searchRequest.source().seqNoAndPrimaryTerm(true);


Add size ?

searchRequest.source().size(MAX_TOOL_NUMBER);

ylwu-amzn · 2025-09-22T23:04:28Z

...src/main/java/org/opensearch/ml/action/mcpserver/TransportMcpToolsRegisterOnNodesAction.java

-                    .getMcpAsyncServerInstance()
+            // Use putIfAbsent to make check-and-act atomic
+            Long previousVersion = McpStatelessServerHolder.IN_MEMORY_MCP_TOOLS.putIfAbsent(tool.getName(), tool.getVersion());
+            if (previousVersion == null) {


What if previousVersion != null ? I think we should delete the old version and add the new version, right?

ylwu-amzn · 2025-09-22T23:12:55Z

The current design will only start one MCP server inside OpenSearch cluster with one custom set of tools. That sounds not flexible if multiple users want to have different MCP servers with different sets of tools. No need to change the design in this PR. But some draft idea for supporting multiple MCP servers:

Register MCP server

POST /_plugins/_ml/mcp_servers/_register
{
    "name": "search_index_server",
    "description": "This is a MCP server for searching OpenSearch index.",
    "tools": [
        {
            "type": "ListIndexTool"
        },
        {
            "type": "SearchIndexTool"
        }
    ]
}

Sample response

{
"mcp_server_id": "abc123"
}

Initialize connection

POST /_plugins/_ml/mcp_servers/abc123/mcp

{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "initialize",
    "params": {
        "protocolVersion": "2025-03-26",
        "capabilities": {
            "roots": {
                "listChanged": true
            },
            "sampling": {}
        },
        "clientInfo": {
            "name": "test-client",
            "version": "1.0.0"
        }
    }
}

rithin-pullela-aws requested a deployment to ml-commons-cicd-env-require-approval September 10, 2025 16:54 — with GitHub Actions Waiting

rithin-pullela-aws had a problem deploying to ml-commons-cicd-env-require-approval September 10, 2025 18:12 — with GitHub Actions Failure

rithin-pullela-aws had a problem deploying to ml-commons-cicd-env-require-approval September 10, 2025 18:12 — with GitHub Actions Error

rithin-pullela-aws commented Sep 10, 2025

View reviewed changes

plugin/src/main/java/org/opensearch/ml/plugin/MachineLearningPlugin.java Outdated Show resolved Hide resolved

rithin-pullela-aws marked this pull request as ready for review September 10, 2025 19:51

rithin-pullela-aws requested review from b4sjoo, dhrubo-os, mingshl, jngz-es, model-collapse, rbhavna, ylwu-amzn, zane-neo, Zhangxunmt, austintlee, HenryL27, sam-herman, xinyual and pyek-bot as code owners September 10, 2025 19:51

ylwu-amzn reviewed Sep 10, 2025

View reviewed changes

spotless

c2c9dbc

Signed-off-by: rithin-pullela-aws <[email protected]>

rithin-pullela-aws temporarily deployed to ml-commons-cicd-env-require-approval September 18, 2025 19:42 — with GitHub Actions Inactive

rithin-pullela-aws had a problem deploying to ml-commons-cicd-env-require-approval September 18, 2025 19:42 — with GitHub Actions Failure

rithin-pullela-aws temporarily deployed to ml-commons-cicd-env-require-approval September 18, 2025 23:53 — with GitHub Actions Inactive

rithin-pullela-aws temporarily deployed to ml-commons-cicd-env-require-approval September 19, 2025 00:57 — with GitHub Actions Inactive

rithin-pullela-aws added 2 commits September 19, 2025 10:35

Merge branch 'main' into streamable-http

6e44bf1

Signed-off-by: Rithin Pullela <[email protected]>

remove older imports

69e9bc1

Signed-off-by: rithin-pullela-aws <[email protected]>

rithin-pullela-aws requested a deployment to ml-commons-cicd-env-require-approval September 19, 2025 23:23 — with GitHub Actions Waiting

Merge branch 'main' into streamable-http

0a73c33

rithin-pullela-aws temporarily deployed to ml-commons-cicd-env-require-approval September 22, 2025 20:30 — with GitHub Actions Inactive

ylwu-amzn reviewed Sep 22, 2025

View reviewed changes

rithin-pullela-aws requested a deployment to ml-commons-cicd-env-require-approval September 23, 2025 00:16 — with GitHub Actions Waiting

[MCP Server] Support Streamable HTTP and deprecate SSE in MCP server #4162

Are you sure you want to change the base?

[MCP Server] Support Streamable HTTP and deprecate SSE in MCP server #4162

Uh oh!

Conversation

rithin-pullela-aws commented Sep 10, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Testing:

Other ways of Testing:

Related Issues

Check List

Uh oh!

Uh oh!

rithin-pullela-aws commented Sep 10, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ylwu-amzn Sep 10, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

codecov bot commented Sep 18, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ylwu-amzn Sep 22, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rithin-pullela-aws commented Sep 10, 2025 •

edited

Loading

ylwu-amzn Sep 10, 2025 •

edited

Loading

codecov bot commented Sep 18, 2025 •

edited

Loading

ylwu-amzn Sep 22, 2025 •

edited

Loading