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..0b84211f8 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,96 @@ 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 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, "{}"); + } else if (McpHttpBinding.isInitialize(request)) { + writeJson(exchange, + 200, + "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"protocolVersion\":\"2025-03-26\"}}"); + } else { + writeJson(exchange, 200, "{\"jsonrpc\":\"2.0\",\"id\":2,\"result\":\"pong\"}"); + } + }); + + 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");