fix: propagate Reactor Context into client transport chain (#154)

This commit is contained in:
Francis Hodianto 2025-04-23 23:09:41 +08:00 • committed by GitHub
commit 261554bb7f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194

View file

@ -230,18 +230,19 @@ public class McpClientSession implements McpSession {
public <T> Mono<T> sendRequest(String method, Object requestParams, TypeReference<T> typeRef) {
String requestId = this.generateRequestId();
return Mono.<McpSchema.JSONRPCResponse>create(sink -> {
return Mono.deferContextual(ctx -> Mono.<McpSchema.JSONRPCResponse>create(sink -> {
this.pendingResponses.put(requestId, sink);
McpSchema.JSONRPCRequest jsonrpcRequest = new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION, method,
requestId, requestParams);
this.transport.sendMessage(jsonrpcRequest)
.contextWrite(ctx)
// TODO: It's most efficient to create a dedicated Subscriber here
.subscribe(v -> {
}, error -> {
this.pendingResponses.remove(requestId);
sink.error(error);
});
}).timeout(this.requestTimeout).handle((jsonRpcResponse, sink) -> {
})).timeout(this.requestTimeout).handle((jsonRpcResponse, sink) -> {
if (jsonRpcResponse.error() != null) {
logger.error("Error handling request: {}", jsonRpcResponse.error());
sink.error(new McpError(jsonRpcResponse.error()));