Adding StreamableHttpServerTransportProvider class along with unit & integration 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

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.

[EDIT]

For the record, the above was wishful thinking. I agree we are only able to generate the session ID upon POST of the initialize request. The spec is in fact melding the json-rpc layer with http, so the session ID gets generated only when processing a initialization request:

A server using the Streamable HTTP transport MAY assign a session ID at initialization time, by including it in an Mcp-Session-Id header on the HTTP response containing the InitializeResult

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.

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

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.

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

mcp/src/main/java/io/modelcontextprotocol/server/McpAsyncStreamableHttpServer.java

mcp/src/main/java/io/modelcontextprotocol/spec/McpServerSession.java

mcp/src/main/java/io/modelcontextprotocol/server/McpAsyncStreamableHttpServer.java

mcp/src/main/java/io/modelcontextprotocol/server/McpAsyncServer.java

sivankri · 2025-07-01T15:59:02Z

@ZachGerman I tried running my mcp remote server against the mcp sdk generated from your branch. When i used the MCP inspector tool by setting query parameter transport=streamable-http, i could not get the connection established. Can you please help me with this?

ZachGerman · 2025-07-01T18:29:23Z

@sivankri I'm unfamiliar with the MCP inspector tool, but I imagine it's related to the fact that this JDK has public static final String LATEST_PROTOCOL_VERSION = "2024-11-05"; in the McpSchema spec class & that is currently used for all server instantiation. You can update it locally and the server from my branch, as well as @chemicL's webClient, should work with the update, but it will break a lot of tests. I plan on tackling protocol version negotiation soon if nobody else does.

mcp/src/main/java/io/modelcontextprotocol/spec/McpServerSession.java

… with Last-Event-Id header

ZachGerman · 2025-07-08T06:48:52Z

Once I realized that I needed to identify the transports in the exchange's session from the tool handlers, I realized I need to either pass the entire JSONRPCRequest to the handler (instead of the request.params()), or add a third request.id() parameter to tool handlers.

Either way, it brought more attention to the fact that we were using Object for JSONRPCRequest/Response id, so I added a little sub-class in the McpSchema to enforce the "(String | Number) && Non-null" requirements of that field.

Tomorrow I will be making the JSONRPCRequest id available to tool handlers and it can be passed to the McpAsyncServerExchange class methods in order to route notifications and requests to the related SSE stream.

Edit: Ended up going with a per-request exchange instance instead of passing transport ID to tool call handlers. This allows tool implementers to not worry about transport routing with their exchange method calls.

…nsportProvider

…age to their related transport streams

… Added extensive integration testing.

ZachGerman · 2025-07-13T08:32:36Z

Removed streamTools in favor of upgrading transport instances to SSE based on McpAsyncServerExchange method usage.
Also added a more extensive suite of integration tests.
Will speak to @chemicL about adding a config option at the server level to declare default transport type, as I believe he wants that to be SSE for session-bound requests.
Also planning to make event histories some kind of append-only ordered iterable struct, as using Map.keySet().stream().sorted() when replaying is a bit gross in hind-sight.

etherandrius · 2025-07-14T09:13:15Z

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

+	/** Sessions map, keyed by Session ID */
+	private static final Map<String, McpServerSession> sessions = new ConcurrentHashMap<>();


This should not be static in order to allow having multiple independent MCP Server instances on the same Java Server. For example if we want to support having an MCP on paths /mcp/v1/ and /mcp/v2/.

chemicL

Thank you for iterating over the Streamable HTTP server-side. I left a few comments. For now, please don't push more updates. Based on our exploratory work I am working on a parallel branch to have another transport type. I will focus on the WebFlux server and will let you know once I have all the architectural changes in so you can incorporate them into your PR and we can then make further progress.

chemicL · 2025-07-14T14:35:29Z

...src/main/java/io/modelcontextprotocol/client/transport/WebClientStreamableHttpTransport.java

+					if (response.statusCode().is2xxSuccessful()
+							&& message instanceof JSONRPCNotification notification) {
+						if (notification.method().equals("notifications/initialized")) {


It is not required to wait until the notification is sent out. If you see something in the spec, please let me know, but what I can see is this:

A server using the Streamable HTTP transport MAY assign a session ID at initialization time, by including it in an Mcp-Session-Id header on the HTTP response containing the InitializeResult

After we split the server session class, should we switch them all to mark the session as initialized once they send the InitializeResult response? (They currently only mark them as initialized upon receiving the initialized notification) Without that change or making the client wait to get the 200 from its initialized notification, we will have clients trying to establish streams prior to initialization being complete.

chemicL · 2025-07-14T14:36:25Z

mcp/src/main/java/io/modelcontextprotocol/spec/McpSchema.java

+	 */
+	@JsonSerialize(using = McpId.Serializer.class)
+	@JsonDeserialize(using = McpId.Deserializer.class)
+	public static final class McpId {


As I understand this change is not strictly required for the Streamable HTTP Server implementation to work, so I'd prefer this to not be a part of this already big PR.

chemicL · 2025-07-14T14:40:47Z

mcp/src/main/java/io/modelcontextprotocol/server/McpAsyncServerExchange.java

+	 * This is for tool writers to use if they want to send their tool response over an
+	 * SSE stream without using any other McpAsyncServerExchange methods
+	 */
+	public void upgradeTransport() {


IMO this is not a correct way to go about it. I'd suggest that if in the future someone asks for dynamic upgrade we should do it differently by providing a hook to inspect the request first, instruct the consuming session and transport that it will not only be a response but a bit more, and only then feed an exchange that has the SSE stream producer to feed with user interactions, but not in a stateful way like this.

For now I suggest we keep it simple as other SDKs:

notifications and responses -> handle as 201 ACCEPTED

requests

sessionless -> application/json response expected

with sessions -> always upgrade to SSE

chemicL · 2025-07-14T14:42:13Z

mcp/src/main/java/io/modelcontextprotocol/server/McpAsyncServerExchange.java

@@ -92,14 +121,17 @@ public McpSchema.Implementation getClientInfo() {
 	 * Specification</a>
 	 */
 	public Mono<McpSchema.CreateMessageResult> createMessage(McpSchema.CreateMessageRequest createMessageRequest) {
+
+		establishSseStream();


Methods returning Flux/Mono need to be lazy, even if they have side-effects. Another issue with this is explained in the comment above.

mcp/src/main/java/io/modelcontextprotocol/spec/McpServerSession.java

chemicL · 2025-07-14T14:46:39Z

mcp/src/main/java/io/modelcontextprotocol/spec/McpServerSession.java

+	 * @param <T> the type of the response that is expected as a result of handling the
+	 * request.
+	 */
+	public interface StreamingRequestHandler<T> extends RequestHandler<T> {


This is unused and should be removed I suppose?

Yes. Forgot to remove this when I removed streamTools.

chemicL · 2025-07-14T14:47:31Z

mcp/src/main/java/io/modelcontextprotocol/spec/McpServerSession.java

+	 * with a connected client.
+	 */
+	@FunctionalInterface
+	public interface StreamableHttpSessionFactory {


Adding the boolean flag causes adding another factory and infects the transports with empty implementations. Unfortunately, we will need a parallel type hierarchy.

chemicL · 2025-07-14T14:48:32Z

...src/main/java/io/modelcontextprotocol/client/transport/WebClientStreamableHttpTransport.java

@@ -117,10 +118,6 @@ public static Builder builder(WebClient.Builder webClientBuilder) {
 	public Mono<Void> connect(Function<Mono<McpSchema.JSONRPCMessage>, Mono<McpSchema.JSONRPCMessage>> handler) {
 		return Mono.deferContextual(ctx -> {
 			this.handler.set(handler);
-			if (openConnectionOnStartup) {


I agree this can be removed but we also have to remove it from the constructor.

chemicL · 2025-07-14T14:49:35Z

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

+
+	@Override
+	public void setSessionFactory(McpServerSession.Factory sessionFactory) {
+		// Required but not used for this implementation


We can't implement it this way. This transport needs to be of a different kind.

chemicL · 2025-07-14T14:51:44Z

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

+
+					// Only set content type for requests
+					if (message instanceof McpSchema.JSONRPCRequest) {
+						response.setContentType(APPLICATION_JSON);


So here you set it to application/json but later you let the user change this to text/event-stream. Unfortunately with other transports it might not be as straightforward and the builders are more type-safe and will forbid rolling back in such a way. That's why I am proposing to keep it simple for now as in another comment and always upgrade to SSE for requests.

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 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

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

ZachGerman commented Jun 30, 2025

View reviewed changes

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

ZachGerman commented Jun 30, 2025

View reviewed changes

mcp/src/main/java/io/modelcontextprotocol/server/McpAsyncStreamableHttpServer.java Outdated Show resolved Hide resolved

ZachGerman commented Jun 30, 2025

View reviewed changes

mcp/src/main/java/io/modelcontextprotocol/spec/McpServerSession.java Outdated Show resolved Hide resolved

sivankri reviewed Jul 1, 2025

View reviewed changes

mcp/src/main/java/io/modelcontextprotocol/server/McpAsyncStreamableHttpServer.java Outdated Show resolved Hide resolved

sivankri reviewed Jul 1, 2025

View reviewed changes

mcp/src/main/java/io/modelcontextprotocol/server/McpAsyncServer.java Outdated Show resolved Hide resolved

stantonk mentioned this pull request Jul 1, 2025

feat: implement HttpClient Streamable HTTP transport #337

Closed

ZachGerman changed the title ~~Adding StreamableHttpServerTransportProvider class and unit tests~~ Adding StreamableHttpServerTransportProvider class along with unit & integration tests Jul 1, 2025

sivankri reviewed Jul 2, 2025

View reviewed changes

mcp/src/main/java/io/modelcontextprotocol/spec/McpServerSession.java Outdated Show resolved Hide resolved

ZachGerman and others added 3 commits July 8, 2025 06:17

Adding StreamableHttp server support via HTTPServlet with async support

2c07ee8

Integrated dedicated GET/listening stream and multi-transport management

a01872d

Fixed backward-compatibility issues

2a97519

Adding Session-owned event histories and stream replayability via GET…

09d47b2

… with Last-Event-Id header

ZachGerman force-pushed the StreamableHttpServerTransportProvider branch from 77333d1 to 244208d Compare July 8, 2025 06:18

Enforce "String | Number" for JSONRPC message IDs

25ff3dc

ZachGerman force-pushed the StreamableHttpServerTransportProvider branch from 244208d to 25ff3dc Compare July 8, 2025 06:40

ZachGerman and others added 3 commits July 8, 2025 22:42

Merge branch 'modelcontextprotocol:main' into StreamableHttpServerTra…

91cacc1

…nsportProvider

Added per-request exchange instances to route server-sent JSONRPCMess…

9d7f9f7

…age to their related transport streams

Removing streamTools in favor of exchange-driven transport upgrading.…

29a39f7

… Added extensive integration testing.

etherandrius reviewed Jul 14, 2025

View reviewed changes

chemicL requested changes Jul 14, 2025

View reviewed changes

		/** Sessions map, keyed by Session ID */
		private static final Map<String, McpServerSession> sessions = new ConcurrentHashMap<>();

Adding StreamableHttpServerTransportProvider class along with unit & integration tests #290

Are you sure you want to change the base?

Adding StreamableHttpServerTransportProvider class along with unit & integration 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!

Uh oh!

Uh oh!

Uh oh!

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!

chemicL Jun 26, 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!

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!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

sivankri commented Jul 1, 2025

Uh oh!

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

Uh oh!

Uh oh!

Uh oh!

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

Uh oh!

Uh oh!

ZachGerman commented Jul 13, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chemicL left a comment

Choose a reason for hiding this comment

Uh oh!

ZachGerman commented Jun 2, 2025 •

edited

Loading

tzolov commented Jun 26, 2025 •

edited

Loading

chemicL 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

ZachGerman commented Jul 1, 2025 •

edited

Loading

ZachGerman commented Jul 8, 2025 •

edited

Loading

chemicL Jul 14, 2025 •

edited

Loading

ZachGerman Jul 14, 2025 •

edited

Loading

ZachGerman Jul 14, 2025 •

edited

Loading