Fix: Move markInitialized() and reconnect() after status code check

When connecting to an MCP server that doesn't support Streamable HTTP
transport (returns 405 Method Not Allowed), the transports still called
markInitialized() and reconnect() before checking the status code,
which triggered unnecessary GET requests.

This caused issues when using transport fallback (Streamable HTTP → SSE),
as duplicate SSE sessions were created on the upstream server.

This fix moves markInitialized() and reconnect() inside the success
status code check (2xx), so they are only called for successful responses.

Fixes #773
Related to #362

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: irwin-chequer

mcp-core/src/main/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransport.java

@@ -466,19 +466,21 @@ public Mono<Void> sendMessage(McpSchema.JSONRPCMessage sentMessage) {
 					})).onErrorMap(CompletionException.class, t -> t.getCause()).onErrorComplete().subscribe();
 
 			})).flatMap(responseEvent -> {
-				if (transportSession.markInitialized(
-						responseEvent.responseInfo().headers().firstValue("mcp-session-id").orElseGet(() -> null))) {
-					// Once we have a session, we try to open an async stream for
-					// the server to send notifications and requests out-of-band.
-
-					reconnect(null).contextWrite(deliveredSink.contextView()).subscribe();
-				}
-
 				String sessionRepresentation = sessionIdOrPlaceholder(transportSession);
 
 				int statusCode = responseEvent.responseInfo().statusCode();
 
 				if (statusCode >= 200 && statusCode < 300) {
+					// Only initialize session and open async stream for successful
+					// responses
+					if (transportSession.markInitialized(responseEvent.responseInfo()
+						.headers()
+						.firstValue("mcp-session-id")
+						.orElseGet(() -> null))) {
+						// Once we have a session, we try to open an async stream for
+						// the server to send notifications and requests out-of-band.
+						reconnect(null).contextWrite(deliveredSink.contextView()).subscribe();
+					}
 
 					String contentType = responseEvent.responseInfo()
 						.headers()

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

@@ -310,18 +310,19 @@ public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message) {
 				})
 				.bodyValue(jsonText)
 				.exchangeToFlux(response -> {
-					if (transportSession
-						.markInitialized(response.headers().asHttpHeaders().getFirst(HttpHeaders.MCP_SESSION_ID))) {
-						// Once we have a session, we try to open an async stream for
-						// the server to send notifications and requests out-of-band.
-						reconnect(null).contextWrite(sink.contextView()).subscribe();
-					}
-
 					String sessionRepresentation = sessionIdOrPlaceholder(transportSession);
 
 					// The spec mentions only ACCEPTED, but the existing SDKs can return
 					// 200 OK for notifications
 					if (response.statusCode().is2xxSuccessful()) {
+						// Only initialize session and open async stream for successful
+						// responses
+						if (transportSession
+							.markInitialized(response.headers().asHttpHeaders().getFirst(HttpHeaders.MCP_SESSION_ID))) {
+							// Once we have a session, we try to open an async stream for
+							// the server to send notifications and requests out-of-band.
+							reconnect(null).contextWrite(sink.contextView()).subscribe();
+						}
 						Optional<MediaType> contentType = response.headers().contentType();
 						long contentLength = response.headers().contentLength().orElse(-1);
 						// Existing SDKs consume notifications with no response body nor