fix: keep stdio server handling messages after one fails

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();

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