From 0874afe918bd0267d123e225207ddae152faa67f Mon Sep 17 00:00:00 2001
From: prathamgoyal787 <prathamg2003@gmail.com>
Date: Tue, 20 Oct 2026 20:34:58 +0530
Subject: [PATCH] fix: keep stdio server handling messages after one fails

StdioServerTransportProvider fed every inbound message through one flatMap, so an error from a single message (for example a failing notification handler) terminated the inbound stream. The server process stayed alive but never answered another request, not even ping. Errors are now handled per message, logged, and the stream keeps going.
---
 .../StdioServerTransportProvider.java         |  7 ++++-
 .../StdioServerTransportProviderTests.java    | 31 +++++++++++++++++++
 2 files changed, 37 insertions(+), 1 deletion(-)

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é 🎉";