Skip to content

Fix non-OK HTTP POST responses handling #374

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 1 commit into
base: main
Choose a base branch
from
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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();
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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();
}
Expand Down Expand Up @@ -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() {

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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());
Expand Down
Loading