From 81909b61467b187858c893914a340a5cd563d42a Mon Sep 17 00:00:00 2001 From: kele5555 <464737194@qq.com> Date: Wed, 30 Sep 2026 10:45:49 +0800 Subject: [PATCH] Fix parsing error propagation in Streamable HTTP JSON responses --- .../HttpClientStreamableHttpTransport.java | 6 +- ...bleHttpTransportEmptyJsonResponseTest.java | 13 +- ...StreamableHttpTransportJsonErrorTests.java | 231 ++++++++++++++++++ 3 files changed, 239 insertions(+), 11 deletions(-) create mode 100644 mcp-test/src/test/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransportJsonErrorTests.java diff --git a/mcp-core/src/main/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransport.java b/mcp-core/src/main/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransport.java index 5517823b6..d44b63e24 100644 --- a/mcp-core/src/main/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransport.java +++ b/mcp-core/src/main/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransport.java @@ -645,16 +645,18 @@ else if (contentType.contains(TEXT_EVENT_STREAM)) { }); } else if (contentType.contains(APPLICATION_JSON)) { - deliveredSink.success(); String data = ((ResponseSubscribers.AggregateResponseEvent) responseEvent).data(); if (sentMessage instanceof McpSchema.JSONRPCNotification) { logger.warn("Notification: {} received non-compliant response: {}", sentMessage, Utils.hasText(data) ? data : "[empty]"); + deliveredSink.success(); return Mono.empty(); } try { - return Mono.just(McpSchema.deserializeJsonRpcMessage(jsonMapper, data)); + McpSchema.JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(jsonMapper, data); + deliveredSink.success(); + return Mono.just(message); } catch (IOException e) { return Mono.error(new McpTransportException( diff --git a/mcp-test/src/test/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransportEmptyJsonResponseTest.java b/mcp-test/src/test/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransportEmptyJsonResponseTest.java index c2d19ef67..1c96bd158 100644 --- a/mcp-test/src/test/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransportEmptyJsonResponseTest.java +++ b/mcp-test/src/test/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransportEmptyJsonResponseTest.java @@ -25,7 +25,6 @@ import io.modelcontextprotocol.client.transport.customizer.McpSyncHttpClientRequestCustomizer; import io.modelcontextprotocol.server.transport.TomcatTestUtil; import io.modelcontextprotocol.spec.McpSchema; -import io.modelcontextprotocol.spec.ProtocolVersions; import reactor.test.StepVerifier; /** @@ -76,18 +75,14 @@ void testNotificationInitialized() throws URISyntaxException { .httpRequestCustomizer(mockRequestCustomizer) .build(); - var initializeRequest = McpSchema.InitializeRequest - .builder(ProtocolVersions.MCP_2025_03_26, McpSchema.ClientCapabilities.builder().roots(true).build(), - McpSchema.Implementation.builder("MCP Client", "0.3.1").build()) - .build(); - var testMessage = new McpSchema.JSONRPCRequest(McpSchema.METHOD_INITIALIZE, "test-id", initializeRequest); + var testMessage = new McpSchema.JSONRPCNotification(McpSchema.METHOD_NOTIFICATION_INITIALIZED); StepVerifier.create(transport.sendMessage(testMessage)).verifyComplete(); // Verify the customizer was called - verify(mockRequestCustomizer, atLeastOnce()).customize(any(), eq("POST"), eq(uri), eq( - "{\"jsonrpc\":\"2.0\",\"method\":\"initialize\",\"id\":\"test-id\",\"params\":{\"protocolVersion\":\"2025-03-26\",\"capabilities\":{\"roots\":{\"listChanged\":true}},\"clientInfo\":{\"name\":\"MCP Client\",\"version\":\"0.3.1\"}}}"), - any()); + verify(mockRequestCustomizer, atLeastOnce()).customize(any(), eq("POST"), eq(uri), + eq("{\"jsonrpc\":\"2.0\",\"method\":\"notifications/initialized\"}"), any()); + transport.closeGracefully().block(); } diff --git a/mcp-test/src/test/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransportJsonErrorTests.java b/mcp-test/src/test/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransportJsonErrorTests.java new file mode 100644 index 000000000..010d77e1b --- /dev/null +++ b/mcp-test/src/test/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransportJsonErrorTests.java @@ -0,0 +1,231 @@ +/* + * Copyright 2026 the original author or authors. + */ + +package io.modelcontextprotocol.client.transport; + +import java.io.IOException; +import java.net.ConnectException; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Map; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpServer; +import io.modelcontextprotocol.client.McpClient; +import io.modelcontextprotocol.json.McpJsonDefaults; +import io.modelcontextprotocol.json.TypeRef; +import io.modelcontextprotocol.spec.McpClientSession; +import io.modelcontextprotocol.spec.McpSchema; +import io.modelcontextprotocol.spec.McpTransportException; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.springframework.test.util.ReflectionTestUtils; +import reactor.test.StepVerifier; + +import static org.assertj.core.api.Assertions.assertThat; + +@Timeout(15) +class HttpClientStreamableHttpTransportJsonErrorTests { + + private HttpServer server; + + private ExecutorService executor; + + private HttpClientStreamableHttpTransport transport; + + private volatile String responseBody = "{broken"; + + private volatile int responseStatus = 200; + + private volatile boolean unresponsive; + + private final CountDownLatch releaseResponse = new CountDownLatch(1); + + private final ConcurrentLinkedQueue methods = new ConcurrentLinkedQueue<>(); + + @BeforeEach + void startServer() throws IOException { + server = HttpServer.create(new InetSocketAddress("localhost", 0), 0); + executor = Executors.newCachedThreadPool(); + server.setExecutor(executor); + server.createContext("/mcp", this::respond); + server.start(); + transport = HttpClientStreamableHttpTransport.builder("http://localhost:" + server.getAddress().getPort()) + .openConnectionOnStartup(false) + .resumableStreams(false) + .build(); + } + + @AfterEach + void stopServer() { + releaseResponse.countDown(); + transport.closeGracefully().block(Duration.ofSeconds(2)); + server.stop(0); + executor.shutdownNow(); + } + + @ParameterizedTest + @ValueSource(strings = { "{broken", "", " " }) + void malformedJsonFailsSendMessageWithParsingCause(String body) { + responseBody = body; + transport.connect(message -> message).block(); + StepVerifier.create(transport.sendMessage(new McpSchema.JSONRPCRequest("ping", "test-id", null))) + .expectErrorSatisfies(HttpClientStreamableHttpTransportJsonErrorTests::assertParsingFailure) + .verify(Duration.ofSeconds(3)); + } + + @Test + void malformedJsonFailsPendingRequestAndRemovesItBeforeTimeout() { + var session = new McpClientSession(Duration.ofSeconds(5), transport, Map.of(), Map.of(), + connection -> connection); + try { + StepVerifier.create(session.sendRequest("ping", null, new TypeRef>() { + })) + .expectErrorSatisfies(HttpClientStreamableHttpTransportJsonErrorTests::assertParsingFailure) + .verify(Duration.ofSeconds(3)); + assertThat((Map) ReflectionTestUtils.getField(session, "pendingResponses")).isEmpty(); + } + finally { + session.close(); + } + } + + @Test + void malformedInitializeFailsWithoutCallingTools() { + var client = McpClient.async(transport) + .initializationTimeout(Duration.ofSeconds(5)) + .requestTimeout(Duration.ofSeconds(5)) + .build(); + try { + StepVerifier.create(client.initialize()) + .expectErrorSatisfies(error -> assertParsingFailure(error.getCause())) + .verify(Duration.ofSeconds(3)); + assertThat(methods).containsExactly("initialize"); + } + finally { + client.close(); + } + } + + @Test + void notificationStillAcceptsNonCompliantJsonResponse() { + transport.connect(message -> message).block(); + StepVerifier.create(transport.sendMessage(new McpSchema.JSONRPCNotification("notifications/initialized"))) + .verifyComplete(); + } + + @Test + void validInitializeListAndCallStillSucceed() { + responseBody = null; + var client = McpClient.async(transport).build(); + try { + StepVerifier.create(client.initialize().then(client.listTools())) + .assertNext( + result -> assertThat(result.tools()).extracting(McpSchema.Tool::name).containsExactly("probe")) + .verifyComplete(); + StepVerifier.create(client.callTool(McpSchema.CallToolRequest.builder("probe").build())) + .assertNext(result -> { + assertThat(result.isError()).isFalse(); + assertThat(result.content()).containsExactly(new McpSchema.TextContent("ok")); + }) + .verifyComplete(); + assertThat(methods).containsExactly("initialize", "notifications/initialized", "tools/list", "tools/call"); + } + finally { + client.close(); + } + } + + @Test + void unresponsiveServerStillTimesOut() { + unresponsive = true; + var session = new McpClientSession(Duration.ofMillis(300), transport, Map.of(), Map.of(), + connection -> connection); + try { + StepVerifier.create(session.sendRequest("ping", null, new TypeRef>() { + })).expectError(TimeoutException.class).verify(Duration.ofSeconds(3)); + } + finally { + session.close(); + } + } + + @Test + void serviceUnavailableStillReportsHttpFailure() { + responseStatus = 503; + transport.connect(message -> message).block(); + StepVerifier.create(transport.sendMessage(new McpSchema.JSONRPCRequest("ping", "test-id", null))) + .expectErrorSatisfies(error -> assertThat(error).isNotInstanceOf(TimeoutException.class) + .hasMessageContaining("Failed to send message")) + .verify(Duration.ofSeconds(3)); + } + + @Test + void connectionRefusalStillReportsConnectionFailure() { + server.stop(0); + transport.connect(message -> message).block(); + StepVerifier.create(transport.sendMessage(new McpSchema.JSONRPCRequest("ping", "test-id", null))) + .expectErrorSatisfies(error -> assertThat(error).isInstanceOf(ConnectException.class)) + .verify(Duration.ofSeconds(3)); + } + + private static void assertParsingFailure(Throwable error) { + assertThat(error).isInstanceOf(McpTransportException.class).hasCauseInstanceOf(IOException.class); + } + + private void respond(HttpExchange exchange) throws IOException { + try (exchange) { + if (!"POST".equals(exchange.getRequestMethod())) { + exchange.sendResponseHeaders(405, -1); + return; + } + var request = McpJsonDefaults.getMapper().readValue(exchange.getRequestBody().readAllBytes(), Map.class); + String method = (String) request.get("method"); + methods.add(method); + if (unresponsive) { + try { + releaseResponse.await(10, TimeUnit.SECONDS); + } + catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + } + return; + } + String body = responseBody; + if (body == null) { + if (!request.containsKey("id")) { + exchange.sendResponseHeaders(202, -1); + return; + } + Object result = switch (method) { + case "initialize" -> Map.of("protocolVersion", "2025-11-25", "capabilities", + Map.of("tools", Map.of()), "serverInfo", Map.of("name", "test", "version", "1")); + case "tools/list" -> Map.of("tools", + java.util.List.of(Map.of("name", "probe", "inputSchema", Map.of("type", "object")))); + case "tools/call" -> + Map.of("content", java.util.List.of(Map.of("type", "text", "text", "ok")), "isError", false); + default -> Map.of(); + }; + body = McpJsonDefaults.getMapper() + .writeValueAsString(Map.of("jsonrpc", "2.0", "id", request.get("id"), "result", result)); + } + byte[] bytes = body.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().set("Content-Type", "application/json"); + exchange.sendResponseHeaders(responseStatus, bytes.length); + exchange.getResponseBody().write(bytes); + } + } + +}