Adding StreamableHttpServerTransportProvider class and unit tests #290

ZachGerman · 2025-06-02T20:49:39Z

Not planned:

Backward-compatible endpoint combo

Motivation and Context

Trying to reach spec parity with TS and Python for Java. Will continue working on other aspects of this.

How Has This Been Tested?

Unit tests and integ tests using the SHTTP web client in mcp-spring.

Breaking Changes

N/A

Types of changes

Bug fix (non-breaking change which fixes an issue)
New feature (non-breaking change which adds functionality)
Breaking change (fix or feature that would cause existing functionality to change)
Documentation update

Checklist

I have read the MCP Documentation
My code follows the repository's style guidelines
New and existing tests pass locally
I have added appropriate error handling
I have added or updated documentation as needed

...ain/java/io/modelcontextprotocol/server/transport/StreamableHttpServerTransportProvider.java

chemicL

Hey 👋 Thanks for a comprehensive PR! I did my first round focusing on the main themes. Happy to offer guidance to cover the essential aspects (simple/stateful servers, multiple streams per session, lifecycle) if you'd like to push this forward.

...ain/java/io/modelcontextprotocol/server/transport/StreamableHttpServerTransportProvider.java

chemicL · 2025-06-17T19:44:52Z

...ain/java/io/modelcontextprotocol/server/transport/StreamableHttpServerTransportProvider.java

+				AsyncContext asyncContext = request.startAsync();
+				asyncContext.setTimeout(0); // No timeout
+
+				StreamableHttpSseStream sseStream = getOrCreateSseStream(sessionId);


Here the spec is being slightly violated IMO. We should aim to respect the MAY keyword to the best of our ability. The responses should go on a dedicated SSE stream if a stream is to be used, not the one opened initially with GET. The free hanging GET stream is meant for notifications or requests from the server. The SSE events that deal with the originating request should be sent over the stream associated with this request.

Here's some explanation from the specification on POST:

If the server initiates an SSE stream: ... The server MAY send JSON-RPC requests and notifications before sending a JSON-RPC response. These messages SHOULD relate to the originating client request. These requests and notifications MAY be [batched](https://www.jsonrpc.org/specification#batch). The server SHOULD NOT close the SSE stream before sending a JSON-RPC response per each received JSON-RPC request, unless the [session](https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#session-management) expires

and for GET:

If the server initiates an SSE stream: The server MAY send JSON-RPC requests and notifications on the stream. These requests and notifications MAY be [batched](https://www.jsonrpc.org/specification#batch). These messages SHOULD be unrelated to any concurrently-running JSON-RPC request from the client.

Same comment:
We can probably just add a single stream session for GET requests into the StreamableHttpSession class. I can add that to my list today.

chemicL · 2025-06-17T19:46:24Z

...ain/java/io/modelcontextprotocol/server/transport/StreamableHttpServerTransportProvider.java

+		// Create or get SSE stream for this session
+		StreamableHttpSseStream sseStream = getOrCreateSseStream(sessionId);
+		if (lastEventId != null) {
+			sseStream.replayEventsAfter(lastEventId);


As I understand it, in case of resumption, once the final response is streamed, the SSE stream should be closed.

Either client or server can close it at any time. Should we add a timer of sorts to hold the stream open and have it default to end right after stream completes?

My understanding is that we should close the stream on the server side once the Flux containing the json payloads completes. Not a timer though, just giving the user the ability to return a Flux that contains a set of notification/request/response payloads, where the response matching the initial request should be the last message from the Flux, followed by the onComplete signal. The other option would be for the user handling to return a Mono that signifies a non-SSE response directly to the POST message.

...ain/java/io/modelcontextprotocol/server/transport/StreamableHttpServerTransportProvider.java

chemicL · 2025-06-17T21:35:48Z

...ain/java/io/modelcontextprotocol/server/transport/StreamableHttpServerTransportProvider.java

+		String lastEventId = request.getHeader(LAST_EVENT_ID_HEADER);
+
+		// Create or get SSE stream for this session
+		StreamableHttpSseStream sseStream = getOrCreateSseStream(sessionId);


Resumption needs to happen on a brand new SSE stream, otherwise the client is unable to distinguish between different streams.

...ain/java/io/modelcontextprotocol/server/transport/StreamableHttpServerTransportProvider.java

ZachGerman · 2025-06-17T23:46:15Z

Thank you very much for all of the input @chemicL! I will begin making changes accordingly this afternoon.

ZachGerman · 2025-06-25T16:05:33Z

Today I'm targeting origin header validation and moving the dedicated GET stream to the StreamableHttpSession class, then adding an integ test for GET on /mcp to start the listening stream.
After that, I believe we should have all core functionality except sessionless and proper SSE response upgrade logic.

sivankri · 2025-06-26T02:56:45Z

Hi @ZachGerman I tried using your StreamableHttpServerTransportProvider.java file + java MCP SDK 0.10.0. My backend server is Jetty 12. The request from MCP Interceptor hangs in the below call.

return streamSession.handle(message).then(Mono.just(responseType)).onErrorReturn(ResponseType.IMMEDIATE);

This is the actuall line which gets blocked
--McpServerSession.java --> return var10000.flatMap(var10001::sendMessage);

Can you please check this?

tzolov · 2025-06-26T10:20:13Z

Today I'm targeting origin header validation and moving the dedicated GET stream to the StreamableHttpSession class, then adding an integ test for GET on /mcp to start the listening stream. After that, I believe we should have all core functionality except sessionless and proper SSE response upgrade logic.

@ZachGerman what do you mean by origin header validation? I hope it is not overlapping with the #284

chemicL

Added a few more comments.

chemicL · 2025-06-26T12:25:36Z

...ain/java/io/modelcontextprotocol/server/transport/StreamableHttpServerTransportProvider.java

+		if (sessionId == null) {
+			response.setContentType(APPLICATION_JSON);
+			response.setStatus(HttpServletResponse.SC_BAD_REQUEST);
+			response.getWriter().write(createErrorJson("Session ID missing in request header"));


modelcontextprotocol/modelcontextprotocol#282 please check this issue. I'm leaning towards separating the JSON-RPC lifecycle from the lower level transport lifecycle. In that world, GET and POST and session concepts are not at the JSON-RPC layer, so the lifecycle does not apply. In fact, the previous SSE transport did begin with a HTTP GET request before the initialization request could have been sent. For the stateless servers it should definitely be improved in the spec to mention that initialization is not required before other requests.

chemicL · 2025-06-26T14:06:23Z

...ain/java/io/modelcontextprotocol/server/transport/StreamableHttpServerTransportProvider.java

+		// Subscribe to the SSE stream and write events to the response
+		sseStream.getEventFlux().doOnNext(event -> {
+			try {
+				if (event.id() != null) {


Right, but do we emit events with no ID ? Reading the code I don't see a situation like this. My understanding of the mentioned spec is that if the ID is missing then the client would not use this id for Last-Event-ID tracking, but we are on the server side and we always generate an ID.

chemicL · 2025-06-26T14:11:45Z

...ain/java/io/modelcontextprotocol/server/transport/StreamableHttpServerTransportProvider.java

+		// Create or get SSE stream for this session
+		StreamableHttpSseStream sseStream = getOrCreateSseStream(sessionId);
+		if (lastEventId != null) {
+			sseStream.replayEventsAfter(lastEventId);


My understanding is that we should close the stream on the server side once the Flux containing the json payloads completes. Not a timer though, just giving the user the ability to return a Flux that contains a set of notification/request/response payloads, where the response matching the initial request should be the last message from the Flux, followed by the onComplete signal. The other option would be for the user handling to return a Mono that signifies a non-SSE response directly to the POST message.

ZachGerman · 2025-06-26T18:57:56Z

@sivankri: Changing the related logic after my meeting with Dariusz this morning, so (hopefully) your issue goes away with the new response type mapping.
@tzolov: It's related, but I don't think it's overlapping as I'm just adding the ability to set a list of allowed origins to the server transport provider and enforcing it if it's set.

viyaviya · 2025-06-27T08:40:03Z

merge commit 6c4830b is bad

ZachGerman · 2025-06-27T15:33:59Z

merge commit 6c4830b is bad

Oops! Thanks for the heads up! Fixed!

ZachGerman · 2025-06-27T18:07:35Z

@sivankri the new logic added streamTools to McpServerFeatures, used in the Async constructor, which facilitates a list of tool specifications that return Flux instead of Mono via the new AsyncStreamingToolSpecification record type and uses the return type of the call to differentiate between direct-HTTP and SSE-stream responses.

ZachGerman commented Jun 2, 2025

View reviewed changes

...ain/java/io/modelcontextprotocol/server/transport/StreamableHttpServerTransportProvider.java Outdated Show resolved Hide resolved

ZachGerman commented Jun 2, 2025

View reviewed changes

...ain/java/io/modelcontextprotocol/server/transport/StreamableHttpServerTransportProvider.java Outdated Show resolved Hide resolved

ZachGerman mentioned this pull request Jun 4, 2025

Support "Streamable HTTP" transport #72

Open

7 tasks

oliveregger mentioned this pull request Jun 16, 2025

MCP Server integration ahdis/matchbox#398

Open

chemicL requested changes Jun 17, 2025

View reviewed changes

chemicL reviewed Jun 17, 2025

View reviewed changes

...ain/java/io/modelcontextprotocol/server/transport/StreamableHttpServerTransportProvider.java Outdated Show resolved Hide resolved

chemicL reviewed Jun 26, 2025

View reviewed changes

Adding StreamableHttp server support via HTTPServlet with async support

7817da6

ZachGerman force-pushed the StreamableHttpServerTransportProvider branch from 6c4830b to 7817da6 Compare June 27, 2025 16:45

Adding StreamableHttpServerTransportProvider class and unit tests #290

Are you sure you want to change the base?

Adding StreamableHttpServerTransportProvider class and unit tests #290

Conversation

ZachGerman commented Jun 2, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation and Context

How Has This Been Tested?

Breaking Changes

Types of changes

Checklist

Uh oh!

Uh oh!

Uh oh!

chemicL left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

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!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

ZachGerman commented Jun 17, 2025

Uh oh!

ZachGerman commented Jun 25, 2025

Uh oh!

sivankri commented Jun 26, 2025

Uh oh!

tzolov commented Jun 26, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

chemicL left a comment

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!

ZachGerman commented Jun 26, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

viyaviya commented Jun 27, 2025

Uh oh!

ZachGerman commented Jun 27, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ZachGerman commented Jun 27, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

ZachGerman commented Jun 2, 2025 •

edited

Loading

tzolov commented Jun 26, 2025 •

edited

Loading

ZachGerman commented Jun 26, 2025 •

edited

Loading

ZachGerman commented Jun 27, 2025 •

edited

Loading

ZachGerman commented Jun 27, 2025 •

edited

Loading