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 <dariusz.jedrzejczyk@broadcom.com>
This commit is contained in:
parent
b37c32cdba
commit
95df67ec05
4 changed files with 42 additions and 6 deletions
|
|
@ -354,7 +354,7 @@ public class WebClientStreamableHttpTransport implements McpClientTransport {
|
|||
if (responseException.getStatusCode().isSameCodeAs(HttpStatus.BAD_REQUEST)) {
|
||||
return Mono.error(new McpTransportSessionNotFoundException(sessionRepresentation, toPropagate));
|
||||
}
|
||||
return Mono.empty();
|
||||
return Mono.error(toPropagate);
|
||||
}).flux();
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -32,6 +32,7 @@ import org.junit.jupiter.api.AfterEach;
|
|||
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 @@ class WebMvcSseIntegrationTests {
|
|||
|
||||
@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 @@ class WebMvcSseIntegrationTests {
|
|||
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() {
|
||||
|
||||
|
|
|
|||
|
|
@ -421,13 +421,17 @@ public class HttpClientSseClientTransport implements McpClientTransport {
|
|||
}
|
||||
|
||||
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());
|
||||
|
|
|
|||
|
|
@ -59,7 +59,9 @@ class HttpClientSseClientTransportTests {
|
|||
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() {
|
||||
|
|
|
|||
Loading…
Reference in a new issue