Client transports: make #protocolVersions() configurable (#669)
Signed-off-by: Daniel Garnier-Moiroux <git@garnier.wf>
This commit is contained in:
parent
bc308573f3
commit
87bdf1ee49
8 changed files with 580 additions and 80 deletions
|
|
@ -11,6 +11,8 @@ import java.net.http.HttpRequest;
|
||||||
import java.net.http.HttpResponse;
|
import java.net.http.HttpResponse;
|
||||||
import java.net.http.HttpResponse.BodyHandler;
|
import java.net.http.HttpResponse.BodyHandler;
|
||||||
import java.time.Duration;
|
import java.time.Duration;
|
||||||
|
import java.util.Collections;
|
||||||
|
import java.util.Comparator;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Optional;
|
import java.util.Optional;
|
||||||
import java.util.concurrent.CompletionException;
|
import java.util.concurrent.CompletionException;
|
||||||
|
|
@ -18,17 +20,12 @@ import java.util.concurrent.atomic.AtomicReference;
|
||||||
import java.util.function.Consumer;
|
import java.util.function.Consumer;
|
||||||
import java.util.function.Function;
|
import java.util.function.Function;
|
||||||
|
|
||||||
import org.reactivestreams.Publisher;
|
import io.modelcontextprotocol.client.transport.ResponseSubscribers.ResponseEvent;
|
||||||
import org.slf4j.Logger;
|
|
||||||
import org.slf4j.LoggerFactory;
|
|
||||||
|
|
||||||
import io.modelcontextprotocol.json.TypeRef;
|
|
||||||
import io.modelcontextprotocol.json.McpJsonMapper;
|
|
||||||
|
|
||||||
import io.modelcontextprotocol.client.transport.customizer.McpAsyncHttpClientRequestCustomizer;
|
import io.modelcontextprotocol.client.transport.customizer.McpAsyncHttpClientRequestCustomizer;
|
||||||
import io.modelcontextprotocol.client.transport.customizer.McpSyncHttpClientRequestCustomizer;
|
import io.modelcontextprotocol.client.transport.customizer.McpSyncHttpClientRequestCustomizer;
|
||||||
import io.modelcontextprotocol.client.transport.ResponseSubscribers.ResponseEvent;
|
|
||||||
import io.modelcontextprotocol.common.McpTransportContext;
|
import io.modelcontextprotocol.common.McpTransportContext;
|
||||||
|
import io.modelcontextprotocol.json.McpJsonMapper;
|
||||||
|
import io.modelcontextprotocol.json.TypeRef;
|
||||||
import io.modelcontextprotocol.spec.ClosedMcpTransportSession;
|
import io.modelcontextprotocol.spec.ClosedMcpTransportSession;
|
||||||
import io.modelcontextprotocol.spec.DefaultMcpTransportSession;
|
import io.modelcontextprotocol.spec.DefaultMcpTransportSession;
|
||||||
import io.modelcontextprotocol.spec.DefaultMcpTransportStream;
|
import io.modelcontextprotocol.spec.DefaultMcpTransportStream;
|
||||||
|
|
@ -42,6 +39,9 @@ import io.modelcontextprotocol.spec.McpTransportStream;
|
||||||
import io.modelcontextprotocol.spec.ProtocolVersions;
|
import io.modelcontextprotocol.spec.ProtocolVersions;
|
||||||
import io.modelcontextprotocol.util.Assert;
|
import io.modelcontextprotocol.util.Assert;
|
||||||
import io.modelcontextprotocol.util.Utils;
|
import io.modelcontextprotocol.util.Utils;
|
||||||
|
import org.reactivestreams.Publisher;
|
||||||
|
import org.slf4j.Logger;
|
||||||
|
import org.slf4j.LoggerFactory;
|
||||||
import reactor.core.Disposable;
|
import reactor.core.Disposable;
|
||||||
import reactor.core.publisher.Flux;
|
import reactor.core.publisher.Flux;
|
||||||
import reactor.core.publisher.FluxSink;
|
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 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";
|
private static final String DEFAULT_ENDPOINT = "/mcp";
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
@ -125,9 +123,14 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
|
||||||
|
|
||||||
private final AtomicReference<Consumer<Throwable>> exceptionHandler = new AtomicReference<>();
|
private final AtomicReference<Consumer<Throwable>> exceptionHandler = new AtomicReference<>();
|
||||||
|
|
||||||
|
private final List<String> supportedProtocolVersions;
|
||||||
|
|
||||||
|
private final String latestSupportedProtocolVersion;
|
||||||
|
|
||||||
private HttpClientStreamableHttpTransport(McpJsonMapper jsonMapper, HttpClient httpClient,
|
private HttpClientStreamableHttpTransport(McpJsonMapper jsonMapper, HttpClient httpClient,
|
||||||
HttpRequest.Builder requestBuilder, String baseUri, String endpoint, boolean resumableStreams,
|
HttpRequest.Builder requestBuilder, String baseUri, String endpoint, boolean resumableStreams,
|
||||||
boolean openConnectionOnStartup, McpAsyncHttpClientRequestCustomizer httpRequestCustomizer) {
|
boolean openConnectionOnStartup, McpAsyncHttpClientRequestCustomizer httpRequestCustomizer,
|
||||||
|
List<String> supportedProtocolVersions) {
|
||||||
this.jsonMapper = jsonMapper;
|
this.jsonMapper = jsonMapper;
|
||||||
this.httpClient = httpClient;
|
this.httpClient = httpClient;
|
||||||
this.requestBuilder = requestBuilder;
|
this.requestBuilder = requestBuilder;
|
||||||
|
|
@ -137,12 +140,16 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
|
||||||
this.openConnectionOnStartup = openConnectionOnStartup;
|
this.openConnectionOnStartup = openConnectionOnStartup;
|
||||||
this.activeSession.set(createTransportSession());
|
this.activeSession.set(createTransportSession());
|
||||||
this.httpRequestCustomizer = httpRequestCustomizer;
|
this.httpRequestCustomizer = httpRequestCustomizer;
|
||||||
|
this.supportedProtocolVersions = Collections.unmodifiableList(supportedProtocolVersions);
|
||||||
|
this.latestSupportedProtocolVersion = this.supportedProtocolVersions.stream()
|
||||||
|
.sorted(Comparator.reverseOrder())
|
||||||
|
.findFirst()
|
||||||
|
.get();
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public List<String> protocolVersions() {
|
public List<String> protocolVersions() {
|
||||||
return List.of(ProtocolVersions.MCP_2024_11_05, ProtocolVersions.MCP_2025_03_26,
|
return supportedProtocolVersions;
|
||||||
ProtocolVersions.MCP_2025_06_18);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public static Builder builder(String baseUri) {
|
public static Builder builder(String baseUri) {
|
||||||
|
|
@ -186,7 +193,7 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
|
||||||
.uri(uri)
|
.uri(uri)
|
||||||
.header("Cache-Control", "no-cache")
|
.header("Cache-Control", "no-cache")
|
||||||
.header(HttpHeaders.MCP_SESSION_ID, sessionId)
|
.header(HttpHeaders.MCP_SESSION_ID, sessionId)
|
||||||
.header(HttpHeaders.PROTOCOL_VERSION, MCP_PROTOCOL_VERSION)
|
.header(HttpHeaders.PROTOCOL_VERSION, this.latestSupportedProtocolVersion)
|
||||||
.DELETE();
|
.DELETE();
|
||||||
var transportContext = ctx.getOrDefault(McpTransportContext.KEY, McpTransportContext.EMPTY);
|
var transportContext = ctx.getOrDefault(McpTransportContext.KEY, McpTransportContext.EMPTY);
|
||||||
return Mono.from(this.httpRequestCustomizer.customize(builder, "DELETE", uri, null, transportContext));
|
return Mono.from(this.httpRequestCustomizer.customize(builder, "DELETE", uri, null, transportContext));
|
||||||
|
|
@ -257,7 +264,7 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
|
||||||
var builder = requestBuilder.uri(uri)
|
var builder = requestBuilder.uri(uri)
|
||||||
.header(HttpHeaders.ACCEPT, TEXT_EVENT_STREAM)
|
.header(HttpHeaders.ACCEPT, TEXT_EVENT_STREAM)
|
||||||
.header("Cache-Control", "no-cache")
|
.header("Cache-Control", "no-cache")
|
||||||
.header(HttpHeaders.PROTOCOL_VERSION, MCP_PROTOCOL_VERSION)
|
.header(HttpHeaders.PROTOCOL_VERSION, this.latestSupportedProtocolVersion)
|
||||||
.GET();
|
.GET();
|
||||||
var transportContext = connectionCtx.getOrDefault(McpTransportContext.KEY, McpTransportContext.EMPTY);
|
var transportContext = connectionCtx.getOrDefault(McpTransportContext.KEY, McpTransportContext.EMPTY);
|
||||||
return Mono.from(this.httpRequestCustomizer.customize(builder, "GET", uri, null, transportContext));
|
return Mono.from(this.httpRequestCustomizer.customize(builder, "GET", uri, null, transportContext));
|
||||||
|
|
@ -432,7 +439,7 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
|
||||||
.header(HttpHeaders.ACCEPT, APPLICATION_JSON + ", " + TEXT_EVENT_STREAM)
|
.header(HttpHeaders.ACCEPT, APPLICATION_JSON + ", " + TEXT_EVENT_STREAM)
|
||||||
.header(HttpHeaders.CONTENT_TYPE, APPLICATION_JSON)
|
.header(HttpHeaders.CONTENT_TYPE, APPLICATION_JSON)
|
||||||
.header(HttpHeaders.CACHE_CONTROL, "no-cache")
|
.header(HttpHeaders.CACHE_CONTROL, "no-cache")
|
||||||
.header(HttpHeaders.PROTOCOL_VERSION, MCP_PROTOCOL_VERSION)
|
.header(HttpHeaders.PROTOCOL_VERSION, this.latestSupportedProtocolVersion)
|
||||||
.POST(HttpRequest.BodyPublishers.ofString(jsonBody));
|
.POST(HttpRequest.BodyPublishers.ofString(jsonBody));
|
||||||
var transportContext = ctx.getOrDefault(McpTransportContext.KEY, McpTransportContext.EMPTY);
|
var transportContext = ctx.getOrDefault(McpTransportContext.KEY, McpTransportContext.EMPTY);
|
||||||
return Mono
|
return Mono
|
||||||
|
|
@ -624,6 +631,9 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
|
||||||
|
|
||||||
private Duration connectTimeout = Duration.ofSeconds(10);
|
private Duration connectTimeout = Duration.ofSeconds(10);
|
||||||
|
|
||||||
|
private List<String> 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.
|
* Creates a new builder with the specified base URI.
|
||||||
* @param baseUri the base URI of the MCP server
|
* @param baseUri the base URI of the MCP server
|
||||||
|
|
@ -772,6 +782,30 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
|
||||||
return this;
|
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.
|
||||||
|
* <p>
|
||||||
|
* 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 <a href=
|
||||||
|
* "https://modelcontextprotocol.io/specification/2024-11-05/basic/lifecycle#version-negotiation">version
|
||||||
|
* negotiation specification</a>
|
||||||
|
* @see <a href=
|
||||||
|
* "https://modelcontextprotocol.io/specification/2025-06-18/basic/transports#protocol-version-header">Protocol
|
||||||
|
* Version Header</a>
|
||||||
|
*/
|
||||||
|
public Builder supportedProtocolVersions(List<String> supportedProtocolVersions) {
|
||||||
|
Assert.notEmpty(supportedProtocolVersions, "supportedProtocolVersions must not be empty");
|
||||||
|
this.supportedProtocolVersions = Collections.unmodifiableList(supportedProtocolVersions);
|
||||||
|
return this;
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Construct a fresh instance of {@link HttpClientStreamableHttpTransport} using
|
* Construct a fresh instance of {@link HttpClientStreamableHttpTransport} using
|
||||||
* the current builder configuration.
|
* the current builder configuration.
|
||||||
|
|
@ -781,7 +815,7 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
|
||||||
HttpClient httpClient = this.clientBuilder.connectTimeout(this.connectTimeout).build();
|
HttpClient httpClient = this.clientBuilder.connectTimeout(this.connectTimeout).build();
|
||||||
return new HttpClientStreamableHttpTransport(jsonMapper == null ? McpJsonMapper.getDefault() : jsonMapper,
|
return new HttpClientStreamableHttpTransport(jsonMapper == null ? McpJsonMapper.getDefault() : jsonMapper,
|
||||||
httpClient, requestBuilder, baseUri, endpoint, resumableStreams, openConnectionOnStartup,
|
httpClient, requestBuilder, baseUri, endpoint, resumableStreams, openConnectionOnStartup,
|
||||||
httpRequestCustomizer);
|
httpRequestCustomizer, supportedProtocolVersions);
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -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<McpSyncServerExchange, McpSchema.CallToolRequest, McpSchema.CallToolResult> 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);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
@ -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<Call> 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<Call> getCalls() {
|
||||||
|
|
||||||
|
return List.copyOf(calls);
|
||||||
|
}
|
||||||
|
|
||||||
|
public record Call(Map<String, String> 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();
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
@ -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<String> 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();
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
@ -8,6 +8,7 @@ import java.io.IOException;
|
||||||
import java.net.InetSocketAddress;
|
import java.net.InetSocketAddress;
|
||||||
import java.net.ServerSocket;
|
import java.net.ServerSocket;
|
||||||
|
|
||||||
|
import jakarta.servlet.Filter;
|
||||||
import jakarta.servlet.Servlet;
|
import jakarta.servlet.Servlet;
|
||||||
import org.apache.catalina.Context;
|
import org.apache.catalina.Context;
|
||||||
import org.apache.catalina.startup.Tomcat;
|
import org.apache.catalina.startup.Tomcat;
|
||||||
|
|
@ -24,7 +25,8 @@ public class TomcatTestUtil {
|
||||||
// Prevent instantiation
|
// 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();
|
var tomcat = new Tomcat();
|
||||||
tomcat.setPort(port);
|
tomcat.setPort(port);
|
||||||
|
|
@ -43,15 +45,17 @@ public class TomcatTestUtil {
|
||||||
context.addChild(wrapper);
|
context.addChild(wrapper);
|
||||||
context.addServletMappingDecoded("/*", "mcpServlet");
|
context.addServletMappingDecoded("/*", "mcpServlet");
|
||||||
|
|
||||||
|
for (var filter : additionalFilters) {
|
||||||
var filterDef = new FilterDef();
|
var filterDef = new FilterDef();
|
||||||
filterDef.setFilterClass(McpTestServletFilter.class.getName());
|
filterDef.setFilter(filter);
|
||||||
filterDef.setFilterName(McpTestServletFilter.class.getSimpleName());
|
filterDef.setFilterName(McpTestRequestRecordingServletFilter.class.getSimpleName());
|
||||||
context.addFilterDef(filterDef);
|
context.addFilterDef(filterDef);
|
||||||
|
|
||||||
var filterMap = new FilterMap();
|
var filterMap = new FilterMap();
|
||||||
filterMap.setFilterName(McpTestServletFilter.class.getSimpleName());
|
filterMap.setFilterName(McpTestRequestRecordingServletFilter.class.getSimpleName());
|
||||||
filterMap.addURLPattern("/*");
|
filterMap.addURLPattern("/*");
|
||||||
context.addFilterMap(filterMap);
|
context.addFilterMap(filterMap);
|
||||||
|
}
|
||||||
|
|
||||||
var connector = tomcat.getConnector();
|
var connector = tomcat.getConnector();
|
||||||
connector.setAsyncTimeout(3000);
|
connector.setAsyncTimeout(3000);
|
||||||
|
|
|
||||||
|
|
@ -5,6 +5,8 @@
|
||||||
package io.modelcontextprotocol.client.transport;
|
package io.modelcontextprotocol.client.transport;
|
||||||
|
|
||||||
import java.io.IOException;
|
import java.io.IOException;
|
||||||
|
import java.util.Collections;
|
||||||
|
import java.util.Comparator;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Optional;
|
import java.util.Optional;
|
||||||
import java.util.concurrent.atomic.AtomicReference;
|
import java.util.concurrent.atomic.AtomicReference;
|
||||||
|
|
@ -22,9 +24,8 @@ import org.springframework.web.reactive.function.client.ClientResponse;
|
||||||
import org.springframework.web.reactive.function.client.WebClient;
|
import org.springframework.web.reactive.function.client.WebClient;
|
||||||
import org.springframework.web.reactive.function.client.WebClientResponseException;
|
import org.springframework.web.reactive.function.client.WebClientResponseException;
|
||||||
|
|
||||||
import io.modelcontextprotocol.json.TypeRef;
|
|
||||||
import io.modelcontextprotocol.json.McpJsonMapper;
|
import io.modelcontextprotocol.json.McpJsonMapper;
|
||||||
|
import io.modelcontextprotocol.json.TypeRef;
|
||||||
import io.modelcontextprotocol.spec.ClosedMcpTransportSession;
|
import io.modelcontextprotocol.spec.ClosedMcpTransportSession;
|
||||||
import io.modelcontextprotocol.spec.DefaultMcpTransportSession;
|
import io.modelcontextprotocol.spec.DefaultMcpTransportSession;
|
||||||
import io.modelcontextprotocol.spec.DefaultMcpTransportStream;
|
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 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";
|
private static final String DEFAULT_ENDPOINT = "/mcp";
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
@ -105,20 +104,29 @@ public class WebClientStreamableHttpTransport implements McpClientTransport {
|
||||||
|
|
||||||
private final AtomicReference<Consumer<Throwable>> exceptionHandler = new AtomicReference<>();
|
private final AtomicReference<Consumer<Throwable>> exceptionHandler = new AtomicReference<>();
|
||||||
|
|
||||||
|
private final List<String> supportedProtocolVersions;
|
||||||
|
|
||||||
|
private final String latestSupportedProtocolVersion;
|
||||||
|
|
||||||
private WebClientStreamableHttpTransport(McpJsonMapper jsonMapper, WebClient.Builder webClientBuilder,
|
private WebClientStreamableHttpTransport(McpJsonMapper jsonMapper, WebClient.Builder webClientBuilder,
|
||||||
String endpoint, boolean resumableStreams, boolean openConnectionOnStartup) {
|
String endpoint, boolean resumableStreams, boolean openConnectionOnStartup,
|
||||||
|
List<String> supportedProtocolVersions) {
|
||||||
this.jsonMapper = jsonMapper;
|
this.jsonMapper = jsonMapper;
|
||||||
this.webClient = webClientBuilder.build();
|
this.webClient = webClientBuilder.build();
|
||||||
this.endpoint = endpoint;
|
this.endpoint = endpoint;
|
||||||
this.resumableStreams = resumableStreams;
|
this.resumableStreams = resumableStreams;
|
||||||
this.openConnectionOnStartup = openConnectionOnStartup;
|
this.openConnectionOnStartup = openConnectionOnStartup;
|
||||||
this.activeSession.set(createTransportSession());
|
this.activeSession.set(createTransportSession());
|
||||||
|
this.supportedProtocolVersions = List.copyOf(supportedProtocolVersions);
|
||||||
|
this.latestSupportedProtocolVersion = this.supportedProtocolVersions.stream()
|
||||||
|
.sorted(Comparator.reverseOrder())
|
||||||
|
.findFirst()
|
||||||
|
.get();
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public List<String> protocolVersions() {
|
public List<String> protocolVersions() {
|
||||||
return List.of(ProtocolVersions.MCP_2024_11_05, ProtocolVersions.MCP_2025_03_26,
|
return supportedProtocolVersions;
|
||||||
ProtocolVersions.MCP_2025_06_18);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
@ -149,7 +157,7 @@ public class WebClientStreamableHttpTransport implements McpClientTransport {
|
||||||
: webClient.delete()
|
: webClient.delete()
|
||||||
.uri(this.endpoint)
|
.uri(this.endpoint)
|
||||||
.header(HttpHeaders.MCP_SESSION_ID, sessionId)
|
.header(HttpHeaders.MCP_SESSION_ID, sessionId)
|
||||||
.header(HttpHeaders.PROTOCOL_VERSION, MCP_PROTOCOL_VERSION)
|
.header(HttpHeaders.PROTOCOL_VERSION, this.latestSupportedProtocolVersion)
|
||||||
.retrieve()
|
.retrieve()
|
||||||
.toBodilessEntity()
|
.toBodilessEntity()
|
||||||
.onErrorComplete(e -> {
|
.onErrorComplete(e -> {
|
||||||
|
|
@ -217,7 +225,7 @@ public class WebClientStreamableHttpTransport implements McpClientTransport {
|
||||||
Disposable connection = webClient.get()
|
Disposable connection = webClient.get()
|
||||||
.uri(this.endpoint)
|
.uri(this.endpoint)
|
||||||
.accept(MediaType.TEXT_EVENT_STREAM)
|
.accept(MediaType.TEXT_EVENT_STREAM)
|
||||||
.header(HttpHeaders.PROTOCOL_VERSION, MCP_PROTOCOL_VERSION)
|
.header(HttpHeaders.PROTOCOL_VERSION, this.latestSupportedProtocolVersion)
|
||||||
.headers(httpHeaders -> {
|
.headers(httpHeaders -> {
|
||||||
transportSession.sessionId().ifPresent(id -> httpHeaders.add(HttpHeaders.MCP_SESSION_ID, id));
|
transportSession.sessionId().ifPresent(id -> httpHeaders.add(HttpHeaders.MCP_SESSION_ID, id));
|
||||||
if (stream != null) {
|
if (stream != null) {
|
||||||
|
|
@ -283,7 +291,7 @@ public class WebClientStreamableHttpTransport implements McpClientTransport {
|
||||||
Disposable connection = webClient.post()
|
Disposable connection = webClient.post()
|
||||||
.uri(this.endpoint)
|
.uri(this.endpoint)
|
||||||
.accept(MediaType.APPLICATION_JSON, MediaType.TEXT_EVENT_STREAM)
|
.accept(MediaType.APPLICATION_JSON, MediaType.TEXT_EVENT_STREAM)
|
||||||
.header(HttpHeaders.PROTOCOL_VERSION, MCP_PROTOCOL_VERSION)
|
.header(HttpHeaders.PROTOCOL_VERSION, this.latestSupportedProtocolVersion)
|
||||||
.headers(httpHeaders -> {
|
.headers(httpHeaders -> {
|
||||||
transportSession.sessionId().ifPresent(id -> httpHeaders.add(HttpHeaders.MCP_SESSION_ID, id));
|
transportSession.sessionId().ifPresent(id -> httpHeaders.add(HttpHeaders.MCP_SESSION_ID, id));
|
||||||
})
|
})
|
||||||
|
|
@ -495,6 +503,9 @@ public class WebClientStreamableHttpTransport implements McpClientTransport {
|
||||||
|
|
||||||
private boolean openConnectionOnStartup = false;
|
private boolean openConnectionOnStartup = false;
|
||||||
|
|
||||||
|
private List<String> supportedProtocolVersions = List.of(ProtocolVersions.MCP_2024_11_05,
|
||||||
|
ProtocolVersions.MCP_2025_03_26, ProtocolVersions.MCP_2025_06_18);
|
||||||
|
|
||||||
private Builder(WebClient.Builder webClientBuilder) {
|
private Builder(WebClient.Builder webClientBuilder) {
|
||||||
Assert.notNull(webClientBuilder, "WebClient.Builder must not be null");
|
Assert.notNull(webClientBuilder, "WebClient.Builder must not be null");
|
||||||
this.webClientBuilder = webClientBuilder;
|
this.webClientBuilder = webClientBuilder;
|
||||||
|
|
@ -560,6 +571,30 @@ public class WebClientStreamableHttpTransport implements McpClientTransport {
|
||||||
return this;
|
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.
|
||||||
|
* <p>
|
||||||
|
* 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 <a href=
|
||||||
|
* "https://modelcontextprotocol.io/specification/2024-11-05/basic/lifecycle#version-negotiation">version
|
||||||
|
* negotiation specification</a>
|
||||||
|
* @see <a href=
|
||||||
|
* "https://modelcontextprotocol.io/specification/2025-06-18/basic/transports#protocol-version-header">Protocol
|
||||||
|
* Version Header</a>
|
||||||
|
*/
|
||||||
|
public Builder supportedProtocolVersions(List<String> supportedProtocolVersions) {
|
||||||
|
Assert.notEmpty(supportedProtocolVersions, "supportedProtocolVersions must not be empty");
|
||||||
|
this.supportedProtocolVersions = Collections.unmodifiableList(supportedProtocolVersions);
|
||||||
|
return this;
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Construct a fresh instance of {@link WebClientStreamableHttpTransport} using
|
* Construct a fresh instance of {@link WebClientStreamableHttpTransport} using
|
||||||
* the current builder configuration.
|
* the current builder configuration.
|
||||||
|
|
@ -567,7 +602,7 @@ public class WebClientStreamableHttpTransport implements McpClientTransport {
|
||||||
*/
|
*/
|
||||||
public WebClientStreamableHttpTransport build() {
|
public WebClientStreamableHttpTransport build() {
|
||||||
return new WebClientStreamableHttpTransport(jsonMapper == null ? McpJsonMapper.getDefault() : jsonMapper,
|
return new WebClientStreamableHttpTransport(jsonMapper == null ? McpJsonMapper.getDefault() : jsonMapper,
|
||||||
webClientBuilder, endpoint, resumableStreams, openConnectionOnStartup);
|
webClientBuilder, endpoint, resumableStreams, openConnectionOnStartup, supportedProtocolVersions);
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -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<McpSyncServerExchange, McpSchema.CallToolRequest, McpSchema.CallToolResult> 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<ServerResponse> 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();
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
@ -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<Call> calls = new ArrayList<>();
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public Mono<ServerResponse> filter(ServerRequest request, HandlerFunction next) {
|
||||||
|
Map<String, String> 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<Call> getCalls() {
|
||||||
|
return List.copyOf(calls);
|
||||||
|
}
|
||||||
|
|
||||||
|
public record Call(Map<String, String> headers, String body) {
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
Loading…
Reference in a new issue