From c293f09e7e47155f29d5dad19bddf0209ea40659 Mon Sep 17 00:00:00 2001 From: Daniel Garnier-Moiroux Date: Mon, 17 Nov 2025 15:35:53 +0100 Subject: [PATCH] Client transports: make #protocolVersions() configurable Signed-off-by: Daniel Garnier-Moiroux --- .../HttpClientStreamableHttpTransport.java | 68 ++++++-- ...ttpVersionNegotiationIntegrationTests.java | 143 +++++++++++++++++ .../McpTestRequestRecordingServletFilter.java | 128 +++++++++++++++ .../transport/McpTestServletFilter.java | 43 ------ .../server/transport/TomcatTestUtil.java | 22 +-- .../WebClientStreamableHttpTransport.java | 57 +++++-- ...ttpVersionNegotiationIntegrationTests.java | 146 ++++++++++++++++++ ...equestRecordingExchangeFilterFunction.java | 53 +++++++ 8 files changed, 580 insertions(+), 80 deletions(-) create mode 100644 mcp-core/src/test/java/io/modelcontextprotocol/common/HttpClientStreamableHttpVersionNegotiationIntegrationTests.java create mode 100644 mcp-core/src/test/java/io/modelcontextprotocol/server/transport/McpTestRequestRecordingServletFilter.java delete mode 100644 mcp-core/src/test/java/io/modelcontextprotocol/server/transport/McpTestServletFilter.java create mode 100644 mcp-spring/mcp-spring-webflux/src/test/java/io/modelcontextprotocol/WebFluxStreamableHttpVersionNegotiationIntegrationTests.java create mode 100644 mcp-spring/mcp-spring-webflux/src/test/java/io/modelcontextprotocol/utils/McpTestRequestRecordingExchangeFilterFunction.java diff --git a/mcp-core/src/main/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransport.java b/mcp-core/src/main/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransport.java index cd8fa171f..c48aedbcf 100644 --- a/mcp-core/src/main/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransport.java +++ b/mcp-core/src/main/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransport.java @@ -11,6 +11,8 @@ import java.net.http.HttpResponse; import java.net.http.HttpResponse.BodyHandler; import java.time.Duration; +import java.util.Collections; +import java.util.Comparator; import java.util.List; import java.util.Optional; import java.util.concurrent.CompletionException; @@ -18,17 +20,12 @@ import java.util.function.Consumer; import java.util.function.Function; -import org.reactivestreams.Publisher; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import io.modelcontextprotocol.json.TypeRef; -import io.modelcontextprotocol.json.McpJsonMapper; - +import io.modelcontextprotocol.client.transport.ResponseSubscribers.ResponseEvent; import io.modelcontextprotocol.client.transport.customizer.McpAsyncHttpClientRequestCustomizer; import io.modelcontextprotocol.client.transport.customizer.McpSyncHttpClientRequestCustomizer; -import io.modelcontextprotocol.client.transport.ResponseSubscribers.ResponseEvent; import io.modelcontextprotocol.common.McpTransportContext; +import io.modelcontextprotocol.json.McpJsonMapper; +import io.modelcontextprotocol.json.TypeRef; import io.modelcontextprotocol.spec.ClosedMcpTransportSession; import io.modelcontextprotocol.spec.DefaultMcpTransportSession; import io.modelcontextprotocol.spec.DefaultMcpTransportStream; @@ -42,6 +39,9 @@ import io.modelcontextprotocol.spec.ProtocolVersions; import io.modelcontextprotocol.util.Assert; import io.modelcontextprotocol.util.Utils; +import org.reactivestreams.Publisher; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import reactor.core.Disposable; import reactor.core.publisher.Flux; import reactor.core.publisher.FluxSink; @@ -78,8 +78,6 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport { private static final Logger logger = LoggerFactory.getLogger(HttpClientStreamableHttpTransport.class); - private static final String MCP_PROTOCOL_VERSION = ProtocolVersions.MCP_2025_06_18; - private static final String DEFAULT_ENDPOINT = "/mcp"; /** @@ -125,9 +123,14 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport { private final AtomicReference> exceptionHandler = new AtomicReference<>(); + private final List supportedProtocolVersions; + + private final String latestSupportedProtocolVersion; + private HttpClientStreamableHttpTransport(McpJsonMapper jsonMapper, HttpClient httpClient, HttpRequest.Builder requestBuilder, String baseUri, String endpoint, boolean resumableStreams, - boolean openConnectionOnStartup, McpAsyncHttpClientRequestCustomizer httpRequestCustomizer) { + boolean openConnectionOnStartup, McpAsyncHttpClientRequestCustomizer httpRequestCustomizer, + List supportedProtocolVersions) { this.jsonMapper = jsonMapper; this.httpClient = httpClient; this.requestBuilder = requestBuilder; @@ -137,12 +140,16 @@ private HttpClientStreamableHttpTransport(McpJsonMapper jsonMapper, HttpClient h this.openConnectionOnStartup = openConnectionOnStartup; this.activeSession.set(createTransportSession()); this.httpRequestCustomizer = httpRequestCustomizer; + this.supportedProtocolVersions = Collections.unmodifiableList(supportedProtocolVersions); + this.latestSupportedProtocolVersion = this.supportedProtocolVersions.stream() + .sorted(Comparator.reverseOrder()) + .findFirst() + .get(); } @Override public List protocolVersions() { - return List.of(ProtocolVersions.MCP_2024_11_05, ProtocolVersions.MCP_2025_03_26, - ProtocolVersions.MCP_2025_06_18); + return supportedProtocolVersions; } public static Builder builder(String baseUri) { @@ -186,7 +193,7 @@ private Publisher createDelete(String sessionId) { .uri(uri) .header("Cache-Control", "no-cache") .header(HttpHeaders.MCP_SESSION_ID, sessionId) - .header(HttpHeaders.PROTOCOL_VERSION, MCP_PROTOCOL_VERSION) + .header(HttpHeaders.PROTOCOL_VERSION, this.latestSupportedProtocolVersion) .DELETE(); var transportContext = ctx.getOrDefault(McpTransportContext.KEY, McpTransportContext.EMPTY); return Mono.from(this.httpRequestCustomizer.customize(builder, "DELETE", uri, null, transportContext)); @@ -257,7 +264,7 @@ private Mono reconnect(McpTransportStream stream) { var builder = requestBuilder.uri(uri) .header(HttpHeaders.ACCEPT, TEXT_EVENT_STREAM) .header("Cache-Control", "no-cache") - .header(HttpHeaders.PROTOCOL_VERSION, MCP_PROTOCOL_VERSION) + .header(HttpHeaders.PROTOCOL_VERSION, this.latestSupportedProtocolVersion) .GET(); var transportContext = connectionCtx.getOrDefault(McpTransportContext.KEY, McpTransportContext.EMPTY); return Mono.from(this.httpRequestCustomizer.customize(builder, "GET", uri, null, transportContext)); @@ -432,7 +439,7 @@ public Mono sendMessage(McpSchema.JSONRPCMessage sentMessage) { .header(HttpHeaders.ACCEPT, APPLICATION_JSON + ", " + TEXT_EVENT_STREAM) .header(HttpHeaders.CONTENT_TYPE, APPLICATION_JSON) .header(HttpHeaders.CACHE_CONTROL, "no-cache") - .header(HttpHeaders.PROTOCOL_VERSION, MCP_PROTOCOL_VERSION) + .header(HttpHeaders.PROTOCOL_VERSION, this.latestSupportedProtocolVersion) .POST(HttpRequest.BodyPublishers.ofString(jsonBody)); var transportContext = ctx.getOrDefault(McpTransportContext.KEY, McpTransportContext.EMPTY); return Mono @@ -624,6 +631,9 @@ public static class Builder { private Duration connectTimeout = Duration.ofSeconds(10); + private List supportedProtocolVersions = List.of(ProtocolVersions.MCP_2024_11_05, + ProtocolVersions.MCP_2025_03_26, ProtocolVersions.MCP_2025_06_18); + /** * Creates a new builder with the specified base URI. * @param baseUri the base URI of the MCP server @@ -772,6 +782,30 @@ public Builder connectTimeout(Duration connectTimeout) { return this; } + /** + * Sets the list of supported protocol versions used in version negotiation. By + * default, the client will send the latest of those versions in the + * {@code MCP-Protocol-Version} header. + *

+ * Setting this value only updates the values used in version negotiation, and + * does NOT impact the actual capabilities of the transport. It should only be + * used for compatibility with servers having strict requirements around the + * {@code MCP-Protocol-Version} header. + * @param supportedProtocolVersions protocol versions supported by this transport + * @return this builder + * @see version + * negotiation specification + * @see Protocol + * Version Header + */ + public Builder supportedProtocolVersions(List supportedProtocolVersions) { + Assert.notEmpty(supportedProtocolVersions, "supportedProtocolVersions must not be empty"); + this.supportedProtocolVersions = Collections.unmodifiableList(supportedProtocolVersions); + return this; + } + /** * Construct a fresh instance of {@link HttpClientStreamableHttpTransport} using * the current builder configuration. @@ -781,7 +815,7 @@ public HttpClientStreamableHttpTransport build() { HttpClient httpClient = this.clientBuilder.connectTimeout(this.connectTimeout).build(); return new HttpClientStreamableHttpTransport(jsonMapper == null ? McpJsonMapper.getDefault() : jsonMapper, httpClient, requestBuilder, baseUri, endpoint, resumableStreams, openConnectionOnStartup, - httpRequestCustomizer); + httpRequestCustomizer, supportedProtocolVersions); } } diff --git a/mcp-core/src/test/java/io/modelcontextprotocol/common/HttpClientStreamableHttpVersionNegotiationIntegrationTests.java b/mcp-core/src/test/java/io/modelcontextprotocol/common/HttpClientStreamableHttpVersionNegotiationIntegrationTests.java new file mode 100644 index 000000000..12a3ef9c6 --- /dev/null +++ b/mcp-core/src/test/java/io/modelcontextprotocol/common/HttpClientStreamableHttpVersionNegotiationIntegrationTests.java @@ -0,0 +1,143 @@ +/* + * Copyright 2025-2025 the original author or authors. + */ + +package io.modelcontextprotocol.common; + +import java.util.List; +import java.util.Map; +import java.util.function.BiFunction; + +import io.modelcontextprotocol.client.McpClient; +import io.modelcontextprotocol.client.transport.HttpClientStreamableHttpTransport; +import io.modelcontextprotocol.server.McpServer; +import io.modelcontextprotocol.server.McpServerFeatures; +import io.modelcontextprotocol.server.McpSyncServer; +import io.modelcontextprotocol.server.McpSyncServerExchange; +import io.modelcontextprotocol.server.transport.HttpServletStreamableServerTransportProvider; +import io.modelcontextprotocol.server.transport.McpTestRequestRecordingServletFilter; +import io.modelcontextprotocol.server.transport.TomcatTestUtil; +import io.modelcontextprotocol.spec.McpSchema; +import io.modelcontextprotocol.spec.ProtocolVersions; +import org.apache.catalina.LifecycleException; +import org.apache.catalina.LifecycleState; +import org.apache.catalina.startup.Tomcat; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +class HttpClientStreamableHttpVersionNegotiationIntegrationTests { + + private Tomcat tomcat; + + private static final int PORT = TomcatTestUtil.findAvailablePort(); + + private final McpTestRequestRecordingServletFilter requestRecordingFilter = new McpTestRequestRecordingServletFilter(); + + private final HttpServletStreamableServerTransportProvider transport = HttpServletStreamableServerTransportProvider + .builder() + .contextExtractor( + req -> McpTransportContext.create(Map.of("protocol-version", req.getHeader("MCP-protocol-version")))) + .build(); + + private final McpSchema.Tool toolSpec = McpSchema.Tool.builder() + .name("test-tool") + .description("return the protocol version used") + .build(); + + private final BiFunction toolHandler = ( + exchange, request) -> new McpSchema.CallToolResult( + exchange.transportContext().get("protocol-version").toString(), null); + + McpSyncServer mcpServer = McpServer.sync(transport) + .capabilities(McpSchema.ServerCapabilities.builder().tools(false).build()) + .tools(new McpServerFeatures.SyncToolSpecification(toolSpec, null, toolHandler)) + .build(); + + @AfterEach + void tearDown() { + stopTomcat(); + } + + @Test + void usesLatestVersion() { + startTomcat(); + + var client = McpClient.sync(HttpClientStreamableHttpTransport.builder("http://localhost:" + PORT).build()) + .build(); + + client.initialize(); + McpSchema.CallToolResult response = client.callTool(new McpSchema.CallToolRequest("test-tool", Map.of())); + + var calls = requestRecordingFilter.getCalls(); + + assertThat(calls).filteredOn(c -> !c.body().contains("\"method\":\"initialize\"")) + // GET /mcp ; POST notification/initialized ; POST tools/call + .hasSize(3) + .map(McpTestRequestRecordingServletFilter.Call::headers) + .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(ProtocolVersions.MCP_2025_06_18); + mcpServer.close(); + } + + @Test + void usesCustomLatestVersion() { + startTomcat(); + + var transport = HttpClientStreamableHttpTransport.builder("http://localhost:" + PORT) + .supportedProtocolVersions(List.of(ProtocolVersions.MCP_2025_06_18, "2263-03-18")) + .build(); + var client = McpClient.sync(transport).build(); + + client.initialize(); + McpSchema.CallToolResult response = client.callTool(new McpSchema.CallToolRequest("test-tool", Map.of())); + + var calls = requestRecordingFilter.getCalls(); + + assertThat(calls).filteredOn(c -> !c.body().contains("\"method\":\"initialize\"")) + // GET /mcp ; POST notification/initialized ; POST tools/call + .hasSize(3) + .map(McpTestRequestRecordingServletFilter.Call::headers) + .allSatisfy(headers -> assertThat(headers).containsEntry("mcp-protocol-version", "2263-03-18")); + + assertThat(response).isNotNull(); + assertThat(response.content()).hasSize(1) + .first() + .extracting(McpSchema.TextContent.class::cast) + .extracting(McpSchema.TextContent::text) + .isEqualTo("2263-03-18"); + mcpServer.close(); + } + + private void startTomcat() { + tomcat = TomcatTestUtil.createTomcatServer("", PORT, transport, requestRecordingFilter); + try { + tomcat.start(); + assertThat(tomcat.getServer().getState()).isEqualTo(LifecycleState.STARTED); + } + catch (Exception e) { + throw new RuntimeException("Failed to start Tomcat", e); + } + } + + private void stopTomcat() { + if (tomcat != null) { + try { + tomcat.stop(); + tomcat.destroy(); + } + catch (LifecycleException e) { + throw new RuntimeException("Failed to stop Tomcat", e); + } + } + } + +} diff --git a/mcp-core/src/test/java/io/modelcontextprotocol/server/transport/McpTestRequestRecordingServletFilter.java b/mcp-core/src/test/java/io/modelcontextprotocol/server/transport/McpTestRequestRecordingServletFilter.java new file mode 100644 index 000000000..09f0d305d --- /dev/null +++ b/mcp-core/src/test/java/io/modelcontextprotocol/server/transport/McpTestRequestRecordingServletFilter.java @@ -0,0 +1,128 @@ +/* + * Copyright 2025 - 2025 the original author or authors. + */ + +package io.modelcontextprotocol.server.transport; + +import java.io.BufferedReader; +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.function.Function; +import java.util.stream.Collectors; + +import jakarta.servlet.Filter; +import jakarta.servlet.FilterChain; +import jakarta.servlet.ReadListener; +import jakarta.servlet.ServletException; +import jakarta.servlet.ServletInputStream; +import jakarta.servlet.ServletRequest; +import jakarta.servlet.ServletResponse; +import jakarta.servlet.http.HttpServletRequest; +import jakarta.servlet.http.HttpServletRequestWrapper; + +/** + * Simple {@link Filter} which records calls made to an MCP server. + * + * @author Daniel Garnier-Moiroux + */ +public class McpTestRequestRecordingServletFilter implements Filter { + + private final List calls = new ArrayList<>(); + + @Override + public void doFilter(ServletRequest servletRequest, ServletResponse servletResponse, FilterChain filterChain) + throws IOException, ServletException { + + if (servletRequest instanceof HttpServletRequest req) { + var headers = Collections.list(req.getHeaderNames()) + .stream() + .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())); + filterChain.doFilter(request, servletResponse); + } + else { + filterChain.doFilter(servletRequest, servletResponse); + } + + } + + public List getCalls() { + + return List.copyOf(calls); + } + + public record Call(Map headers, String body) { + + } + + public static class CachedBodyHttpServletRequest extends HttpServletRequestWrapper { + + private final byte[] cachedBody; + + public CachedBodyHttpServletRequest(HttpServletRequest request) throws IOException { + super(request); + this.cachedBody = request.getInputStream().readAllBytes(); + } + + @Override + public ServletInputStream getInputStream() { + return new CachedBodyServletInputStream(cachedBody); + } + + @Override + public BufferedReader getReader() { + return new BufferedReader(new InputStreamReader(getInputStream(), StandardCharsets.UTF_8)); + } + + public String getBodyAsString() { + return new String(cachedBody, StandardCharsets.UTF_8); + } + + } + + public static class CachedBodyServletInputStream extends ServletInputStream { + + private InputStream cachedBodyInputStream; + + public CachedBodyServletInputStream(byte[] cachedBody) { + this.cachedBodyInputStream = new ByteArrayInputStream(cachedBody); + } + + @Override + public boolean isFinished() { + try { + return cachedBodyInputStream.available() == 0; + } + catch (IOException e) { + e.printStackTrace(); + } + return false; + } + + @Override + public boolean isReady() { + return true; + } + + @Override + public void setReadListener(ReadListener readListener) { + throw new UnsupportedOperationException(); + } + + @Override + public int read() throws IOException { + return cachedBodyInputStream.read(); + } + + } + +} diff --git a/mcp-core/src/test/java/io/modelcontextprotocol/server/transport/McpTestServletFilter.java b/mcp-core/src/test/java/io/modelcontextprotocol/server/transport/McpTestServletFilter.java deleted file mode 100644 index cc2543aa9..000000000 --- a/mcp-core/src/test/java/io/modelcontextprotocol/server/transport/McpTestServletFilter.java +++ /dev/null @@ -1,43 +0,0 @@ -/* - * Copyright 2025 - 2025 the original author or authors. - */ - -package io.modelcontextprotocol.server.transport; - -import java.io.IOException; - -import jakarta.servlet.Filter; -import jakarta.servlet.FilterChain; -import jakarta.servlet.ServletException; -import jakarta.servlet.ServletRequest; -import jakarta.servlet.ServletResponse; - -/** - * Simple {@link Filter} which sets a value in a thread local. Used to verify whether MCP - * executions happen on the thread processing the request or are offloaded. - * - * @author Daniel Garnier-Moiroux - */ -public class McpTestServletFilter implements Filter { - - public static final String THREAD_LOCAL_VALUE = McpTestServletFilter.class.getName(); - - private static final ThreadLocal holder = new ThreadLocal<>(); - - @Override - public void doFilter(ServletRequest servletRequest, ServletResponse servletResponse, FilterChain filterChain) - throws IOException, ServletException { - holder.set(THREAD_LOCAL_VALUE); - try { - filterChain.doFilter(servletRequest, servletResponse); - } - finally { - holder.remove(); - } - } - - public static String getThreadLocalValue() { - return holder.get(); - } - -} diff --git a/mcp-core/src/test/java/io/modelcontextprotocol/server/transport/TomcatTestUtil.java b/mcp-core/src/test/java/io/modelcontextprotocol/server/transport/TomcatTestUtil.java index 2cf95dc94..490e29838 100644 --- a/mcp-core/src/test/java/io/modelcontextprotocol/server/transport/TomcatTestUtil.java +++ b/mcp-core/src/test/java/io/modelcontextprotocol/server/transport/TomcatTestUtil.java @@ -8,6 +8,7 @@ import java.net.InetSocketAddress; import java.net.ServerSocket; +import jakarta.servlet.Filter; import jakarta.servlet.Servlet; import org.apache.catalina.Context; import org.apache.catalina.startup.Tomcat; @@ -24,7 +25,8 @@ public class TomcatTestUtil { // Prevent instantiation } - public static Tomcat createTomcatServer(String contextPath, int port, Servlet servlet) { + public static Tomcat createTomcatServer(String contextPath, int port, Servlet servlet, + Filter... additionalFilters) { var tomcat = new Tomcat(); tomcat.setPort(port); @@ -43,15 +45,17 @@ public static Tomcat createTomcatServer(String contextPath, int port, Servlet se context.addChild(wrapper); context.addServletMappingDecoded("/*", "mcpServlet"); - var filterDef = new FilterDef(); - filterDef.setFilterClass(McpTestServletFilter.class.getName()); - filterDef.setFilterName(McpTestServletFilter.class.getSimpleName()); - context.addFilterDef(filterDef); + for (var filter : additionalFilters) { + var filterDef = new FilterDef(); + filterDef.setFilter(filter); + filterDef.setFilterName(McpTestRequestRecordingServletFilter.class.getSimpleName()); + context.addFilterDef(filterDef); - var filterMap = new FilterMap(); - filterMap.setFilterName(McpTestServletFilter.class.getSimpleName()); - filterMap.addURLPattern("/*"); - context.addFilterMap(filterMap); + var filterMap = new FilterMap(); + filterMap.setFilterName(McpTestRequestRecordingServletFilter.class.getSimpleName()); + filterMap.addURLPattern("/*"); + context.addFilterMap(filterMap); + } var connector = tomcat.getConnector(); connector.setAsyncTimeout(3000); diff --git a/mcp-spring/mcp-spring-webflux/src/main/java/io/modelcontextprotocol/client/transport/WebClientStreamableHttpTransport.java b/mcp-spring/mcp-spring-webflux/src/main/java/io/modelcontextprotocol/client/transport/WebClientStreamableHttpTransport.java index 5ec272961..b67d34f6b 100644 --- a/mcp-spring/mcp-spring-webflux/src/main/java/io/modelcontextprotocol/client/transport/WebClientStreamableHttpTransport.java +++ b/mcp-spring/mcp-spring-webflux/src/main/java/io/modelcontextprotocol/client/transport/WebClientStreamableHttpTransport.java @@ -5,6 +5,8 @@ package io.modelcontextprotocol.client.transport; import java.io.IOException; +import java.util.Collections; +import java.util.Comparator; import java.util.List; import java.util.Optional; import java.util.concurrent.atomic.AtomicReference; @@ -22,9 +24,8 @@ import org.springframework.web.reactive.function.client.WebClient; import org.springframework.web.reactive.function.client.WebClientResponseException; -import io.modelcontextprotocol.json.TypeRef; import io.modelcontextprotocol.json.McpJsonMapper; - +import io.modelcontextprotocol.json.TypeRef; import io.modelcontextprotocol.spec.ClosedMcpTransportSession; import io.modelcontextprotocol.spec.DefaultMcpTransportSession; import io.modelcontextprotocol.spec.DefaultMcpTransportStream; @@ -76,8 +77,6 @@ public class WebClientStreamableHttpTransport implements McpClientTransport { private static final Logger logger = LoggerFactory.getLogger(WebClientStreamableHttpTransport.class); - private static final String MCP_PROTOCOL_VERSION = ProtocolVersions.MCP_2025_06_18; - private static final String DEFAULT_ENDPOINT = "/mcp"; /** @@ -105,20 +104,29 @@ public class WebClientStreamableHttpTransport implements McpClientTransport { private final AtomicReference> exceptionHandler = new AtomicReference<>(); + private final List supportedProtocolVersions; + + private final String latestSupportedProtocolVersion; + private WebClientStreamableHttpTransport(McpJsonMapper jsonMapper, WebClient.Builder webClientBuilder, - String endpoint, boolean resumableStreams, boolean openConnectionOnStartup) { + String endpoint, boolean resumableStreams, boolean openConnectionOnStartup, + List supportedProtocolVersions) { this.jsonMapper = jsonMapper; this.webClient = webClientBuilder.build(); this.endpoint = endpoint; this.resumableStreams = resumableStreams; this.openConnectionOnStartup = openConnectionOnStartup; this.activeSession.set(createTransportSession()); + this.supportedProtocolVersions = List.copyOf(supportedProtocolVersions); + this.latestSupportedProtocolVersion = this.supportedProtocolVersions.stream() + .sorted(Comparator.reverseOrder()) + .findFirst() + .get(); } @Override public List protocolVersions() { - return List.of(ProtocolVersions.MCP_2024_11_05, ProtocolVersions.MCP_2025_03_26, - ProtocolVersions.MCP_2025_06_18); + return supportedProtocolVersions; } /** @@ -149,7 +157,7 @@ private McpTransportSession createTransportSession() { : webClient.delete() .uri(this.endpoint) .header(HttpHeaders.MCP_SESSION_ID, sessionId) - .header(HttpHeaders.PROTOCOL_VERSION, MCP_PROTOCOL_VERSION) + .header(HttpHeaders.PROTOCOL_VERSION, this.latestSupportedProtocolVersion) .retrieve() .toBodilessEntity() .onErrorComplete(e -> { @@ -217,7 +225,7 @@ private Mono reconnect(McpTransportStream stream) { Disposable connection = webClient.get() .uri(this.endpoint) .accept(MediaType.TEXT_EVENT_STREAM) - .header(HttpHeaders.PROTOCOL_VERSION, MCP_PROTOCOL_VERSION) + .header(HttpHeaders.PROTOCOL_VERSION, this.latestSupportedProtocolVersion) .headers(httpHeaders -> { transportSession.sessionId().ifPresent(id -> httpHeaders.add(HttpHeaders.MCP_SESSION_ID, id)); if (stream != null) { @@ -283,7 +291,7 @@ public Mono sendMessage(McpSchema.JSONRPCMessage message) { Disposable connection = webClient.post() .uri(this.endpoint) .accept(MediaType.APPLICATION_JSON, MediaType.TEXT_EVENT_STREAM) - .header(HttpHeaders.PROTOCOL_VERSION, MCP_PROTOCOL_VERSION) + .header(HttpHeaders.PROTOCOL_VERSION, this.latestSupportedProtocolVersion) .headers(httpHeaders -> { transportSession.sessionId().ifPresent(id -> httpHeaders.add(HttpHeaders.MCP_SESSION_ID, id)); }) @@ -495,6 +503,9 @@ public static class Builder { private boolean openConnectionOnStartup = false; + private List supportedProtocolVersions = List.of(ProtocolVersions.MCP_2024_11_05, + ProtocolVersions.MCP_2025_03_26, ProtocolVersions.MCP_2025_06_18); + private Builder(WebClient.Builder webClientBuilder) { Assert.notNull(webClientBuilder, "WebClient.Builder must not be null"); this.webClientBuilder = webClientBuilder; @@ -560,6 +571,30 @@ public Builder openConnectionOnStartup(boolean openConnectionOnStartup) { return this; } + /** + * Sets the list of supported protocol versions used in version negotiation. By + * default, the client will send the latest of those versions in the + * {@code MCP-Protocol-Version} header. + *

+ * Setting this value only updates the values used in version negotiation, and + * does NOT impact the actual capabilities of the transport. It should only be + * used for compatibility with servers having strict requirements around the + * {@code MCP-Protocol-Version} header. + * @param supportedProtocolVersions protocol versions supported by this transport + * @return this builder + * @see version + * negotiation specification + * @see Protocol + * Version Header + */ + public Builder supportedProtocolVersions(List supportedProtocolVersions) { + Assert.notEmpty(supportedProtocolVersions, "supportedProtocolVersions must not be empty"); + this.supportedProtocolVersions = Collections.unmodifiableList(supportedProtocolVersions); + return this; + } + /** * Construct a fresh instance of {@link WebClientStreamableHttpTransport} using * the current builder configuration. @@ -567,7 +602,7 @@ public Builder openConnectionOnStartup(boolean openConnectionOnStartup) { */ public WebClientStreamableHttpTransport build() { return new WebClientStreamableHttpTransport(jsonMapper == null ? McpJsonMapper.getDefault() : jsonMapper, - webClientBuilder, endpoint, resumableStreams, openConnectionOnStartup); + webClientBuilder, endpoint, resumableStreams, openConnectionOnStartup, supportedProtocolVersions); } } diff --git a/mcp-spring/mcp-spring-webflux/src/test/java/io/modelcontextprotocol/WebFluxStreamableHttpVersionNegotiationIntegrationTests.java b/mcp-spring/mcp-spring-webflux/src/test/java/io/modelcontextprotocol/WebFluxStreamableHttpVersionNegotiationIntegrationTests.java new file mode 100644 index 000000000..7627bd419 --- /dev/null +++ b/mcp-spring/mcp-spring-webflux/src/test/java/io/modelcontextprotocol/WebFluxStreamableHttpVersionNegotiationIntegrationTests.java @@ -0,0 +1,146 @@ +/* + * Copyright 2025-2025 the original author or authors. + */ + +package io.modelcontextprotocol; + +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.function.BiFunction; + +import io.modelcontextprotocol.client.McpClient; +import io.modelcontextprotocol.client.transport.WebClientStreamableHttpTransport; +import io.modelcontextprotocol.common.McpTransportContext; +import io.modelcontextprotocol.server.McpServer; +import io.modelcontextprotocol.server.McpServerFeatures; +import io.modelcontextprotocol.server.McpSyncServer; +import io.modelcontextprotocol.server.McpSyncServerExchange; +import io.modelcontextprotocol.server.TestUtil; +import io.modelcontextprotocol.server.transport.WebFluxStreamableServerTransportProvider; +import io.modelcontextprotocol.spec.McpSchema; +import io.modelcontextprotocol.spec.ProtocolVersions; +import io.modelcontextprotocol.utils.McpTestRequestRecordingExchangeFilterFunction; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import reactor.netty.DisposableServer; +import reactor.netty.http.server.HttpServer; + +import org.springframework.http.server.reactive.HttpHandler; +import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter; +import org.springframework.web.reactive.function.client.WebClient; +import org.springframework.web.reactive.function.server.RouterFunction; +import org.springframework.web.reactive.function.server.RouterFunctions; +import org.springframework.web.reactive.function.server.ServerResponse; + +import static org.assertj.core.api.Assertions.assertThat; + +class WebFluxStreamableHttpVersionNegotiationIntegrationTests { + + private static final int PORT = TestUtil.findAvailablePort(); + + private DisposableServer httpServer; + + private final McpTestRequestRecordingExchangeFilterFunction recordingFilterFunction = new McpTestRequestRecordingExchangeFilterFunction(); + + private final McpSchema.Tool toolSpec = McpSchema.Tool.builder() + .name("test-tool") + .description("return the protocol version used") + .build(); + + private final BiFunction toolHandler = ( + exchange, request) -> new McpSchema.CallToolResult( + exchange.transportContext().get("protocol-version").toString(), null); + + private final WebFluxStreamableServerTransportProvider mcpStreamableServerTransportProvider = WebFluxStreamableServerTransportProvider + .builder() + .contextExtractor(req -> McpTransportContext + .create(Map.of("protocol-version", req.headers().firstHeader("MCP-protocol-version")))) + .build(); + + private final McpSyncServer mcpServer = McpServer.sync(mcpStreamableServerTransportProvider) + .capabilities(McpSchema.ServerCapabilities.builder().tools(false).build()) + .tools(new McpServerFeatures.SyncToolSpecification(toolSpec, null, toolHandler)) + .build(); + + @BeforeEach + void setUp() { + RouterFunction filteredRouter = mcpStreamableServerTransportProvider.getRouterFunction() + .filter(recordingFilterFunction); + + HttpHandler httpHandler = RouterFunctions.toHttpHandler(filteredRouter); + + ReactorHttpHandlerAdapter adapter = new ReactorHttpHandlerAdapter(httpHandler); + + this.httpServer = HttpServer.create().port(PORT).handle(adapter).bindNow(); + } + + @AfterEach + public void after() { + if (httpServer != null) { + httpServer.disposeNow(); + } + if (mcpServer != null) { + mcpServer.close(); + } + } + + @Test + void usesLatestVersion() { + var client = McpClient + .sync(WebClientStreamableHttpTransport.builder(WebClient.builder().baseUrl("http://localhost:" + PORT)) + .build()) + .requestTimeout(Duration.ofHours(10)) + .build(); + + client.initialize(); + + 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\"")) + // GET /mcp ; POST notification/initialized ; POST tools/call + .hasSize(3) + .map(McpTestRequestRecordingExchangeFilterFunction.Call::headers) + .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(ProtocolVersions.MCP_2025_06_18); + mcpServer.close(); + } + + @Test + void usesCustomLatestVersion() { + var transport = WebClientStreamableHttpTransport + .builder(WebClient.builder().baseUrl("http://localhost:" + PORT)) + .supportedProtocolVersions(List.of(ProtocolVersions.MCP_2025_06_18, "2263-03-18")) + .build(); + var client = McpClient.sync(transport).requestTimeout(Duration.ofHours(10)).build(); + + client.initialize(); + + 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\"")) + // GET /mcp ; POST notification/initialized ; POST tools/call + .hasSize(3) + .map(McpTestRequestRecordingExchangeFilterFunction.Call::headers) + .allSatisfy(headers -> assertThat(headers).containsEntry("mcp-protocol-version", "2263-03-18")); + + assertThat(response).isNotNull(); + assertThat(response.content()).hasSize(1) + .first() + .extracting(McpSchema.TextContent.class::cast) + .extracting(McpSchema.TextContent::text) + .isEqualTo("2263-03-18"); + mcpServer.close(); + } + +} diff --git a/mcp-spring/mcp-spring-webflux/src/test/java/io/modelcontextprotocol/utils/McpTestRequestRecordingExchangeFilterFunction.java b/mcp-spring/mcp-spring-webflux/src/test/java/io/modelcontextprotocol/utils/McpTestRequestRecordingExchangeFilterFunction.java new file mode 100644 index 000000000..5600795c1 --- /dev/null +++ b/mcp-spring/mcp-spring-webflux/src/test/java/io/modelcontextprotocol/utils/McpTestRequestRecordingExchangeFilterFunction.java @@ -0,0 +1,53 @@ +/* + * Copyright 2025-2025 the original author or authors. + */ + +package io.modelcontextprotocol.utils; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import reactor.core.publisher.Mono; + +import org.springframework.web.reactive.function.server.HandlerFilterFunction; +import org.springframework.web.reactive.function.server.HandlerFunction; +import org.springframework.web.reactive.function.server.ServerRequest; +import org.springframework.web.reactive.function.server.ServerResponse; + +/** + * Simple {@link HandlerFilterFunction} which records calls made to an MCP server. + * + * @author Daniel Garnier-Moiroux + */ +public class McpTestRequestRecordingExchangeFilterFunction implements HandlerFilterFunction { + + private final List calls = new ArrayList<>(); + + @Override + public Mono filter(ServerRequest request, HandlerFunction next) { + Map headers = request.headers() + .asHttpHeaders() + .keySet() + .stream() + .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)); + return ServerRequest.from(request).body(body).build(); + }); + + return cr.flatMap(next::handle); + + } + + public List getCalls() { + return List.copyOf(calls); + } + + public record Call(Map headers, String body) { + + } + +}