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..9e59bea80 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 @@ -212,7 +212,12 @@ private void initProcessing() { } private void handleIncomingMessages() { - this.inboundSink.asFlux().flatMap(message -> session.handle(message)).doOnTerminate(() -> { + this.inboundSink.asFlux().flatMap(message -> session.handle(message).onErrorResume(error -> { + // A failure in one message must not terminate the inbound stream, + // otherwise the server stops answering all further messages + logger.error("Error handling message: {}", error.getMessage()); + return Mono.empty(); + })).doOnTerminate(() -> { // The outbound processing will dispose its scheduler upon completion this.outboundSink.tryEmitComplete(); this.inboundScheduler.dispose(); diff --git a/mcp-test/src/test/java/io/modelcontextprotocol/server/transport/StdioServerTransportProviderTests.java b/mcp-test/src/test/java/io/modelcontextprotocol/server/transport/StdioServerTransportProviderTests.java index 72d0dfcf8..4f40c92c3 100644 --- a/mcp-test/src/test/java/io/modelcontextprotocol/server/transport/StdioServerTransportProviderTests.java +++ b/mcp-test/src/test/java/io/modelcontextprotocol/server/transport/StdioServerTransportProviderTests.java @@ -143,6 +143,37 @@ void shouldHandleIncomingMessages() { }).verifyComplete(); } + @Test + void shouldKeepHandlingMessagesAfterOneFails() { + + String jsonMessages = "{\"jsonrpc\":\"2.0\",\"method\":\"notifications/fails\"}\n" + + "{\"jsonrpc\":\"2.0\",\"method\":\"ping\",\"id\":2}\n"; + InputStream stream = new ByteArrayInputStream(jsonMessages.getBytes(StandardCharsets.UTF_8)); + + transportProvider = new StdioServerTransportProvider(McpJsonDefaults.getMapper(), stream, System.out); + CountDownLatch pingLatch = new CountDownLatch(1); + + McpServerSession.Factory realSessionFactory = transport -> { + McpServerSession session = mock(McpServerSession.class); + when(session.handle(any())).thenAnswer(invocation -> { + McpSchema.JSONRPCMessage message = invocation.getArgument(0); + if (message instanceof McpSchema.JSONRPCNotification) { + return Mono.error(new RuntimeException("notification handler failed")); + } + pingLatch.countDown(); + return Mono.empty(); + }); + when(session.closeGracefully()).thenReturn(Mono.empty()); + return session; + }; + + transportProvider.setSessionFactory(realSessionFactory); + + StepVerifier.create(Mono.fromCallable(() -> pingLatch.await(5, TimeUnit.SECONDS))) + .expectNext(true) + .verifyComplete(); + } + @Test void shouldHandleUtf8MessagesWithNonUtf8DefaultCharset() throws Exception { String utf8Content = "한글 漢字 café 🎉";