Skip to content

Streamable HTTP Server abstractions and WebFlux transport provider #420

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 19 commits into from
Jul 30, 2025
+6,578 −704
Merged

Streamable HTTP Server abstractions and WebFlux transport provider #420

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
98990e9
WIP
chemicL Jul 18, 2025
8e4c745
Merge branch 'main' into streamable-server
chemicL Jul 21, 2025
f83db45
More WIP: renaming, steteless server abstractions, client context ext…
chemicL Jul 22, 2025
1645ecb
Formatting
chemicL Jul 22, 2025
a3ddb75
WIP: make it compile
chemicL Jul 22, 2025
cef7aac
WIP: Add session ID to exchange
chemicL Jul 22, 2025
6423fcd
WIP: make it compile
chemicL Jul 22, 2025
8e9ab52
WIP: add SSE id generation and DELETE handling
chemicL Jul 23, 2025
92ca81f
WIP: Handle logging notifications, start with integration tests and a…
chemicL Jul 25, 2025
c0e3129
WIP: Sync server builder tests
chemicL Jul 28, 2025
a0ad5a5
WIP: resolve integration test issues
chemicL Jul 28, 2025
d98b8c9
WIP: Logging notifications tests on exchange adapted
chemicL Jul 28, 2025
6e66602
WIP: Get rid of initialization notification handler
chemicL Jul 28, 2025
b382ac5
refactor: improve MCP server test architecture with parameterized tests
tzolov Jul 28, 2025
0a4ecb0
WIP: stateless server implementation
chemicL Jul 29, 2025
87a018f
WIP: context extractor usage, some javadocs, HttpHeaders constants
chemicL Jul 30, 2025
92be635
WIP: Add remaining javadocs, tiny corrections here and there
chemicL Jul 30, 2025
08900bc
Add accept header check for WebFlux server
chemicL Jul 30, 2025
ca7f870
Use random port in McpCompletionTests
chemicL Jul 30, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion .../main/java/io/modelcontextprotocol/client/transport/WebClientStreamableHttpTransport.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -340,7 +340,8 @@ private Flux<McpSchema.JSONRPCMessage> extractError(ClientResponse response, Str
McpSchema.JSONRPCResponse jsonRpcResponse = objectMapper.readValue(body,
McpSchema.JSONRPCResponse.class);
jsonRpcError = jsonRpcResponse.error();
toPropagate = new McpError(jsonRpcError);
toPropagate = jsonRpcError != null ? new McpError(jsonRpcError)
: new McpError("Can't parse the jsonResponse " + jsonRpcResponse);
}
catch (IOException ex) {
toPropagate = new RuntimeException("Sending request failed", e);
Expand Down
23 changes: 17 additions & 6 deletions ...va/io/modelcontextprotocol/server/transport/WebFluxStreamableServerTransportProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,8 +35,6 @@ public class WebFluxStreamableServerTransportProvider implements McpStreamableSe

public static final String MESSAGE_EVENT_TYPE = "message";

public static final String ENDPOINT_EVENT_TYPE = "endpoint";

public static final String DEFAULT_BASE_URL = "";

private final ObjectMapper objectMapper;
Expand Down Expand Up @@ -263,17 +261,28 @@ private Mono<ServerResponse> handlePost(ServerRequest request) {
McpStreamableServerSession.McpStreamableServerSessionInit init = this.sessionFactory
.startSession(initializeRequest);
sessions.put(init.session().getId(), init.session());
return init.initResult()
return init.initResult().map(initializeResult -> {
McpSchema.JSONRPCResponse jsonrpcResponse = new McpSchema.JSONRPCResponse(
McpSchema.JSONRPC_VERSION, jsonrpcRequest.id(), initializeResult, null);
try {
return this.objectMapper.writeValueAsString(jsonrpcResponse);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This seems redundant as the WebFlux would do the conversion?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I suppose so! Although it's a temporary mitigation, I think we should deal with the init result conversion into JSON at an earlier layer. Need to come back to this.

}
catch (IOException e) {
logger.warn("Failed to serialize initResponse", e);
throw Exceptions.propagate(e);
}
})
.flatMap(initResult -> ServerResponse.ok()
.contentType(MediaType.APPLICATION_JSON)
.header("mcp-session-id", init.session().getId())
.bodyValue(initResult));
}

if (!request.headers().asHttpHeaders().containsKey("sessionId")) {
if (!request.headers().asHttpHeaders().containsKey("mcp-session-id")) {
return ServerResponse.badRequest().bodyValue(new McpError("Session ID missing"));
}

String sessionId = request.headers().asHttpHeaders().getFirst("sessionId");
String sessionId = request.headers().asHttpHeaders().getFirst("mcp-session-id");
McpStreamableServerSession session = sessions.get(sessionId);

if (session == null) {
Expand Down Expand Up @@ -308,7 +317,9 @@ else if (message instanceof McpSchema.JSONRPCRequest jsonrpcRequest) {
logger.error("Failed to deserialize message: {}", e.getMessage());
return ServerResponse.badRequest().bodyValue(new McpError("Invalid message format"));
}
}).contextWrite(ctx -> ctx.put(McpTransportContext.KEY, transportContext));
})
.switchIfEmpty(ServerResponse.badRequest().build())
.contextWrite(ctx -> ctx.put(McpTransportContext.KEY, transportContext));
}

private Mono<ServerResponse> handleDelete(ServerRequest request) {
Expand Down
Loading
Loading