modelcontextprotocol · Lubaoshuai · Sep 13, 2026 · Sep 26, 2026
diff --git a/mcp-core/src/main/java/io/modelcontextprotocol/spec/McpClientSession.java b/mcp-core/src/main/java/io/modelcontextprotocol/spec/McpClientSession.java
@@ -264,21 +264,25 @@ public <T> Mono<T> sendRequest(String method, Object requestParams, TypeRef<T> t
 				this.pendingResponses.remove(requestId);
 				pendingResponseSink.error(error);
 			});
-		})).timeout(this.requestTimeout).handle((jsonRpcResponse, deliveredResponseSink) -> {
-			if (jsonRpcResponse.error() != null) {
-				logger.info("Server returned a JSON-RPC error when calling method {}: {}", method,
-						jsonRpcResponse.error());
-				deliveredResponseSink.error(new McpError(jsonRpcResponse.error()));
-			}
-			else {
-				if (typeRef.getType().equals(Void.class)) {
-					deliveredResponseSink.complete();
+		}))
+			.timeout(this.requestTimeout)
+			.doOnError(e -> this.pendingResponses.remove(requestId))
+			.doOnCancel(() -> this.pendingResponses.remove(requestId))
+			.handle((jsonRpcResponse, deliveredResponseSink) -> {
+				if (jsonRpcResponse.error() != null) {
+					logger.info("Server returned a JSON-RPC error when calling method {}: {}", method,
+							jsonRpcResponse.error());
+					deliveredResponseSink.error(new McpError(jsonRpcResponse.error()));
 				}
 				else {
-					deliveredResponseSink.next(this.transport.unmarshalFrom(jsonRpcResponse.result(), typeRef));
+					if (typeRef.getType().equals(Void.class)) {
+						deliveredResponseSink.complete();
+					}
+					else {
+						deliveredResponseSink.next(this.transport.unmarshalFrom(jsonRpcResponse.result(), typeRef));
+					}
 				}
-			}
-		});
+			});
 	}
 
 	/**

diff --git a/mcp-core/src/main/java/io/modelcontextprotocol/spec/McpServerSession.java b/mcp-core/src/main/java/io/modelcontextprotocol/spec/McpServerSession.java
@@ -184,19 +184,23 @@ public <T> Mono<T> sendRequest(String method, Object requestParams, TypeRef<T> t
 				this.pendingResponses.remove(requestId);
 				sink.error(error);
 			});
-		}).timeout(requestTimeout).handle((jsonRpcResponse, sink) -> {
-			if (jsonRpcResponse.error() != null) {
-				sink.error(new McpError(jsonRpcResponse.error()));
-			}
-			else {
-				if (typeRef.getType().equals(Void.class)) {
-					sink.complete();
+		})
+			.timeout(requestTimeout)
+			.doOnError(e -> this.pendingResponses.remove(requestId))
+			.doOnCancel(() -> this.pendingResponses.remove(requestId))
+			.handle((jsonRpcResponse, sink) -> {
+				if (jsonRpcResponse.error() != null) {
+					sink.error(new McpError(jsonRpcResponse.error()));
 				}
 				else {
-					sink.next(this.transport.unmarshalFrom(jsonRpcResponse.result(), typeRef));
+					if (typeRef.getType().equals(Void.class)) {
+						sink.complete();
+					}
+					else {
+						sink.next(this.transport.unmarshalFrom(jsonRpcResponse.result(), typeRef));
+					}
 				}
-			}
-		});
+			});
 	}
 
 	@Override

diff --git a/mcp-core/src/test/java/io/modelcontextprotocol/spec/McpClientSessionTests.java b/mcp-core/src/test/java/io/modelcontextprotocol/spec/McpClientSessionTests.java
@@ -303,4 +303,43 @@ void testGracefulShutdown() {
 		StepVerifier.create(session.closeGracefully()).verifyComplete();
 	}
 
+	@Test
+	void testRequestTimeoutRemovesPendingResponse() throws Exception {
+		var transport = new MockMcpClientTransport();
+		var session = new McpClientSession(Duration.ofMillis(50), transport, Map.of(), Map.of(), Function.identity());
+
+		Mono<String> responseMono = session.sendRequest(TEST_METHOD, "test", responseType);
+
+		StepVerifier.create(responseMono).expectError(java.util.concurrent.TimeoutException.class).verify();
+
+		var field = McpClientSession.class.getDeclaredField("pendingResponses");
+		field.setAccessible(true);
+		@SuppressWarnings("unchecked")
+		var pendingResponses = (java.util.Map<Object, ?>) field.get(session);
+		assertThat(pendingResponses).isEmpty();
+
+		session.close();
+	}
+
+	@Test
+	void testRequestCancellationRemovesPendingResponse() throws Exception {
+		var transport = new MockMcpClientTransport();
+		var session = new McpClientSession(TIMEOUT, transport, Map.of(), Map.of(), Function.identity());
+
+		Mono<String> responseMono = session.sendRequest(TEST_METHOD, "test", responseType);
+
+		var subscription = responseMono.subscribe();
+		var field = McpClientSession.class.getDeclaredField("pendingResponses");
+		field.setAccessible(true);
+		@SuppressWarnings("unchecked")
+		var pendingResponses = (java.util.Map<Object, ?>) field.get(session);
+		assertThat(pendingResponses).hasSize(1);
+
+		subscription.dispose();
+
+		assertThat(pendingResponses).isEmpty();
+
+		session.close();
+	}
+
 }
diff --git a/mcp-core/src/test/java/io/modelcontextprotocol/spec/McpServerSessionTests.java b/mcp-core/src/test/java/io/modelcontextprotocol/spec/McpServerSessionTests.java
@@ -0,0 +1,106 @@
+/*
+ * Copyright 2024-2024 the original author or authors.
+ */
+
+package io.modelcontextprotocol.spec;
+
+import java.time.Duration;
+import java.util.Map;
+
+import io.modelcontextprotocol.json.TypeRef;
+import io.modelcontextprotocol.json.schema.JsonSchemaValidator;
+import io.modelcontextprotocol.server.McpInitRequestHandler;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Test suite for {@link McpServerSession} request bookkeeping: pending response entries
+ * must not leak when a request times out or the subscriber cancels.
+ */
+class McpServerSessionTests {
+
+	private static final Duration TIMEOUT = Duration.ofMillis(50);
+
+	private static final String TEST_METHOD = "test.method";
+
+	TypeRef<String> responseType = new TypeRef<>() {
+	};
+
+	private static McpServerTransport neverRespondingTransport() {
+		return new McpServerTransport() {
+			@Override
+			public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message) {
+				// Accept the message but never deliver a response.
+				return Mono.empty();
+			}
+
+			@Override
+			public Mono<Void> closeGracefully() {
+				return Mono.empty();
+			}
+
+			@Override
+			public <T> T unmarshalFrom(Object data, TypeRef<T> typeRef) {
+				return (T) data;
+			}
+		};
+	}
+
+	private static McpServerSession newSession(Duration requestTimeout) {
+		McpInitRequestHandler initHandler = params -> Mono
+			.error(new UnsupportedOperationException("not used in this test"));
+		JsonSchemaValidator validator = new JsonSchemaValidator() {
+			@Override
+			public ValidationResponse validate(Map<String, Object> schema, Object content) {
+				return ValidationResponse.asValid(null);
+			}
+
+			@Override
+			public ValidationResponse validateSchema(Map<String, Object> schema) {
+				return ValidationResponse.asValid(null);
+			}
+		};
+		return new McpServerSession("test-session", requestTimeout, neverRespondingTransport(), initHandler, Map.of(),
+				Map.of(), () -> Mono.empty(), validator);
+	}
+
+	@SuppressWarnings("unchecked")
+	private static Map<Object, Object> pendingResponses(McpServerSession session) throws Exception {
+		var field = McpServerSession.class.getDeclaredField("pendingResponses");
+		field.setAccessible(true);
+		return (Map<Object, Object>) field.get(session);
+	}
+
+	@Test
+	void testRequestTimeoutRemovesPendingResponse() throws Exception {
+		var session = newSession(TIMEOUT);
+
+		Mono<String> responseMono = session.sendRequest(TEST_METHOD, "test", responseType);
+
+		StepVerifier.create(responseMono).expectError(java.util.concurrent.TimeoutException.class).verify();
+
+		assertThat(pendingResponses(session)).isEmpty();
+
+		session.close();
+	}
+
+	@Test
+	void testRequestCancellationRemovesPendingResponse() throws Exception {
+		var session = newSession(Duration.ofSeconds(5));
+
+		Mono<String> responseMono = session.sendRequest(TEST_METHOD, "test", responseType);
+
+		var subscription = responseMono.subscribe();
+		assertThat(pendingResponses(session)).hasSize(1);
+
+		subscription.dispose();
+
+		assertThat(pendingResponses(session)).isEmpty();
+
+		session.close();
+	}
+
+}