modelcontextprotocol/java-sdk
Watch
1
0
Fork
You've already forked java-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: improve streamable HTTP session reinitialization (#459)

Browse source 
Implements the MCP spec guidelines for streamable HTTP (re)initialization:
- Server MAY terminate session and MUST respond with HTTP 404 for terminated session IDs
- Client MUST start new session when receiving HTTP 404 for requests with session ID

Changes:
- Replace generic McpError with McpTransportException for transport-layer errors
- Only throw McpTransportSessionNotFoundException when session ID is present in request
  (per spec: 404 with session ID means session terminated, without means general error)
- Enhance error messages with more context (status codes, response events)
- Use RuntimeException for non-transport specific SSE endpoint failures
- Ensure consistent error handling across HTTP client transports

- Improve error handling with standard Java exceptions.
  Replace generic McpError with appropriate standard exceptions:
  - Use IllegalArgumentException for invalid input parameters
  - Use IllegalStateException for state-related issues
  - Use RuntimeException wrapper for initialization failures
  - Use McpError.builder() with proper error codes for protocol errors

Fixes #459

Signed-off-by: Christian Tzolov <christian.tzolov@broadcom.com>
This commit is contained in:
Christian Tzolov 2025-08-07 18:35:50 +01:00
commit 6c38f37637
6 changed files with 955 additions and 90 deletions
Download patch file Download diff file Expand all files Collapse all files

10
mcp/src/main/java/io/modelcontextprotocol/client/LifecycleInitializer.java
View file

@ -289,9 +289,7 @@ class LifecycleInitializer {
return initializationJob.map(initializeResult -> this.initializationRef.get())
.timeout(this.initializationTimeout)
.onErrorResume(ex -> {
logger.warn("Failed to initialize", ex);
return Mono.error(
new McpError("Client failed to initialize " + actionName + " due to: " + ex.getMessage()));
return Mono.error(new RuntimeException("Client failed to initialize " + actionName, ex));
})
.flatMap(operation);
});
@ -316,8 +314,10 @@ class LifecycleInitializer {
initializeResult.instructions());
if (!this.protocolVersions.contains(initializeResult.protocolVersion())) {
return Mono.error(new McpError(
"Unsupported protocol version from the server: " + initializeResult.protocolVersion()));
return Mono.error(McpError.builder(-32602)
.message("Unsupported protocol version")
.data("Unsupported protocol version from the server: " + initializeResult.protocolVersion())
.build());
}
return mcpClientSession.sendNotification(McpSchema.METHOD_NOTIFICATION_INITIALIZED, null)

30
mcp/src/main/java/io/modelcontextprotocol/client/McpAsyncClient.java
View file

@ -190,7 +190,8 @@ public class McpAsyncClient {
// Sampling Handler
if (this.clientCapabilities.sampling() != null) {
if (features.samplingHandler() == null) {
throw new McpError("Sampling handler must not be null when client capabilities include sampling");
throw new IllegalArgumentException(
"Sampling handler must not be null when client capabilities include sampling");
}
this.samplingHandler = features.samplingHandler();
requestHandlers.put(McpSchema.METHOD_SAMPLING_CREATE_MESSAGE, samplingCreateMessageHandler());
@ -199,7 +200,8 @@ public class McpAsyncClient {
// Elicitation Handler
if (this.clientCapabilities.elicitation() != null) {
if (features.elicitationHandler() == null) {
throw new McpError("Elicitation handler must not be null when client capabilities include elicitation");
throw new IllegalArgumentException(
"Elicitation handler must not be null when client capabilities include elicitation");
}
this.elicitationHandler = features.elicitationHandler();
requestHandlers.put(McpSchema.METHOD_ELICITATION_CREATE, elicitationCreateHandler());
@ -413,15 +415,15 @@ public class McpAsyncClient {
public Mono<Void> addRoot(Root root) {
if (root == null) {
return Mono.error(new McpError("Root must not be null"));
return Mono.error(new IllegalArgumentException("Root must not be null"));
}
if (this.clientCapabilities.roots() == null) {
return Mono.error(new McpError("Client must be configured with roots capabilities"));
return Mono.error(new IllegalStateException("Client must be configured with roots capabilities"));
}
if (this.roots.containsKey(root.uri())) {
return Mono.error(new McpError("Root with uri '" + root.uri() + "' already exists"));
return Mono.error(new IllegalStateException("Root with uri '" + root.uri() + "' already exists"));
}
this.roots.put(root.uri(), root);
@ -447,11 +449,11 @@ public class McpAsyncClient {
public Mono<Void> removeRoot(String rootUri) {
if (rootUri == null) {
return Mono.error(new McpError("Root uri must not be null"));
return Mono.error(new IllegalArgumentException("Root uri must not be null"));
}
if (this.clientCapabilities.roots() == null) {
return Mono.error(new McpError("Client must be configured with roots capabilities"));
return Mono.error(new IllegalStateException("Client must be configured with roots capabilities"));
}
Root removed = this.roots.remove(rootUri);
@ -469,7 +471,7 @@ public class McpAsyncClient {
}
return Mono.empty();
}
return Mono.error(new McpError("Root with uri '" + rootUri + "' not found"));
return Mono.error(new IllegalStateException("Root with uri '" + rootUri + "' not found"));
}
/**
@ -540,7 +542,7 @@ public class McpAsyncClient {
public Mono<McpSchema.CallToolResult> callTool(McpSchema.CallToolRequest callToolRequest) {
return this.initializer.withIntitialization("calling tools", init -> {
if (init.initializeResult().capabilities().tools() == null) {
return Mono.error(new McpError("Server does not provide tools capability"));
return Mono.error(new IllegalStateException("Server does not provide tools capability"));
}
return init.mcpSession()
.sendRequest(McpSchema.METHOD_TOOLS_CALL, callToolRequest, CALL_TOOL_RESULT_TYPE_REF);
@ -569,7 +571,7 @@ public class McpAsyncClient {
public Mono<McpSchema.ListToolsResult> listTools(String cursor) {
return this.initializer.withIntitialization("listing tools", init -> {
if (init.initializeResult().capabilities().tools() == null) {
return Mono.error(new McpError("Server does not provide tools capability"));
return Mono.error(new IllegalStateException("Server does not provide tools capability"));
}
return init.mcpSession()
.sendRequest(McpSchema.METHOD_TOOLS_LIST, new McpSchema.PaginatedRequest(cursor),
@ -633,7 +635,7 @@ public class McpAsyncClient {
public Mono<McpSchema.ListResourcesResult> listResources(String cursor) {
return this.initializer.withIntitialization("listing resources", init -> {
if (init.initializeResult().capabilities().resources() == null) {
return Mono.error(new McpError("Server does not provide the resources capability"));
return Mono.error(new IllegalStateException("Server does not provide the resources capability"));
}
return init.mcpSession()
.sendRequest(McpSchema.METHOD_RESOURCES_LIST, new McpSchema.PaginatedRequest(cursor),
@ -665,7 +667,7 @@ public class McpAsyncClient {
public Mono<McpSchema.ReadResourceResult> readResource(McpSchema.ReadResourceRequest readResourceRequest) {
return this.initializer.withIntitialization("reading resources", init -> {
if (init.initializeResult().capabilities().resources() == null) {
return Mono.error(new McpError("Server does not provide the resources capability"));
return Mono.error(new IllegalStateException("Server does not provide the resources capability"));
}
return init.mcpSession()
.sendRequest(McpSchema.METHOD_RESOURCES_READ, readResourceRequest, READ_RESOURCE_RESULT_TYPE_REF);
@ -703,7 +705,7 @@ public class McpAsyncClient {
public Mono<McpSchema.ListResourceTemplatesResult> listResourceTemplates(String cursor) {
return this.initializer.withIntitialization("listing resource templates", init -> {
if (init.initializeResult().capabilities().resources() == null) {
return Mono.error(new McpError("Server does not provide the resources capability"));
return Mono.error(new IllegalStateException("Server does not provide the resources capability"));
}
return init.mcpSession()
.sendRequest(McpSchema.METHOD_RESOURCES_TEMPLATES_LIST, new McpSchema.PaginatedRequest(cursor),
@ -863,7 +865,7 @@ public class McpAsyncClient {
*/
public Mono<Void> setLoggingLevel(LoggingLevel loggingLevel) {
if (loggingLevel == null) {
return Mono.error(new McpError("Logging level must not be null"));
return Mono.error(new IllegalArgumentException("Logging level must not be null"));
}
return this.initializer.withIntitialization("setting logging level", init -> {

10
mcp/src/main/java/io/modelcontextprotocol/client/transport/HttpClientSseClientTransport.java
View file

@ -24,10 +24,10 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import io.modelcontextprotocol.client.transport.ResponseSubscribers.ResponseEvent;
import io.modelcontextprotocol.spec.McpClientTransport;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.ProtocolVersions;
import io.modelcontextprotocol.spec.McpSchema.JSONRPCMessage;
import io.modelcontextprotocol.spec.McpTransportException;
import io.modelcontextprotocol.util.Assert;
import io.modelcontextprotocol.util.Utils;
import reactor.core.Disposable;
@ -431,7 +431,7 @@ public class HttpClientSseClientTransport implements McpClientTransport {
return Flux.empty(); // No further processing needed
}
else {
sink.error(new McpError("Failed to handle SSE endpoint event"));
sink.error(new RuntimeException("Failed to handle SSE endpoint event"));
}
}
else if (MESSAGE_EVENT_TYPE.equals(responseEvent.sseEvent().event())) {
@ -446,8 +446,7 @@ public class HttpClientSseClientTransport implements McpClientTransport {
}
}
catch (IOException e) {
logger.error("Error processing SSE event", e);
sink.error(new McpError("Error processing SSE event"));
sink.error(new McpTransportException("Error processing SSE event", e));
}
}
return Flux.<McpSchema.JSONRPCMessage>error(
@ -520,8 +519,7 @@ public class HttpClientSseClientTransport implements McpClientTransport {
return Mono.just(objectMapper.writeValueAsString(message));
}
catch (IOException e) {
// TODO: why McpError and not RuntimeException?
return Mono.error(new McpError("Failed to serialize message"));
return Mono.error(new McpTransportException("Failed to serialize message", e));
}
});
}

87
mcp/src/main/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransport.java
View file

@ -30,8 +30,8 @@ import io.modelcontextprotocol.spec.DefaultMcpTransportSession;
import io.modelcontextprotocol.spec.DefaultMcpTransportStream;
import io.modelcontextprotocol.spec.HttpHeaders;
import io.modelcontextprotocol.spec.McpClientTransport;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpTransportException;
import io.modelcontextprotocol.spec.McpTransportSession;
import io.modelcontextprotocol.spec.McpTransportSessionNotFoundException;
import io.modelcontextprotocol.spec.McpTransportStream;
@ -288,9 +288,8 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
}
catch (IOException ioException) {
return Flux.<McpSchema.JSONRPCMessage>error(
new McpError("Error parsing JSON-RPC message: "
+ responseEvent.sseEvent().data()));
return Flux.<McpSchema.JSONRPCMessage>error(new McpTransportException(
"Error parsing JSON-RPC message: " + responseEvent, ioException));
}
}
else {
@ -304,19 +303,39 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
return Flux.empty();
}
else if (statusCode == NOT_FOUND) {
String sessionIdRepresentation = sessionIdOrPlaceholder(transportSession);
McpTransportSessionNotFoundException exception = new McpTransportSessionNotFoundException(
"Session not found for session ID: " + sessionIdRepresentation);
return Flux.<McpSchema.JSONRPCMessage>error(exception);
if (transportSession != null && transportSession.sessionId().isPresent()) {
// only if the request was sent with a session id
// and the response is 404, we consider it a
// session not found error.
logger.debug("Session not found for session ID: {}",
transportSession.sessionId().get());
String sessionIdRepresentation = sessionIdOrPlaceholder(transportSession);
McpTransportSessionNotFoundException exception = new McpTransportSessionNotFoundException(
"Session not found for session ID: " + sessionIdRepresentation);
return Flux.<McpSchema.JSONRPCMessage>error(exception);
}
return Flux.<McpSchema.JSONRPCMessage>error(
new McpTransportException("Server Not Found. Status code:" + statusCode
+ ", response-event:" + responseEvent));
}
else if (statusCode == BAD_REQUEST) {
String sessionIdRepresentation = sessionIdOrPlaceholder(transportSession);
McpTransportSessionNotFoundException exception = new McpTransportSessionNotFoundException(
"Session not found for session ID: " + sessionIdRepresentation);
return Flux.<McpSchema.JSONRPCMessage>error(exception);
if (transportSession != null && transportSession.sessionId().isPresent()) {
// only if the request was sent with a session id
// and thre response is 404, we consider it a
// session not found error.
String sessionIdRepresentation = sessionIdOrPlaceholder(transportSession);
McpTransportSessionNotFoundException exception = new McpTransportSessionNotFoundException(
"Session not found for session ID: " + sessionIdRepresentation);
return Flux.<McpSchema.JSONRPCMessage>error(exception);
}
return Flux.<McpSchema.JSONRPCMessage>error(
new McpTransportException("Bad Request. Status code:" + statusCode
+ ", response-event:" + responseEvent));
}
return Flux.<McpSchema.JSONRPCMessage>error(new McpError(
return Flux.<McpSchema.JSONRPCMessage>error(new McpTransportException(
"Received unrecognized SSE event type: " + responseEvent.sseEvent().event()));
}).<McpSchema
.JSONRPCMessage>flatMap(
@ -468,8 +487,8 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
return Flux.from(sessionStream.consumeSseStream(Flux.just(idWithMessages)));
}
catch (IOException ioException) {
return Flux.<McpSchema.JSONRPCMessage>error(
new McpError("Error parsing JSON-RPC message: " + sseEvent.data()));
return Flux.<McpSchema.JSONRPCMessage>error(new McpTransportException(
"Error parsing JSON-RPC message: " + responseEvent, ioException));
}
});
}
@ -485,8 +504,8 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
return Mono.just(McpSchema.deserializeJsonRpcMessage(objectMapper, data));
}
catch (IOException e) {
// TODO: this should be a McpTransportError
return Mono.error(e);
return Mono.error(new McpTransportException(
"Error deserializing JSON-RPC message: " + responseEvent, e));
}
}
logger.warn("Unknown media type {} returned for POST in session {}", contentType,
@ -496,18 +515,32 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
new RuntimeException("Unknown media type returned: " + contentType));
}
else if (statusCode == NOT_FOUND) {
McpTransportSessionNotFoundException exception = new McpTransportSessionNotFoundException(
"Session not found for session ID: " + sessionRepresentation);
return Flux.<McpSchema.JSONRPCMessage>error(exception);
if (transportSession != null && transportSession.sessionId().isPresent()) {
// only if the request was sent with a session id and the
// response is 404, we consider it a session not found error.
logger.debug("Session not found for session ID: {}", transportSession.sessionId().get());
McpTransportSessionNotFoundException exception = new McpTransportSessionNotFoundException(
"Session not found for session ID: " + sessionRepresentation);
return Flux.<McpSchema.JSONRPCMessage>error(exception);
}
return Flux.<McpSchema.JSONRPCMessage>error(new McpTransportException(
"Server Not Found. Status code:" + statusCode + ", response-event:" + responseEvent));
}
// Some implementations can return 400 when presented with a
// session id that it doesn't know about, so we will
// invalidate the session
// https://github.com/modelcontextprotocol/typescript-sdk/issues/389
else if (statusCode == BAD_REQUEST) {
McpTransportSessionNotFoundException exception = new McpTransportSessionNotFoundException(
"Session not found for session ID: " + sessionRepresentation);
return Flux.<McpSchema.JSONRPCMessage>error(exception);
// Some implementations can return 400 when presented with a
// session id that it doesn't know about, so we will
// invalidate the session
// https://github.com/modelcontextprotocol/typescript-sdk/issues/389
if (transportSession != null && transportSession.sessionId().isPresent()) {
// only if the request was sent with a session id and the
// response is 404, we consider it a session not found error.
McpTransportSessionNotFoundException exception = new McpTransportSessionNotFoundException(
"Session not found for session ID: " + sessionRepresentation);
return Flux.<McpSchema.JSONRPCMessage>error(exception);
}
return Flux.<McpSchema.JSONRPCMessage>error(new McpTransportException(
"Bad Request. Status code:" + statusCode + ", response-event:" + responseEvent));
}
return Flux.<McpSchema.JSONRPCMessage>error(

5
mcp/src/main/java/io/modelcontextprotocol/client/transport/ResponseSubscribers.java
View file

@ -15,7 +15,7 @@ import org.reactivestreams.Subscription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpTransportException;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.FluxSink;
@ -178,8 +178,7 @@ class ResponseSubscribers {
}
else {
// If the response is not successful, emit an error
// TODO: This should be a McpTransportError
this.sink.error(new McpError(
this.sink.error(new McpTransportException(
"Invalid SSE response. Status code: " + this.responseInfo.statusCode() + " Line: " + line));
}

38
mcp/src/main/java/io/modelcontextprotocol/spec/McpTransportException.java Normal file
View file

@ -0,0 +1,38 @@
/*
* Copyright 2025 - 2025 the original author or authors.
*/
package io.modelcontextprotocol.spec;
/**
* Exception thrown when there is an issue with the transport layer of the Model Context
* Protocol (MCP).
*
* <p>
* This exception is used to indicate errors that occur during communication between the
* MCP client and server, such as connection failures, protocol violations, or unexpected
* responses.
*
* @author Christian Tzolov
*/
public class McpTransportException extends RuntimeException {
private static final long serialVersionUID = 1L;
public McpTransportException(String message) {
super(message);
}
public McpTransportException(String message, Throwable cause) {
super(message, cause);
}
public McpTransportException(Throwable cause) {
super(cause);
}
public McpTransportException(String message, Throwable cause, boolean enableSuppression,
boolean writableStackTrace) {
super(message, cause, enableSuppression, writableStackTrace);
}
}