From ac2c2fb8f5c3fdc28b2146ba1b4d162c8d5fa061 Mon Sep 17 00:00:00 2001 From: Param Parikh Date: Tue, 13 Oct 2026 22:36:00 +0000 Subject: [PATCH] Ignore the body of an HTTP notification response Streamable HTTP answers an accepted notification with 202 and no body, but some servers send a body anyway, such as {}. HttpMcpClient parsed that body as a JSON-RPC response and failed with "Missing required members: [jsonrpc]". For notifications/initialized that failed McpRemoteClient.initialize, so the remote server contributed no tools. A notification has no JSON-RPC response, so discard the body, as StdioMcpClient does. A short body of known length is read so the connection can be reused; anything else, including a stream, is closed unread so a misbehaving server cannot block the client or make it buffer without bound. --- .../smithy/java/mcp/server/HttpMcpClient.java | 21 ++++ .../java/mcp/server/HttpMcpClientTest.java | 117 ++++++++++++++++++ 2 files changed, 138 insertions(+) diff --git a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/HttpMcpClient.java b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/HttpMcpClient.java index a1078f613..5bb414c02 100644 --- a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/HttpMcpClient.java +++ b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/HttpMcpClient.java @@ -41,6 +41,7 @@ public final class HttpMcpClient extends McpRemoteClient { private static final InternalLogger LOG = InternalLogger.getLogger(HttpMcpClient.class); private static final int UPSTREAM_HTTP_ERROR_CODE = -32000; + private static final long MAX_DISCARDED_BODY_BYTES = 8 * 1024; private final ClientTransport transport; private final URI endpoint; @@ -246,6 +247,13 @@ private JsonRpcResponse exchange(JsonRpcRequest request, boolean allowSessionRec return handleErrorResponse(response, request); } + // A notification has no JSON-RPC response. Streamable HTTP accepts one with 202 and no body, + // but some servers send a body anyway (e.g. "{}"), so discard it instead of parsing it. + if (request.getId() == null) { + discardBody(response); + return null; + } + // Check if response is SSE String contentType = response.body().contentType(); if (contentType != null && contentType.startsWith("text/event-stream")) { @@ -264,6 +272,19 @@ private JsonRpcResponse exchange(JsonRpcRequest request, boolean allowSessionRec } } + /** + * Reads a short body of known length so the connection can be reused, and closes anything longer, + * of unknown length, or streaming without reading it. + */ + private static void discardBody(HttpResponse response) { + var body = response.body(); + if (body != null && body.hasKnownLength() && body.contentLength() <= MAX_DISCARDED_BODY_BYTES) { + ByteBufferUtils.getBytes(body.asByteBuffer()); + } else { + response.close(); + } + } + private JsonRpcResponse sessionRecoveryFailure( JsonRpcRequest request, boolean ownedRecovery diff --git a/mcp/mcp-server/src/test/java/software/amazon/smithy/java/mcp/server/HttpMcpClientTest.java b/mcp/mcp-server/src/test/java/software/amazon/smithy/java/mcp/server/HttpMcpClientTest.java index 5619a66fd..3178bc9f5 100644 --- a/mcp/mcp-server/src/test/java/software/amazon/smithy/java/mcp/server/HttpMcpClientTest.java +++ b/mcp/mcp-server/src/test/java/software/amazon/smithy/java/mcp/server/HttpMcpClientTest.java @@ -15,6 +15,7 @@ import java.math.BigInteger; import java.net.InetSocketAddress; import java.nio.charset.StandardCharsets; +import java.time.Duration; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -300,6 +301,122 @@ void testNotificationAcceptsEmptySuccessfulResponse() throws IOException { assertNull(proxy.exchange(notification)); } + @Test + void testNotificationIgnoresBodyOfAcceptedResponse() throws IOException { + // Some servers answer an accepted notification with 202 and a JSON body such as "{}" + // instead of no body. + mockServer.removeContext("/mcp"); + mockServer.createContext("/mcp", exchange -> { + exchange.getRequestBody().readAllBytes(); + writeJson(exchange, 202, "{}"); + }); + + var notification = JsonRpcRequest.builder() + .method(McpMethod.Standard.NOTIFICATIONS_INITIALIZED.wireName()) + .jsonrpc("2.0") + .build(); + + assertNull(proxy.exchange(notification)); + } + + @Test + void testNotificationIgnoresChunkedNonObjectBodyOfAcceptedResponse() throws IOException { + // A body of unknown length that is not a JSON object, here "" sent chunked with no + // Content-Length or content type, is discarded like any other notification response body. + mockServer.removeContext("/mcp"); + mockServer.createContext("/mcp", exchange -> { + exchange.getRequestBody().readAllBytes(); + exchange.sendResponseHeaders(202, 0); + try (var output = exchange.getResponseBody()) { + output.write("\"\"".getBytes(StandardCharsets.UTF_8)); + } + }); + + var notification = JsonRpcRequest.builder() + .method(McpMethod.Standard.NOTIFICATIONS_INITIALIZED.wireName()) + .jsonrpc("2.0") + .build(); + + assertNull(proxy.exchange(notification)); + } + + @Test + void testNotificationDoesNotReadAStreamingResponseBody() throws Exception { + // A server that answers a notification with a never-ending stream must not block the + // client or make it buffer the stream. + var release = new CountDownLatch(1); + mockServer.removeContext("/mcp"); + mockServer.createContext("/mcp", exchange -> { + exchange.getRequestBody().readAllBytes(); + exchange.getResponseHeaders().set("Content-Type", "text/event-stream"); + exchange.sendResponseHeaders(202, 0); + try (var output = exchange.getResponseBody()) { + output.write("data: {}\n\n".getBytes(StandardCharsets.UTF_8)); + output.flush(); + release.await(10, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } catch (IOException ignored) { + // The client closed the stream. + } + }); + + var notification = JsonRpcRequest.builder() + .method(McpMethod.Standard.NOTIFICATIONS_INITIALIZED.wireName()) + .jsonrpc("2.0") + .build(); + try { + assertTimeoutPreemptively(Duration.ofSeconds(5), () -> assertNull(proxy.exchange(notification))); + } finally { + release.countDown(); + } + } + + @Test + void initializeSucceedsWhenInitializedNotificationIsAcceptedWithABody() throws IOException { + mockServer.removeContext("/mcp"); + mockServer.createContext("/mcp", exchange -> { + var request = JsonRpcRequest.builder() + .deserialize(JSON_CODEC.createDeserializer(exchange.getRequestBody().readAllBytes())) + .build(); + if (request.getId() == null) { + writeJson(exchange, 202, "{}"); + return; + } + var result = McpHttpBinding.isInitialize(request) + ? Document.of(Map.of("protocolVersion", Document.of("2025-03-26"))) + : Document.of("pong"); + writeJson(exchange, + 200, + JSON_CODEC.serializeToString(JsonRpcResponse.builder() + .jsonrpc("2.0") + .id(request.getId()) + .result(result) + .build())); + }); + + var protocol = BuiltInProtocols.protocol(KnownProtocolVersion.V2025_06_18); + assertDoesNotThrow(() -> proxy.initialize( + ignored -> {}, + ignored -> {}, + JsonRpcRequest.builder() + .method(McpMethod.Standard.INITIALIZE.wireName()) + .id(Document.of(1)) + .jsonrpc("2.0") + .params(Document.of(Map.of( + "protocolVersion", + Document.of(protocol.id().identifier())))) + .build(), + protocol)); + + var response = proxy.exchange(JsonRpcRequest.builder() + .method(McpMethod.Standard.PING.wireName()) + .id(Document.of(2)) + .jsonrpc("2.0") + .build()); + assertEquals("pong", response.getResult().asString()); + } + @Test void testRpcHttpError() throws IOException { mockServer.removeContext("/mcp");