Adding StreamableHttpServerTransportProvider class and unit tests #290

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

Still missing:

Origin header validation (ongoing discussions about spec for that in Discord)
Backward-compatible endpoint combo
Session & Transport spec class changes to actually use the new transport

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 a partially-complete sHTTP client transport class, but I didn't include the integ test file as the client isn't complete.

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

Additional context

This is part of a larger commit seen here: #289

ZachGerman · 2025-06-02T20:53:14Z

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

+	 */
+	public class StreamableHttpSseStream {
+
+		private final Sinks.Many<SseEvent> eventSink = Sinks.many().multicast().onBackpressureBuffer();


Meant to change this to unicast. Will address.

ZachGerman · 2025-06-02T23:22:48Z

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

+
+		private String mcpEndpoint;
+
+		private Supplier<String> sessionIdProvider = () -> UUID.randomUUID().toString();


Should change the value here to DEFAULT_SESSION_ID_PROVIDER. Will address.

…nsportProvider

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.

chemicL · 2025-06-17T19:13:14Z

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

+
+		return Flux.fromIterable(sessions.values())
+			.flatMap(session -> session.sendNotification(method, params)
+				.doOnError(


I think we should provide a handler so users can provide a callback here.

chemicL · 2025-06-17T19:13:48Z

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

+
+	@Override
+	public Mono<Void> closeGracefully() {
+		isClosing.set(true);


This needs to be lazy, consider using Mono.defer().

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

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

+		isClosing.set(true);
+		logger.debug("Initiating graceful shutdown with {} active sessions", sessions.size());
+
+		return Flux.fromIterable(sessions.values()).flatMap(McpServerSession::closeGracefully).then();


Errors should be handled and onErrorComplete should be added for each inner chain

chemicL · 2025-06-17T19:15:42Z

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

+	 * @throws IOException If an I/O error occurs
+	 */
+	@Override
+	protected void doGet(HttpServletRequest request, HttpServletResponse response)


There needs to be a configuration that allows session-less servers.

chemicL · 2025-06-17T19:16:58Z

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

+		Enumeration<String> headerNames = request.getHeaderNames();
+		while (headerNames.hasMoreElements()) {
+			String headerName = headerNames.nextElement();
+			logger.debug("Header: {} = {}", headerName, request.getHeader(headerName));


This log line could be aggregated into one with the debug log for receiving a request so only one line would be appended to the log.

chemicL · 2025-06-17T19:33:54Z

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

+			}
+
+			// Handle the message
+			session.handle(message).block(); // Block for servlet compatibility


Shouldn't the headers be written first? If we call into the handler first, it should attempt to respond, which might trigger flushing the response before headers are set.

chemicL · 2025-06-17T19:38:31Z

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

+
+			// For requests that expect responses, we need to set up an SSE stream
+			if (message instanceof McpSchema.JSONRPCRequest && acceptsEventStream) {
+				// Set up SSE connection


This logic should be encapsulated into how the session decides to go about it. If it decides it's a simple response, no SSE should be started.

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.

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.

chemicL · 2025-06-17T19:47:48Z

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

+				}).subscribe();
+			}
+			else if (message instanceof McpSchema.JSONRPCRequest) {
+				// Client doesn't accept SSE, we'll return a regular JSON response


This should not be possible. It's up to the server to decide, the clients are obliged to support SSE.

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.

chemicL · 2025-06-17T21:39:12Z

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

+	/**
+	 * Implementation of McpServerTransport for Streamable HTTP sessions.
+	 */
+	private class StreamableHttpServerTransport implements McpServerTransport {


I am guessing for Streamable HTTP we need a new abstraction, a McpStreamableServerTransport that has a different set of methods allowing opening a new stream and staying with the McpServerTransport in case of simple (sessionless) servers.

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

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

Adding StreamableHttpServerTransportProvider class and unit tests

0576a88

ZachGerman commented Jun 2, 2025

View reviewed changes

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

Merge branch 'modelcontextprotocol:main' into StreamableHttpServerTra…

11a7d3e

…nsportProvider

chemicL requested changes Jun 17, 2025

View reviewed changes

chemicL reviewed Jun 17, 2025

View reviewed changes


		private String mcpEndpoint;

		private Supplier<String> sessionIdProvider = () -> UUID.randomUUID().toString();

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

Additional context

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

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!

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!

ZachGerman commented Jun 17, 2025

Uh oh!

Uh oh!

ZachGerman commented Jun 2, 2025 •

edited

Loading