Add McpTransportContext to McpSyncClient (#522)

* McpSyncClient: introduce McpTransportContext

- McpSyncClient should be considered thread-agnostic, and therefore
  consumers cannot rely on thread locals to propagate "context", e.g.
  pass down the Servlet request reference in a server context.
- This PR introduces a mechanism for populating an McpTransportContext
  before executing client operations, and reworks the HTTP request
  customizers to leverage that McpTransportContext.
- Move McpTransportContext from server to common package for shared client/server usage
- Make McpTransportContext immutable by removing put() and copy() methods
- Add static create() factory method for creating contexts with predefined data
- Update McpTransportContextExtractor to return context instead of modifying one
- Replace DefaultMcpTransportContext mutable implementation with immutable version
- Update all transport implementations to use McpTransportContext.EMPTY as default
- Rename *HttpRequestCustomizer -> *HttpClientRequestCustomizer
- Add end-to-end McpTransportContextIntegrationTests
- This PR introduces a breaking change to the Sync/Async request
  customizers.

Signed-off-by: Daniel Garnier-Moiroux <git@garnier.wf>
This commit is contained in:
Daniel Garnier-Moiroux 2025-09-02 11:09:40 +02:00 • committed by GitHub
commit 629464b1d6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
28 changed files with 746 additions and 327 deletions

View file

@ -11,10 +11,11 @@ import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import io.modelcontextprotocol.common.McpTransportContext;
import io.modelcontextprotocol.spec.McpClientTransport;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpTransport;
import io.modelcontextprotocol.spec.McpSchema.ClientCapabilities;
import io.modelcontextprotocol.spec.McpSchema.CreateMessageRequest;
import io.modelcontextprotocol.spec.McpSchema.CreateMessageResult;
@ -22,6 +23,7 @@ import io.modelcontextprotocol.spec.McpSchema.ElicitRequest;
import io.modelcontextprotocol.spec.McpSchema.ElicitResult;
import io.modelcontextprotocol.spec.McpSchema.Implementation;
import io.modelcontextprotocol.spec.McpSchema.Root;
import io.modelcontextprotocol.spec.McpTransport;
import io.modelcontextprotocol.util.Assert;
import reactor.core.publisher.Mono;
@ -183,6 +185,8 @@ public interface McpClient {
private Function<ElicitRequest, ElicitResult> elicitationHandler;
private Supplier<McpTransportContext> contextProvider = () -> McpTransportContext.EMPTY;
private SyncSpec(McpClientTransport transport) {
Assert.notNull(transport, "Transport must not be null");
this.transport = transport;
@ -409,6 +413,22 @@ public interface McpClient {
return this;
}
/**
* Add a provider of {@link McpTransportContext}, providing a context before
* calling any client operation. This allows to extract thread-locals and hand
* them over to the underlying transport.
* <p>
* There is no direct equivalent in {@link AsyncSpec}. To achieve the same result,
* append {@code contextWrite(McpTransportContext.KEY, context)} to any
* {@link McpAsyncClient} call.
* @param contextProvider A supplier to create a context
* @return This builder for method chaining
*/
public SyncSpec transportContextProvider(Supplier<McpTransportContext> contextProvider) {
this.contextProvider = contextProvider;
return this;
}
/**
* Create an instance of {@link McpSyncClient} with the provided configurations or
* sensible defaults.
@ -423,7 +443,8 @@ public interface McpClient {
McpClientFeatures.Async asyncFeatures = McpClientFeatures.Async.fromSync(syncFeatures);
return new McpSyncClient(
new McpAsyncClient(transport, this.requestTimeout, this.initializationTimeout, asyncFeatures));
new McpAsyncClient(transport, this.requestTimeout, this.initializationTimeout, asyncFeatures),
this.contextProvider);
}
}

View file

@ -5,16 +5,19 @@
package io.modelcontextprotocol.client;
import java.time.Duration;
import java.util.function.Supplier;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.modelcontextprotocol.common.McpTransportContext;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpSchema.ClientCapabilities;
import io.modelcontextprotocol.spec.McpSchema.GetPromptRequest;
import io.modelcontextprotocol.spec.McpSchema.GetPromptResult;
import io.modelcontextprotocol.spec.McpSchema.ListPromptsResult;
import io.modelcontextprotocol.util.Assert;
import reactor.core.publisher.Mono;
/**
* A synchronous client implementation for the Model Context Protocol (MCP) that wraps an
@ -63,14 +66,20 @@ public class McpSyncClient implements AutoCloseable {
private final McpAsyncClient delegate;
private final Supplier<McpTransportContext> contextProvider;
/**
* Create a new McpSyncClient with the given delegate.
* @param delegate the asynchronous kernel on top of which this synchronous client
* provides a blocking API.
* @param contextProvider the supplier of context before calling any non-blocking
* operation on underlying delegate
*/
McpSyncClient(McpAsyncClient delegate) {
McpSyncClient(McpAsyncClient delegate, Supplier<McpTransportContext> contextProvider) {
Assert.notNull(delegate, "The delegate can not be null");
Assert.notNull(contextProvider, "The contextProvider can not be null");
this.delegate = delegate;
this.contextProvider = contextProvider;
}
/**
@ -177,14 +186,14 @@ public class McpSyncClient implements AutoCloseable {
public McpSchema.InitializeResult initialize() {
// TODO: block takes no argument here as we assume the async client is
// configured with a requestTimeout at all times
return this.delegate.initialize().block();
return withProvidedContext(this.delegate.initialize()).block();
}
/**
* Send a roots/list_changed notification.
*/
public void rootsListChangedNotification() {
this.delegate.rootsListChangedNotification().block();
withProvidedContext(this.delegate.rootsListChangedNotification()).block();
}
/**
@ -206,7 +215,7 @@ public class McpSyncClient implements AutoCloseable {
* @return
*/
public Object ping() {
return this.delegate.ping().block();
return withProvidedContext(this.delegate.ping()).block();
}
// --------------------------
@ -224,7 +233,8 @@ public class McpSyncClient implements AutoCloseable {
* Boolean indicating if the execution failed (true) or succeeded (false/absent)
*/
public McpSchema.CallToolResult callTool(McpSchema.CallToolRequest callToolRequest) {
return this.delegate.callTool(callToolRequest).block();
return withProvidedContext(this.delegate.callTool(callToolRequest)).block();
}
/**
@ -234,7 +244,7 @@ public class McpSyncClient implements AutoCloseable {
* pagination if more tools are available
*/
public McpSchema.ListToolsResult listTools() {
return this.delegate.listTools().block();
return withProvidedContext(this.delegate.listTools()).block();
}
/**
@ -245,7 +255,8 @@ public class McpSyncClient implements AutoCloseable {
* pagination if more tools are available
*/
public McpSchema.ListToolsResult listTools(String cursor) {
return this.delegate.listTools(cursor).block();
return withProvidedContext(this.delegate.listTools(cursor)).block();
}
// --------------------------
@ -257,7 +268,8 @@ public class McpSyncClient implements AutoCloseable {
* @return The list of all resources result
*/
public McpSchema.ListResourcesResult listResources() {
return this.delegate.listResources().block();
return withProvidedContext(this.delegate.listResources()).block();
}
/**
@ -266,7 +278,8 @@ public class McpSyncClient implements AutoCloseable {
* @return The list of resources result
*/
public McpSchema.ListResourcesResult listResources(String cursor) {
return this.delegate.listResources(cursor).block();
return withProvidedContext(this.delegate.listResources(cursor)).block();
}
/**
@ -275,7 +288,8 @@ public class McpSyncClient implements AutoCloseable {
* @return the resource content.
*/
public McpSchema.ReadResourceResult readResource(McpSchema.Resource resource) {
return this.delegate.readResource(resource).block();
return withProvidedContext(this.delegate.readResource(resource)).block();
}
/**
@ -284,7 +298,8 @@ public class McpSyncClient implements AutoCloseable {
* @return the resource content.
*/
public McpSchema.ReadResourceResult readResource(McpSchema.ReadResourceRequest readResourceRequest) {
return this.delegate.readResource(readResourceRequest).block();
return withProvidedContext(this.delegate.readResource(readResourceRequest)).block();
}
/**
@ -292,7 +307,8 @@ public class McpSyncClient implements AutoCloseable {
* @return The list of all resource templates result.
*/
public McpSchema.ListResourceTemplatesResult listResourceTemplates() {
return this.delegate.listResourceTemplates().block();
return withProvidedContext(this.delegate.listResourceTemplates()).block();
}
/**
@ -304,7 +320,8 @@ public class McpSyncClient implements AutoCloseable {
* @return The list of resource templates result.
*/
public McpSchema.ListResourceTemplatesResult listResourceTemplates(String cursor) {
return this.delegate.listResourceTemplates(cursor).block();
return withProvidedContext(this.delegate.listResourceTemplates(cursor)).block();
}
/**
@ -317,7 +334,8 @@ public class McpSyncClient implements AutoCloseable {
* subscribe to.
*/
public void subscribeResource(McpSchema.SubscribeRequest subscribeRequest) {
this.delegate.subscribeResource(subscribeRequest).block();
withProvidedContext(this.delegate.subscribeResource(subscribeRequest)).block();
}
/**
@ -326,7 +344,8 @@ public class McpSyncClient implements AutoCloseable {
* to unsubscribe from.
*/
public void unsubscribeResource(McpSchema.UnsubscribeRequest unsubscribeRequest) {
this.delegate.unsubscribeResource(unsubscribeRequest).block();
withProvidedContext(this.delegate.unsubscribeResource(unsubscribeRequest)).block();
}
// --------------------------
@ -338,7 +357,7 @@ public class McpSyncClient implements AutoCloseable {
* @return The list of all prompts result.
*/
public ListPromptsResult listPrompts() {
return this.delegate.listPrompts().block();
return withProvidedContext(this.delegate.listPrompts()).block();
}
/**
@ -347,11 +366,12 @@ public class McpSyncClient implements AutoCloseable {
* @return The list of prompts result.
*/
public ListPromptsResult listPrompts(String cursor) {
return this.delegate.listPrompts(cursor).block();
return withProvidedContext(this.delegate.listPrompts(cursor)).block();
}
public GetPromptResult getPrompt(GetPromptRequest getPromptRequest) {
return this.delegate.getPrompt(getPromptRequest).block();
return withProvidedContext(this.delegate.getPrompt(getPromptRequest)).block();
}
/**
@ -359,7 +379,8 @@ public class McpSyncClient implements AutoCloseable {
* @param loggingLevel the min logging level
*/
public void setLoggingLevel(McpSchema.LoggingLevel loggingLevel) {
this.delegate.setLoggingLevel(loggingLevel).block();
withProvidedContext(this.delegate.setLoggingLevel(loggingLevel)).block();
}
/**
@ -369,7 +390,18 @@ public class McpSyncClient implements AutoCloseable {
* @return the completion result containing suggested values.
*/
public McpSchema.CompleteResult completeCompletion(McpSchema.CompleteRequest completeRequest) {
return this.delegate.completeCompletion(completeRequest).block();
return withProvidedContext(this.delegate.completeCompletion(completeRequest)).block();
}
/**
* For a given action, on assembly, capture the "context" via the
* {@link #contextProvider} and store it in the Reactor context.
* @param action the action to perform
* @return the result of the action
*/
private <T> Mono<T> withProvidedContext(Mono<T> action) {
return action.contextWrite(ctx -> ctx.put(McpTransportContext.KEY, this.contextProvider.get()));
}
}

View file

@ -22,14 +22,15 @@ import org.slf4j.LoggerFactory;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.modelcontextprotocol.client.transport.customizer.McpAsyncHttpRequestCustomizer;
import io.modelcontextprotocol.client.transport.customizer.McpSyncHttpRequestCustomizer;
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.spec.McpClientTransport;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.ProtocolVersions;
import io.modelcontextprotocol.spec.McpSchema.JSONRPCMessage;
import io.modelcontextprotocol.spec.McpTransportException;
import io.modelcontextprotocol.spec.ProtocolVersions;
import io.modelcontextprotocol.util.Assert;
import io.modelcontextprotocol.util.Utils;
import reactor.core.Disposable;
@ -114,7 +115,7 @@ public class HttpClientSseClientTransport implements McpClientTransport {
/**
* Customizer to modify requests before they are executed.
*/
private final McpAsyncHttpRequestCustomizer httpRequestCustomizer;
private final McpAsyncHttpClientRequestCustomizer httpRequestCustomizer;
/**
* Creates a new transport instance with default HTTP client and object mapper.
@ -188,7 +189,7 @@ public class HttpClientSseClientTransport implements McpClientTransport {
@Deprecated(forRemoval = true)
HttpClientSseClientTransport(HttpClient httpClient, HttpRequest.Builder requestBuilder, String baseUri,
String sseEndpoint, ObjectMapper objectMapper) {
this(httpClient, requestBuilder, baseUri, sseEndpoint, objectMapper, McpAsyncHttpRequestCustomizer.NOOP);
this(httpClient, requestBuilder, baseUri, sseEndpoint, objectMapper, McpAsyncHttpClientRequestCustomizer.NOOP);
}
/**
@ -204,7 +205,7 @@ public class HttpClientSseClientTransport implements McpClientTransport {
* @throws IllegalArgumentException if objectMapper, clientBuilder, or headers is null
*/
HttpClientSseClientTransport(HttpClient httpClient, HttpRequest.Builder requestBuilder, String baseUri,
String sseEndpoint, ObjectMapper objectMapper, McpAsyncHttpRequestCustomizer httpRequestCustomizer) {
String sseEndpoint, ObjectMapper objectMapper, McpAsyncHttpClientRequestCustomizer httpRequestCustomizer) {
Assert.notNull(objectMapper, "ObjectMapper must not be null");
Assert.hasText(baseUri, "baseUri must not be empty");
Assert.hasText(sseEndpoint, "sseEndpoint must not be empty");
@ -248,7 +249,7 @@ public class HttpClientSseClientTransport implements McpClientTransport {
private HttpRequest.Builder requestBuilder = HttpRequest.newBuilder();
private McpAsyncHttpRequestCustomizer httpRequestCustomizer = McpAsyncHttpRequestCustomizer.NOOP;
private McpAsyncHttpClientRequestCustomizer httpRequestCustomizer = McpAsyncHttpClientRequestCustomizer.NOOP;
private Duration connectTimeout = Duration.ofSeconds(10);
@ -354,16 +355,17 @@ public class HttpClientSseClientTransport implements McpClientTransport {
* executing them.
* <p>
* This overrides the customizer from
* {@link #asyncHttpRequestCustomizer(McpAsyncHttpRequestCustomizer)}.
* {@link #asyncHttpRequestCustomizer(McpAsyncHttpClientRequestCustomizer)}.
* <p>
* Do NOT use a blocking {@link McpSyncHttpRequestCustomizer} in a non-blocking
* context. Use {@link #asyncHttpRequestCustomizer(McpAsyncHttpRequestCustomizer)}
* Do NOT use a blocking {@link McpSyncHttpClientRequestCustomizer} in a
* non-blocking context. Use
* {@link #asyncHttpRequestCustomizer(McpAsyncHttpClientRequestCustomizer)}
* instead.
* @param syncHttpRequestCustomizer the request customizer
* @return this builder
*/
public Builder httpRequestCustomizer(McpSyncHttpRequestCustomizer syncHttpRequestCustomizer) {
this.httpRequestCustomizer = McpAsyncHttpRequestCustomizer.fromSync(syncHttpRequestCustomizer);
public Builder httpRequestCustomizer(McpSyncHttpClientRequestCustomizer syncHttpRequestCustomizer) {
this.httpRequestCustomizer = McpAsyncHttpClientRequestCustomizer.fromSync(syncHttpRequestCustomizer);
return this;
}
@ -372,13 +374,13 @@ public class HttpClientSseClientTransport implements McpClientTransport {
* executing them.
* <p>
* This overrides the customizer from
* {@link #httpRequestCustomizer(McpSyncHttpRequestCustomizer)}.
* {@link #httpRequestCustomizer(McpSyncHttpClientRequestCustomizer)}.
* <p>
* Do NOT use a blocking implementation in a non-blocking context.
* @param asyncHttpRequestCustomizer the request customizer
* @return this builder
*/
public Builder asyncHttpRequestCustomizer(McpAsyncHttpRequestCustomizer asyncHttpRequestCustomizer) {
public Builder asyncHttpRequestCustomizer(McpAsyncHttpClientRequestCustomizer asyncHttpRequestCustomizer) {
this.httpRequestCustomizer = asyncHttpRequestCustomizer;
return this;
}
@ -410,14 +412,15 @@ public class HttpClientSseClientTransport implements McpClientTransport {
public Mono<Void> connect(Function<Mono<JSONRPCMessage>, Mono<JSONRPCMessage>> handler) {
var uri = Utils.resolveUri(this.baseUri, this.sseEndpoint);
return Mono.defer(() -> {
return Mono.deferContextual(ctx -> {
var builder = requestBuilder.copy()
.uri(uri)
.header("Accept", "text/event-stream")
.header("Cache-Control", "no-cache")
.header(MCP_PROTOCOL_VERSION_HEADER_NAME, MCP_PROTOCOL_VERSION)
.GET();
return Mono.from(this.httpRequestCustomizer.customize(builder, "GET", uri, null));
var transportContext = ctx.getOrDefault(McpTransportContext.KEY, McpTransportContext.EMPTY);
return Mono.from(this.httpRequestCustomizer.customize(builder, "GET", uri, null, transportContext));
}).flatMap(requestBuilder -> Mono.create(sink -> {
Disposable connection = Flux.<ResponseEvent>create(sseSink -> this.httpClient
.sendAsync(requestBuilder.build(),
@ -538,13 +541,14 @@ public class HttpClientSseClientTransport implements McpClientTransport {
private Mono<HttpResponse<String>> sendHttpPost(final String endpoint, final String body) {
final URI requestUri = Utils.resolveUri(baseUri, endpoint);
return Mono.defer(() -> {
return Mono.deferContextual(ctx -> {
var builder = this.requestBuilder.copy()
.uri(requestUri)
.header("Content-Type", "application/json")
.header(MCP_PROTOCOL_VERSION_HEADER_NAME, MCP_PROTOCOL_VERSION)
.POST(HttpRequest.BodyPublishers.ofString(body));
return Mono.from(this.httpRequestCustomizer.customize(builder, "POST", requestUri, body));
var transportContext = ctx.getOrDefault(McpTransportContext.KEY, McpTransportContext.EMPTY);
return Mono.from(this.httpRequestCustomizer.customize(builder, "POST", requestUri, body, transportContext));
}).flatMap(customizedBuilder -> {
var request = customizedBuilder.build();
return Mono.fromFuture(httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofString()));

View file

@ -25,9 +25,10 @@ import org.slf4j.LoggerFactory;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.modelcontextprotocol.client.transport.customizer.McpAsyncHttpRequestCustomizer;
import io.modelcontextprotocol.client.transport.customizer.McpSyncHttpRequestCustomizer;
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.spec.DefaultMcpTransportSession;
import io.modelcontextprotocol.spec.DefaultMcpTransportStream;
import io.modelcontextprotocol.spec.HttpHeaders;
@ -115,7 +116,7 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
private final boolean resumableStreams;
private final McpAsyncHttpRequestCustomizer httpRequestCustomizer;
private final McpAsyncHttpClientRequestCustomizer httpRequestCustomizer;
private final AtomicReference<DefaultMcpTransportSession> activeSession = new AtomicReference<>();
@ -125,7 +126,7 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
private HttpClientStreamableHttpTransport(ObjectMapper objectMapper, HttpClient httpClient,
HttpRequest.Builder requestBuilder, String baseUri, String endpoint, boolean resumableStreams,
boolean openConnectionOnStartup, McpAsyncHttpRequestCustomizer httpRequestCustomizer) {
boolean openConnectionOnStartup, McpAsyncHttpClientRequestCustomizer httpRequestCustomizer) {
this.objectMapper = objectMapper;
this.httpClient = httpClient;
this.requestBuilder = requestBuilder;
@ -170,14 +171,15 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
private Publisher<Void> createDelete(String sessionId) {
var uri = Utils.resolveUri(this.baseUri, this.endpoint);
return Mono.defer(() -> {
return Mono.deferContextual(ctx -> {
var builder = this.requestBuilder.copy()
.uri(uri)
.header("Cache-Control", "no-cache")
.header(HttpHeaders.MCP_SESSION_ID, sessionId)
.header(HttpHeaders.PROTOCOL_VERSION, MCP_PROTOCOL_VERSION)
.DELETE();
return Mono.from(this.httpRequestCustomizer.customize(builder, "DELETE", uri, null));
var transportContext = ctx.getOrDefault(McpTransportContext.KEY, McpTransportContext.EMPTY);
return Mono.from(this.httpRequestCustomizer.customize(builder, "DELETE", uri, null, transportContext));
}).flatMap(requestBuilder -> {
var request = requestBuilder.build();
return Mono.fromFuture(() -> this.httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofString()));
@ -230,7 +232,7 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
final McpTransportSession<Disposable> transportSession = this.activeSession.get();
var uri = Utils.resolveUri(this.baseUri, this.endpoint);
Disposable connection = Mono.defer(() -> {
Disposable connection = Mono.deferContextual(connectionCtx -> {
HttpRequest.Builder requestBuilder = this.requestBuilder.copy();
if (transportSession != null && transportSession.sessionId().isPresent()) {
@ -247,7 +249,8 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
.header("Cache-Control", "no-cache")
.header(HttpHeaders.PROTOCOL_VERSION, MCP_PROTOCOL_VERSION)
.GET();
return Mono.from(this.httpRequestCustomizer.customize(builder, "GET", uri, null));
var transportContext = connectionCtx.getOrDefault(McpTransportContext.KEY, McpTransportContext.EMPTY);
return Mono.from(this.httpRequestCustomizer.customize(builder, "GET", uri, null, transportContext));
})
.flatMapMany(
requestBuilder -> Flux.<ResponseEvent>create(
@ -407,7 +410,7 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
var uri = Utils.resolveUri(this.baseUri, this.endpoint);
String jsonBody = this.toString(sentMessage);
Disposable connection = Mono.defer(() -> {
Disposable connection = Mono.deferContextual(ctx -> {
HttpRequest.Builder requestBuilder = this.requestBuilder.copy();
if (transportSession != null && transportSession.sessionId().isPresent()) {
@ -421,7 +424,9 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
.header("Cache-Control", "no-cache")
.header(HttpHeaders.PROTOCOL_VERSION, MCP_PROTOCOL_VERSION)
.POST(HttpRequest.BodyPublishers.ofString(jsonBody));
return Mono.from(this.httpRequestCustomizer.customize(builder, "POST", uri, jsonBody));
var transportContext = ctx.getOrDefault(McpTransportContext.KEY, McpTransportContext.EMPTY);
return Mono
.from(this.httpRequestCustomizer.customize(builder, "POST", uri, jsonBody, transportContext));
}).flatMapMany(requestBuilder -> Flux.<ResponseEvent>create(responseEventSink -> {
// Create the async request with proper body subscriber selection
@ -600,7 +605,7 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
private HttpRequest.Builder requestBuilder = HttpRequest.newBuilder();
private McpAsyncHttpRequestCustomizer httpRequestCustomizer = McpAsyncHttpRequestCustomizer.NOOP;
private McpAsyncHttpClientRequestCustomizer httpRequestCustomizer = McpAsyncHttpClientRequestCustomizer.NOOP;
private Duration connectTimeout = Duration.ofSeconds(10);
@ -711,16 +716,17 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
* executing them.
* <p>
* This overrides the customizer from
* {@link #asyncHttpRequestCustomizer(McpAsyncHttpRequestCustomizer)}.
* {@link #asyncHttpRequestCustomizer(McpAsyncHttpClientRequestCustomizer)}.
* <p>
* Do NOT use a blocking {@link McpSyncHttpRequestCustomizer} in a non-blocking
* context. Use {@link #asyncHttpRequestCustomizer(McpAsyncHttpRequestCustomizer)}
* Do NOT use a blocking {@link McpSyncHttpClientRequestCustomizer} in a
* non-blocking context. Use
* {@link #asyncHttpRequestCustomizer(McpAsyncHttpClientRequestCustomizer)}
* instead.
* @param syncHttpRequestCustomizer the request customizer
* @return this builder
*/
public Builder httpRequestCustomizer(McpSyncHttpRequestCustomizer syncHttpRequestCustomizer) {
this.httpRequestCustomizer = McpAsyncHttpRequestCustomizer.fromSync(syncHttpRequestCustomizer);
public Builder httpRequestCustomizer(McpSyncHttpClientRequestCustomizer syncHttpRequestCustomizer) {
this.httpRequestCustomizer = McpAsyncHttpClientRequestCustomizer.fromSync(syncHttpRequestCustomizer);
return this;
}
@ -729,13 +735,13 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
* executing them.
* <p>
* This overrides the customizer from
* {@link #httpRequestCustomizer(McpSyncHttpRequestCustomizer)}.
* {@link #httpRequestCustomizer(McpSyncHttpClientRequestCustomizer)}.
* <p>
* Do NOT use a blocking implementation in a non-blocking context.
* @param asyncHttpRequestCustomizer the request customizer
* @return this builder
*/
public Builder asyncHttpRequestCustomizer(McpAsyncHttpRequestCustomizer asyncHttpRequestCustomizer) {
public Builder asyncHttpRequestCustomizer(McpAsyncHttpClientRequestCustomizer asyncHttpRequestCustomizer) {
this.httpRequestCustomizer = asyncHttpRequestCustomizer;
return this;
}

View file

@ -0,0 +1,42 @@
/*
* Copyright 2024-2025 the original author or authors.
*/
package io.modelcontextprotocol.client.transport.customizer;
import java.net.URI;
import java.net.http.HttpRequest;
import java.util.List;
import org.reactivestreams.Publisher;
import io.modelcontextprotocol.common.McpTransportContext;
import io.modelcontextprotocol.util.Assert;
import reactor.core.publisher.Mono;
/**
* Composable {@link McpAsyncHttpClientRequestCustomizer} that applies multiple
* customizers, in order.
*
* @author Daniel Garnier-Moiroux
*/
public class DelegatingMcpAsyncHttpClientRequestCustomizer implements McpAsyncHttpClientRequestCustomizer {
private final List<McpAsyncHttpClientRequestCustomizer> customizers;
public DelegatingMcpAsyncHttpClientRequestCustomizer(List<McpAsyncHttpClientRequestCustomizer> customizers) {
Assert.notNull(customizers, "Customizers must not be null");
this.customizers = customizers;
}
@Override
public Publisher<HttpRequest.Builder> customize(HttpRequest.Builder builder, String method, URI endpoint,
String body, McpTransportContext context) {
var result = Mono.just(builder);
for (var customizer : this.customizers) {
result = result.flatMap(b -> Mono.from(customizer.customize(b, method, endpoint, body, context)));
}
return result;
}
}

View file

@ -0,0 +1,35 @@
/*
* Copyright 2024-2025 the original author or authors.
*/
package io.modelcontextprotocol.client.transport.customizer;
import java.net.URI;
import java.net.http.HttpRequest;
import java.util.List;
import io.modelcontextprotocol.common.McpTransportContext;
import io.modelcontextprotocol.util.Assert;
/**
* Composable {@link McpSyncHttpClientRequestCustomizer} that applies multiple
* customizers, in order.
*
* @author Daniel Garnier-Moiroux
*/
public class DelegatingMcpSyncHttpClientRequestCustomizer implements McpSyncHttpClientRequestCustomizer {
private final List<McpSyncHttpClientRequestCustomizer> delegates;
public DelegatingMcpSyncHttpClientRequestCustomizer(List<McpSyncHttpClientRequestCustomizer> customizers) {
Assert.notNull(customizers, "Customizers must not be null");
this.delegates = customizers;
}
@Override
public void customize(HttpRequest.Builder builder, String method, URI endpoint, String body,
McpTransportContext context) {
this.delegates.forEach(delegate -> delegate.customize(builder, method, endpoint, body, context));
}
}

View file

@ -1,32 +0,0 @@
/*
* Copyright 2024-2025 the original author or authors.
*/
package io.modelcontextprotocol.client.transport.customizer;
import io.modelcontextprotocol.util.Assert;
import java.net.URI;
import java.net.http.HttpRequest;
import java.util.List;
/**
* Composable {@link McpSyncHttpRequestCustomizer} that applies multiple customizers, in
* order.
*
* @author Daniel Garnier-Moiroux
*/
public class DelegatingMcpSyncHttpRequestCustomizer implements McpSyncHttpRequestCustomizer {
private final List<McpSyncHttpRequestCustomizer> delegates;
public DelegatingMcpSyncHttpRequestCustomizer(List<McpSyncHttpRequestCustomizer> customizers) {
Assert.notNull(customizers, "Customizers must not be null");
this.delegates = customizers;
}
@Override
public void customize(HttpRequest.Builder builder, String method, URI endpoint, String body) {
this.delegates.forEach(delegate -> delegate.customize(builder, method, endpoint, body));
}
}

View file

@ -0,0 +1,55 @@
/*
* Copyright 2024-2025 the original author or authors.
*/
package io.modelcontextprotocol.client.transport.customizer;
import java.net.URI;
import java.net.http.HttpRequest;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import reactor.util.annotation.Nullable;
import io.modelcontextprotocol.common.McpTransportContext;
/**
* Customize {@link HttpRequest.Builder} before executing the request, in either SSE or
* Streamable HTTP transport.
* <p>
* When used in a non-blocking context, implementations MUST be non-blocking.
*
* @author Daniel Garnier-Moiroux
*/
public interface McpAsyncHttpClientRequestCustomizer {
Publisher<HttpRequest.Builder> customize(HttpRequest.Builder builder, String method, URI endpoint,
@Nullable String body, McpTransportContext context);
McpAsyncHttpClientRequestCustomizer NOOP = new Noop();
/**
* Wrap a sync implementation in an async wrapper.
* <p>
* Do NOT wrap a blocking implementation for use in a non-blocking context. For a
* blocking implementation, consider using {@link Schedulers#boundedElastic()}.
*/
static McpAsyncHttpClientRequestCustomizer fromSync(McpSyncHttpClientRequestCustomizer customizer) {
return (builder, method, uri, body, context) -> Mono.fromSupplier(() -> {
customizer.customize(builder, method, uri, body, context);
return builder;
});
}
class Noop implements McpAsyncHttpClientRequestCustomizer {
@Override
public Publisher<HttpRequest.Builder> customize(HttpRequest.Builder builder, String method, URI endpoint,
String body, McpTransportContext context) {
return Mono.just(builder);
}
}
}

View file

@ -0,0 +1,28 @@
/*
* Copyright 2024-2025 the original author or authors.
*/
package io.modelcontextprotocol.client.transport.customizer;
import java.net.URI;
import java.net.http.HttpRequest;
import reactor.util.annotation.Nullable;
import io.modelcontextprotocol.client.McpClient.SyncSpec;
import io.modelcontextprotocol.common.McpTransportContext;
/**
* Customize {@link HttpRequest.Builder} before executing the request, either in SSE or
* Streamable HTTP transport. Do not rely on thread-locals in this implementation, instead
* use {@link SyncSpec#transportContextProvider} to extract context, and then consume it
* through {@link McpTransportContext}.
*
* @author Daniel Garnier-Moiroux
*/
public interface McpSyncHttpClientRequestCustomizer {
void customize(HttpRequest.Builder builder, String method, URI endpoint, @Nullable String body,
McpTransportContext context);
}

View file

@ -1,21 +0,0 @@
/*
* Copyright 2024-2025 the original author or authors.
*/
package io.modelcontextprotocol.client.transport.customizer;
import java.net.URI;
import java.net.http.HttpRequest;
import reactor.util.annotation.Nullable;
/**
* Customize {@link HttpRequest.Builder} before executing the request, either in SSE or
* Streamable HTTP transport.
*
* @author Daniel Garnier-Moiroux
*/
public interface McpSyncHttpRequestCustomizer {
void customize(HttpRequest.Builder builder, String method, URI endpoint, @Nullable String body);
}

View file

@ -0,0 +1,45 @@
/*
* Copyright 2024-2025 the original author or authors.
*/
package io.modelcontextprotocol.common;
import java.util.Map;
import io.modelcontextprotocol.util.Assert;
/**
* Default implementation for {@link McpTransportContext} which uses a map as storage.
*
* @author Dariusz Jędrzejczyk
* @author Daniel Garnier-Moiroux
*/
class DefaultMcpTransportContext implements McpTransportContext {
private final Map<String, Object> metadata;
DefaultMcpTransportContext(Map<String, Object> metadata) {
Assert.notNull(metadata, "The metadata cannot be null");
this.metadata = metadata;
}
@Override
public Object get(String key) {
return this.metadata.get(key);
}
@Override
public boolean equals(Object o) {
if (o == null || getClass() != o.getClass())
return false;
DefaultMcpTransportContext that = (DefaultMcpTransportContext) o;
return this.metadata.equals(that.metadata);
}
@Override
public int hashCode() {
return this.metadata.hashCode();
}
}

View file

@ -0,0 +1,46 @@
/*
* Copyright 2024-2025 the original author or authors.
*/
package io.modelcontextprotocol.common;
import java.util.Collections;
import java.util.Map;
/**
* Context associated with the transport layer. It allows to add transport-level metadata
* for use further down the line. Specifically, it can be beneficial to extract HTTP
* request metadata for use in MCP feature implementations.
*
* @author Dariusz Jędrzejczyk
*/
public interface McpTransportContext {
/**
* Key for use in Reactor Context to transport the context to user land.
*/
String KEY = "MCP_TRANSPORT_CONTEXT";
/**
* An empty, unmodifiable context.
*/
@SuppressWarnings("unchecked")
McpTransportContext EMPTY = new DefaultMcpTransportContext(Collections.EMPTY_MAP);
/**
* Create an unmodifiable context containing the given metadata.
* @param metadata the transport metadata
* @return the context containing the metadata
*/
static McpTransportContext create(Map<String, Object> metadata) {
return new DefaultMcpTransportContext(metadata);
}
/**
* Extract a value from the context.
* @param key the key under the data is expected
* @return the associated value or {@code null} if missing.
*/
Object get(String key);
}

View file

@ -4,6 +4,7 @@
package io.modelcontextprotocol.server;
import io.modelcontextprotocol.common.McpTransportContext;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema;
import org.slf4j.Logger;

View file

@ -1,49 +0,0 @@
/*
* Copyright 2024-2025 the original author or authors.
*/
package io.modelcontextprotocol.server;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* Default implementation for {@link McpTransportContext} which uses a Thread-safe map.
* Objects of this kind are mutable.
*
* @author Dariusz Jędrzejczyk
*/
public class DefaultMcpTransportContext implements McpTransportContext {
private final Map<String, Object> storage;
/**
* Create an empty instance.
*/
public DefaultMcpTransportContext() {
this.storage = new ConcurrentHashMap<>();
}
DefaultMcpTransportContext(Map<String, Object> storage) {
this.storage = storage;
}
@Override
public Object get(String key) {
return this.storage.get(key);
}
@Override
public void put(String key, Object value) {
this.storage.put(key, value);
}
/**
* Allows copying the contents.
* @return new instance with the copy of the underlying map
*/
public McpTransportContext copy() {
return new DefaultMcpTransportContext(new ConcurrentHashMap<>(this.storage));
}
}

View file

@ -4,6 +4,7 @@
package io.modelcontextprotocol.server;
import io.modelcontextprotocol.common.McpTransportContext;
import java.util.ArrayList;
import java.util.Collections;

View file

@ -4,6 +4,7 @@
package io.modelcontextprotocol.server;
import io.modelcontextprotocol.common.McpTransportContext;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;

View file

@ -6,6 +6,7 @@ package io.modelcontextprotocol.server;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.modelcontextprotocol.common.McpTransportContext;
import io.modelcontextprotocol.spec.JsonSchemaValidator;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema;

View file

@ -4,6 +4,7 @@
package io.modelcontextprotocol.server;
import io.modelcontextprotocol.common.McpTransportContext;
import reactor.core.publisher.Mono;
/**

View file

@ -4,6 +4,7 @@
package io.modelcontextprotocol.server;
import io.modelcontextprotocol.common.McpTransportContext;
import reactor.core.publisher.Mono;
/**

View file

@ -4,6 +4,7 @@
package io.modelcontextprotocol.server;
import io.modelcontextprotocol.common.McpTransportContext;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpSchema.CallToolRequest;
import io.modelcontextprotocol.util.Assert;

Some files were not shown because too many files have changed in this diff Show more