Basic version negotiation capabilities

- We do not support full version negotiation, but this commit allows the
  client to the send the correct version to match that of the server.
- There is one limitation: the GET /mcp request happens too early,
  before we have deserialized the initialization response, so we do not
  know the negotiated version yet. This request will have the client
  latest version as Mcp-Protocol-Version value.

Signed-off-by: Daniel Garnier-Moiroux <git@garnier.wf>

mcp-core/src/main/java/io/modelcontextprotocol/client/LifecycleInitializer.java

@@ -287,7 +287,9 @@ class LifecycleInitializer {
 					this.initializationRef.compareAndSet(newInit, null);
 					return Mono.error(new RuntimeException("Client failed to initialize " + actionName, ex));
 				})
-				.flatMap(operation);
+				.flatMap(res -> operation.apply(res)
+					.contextWrite(c -> c.put(McpAsyncClient.NEGOTIATED_PROTOCOL_VERSION,
+							res.initializeResult().protocolVersion())));
 		});
 	}
 
@@ -319,6 +321,8 @@ class LifecycleInitializer {
 			}
 
 			return mcpClientSession.sendNotification(McpSchema.METHOD_NOTIFICATION_INITIALIZED, null)
+				.contextWrite(
+						c -> c.put(McpAsyncClient.NEGOTIATED_PROTOCOL_VERSION, initializeResult.protocolVersion()))
 				.thenReturn(initializeResult);
 		}).flatMap(initializeResult -> {
 			initialization.cacheResult(initializeResult);

mcp-core/src/main/java/io/modelcontextprotocol/client/McpAsyncClient.java

@@ -106,6 +106,8 @@ public class McpAsyncClient {
 	public static final TypeRef<McpSchema.ProgressNotification> PROGRESS_NOTIFICATION_TYPE_REF = new TypeRef<>() {
 	};
 
+	public static final String NEGOTIATED_PROTOCOL_VERSION = "io.modelcontextprotocol.client.negotiated-protocol-version";
+
 	/**
 	 * Client capabilities.
 	 */

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

@@ -20,6 +20,7 @@ import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.function.Function;
 
+import io.modelcontextprotocol.client.McpAsyncClient;
 import io.modelcontextprotocol.client.transport.ResponseSubscribers.ResponseEvent;
 import io.modelcontextprotocol.client.transport.customizer.McpAsyncHttpClientRequestCustomizer;
 import io.modelcontextprotocol.client.transport.customizer.McpSyncHttpClientRequestCustomizer;
@@ -193,7 +194,9 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
 				.uri(uri)
 				.header("Cache-Control", "no-cache")
 				.header(HttpHeaders.MCP_SESSION_ID, sessionId)
-				.header(HttpHeaders.PROTOCOL_VERSION, this.latestSupportedProtocolVersion)
+				.header(HttpHeaders.PROTOCOL_VERSION,
+						ctx.getOrDefault(McpAsyncClient.NEGOTIATED_PROTOCOL_VERSION,
+								this.latestSupportedProtocolVersion))
 				.DELETE();
 			var transportContext = ctx.getOrDefault(McpTransportContext.KEY, McpTransportContext.EMPTY);
 			return Mono.from(this.httpRequestCustomizer.customize(builder, "DELETE", uri, null, transportContext));
@@ -264,7 +267,9 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
 				var builder = requestBuilder.uri(uri)
 					.header(HttpHeaders.ACCEPT, TEXT_EVENT_STREAM)
 					.header("Cache-Control", "no-cache")
-					.header(HttpHeaders.PROTOCOL_VERSION, this.latestSupportedProtocolVersion)
+					.header(HttpHeaders.PROTOCOL_VERSION,
+							connectionCtx.getOrDefault(McpAsyncClient.NEGOTIATED_PROTOCOL_VERSION,
+									this.latestSupportedProtocolVersion))
 					.GET();
 				var transportContext = connectionCtx.getOrDefault(McpTransportContext.KEY, McpTransportContext.EMPTY);
 				return Mono.from(this.httpRequestCustomizer.customize(builder, "GET", uri, null, transportContext));
@@ -439,7 +444,9 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
 					.header(HttpHeaders.ACCEPT, APPLICATION_JSON + ", " + TEXT_EVENT_STREAM)
 					.header(HttpHeaders.CONTENT_TYPE, APPLICATION_JSON)
 					.header(HttpHeaders.CACHE_CONTROL, "no-cache")
-					.header(HttpHeaders.PROTOCOL_VERSION, this.latestSupportedProtocolVersion)
+					.header(HttpHeaders.PROTOCOL_VERSION,
+							ctx.getOrDefault(McpAsyncClient.NEGOTIATED_PROTOCOL_VERSION,
+									this.latestSupportedProtocolVersion))
 					.POST(HttpRequest.BodyPublishers.ofString(jsonBody));
 				var transportContext = ctx.getOrDefault(McpTransportContext.KEY, McpTransportContext.EMPTY);
 				return Mono

mcp-core/src/main/java/io/modelcontextprotocol/spec/McpTransportSession.java

@@ -4,10 +4,10 @@
 
 package io.modelcontextprotocol.spec;
 
-import org.reactivestreams.Publisher;
-
 import java.util.Optional;
 
+import org.reactivestreams.Publisher;
+
 /**
  * An abstraction of the session as perceived from the MCP transport layer. Not to be
  * confused with the {@link McpSession} type that operates at the level of the JSON-RPC

mcp-core/src/test/java/io/modelcontextprotocol/common/HttpClientStreamableHttpVersionNegotiationIntegrationTests.java

@@ -89,7 +89,7 @@ class HttpClientStreamableHttpVersionNegotiationIntegrationTests {
 	}
 
 	@Test
-	void usesCustomLatestVersion() {
+	void usesServerSupportedVersion() {
 		startTomcat();
 
 		var transport = HttpClientStreamableHttpTransport.builder("http://localhost:" + PORT)
@@ -101,19 +101,21 @@ class HttpClientStreamableHttpVersionNegotiationIntegrationTests {
 		McpSchema.CallToolResult response = client.callTool(new McpSchema.CallToolRequest("test-tool", Map.of()));
 
 		var calls = requestRecordingFilter.getCalls();
-
+		// Initialize tells the server the Client's latest supported version
-		assertThat(calls).filteredOn(c -> !c.body().contains("\"method\":\"initialize\""))
+		// FIXME: Set the correct protocol version on GET /mcp
-			// GET /mcp ; POST notification/initialized ; POST tools/call
+		assertThat(calls).filteredOn(c -> c.method().equals("POST") && !c.body().contains("\"method\":\"initialize\""))
-			.hasSize(3)
+			// POST notification/initialized ; POST tools/call
+			.hasSize(2)
 			.map(McpTestRequestRecordingServletFilter.Call::headers)
-			.allSatisfy(headers -> assertThat(headers).containsEntry("mcp-protocol-version", "2263-03-18"));
+			.allSatisfy(headers -> assertThat(headers).containsEntry("mcp-protocol-version",
+					ProtocolVersions.MCP_2025_06_18));
 
 		assertThat(response).isNotNull();
 		assertThat(response.content()).hasSize(1)
 			.first()
 			.extracting(McpSchema.TextContent.class::cast)
 			.extracting(McpSchema.TextContent::text)
-			.isEqualTo("2263-03-18");
+			.isEqualTo(ProtocolVersions.MCP_2025_06_18);
 		mcpServer.close();
 	}
 

mcp-core/src/test/java/io/modelcontextprotocol/server/transport/McpTestRequestRecordingServletFilter.java

@@ -46,7 +46,7 @@ public class McpTestRequestRecordingServletFilter implements Filter {
 				.collect(Collectors.toUnmodifiableMap(Function.identity(),
 						name -> String.join(",", Collections.list(req.getHeaders(name)))));
 			var request = new CachedBodyHttpServletRequest(req);
-			calls.add(new Call(headers, request.getBodyAsString()));
+			calls.add(new Call(req.getMethod(), headers, request.getBodyAsString()));
 			filterChain.doFilter(request, servletResponse);
 		}
 		else {
@@ -60,7 +60,7 @@ public class McpTestRequestRecordingServletFilter implements Filter {
 		return List.copyOf(calls);
 	}
 
-	public record Call(Map<String, String> headers, String body) {
+	public record Call(String method, Map<String, String> headers, String body) {
 
 	}
 

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

@@ -24,6 +24,7 @@ import org.springframework.web.reactive.function.client.ClientResponse;
 import org.springframework.web.reactive.function.client.WebClient;
 import org.springframework.web.reactive.function.client.WebClientResponseException;
 
+import io.modelcontextprotocol.client.McpAsyncClient;
 import io.modelcontextprotocol.json.McpJsonMapper;
 import io.modelcontextprotocol.json.TypeRef;
 import io.modelcontextprotocol.spec.ClosedMcpTransportSession;
@@ -225,7 +226,9 @@ public class WebClientStreamableHttpTransport implements McpClientTransport {
 			Disposable connection = webClient.get()
 				.uri(this.endpoint)
 				.accept(MediaType.TEXT_EVENT_STREAM)
-				.header(HttpHeaders.PROTOCOL_VERSION, this.latestSupportedProtocolVersion)
+				.header(HttpHeaders.PROTOCOL_VERSION,
+						ctx.getOrDefault(McpAsyncClient.NEGOTIATED_PROTOCOL_VERSION,
+								this.latestSupportedProtocolVersion))
 				.headers(httpHeaders -> {
 					transportSession.sessionId().ifPresent(id -> httpHeaders.add(HttpHeaders.MCP_SESSION_ID, id));
 					if (stream != null) {
@@ -288,10 +291,12 @@ public class WebClientStreamableHttpTransport implements McpClientTransport {
 			final AtomicReference<Disposable> disposableRef = new AtomicReference<>();
 			final McpTransportSession<Disposable> transportSession = this.activeSession.get();
 
-			Disposable connection = webClient.post()
+			Disposable connection = Flux.deferContextual(ctx -> webClient.post()
 				.uri(this.endpoint)
 				.accept(MediaType.APPLICATION_JSON, MediaType.TEXT_EVENT_STREAM)
-				.header(HttpHeaders.PROTOCOL_VERSION, this.latestSupportedProtocolVersion)
+				.header(HttpHeaders.PROTOCOL_VERSION,
+						ctx.getOrDefault(McpAsyncClient.NEGOTIATED_PROTOCOL_VERSION,
+								this.latestSupportedProtocolVersion))
 				.headers(httpHeaders -> {
 					transportSession.sessionId().ifPresent(id -> httpHeaders.add(HttpHeaders.MCP_SESSION_ID, id));
 				})
@@ -350,7 +355,7 @@ public class WebClientStreamableHttpTransport implements McpClientTransport {
 						}
 						return this.extractError(response, sessionRepresentation);
 					}
-				})
+				}))
 				.flatMap(jsonRpcMessage -> this.handler.get().apply(Mono.just(jsonRpcMessage)))
 				.onErrorComplete(t -> {
 					// handle the error first

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

@@ -27,6 +27,7 @@ import org.junit.jupiter.api.Test;
 import reactor.netty.DisposableServer;
 import reactor.netty.http.server.HttpServer;
 
+import org.springframework.http.HttpMethod;
 import org.springframework.http.server.reactive.HttpHandler;
 import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter;
 import org.springframework.web.reactive.function.client.WebClient;
@@ -116,7 +117,7 @@ class WebFluxStreamableHttpVersionNegotiationIntegrationTests {
 	}
 
 	@Test
-	void usesCustomLatestVersion() {
+	void usesServerSupportedVersion() {
 		var transport = WebClientStreamableHttpTransport
 			.builder(WebClient.builder().baseUrl("http://localhost:" + PORT))
 			.supportedProtocolVersions(List.of(ProtocolVersions.MCP_2025_06_18, "2263-03-18"))
@@ -128,18 +129,22 @@ class WebFluxStreamableHttpVersionNegotiationIntegrationTests {
 		McpSchema.CallToolResult response = client.callTool(new McpSchema.CallToolRequest("test-tool", Map.of()));
 
 		var calls = recordingFilterFunction.getCalls();
-		assertThat(calls).filteredOn(c -> !c.body().contains("\"method\":\"initialize\""))
+		// Initialize tells the server the Client's latest supported version
-			// GET /mcp ; POST notification/initialized ; POST tools/call
+		// FIXME: Set the correct protocol version on GET /mcp
-			.hasSize(3)
+		assertThat(calls)
+			.filteredOn(c -> !c.body().contains("\"method\":\"initialize\"") && c.method().equals(HttpMethod.POST))
+			// POST notification/initialized ; POST tools/call
+			.hasSize(2)
 			.map(McpTestRequestRecordingExchangeFilterFunction.Call::headers)
-			.allSatisfy(headers -> assertThat(headers).containsEntry("mcp-protocol-version", "2263-03-18"));
+			.allSatisfy(headers -> assertThat(headers).containsEntry("mcp-protocol-version",
+					ProtocolVersions.MCP_2025_06_18));
 
 		assertThat(response).isNotNull();
 		assertThat(response.content()).hasSize(1)
 			.first()
 			.extracting(McpSchema.TextContent.class::cast)
 			.extracting(McpSchema.TextContent::text)
-			.isEqualTo("2263-03-18");
+			.isEqualTo(ProtocolVersions.MCP_2025_06_18);
 		mcpServer.close();
 	}
 

mcp-spring/mcp-spring-webflux/src/test/java/io/modelcontextprotocol/utils/McpTestRequestRecordingExchangeFilterFunction.java

@@ -11,6 +11,7 @@ import java.util.stream.Collectors;
 
 import reactor.core.publisher.Mono;
 
+import org.springframework.http.HttpMethod;
 import org.springframework.web.reactive.function.server.HandlerFilterFunction;
 import org.springframework.web.reactive.function.server.HandlerFunction;
 import org.springframework.web.reactive.function.server.ServerRequest;
@@ -34,7 +35,7 @@ public class McpTestRequestRecordingExchangeFilterFunction implements HandlerFil
 			.collect(Collectors.toMap(String::toLowerCase, k -> String.join(",", request.headers().header(k))));
 
 		var cr = request.bodyToMono(String.class).defaultIfEmpty("").map(body -> {
-			this.calls.add(new Call(headers, body));
+			this.calls.add(new Call(request.method(), headers, body));
 			return ServerRequest.from(request).body(body).build();
 		});
 
@@ -46,7 +47,7 @@ public class McpTestRequestRecordingExchangeFilterFunction implements HandlerFil
 		return List.copyOf(calls);
 	}
 
-	public record Call(Map<String, String> headers, String body) {
+	public record Call(HttpMethod method, Map<String, String> headers, String body) {
 
 	}