fix: improve streamable HTTP session reinitialization (#459)

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 <[email protected]>

Author: tzolov

mcp-spring/mcp-spring-webflux/src/main/java/io/modelcontextprotocol/client/transport/WebClientStreamableHttpTransport.java

@@ -31,6 +31,7 @@
 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;
@@ -70,6 +71,8 @@
  */
 public class WebClientStreamableHttpTransport implements McpClientTransport {
 
+	private static final String MISSING_SESSION_ID = "[missing_session_id]";
+
 	private static final Logger logger = LoggerFactory.getLogger(WebClientStreamableHttpTransport.class);
 
 	private static final String MCP_PROTOCOL_VERSION = ProtocolVersions.MCP_2025_03_26;
@@ -221,8 +224,13 @@ else if (isNotAllowed(response)) {
 						return Flux.empty();
 					}
 					else if (isNotFound(response)) {
-						String sessionIdRepresentation = sessionIdOrPlaceholder(transportSession);
-						return mcpSessionNotFoundError(sessionIdRepresentation);
+						if (transportSession.sessionId().isPresent()) {
+							String sessionIdRepresentation = sessionIdOrPlaceholder(transportSession);
+							return mcpSessionNotFoundError(sessionIdRepresentation);
+						}
+						else {
+							return this.extractError(response, MISSING_SESSION_ID);
+						}
 					}
 					else {
 						return response.<McpSchema.JSONRPCMessage>createError().doOnError(e -> {
@@ -318,10 +326,10 @@ else if (mediaType.isCompatibleWith(MediaType.APPLICATION_JSON)) {
 						}
 					}
 					else {
-						if (isNotFound(response)) {
+						if (isNotFound(response) && !sessionRepresentation.equals(MISSING_SESSION_ID)) {
 							return mcpSessionNotFoundError(sessionRepresentation);
 						}
-						return extractError(response, sessionRepresentation);
+						return this.extractError(response, sessionRepresentation);
 					}
 				})
 				.flatMap(jsonRpcMessage -> this.handler.get().apply(Mono.just(jsonRpcMessage)))
@@ -362,10 +370,10 @@ private Flux<McpSchema.JSONRPCMessage> extractError(ClientResponse response, Str
 						McpSchema.JSONRPCResponse.class);
 				jsonRpcError = jsonRpcResponse.error();
 				toPropagate = jsonRpcError != null ? new McpError(jsonRpcError)
-						: new McpError("Can't parse the jsonResponse " + jsonRpcResponse);
+						: new McpTransportException("Can't parse the jsonResponse " + jsonRpcResponse);
 			}
 			catch (IOException ex) {
-				toPropagate = new RuntimeException("Sending request failed", e);
+				toPropagate = new McpTransportException("Sending request failed, " + e.getMessage(), e);
 				logger.debug("Received content together with {} HTTP code response: {}", response.statusCode(), body);
 			}
 
@@ -374,7 +382,11 @@ private Flux<McpSchema.JSONRPCMessage> extractError(ClientResponse response, Str
 			// invalidate the session
 			// https://github.com/modelcontextprotocol/typescript-sdk/issues/389
 			if (responseException.getStatusCode().isSameCodeAs(HttpStatus.BAD_REQUEST)) {
-				return Mono.error(new McpTransportSessionNotFoundException(sessionRepresentation, toPropagate));
+				if (!sessionRepresentation.equals(MISSING_SESSION_ID)) {
+					return Mono.error(new McpTransportSessionNotFoundException(sessionRepresentation, toPropagate));
+				}
+				return Mono.error(new McpTransportException("Received 400 BAD REQUEST for session "
+						+ sessionRepresentation + ". " + toPropagate.getMessage(), toPropagate));
 			}
 			return Mono.error(toPropagate);
 		}).flux();
@@ -403,7 +415,7 @@ private static boolean isEventStream(ClientResponse response) {
 	}
 
 	private static String sessionIdOrPlaceholder(McpTransportSession<?> transportSession) {
-		return transportSession.sessionId().orElse("[missing_session_id]");
+		return transportSession.sessionId().orElse(MISSING_SESSION_ID);
 	}
 
 	private Flux<McpSchema.JSONRPCMessage> directResponseFlux(McpSchema.JSONRPCMessage sentMessage,
@@ -421,8 +433,7 @@ private Flux<McpSchema.JSONRPCMessage> directResponseFlux(McpSchema.JSONRPCMessa
 				}
 			}
 			catch (IOException e) {
-				// TODO: this should be a McpTransportError
-				s.error(e);
+				s.error(new McpTransportException(e));
 			}
 		}).flatMapIterable(Function.identity());
 	}
@@ -449,7 +460,7 @@ private Tuple2<Optional<String>, Iterable<McpSchema.JSONRPCMessage>> parse(Serve
 				return Tuples.of(Optional.ofNullable(event.id()), List.of(message));
 			}
 			catch (IOException ioException) {
-				throw new McpError("Error parsing JSON-RPC message: " + event.data());
+				throw new McpTransportException("Error parsing JSON-RPC message: " + event.data(), ioException);
 			}
 		}
 		else {

mcp-spring/mcp-spring-webflux/src/main/java/io/modelcontextprotocol/client/transport/WebFluxSseClientTransport.java

@@ -14,7 +14,6 @@
 
 import io.modelcontextprotocol.spec.HttpHeaders;
 import io.modelcontextprotocol.spec.McpClientTransport;
-import io.modelcontextprotocol.spec.McpError;
 import io.modelcontextprotocol.spec.McpSchema;
 import io.modelcontextprotocol.spec.McpSchema.JSONRPCMessage;
 import io.modelcontextprotocol.spec.ProtocolVersions;
@@ -197,8 +196,6 @@ public List<String> protocolVersions() {
 	 * @param handler a function that processes incoming JSON-RPC messages and returns
 	 * responses
 	 * @return a Mono that completes when the connection is fully established
-	 * @throws McpError if there's an error processing SSE events or if an unrecognized
-	 * event type is received
 	 */
 	@Override
 	public Mono<Void> connect(Function<Mono<JSONRPCMessage>, Mono<JSONRPCMessage>> handler) {
@@ -215,7 +212,7 @@ public Mono<Void> connect(Function<Mono<JSONRPCMessage>, Mono<JSONRPCMessage>> h
 				else {
 					// TODO: clarify with the spec if multiple events can be
 					// received
-					s.error(new McpError("Failed to handle SSE endpoint event"));
+					s.error(new RuntimeException("Failed to handle SSE endpoint event"));
 				}
 			}
 			else if (MESSAGE_EVENT_TYPE.equals(event.event())) {

mcp-spring/mcp-spring-webflux/src/test/java/io/modelcontextprotocol/WebFluxSseIntegrationTests.java

@@ -8,6 +8,7 @@
 
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Timeout;
 import org.springframework.http.server.reactive.HttpHandler;
 import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter;
 import org.springframework.web.reactive.function.client.WebClient;
@@ -26,6 +27,7 @@
 import reactor.netty.DisposableServer;
 import reactor.netty.http.server.HttpServer;
 
+@Timeout(15)
 class WebFluxSseIntegrationTests extends AbstractMcpClientServerIntegrationTests {
 
 	private static final int PORT = TestUtil.findAvailablePort();

mcp-spring/mcp-spring-webflux/src/test/java/io/modelcontextprotocol/WebFluxStatelessIntegrationTests.java

@@ -8,6 +8,7 @@
 
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Timeout;
 import org.springframework.http.server.reactive.HttpHandler;
 import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter;
 import org.springframework.web.reactive.function.client.WebClient;
@@ -26,6 +27,7 @@
 import reactor.netty.DisposableServer;
 import reactor.netty.http.server.HttpServer;
 
+@Timeout(15)
 class WebFluxStatelessIntegrationTests extends AbstractStatelessIntegrationTests {
 
 	private static final int PORT = TestUtil.findAvailablePort();

mcp-spring/mcp-spring-webflux/src/test/java/io/modelcontextprotocol/WebFluxStreamableIntegrationTests.java

@@ -8,6 +8,7 @@
 
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Timeout;
 import org.springframework.http.server.reactive.HttpHandler;
 import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter;
 import org.springframework.web.reactive.function.client.WebClient;
@@ -26,6 +27,7 @@
 import reactor.netty.DisposableServer;
 import reactor.netty.http.server.HttpServer;
 
+@Timeout(15)
 class WebFluxStreamableIntegrationTests extends AbstractMcpClientServerIntegrationTests {
 
 	private static final int PORT = TestUtil.findAvailablePort();