Fix non-OK HTTP POST responses handling (#374)

Currently, JDK HttpClient SSE implementation incorrectly handles a
non-2xx HTTP response code for POST. Instead of immediately failing the
caller, it ignores the issue and the user awaits until the timeout kicks
in.

Set HttpClient to use HTTP/1.1 version explicitly for tests

A related problem happened with the WebClient Streamable HTTP
implementation, where the error was swallowed for non-400 responses.

Signed-off-by: Dariusz Jędrzejczyk <[email protected]>

Author: chemicL

mcp-spring/mcp-spring-webflux/src/main/java/io/modelcontextprotocol/client/transport/WebClientStreamableHttpTransport.java

@@ -354,7 +354,7 @@ private Flux<McpSchema.JSONRPCMessage> extractError(ClientResponse response, Str
 			if (responseException.getStatusCode().isSameCodeAs(HttpStatus.BAD_REQUEST)) {
 				return Mono.error(new McpTransportSessionNotFoundException(sessionRepresentation, toPropagate));
 			}
-			return Mono.empty();
+			return Mono.error(toPropagate);
 		}).flux();
 	}
 

mcp-spring/mcp-spring-webmvc/src/test/java/io/modelcontextprotocol/server/WebMvcSseIntegrationTests.java

@@ -32,6 +32,7 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 import reactor.test.StepVerifier;
 
 import org.springframework.context.annotation.Bean;
@@ -96,9 +97,11 @@ public void before() {
 
 	@AfterEach
 	public void after() {
+		reactor.netty.http.HttpResources.disposeLoopsAndConnections();
 		if (mcpServerTransportProvider != null) {
 			mcpServerTransportProvider.closeGracefully().block();
 		}
+		Schedulers.shutdownNow();
 		if (tomcatServer.appContext() != null) {
 			tomcatServer.appContext().close();
 		}
@@ -779,6 +782,33 @@ void testToolCallSuccess() {
 		mcpServer.close();
 	}
 
+	@Test
+	void testThrowingToolCallIsCaughtBeforeTimeout() {
+		McpSyncServer mcpServer = McpServer.sync(mcpServerTransportProvider)
+			.capabilities(ServerCapabilities.builder().tools(true).build())
+			.tools(new McpServerFeatures.SyncToolSpecification(
+					new McpSchema.Tool("tool1", "tool1 description", emptyJsonSchema), (exchange, request) -> {
+						// We trigger a timeout on blocking read, raising an exception
+						Mono.never().block(Duration.ofSeconds(1));
+						return null;
+					}))
+			.build();
+
+		try (var mcpClient = clientBuilder.requestTimeout(Duration.ofMillis(6666)).build()) {
+			InitializeResult initResult = mcpClient.initialize();
+			assertThat(initResult).isNotNull();
+
+			// We expect the tool call to fail immediately with the exception raised by
+			// the offending tool
+			// instead of getting back a timeout.
+			assertThatExceptionOfType(McpError.class)
+				.isThrownBy(() -> mcpClient.callTool(new McpSchema.CallToolRequest("tool1", Map.of())))
+				.withMessageContaining("Timeout on blocking read");
+		}
+
+		mcpServer.close();
+	}
+
 	@Test
 	void testToolListChangeHandlingSuccess() {
 

mcp/src/main/java/io/modelcontextprotocol/client/transport/HttpClientSseClientTransport.java

@@ -421,13 +421,17 @@ public Mono<Void> sendMessage(JSONRPCMessage message) {
 			}
 
 			return this.serializeMessage(message)
-				.flatMap(body -> sendHttpPost(messageEndpointUri, body))
-				.doOnNext(response -> {
+				.flatMap(body -> sendHttpPost(messageEndpointUri, body).handle((response, sink) -> {
 					if (response.statusCode() != 200 && response.statusCode() != 201 && response.statusCode() != 202
 							&& response.statusCode() != 206) {
-						logger.error("Error sending message: {}", response.statusCode());
+						sink.error(new RuntimeException(
+								"Sending message failed with a non-OK HTTP code: " + response.statusCode()));
 					}
-				})
+					else {
+						sink.next(response);
+						sink.complete();
+					}
+				}))
 				.doOnError(error -> {
 					if (!isClosing) {
 						logger.error("Error sending message: {}", error.getMessage());

mcp/src/test/java/io/modelcontextprotocol/client/transport/HttpClientSseClientTransportTests.java

@@ -59,7 +59,9 @@ static class TestHttpClientSseClientTransport extends HttpClientSseClientTranspo
 		private Sinks.Many<ServerSentEvent<String>> events = Sinks.many().unicast().onBackpressureBuffer();
 
 		public TestHttpClientSseClientTransport(final String baseUri) {
-			super(HttpClient.newHttpClient(), HttpRequest.newBuilder(), baseUri, "/sse", new ObjectMapper());
+			super(HttpClient.newBuilder().version(HttpClient.Version.HTTP_1_1).build(),
+					HttpRequest.newBuilder().header("Content-Type", "application/json"), baseUri, "/sse",
+					new ObjectMapper());
 		}
 
 		public int getInboundMessageCount() {