From 621fe2ad11cea9a9c369852dbf86bc8c59dfa904 Mon Sep 17 00:00:00 2001 From: Administrator <1176316138@qq.com> Date: Mon, 12 Oct 2026 19:51:21 +0800 Subject: [PATCH] Fix STDIO server recovery after malformed JSON Return JSON-RPC errors for malformed input and continue processing valid requests through the existing outbound queue. Add wire-level regression coverage for recovery and fatal transport errors. Fixes #1172 --- .../StdioServerTransportProvider.java | 54 +++- .../StdioServerMalformedJsonTests.java | 237 ++++++++++++++++++ 2 files changed, 284 insertions(+), 7 deletions(-) create mode 100644 mcp-test/src/test/java/io/modelcontextprotocol/server/transport/StdioServerMalformedJsonTests.java diff --git a/mcp-core/src/main/java/io/modelcontextprotocol/server/transport/StdioServerTransportProvider.java b/mcp-core/src/main/java/io/modelcontextprotocol/server/transport/StdioServerTransportProvider.java index 8370e41ed..928071454 100644 --- a/mcp-core/src/main/java/io/modelcontextprotocol/server/transport/StdioServerTransportProvider.java +++ b/mcp-core/src/main/java/io/modelcontextprotocol/server/transport/StdioServerTransportProvider.java @@ -12,10 +12,14 @@ import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.List; +import java.util.Map; import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Function; +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.annotation.JsonProperty; + import io.modelcontextprotocol.json.TypeRef; import io.modelcontextprotocol.spec.McpSchema; import io.modelcontextprotocol.spec.McpSchema.JSONRPCMessage; @@ -239,17 +243,17 @@ private void startInboundProcessing() { logger.debug("Received JSON message: {}", line); + McpSchema.JSONRPCMessage message; try { - McpSchema.JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(jsonMapper, - line); - if (!this.inboundSink.tryEmitNext(message).isSuccess()) { - // logIfNotClosing("Failed to enqueue message"); - break; - } - + message = McpSchema.deserializeJsonRpcMessage(jsonMapper, line); } catch (Exception e) { logIfNotClosing("Error processing inbound message", e); + sendInvalidMessageResponse(line); + continue; + } + if (!this.inboundSink.tryEmitNext(message).isSuccess()) { + // logIfNotClosing("Failed to enqueue message"); break; } } @@ -277,6 +281,31 @@ private void startInboundProcessing() { } } + private void sendInvalidMessageResponse(String line) { + Object id = null; + int code = McpSchema.ErrorCodes.PARSE_ERROR; + try { + // Reparse only failed messages as any JSON value, distinguishing syntax + // errors from valid JSON that failed envelope validation. + Object value = jsonMapper.readValue(line, Object.class); + code = McpSchema.ErrorCodes.INVALID_REQUEST; + if (value instanceof Map map) { + Object requestId = map.get("id"); + if (requestId instanceof String || requestId instanceof Integer || requestId instanceof Long) { + id = requestId; + } + } + } + catch (IOException | IllegalArgumentException e) { + // The complete line is not valid JSON. Its request ID cannot be known. + } + String message = code == McpSchema.ErrorCodes.PARSE_ERROR ? "Parse error" : "Invalid Request"; + // Use the normal outbound queue so errors cannot interleave with replies. + sendMessage(new InvalidMessageResponse(McpSchema.JSONRPC_VERSION, id, + new McpSchema.JSONRPCResponse.JSONRPCError(code, message))) + .block(); + } + /** * Starts the outbound processing thread that writes JSON-RPC messages to stdout. * Messages are serialized to JSON and written with a newline delimiter. @@ -367,4 +396,15 @@ private void logIfNotClosing(String message, Exception e) { } + /** + * Transport-only error envelope for input that cannot be decoded as an MCP message. + * Unlike ordinary MCP responses, it must retain an explicit null ID when the request + * identifier cannot be established. + */ + @JsonInclude(JsonInclude.Include.NON_ABSENT) + private record InvalidMessageResponse(@JsonProperty("jsonrpc") String jsonrpc, + @JsonProperty("id") @JsonInclude(JsonInclude.Include.ALWAYS) Object id, + @JsonProperty("error") McpSchema.JSONRPCResponse.JSONRPCError error) implements JSONRPCMessage { + } + } diff --git a/mcp-test/src/test/java/io/modelcontextprotocol/server/transport/StdioServerMalformedJsonTests.java b/mcp-test/src/test/java/io/modelcontextprotocol/server/transport/StdioServerMalformedJsonTests.java new file mode 100644 index 000000000..62bfe1c0a --- /dev/null +++ b/mcp-test/src/test/java/io/modelcontextprotocol/server/transport/StdioServerMalformedJsonTests.java @@ -0,0 +1,237 @@ +/* + * Copyright 2026-2026 the original author or authors. + */ + +package io.modelcontextprotocol.server.transport; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.io.PipedInputStream; +import java.io.PipedOutputStream; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; + +import io.modelcontextprotocol.json.TypeRef; +import io.modelcontextprotocol.server.McpServer; +import io.modelcontextprotocol.server.McpSyncServer; +import io.modelcontextprotocol.spec.McpSchema; +import io.modelcontextprotocol.spec.McpServerSession; +import org.awaitility.Awaitility; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import reactor.core.publisher.Mono; + +import static io.modelcontextprotocol.util.McpJsonMapperUtils.JSON_MAPPER; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Wire-level regression tests for malformed STDIO input followed by valid requests. + */ +class StdioServerMalformedJsonTests { + + @ParameterizedTest + @ValueSource(strings = { "{broken", "", "{\"jsonrpc\":" }) + void respondsToMalformedJsonAndContinuesReading(String line) throws Exception { + try (ServerConnection connection = new ServerConnection()) { + connection.send(line); + assertError(connection.read(), null, McpSchema.ErrorCodes.PARSE_ERROR); + + connection.ping(3); + assertPing(connection.read(), 3); + } + } + + @ParameterizedTest + @ValueSource(strings = { "{}", "null", "[]", "42" }) + void distinguishesInvalidRequestFromParseError(String line) throws Exception { + try (ServerConnection connection = new ServerConnection()) { + connection.send(line); + assertError(connection.read(), null, McpSchema.ErrorCodes.INVALID_REQUEST); + + connection.ping(3); + assertPing(connection.read(), 3); + } + } + + @Test + void preservesUsableIdOnInvalidRequest() throws Exception { + try (ServerConnection connection = new ServerConnection()) { + connection.send("{\"jsonrpc\":\"2.0\",\"id\":7,\"method\":null}"); + assertError(connection.read(), 7, McpSchema.ErrorCodes.INVALID_REQUEST); + + connection.ping(3); + assertPing(connection.read(), 3); + } + } + + @ParameterizedTest + @ValueSource(strings = { "true", "1.5", "[]", "{}" }) + void usesNullIdWhenInvalidRequestHasNoUsableId(String id) throws Exception { + try (ServerConnection connection = new ServerConnection()) { + connection.send("{\"jsonrpc\":\"2.0\",\"id\":" + id + ",\"method\":\"ping\"}"); + assertError(connection.read(), null, McpSchema.ErrorCodes.INVALID_REQUEST); + + connection.ping(3); + assertPing(connection.read(), 3); + } + } + + @Test + void recoversAfterConsecutiveBadLines() throws Exception { + try (ServerConnection connection = new ServerConnection()) { + connection.send("{broken"); + connection.send("{}"); + connection.send("{broken again"); + connection.ping(3); + + assertError(connection.read(), null, McpSchema.ErrorCodes.PARSE_ERROR); + assertError(connection.read(), null, McpSchema.ErrorCodes.INVALID_REQUEST); + assertError(connection.read(), null, McpSchema.ErrorCodes.PARSE_ERROR); + assertPing(connection.read(), 3); + } + } + + @Test + void serializesErrorAndNormalRepliesThroughTheSameQueue() throws Exception { + try (ServerConnection connection = new ServerConnection()) { + connection.send("{broken"); + for (int id = 3; id < 13; id++) { + connection.ping(id); + } + + assertError(connection.read(), null, McpSchema.ErrorCodes.PARSE_ERROR); + Set ids = new HashSet<>(); + for (int i = 0; i < 10; i++) { + Map reply = connection.read(); + assertThat(reply).containsEntry("jsonrpc", "2.0").containsKey("result").doesNotContainKey("error"); + ids.add(((Number) reply.get("id")).intValue()); + } + assertThat(ids).containsExactlyInAnyOrder(3, 4, 5, 6, 7, 8, 9, 10, 11, 12); + } + } + + @Test + void oversizedLineStillClosesSessionWithoutReadingItsRemainder() { + String input = "x".repeat(40) + "\n{\"jsonrpc\":\"2.0\",\"id\":3,\"method\":\"ping\"}\n"; + assertFatalInput(new ByteArrayInputStream(input.getBytes(StandardCharsets.UTF_8)), 32); + } + + @Test + void inputIoFailureStillClosesSession() { + assertFatalInput(new InputStream() { + + @Override + public int read() throws IOException { + throw new IOException("input failed"); + } + + }, 1024); + } + + private static void assertFatalInput(InputStream input, int maxSize) { + McpServerSession session = mock(McpServerSession.class); + when(session.closeGracefully()).thenReturn(Mono.empty()); + var transport = new StdioServerTransportProvider(JSON_MAPPER, input, OutputStream.nullOutputStream(), maxSize); + try { + transport.setSessionFactory(ignored -> session); + Awaitility.await().atMost(Duration.ofSeconds(3)).untilAsserted(() -> verify(session).close()); + verify(session, never()).handle(any()); + } + finally { + transport.closeGracefully().block(); + } + } + + private static void assertError(Map reply, Object id, int code) { + assertThat(reply).containsEntry("jsonrpc", "2.0").containsEntry("id", id).doesNotContainKey("result"); + assertThat(reply.get("error")).isInstanceOfSatisfying(Map.class, + error -> assertThat(error.get("code")).isEqualTo(code)); + } + + private static void assertPing(Map reply, int id) { + assertThat(reply).containsEntry("jsonrpc", "2.0") + .containsEntry("id", id) + .containsEntry("result", Map.of()) + .doesNotContainKey("error"); + } + + private static final class ServerConnection implements AutoCloseable { + + private final PipedOutputStream requests = new PipedOutputStream(); + + private final PipedInputStream input = new PipedInputStream(this.requests); + + private final BlockingQueue replies = new LinkedBlockingQueue<>(); + + private final McpSyncServer server; + + private ServerConnection() throws Exception { + OutputStream output = new OutputStream() { + + private final ByteArrayOutputStream line = new ByteArrayOutputStream(); + + @Override + public void write(int b) { + if (b == '\n') { + replies.add(this.line.toString(StandardCharsets.UTF_8)); + this.line.reset(); + } + else { + this.line.write(b); + } + } + + }; + var transport = new StdioServerTransportProvider(JSON_MAPPER, this.input, output); + this.server = McpServer.sync(transport).serverInfo("stdio-recovery-test", "1.0").build(); + send(""" + {"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-11-25", + "capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}} + """.replace("\n", "")); + assertThat(read()).containsEntry("id", 1).containsKey("result"); + send("{\"jsonrpc\":\"2.0\",\"method\":\"notifications/initialized\"}"); + ping(2); + assertPing(read(), 2); + } + + private void send(String line) throws IOException { + this.requests.write((line + "\n").getBytes(StandardCharsets.UTF_8)); + this.requests.flush(); + } + + private void ping(int id) throws IOException { + send("{\"jsonrpc\":\"2.0\",\"id\":" + id + ",\"method\":\"ping\"}"); + } + + private Map read() throws Exception { + String line = this.replies.poll(3, TimeUnit.SECONDS); + assertThat(line).as("STDIO reply before the timeout").isNotNull(); + return JSON_MAPPER.readValue(line, new TypeRef>() { + }); + } + + @Override + public void close() throws IOException { + this.server.closeGracefully(); + this.requests.close(); + this.input.close(); + } + + } + +}