diff --git a/mcp-core/src/main/java/io/modelcontextprotocol/spec/McpClientSession.java b/mcp-core/src/main/java/io/modelcontextprotocol/spec/McpClientSession.java index 7bbbdb6b2..76a51628a 100644 --- a/mcp-core/src/main/java/io/modelcontextprotocol/spec/McpClientSession.java +++ b/mcp-core/src/main/java/io/modelcontextprotocol/spec/McpClientSession.java @@ -261,9 +261,9 @@ private String generateRequestId() { */ @Override public Mono sendRequest(String method, Object requestParams, TypeRef typeRef) { - String requestId = this.generateRequestId(); - return Mono.deferContextual(ctx -> Mono.create(pendingResponseSink -> { + // A new id per subscription, so that a retried request does not reuse one + String requestId = this.generateRequestId(); logger.debug("Sending message for method {}", method); this.pendingResponses.put(requestId, pendingResponseSink); McpSchema.JSONRPCRequest jsonrpcRequest = new McpSchema.JSONRPCRequest(method, requestId, requestParams); diff --git a/mcp-core/src/test/java/io/modelcontextprotocol/spec/McpClientSessionTests.java b/mcp-core/src/test/java/io/modelcontextprotocol/spec/McpClientSessionTests.java index 1a6a35b62..f1d1f5428 100644 --- a/mcp-core/src/test/java/io/modelcontextprotocol/spec/McpClientSessionTests.java +++ b/mcp-core/src/test/java/io/modelcontextprotocol/spec/McpClientSessionTests.java @@ -5,6 +5,8 @@ package io.modelcontextprotocol.spec; import java.time.Duration; +import java.util.ArrayList; +import java.util.List; import java.util.Map; import java.util.function.Function; @@ -69,6 +71,33 @@ void testSendRequest() { session.close(); } + @Test + void testResubscribedRequestUsesNewRequestId() { + List sentRequestIds = new ArrayList<>(); + var transport = new MockMcpClientTransport((t, message) -> { + if (message instanceof McpSchema.JSONRPCRequest request) { + sentRequestIds.add(request.id()); + // Fail the first attempt, answer the retry + McpSchema.JSONRPCResponse response = sentRequestIds.size() == 1 + ? McpSchema.JSONRPCResponse.error(request.id(), + new McpSchema.JSONRPCResponse.JSONRPCError(McpSchema.ErrorCodes.INTERNAL_ERROR, + "Temporary failure")) + : McpSchema.JSONRPCResponse.result(request.id(), "test response"); + t.simulateIncomingMessage(response); + } + }); + var session = new McpClientSession(TIMEOUT, transport, Map.of(), Map.of(), Function.identity()); + + StepVerifier.create(session.sendRequest(TEST_METHOD, "test", responseType).retry(1)) + .expectNext("test response") + .verifyComplete(); + + // JSON-RPC request ids must not be reused within a session + assertThat(sentRequestIds).hasSize(2).doesNotHaveDuplicates(); + + session.close(); + } + @Test void testSendRequestWithError() { var transport = new MockMcpClientTransport();