HttpClientStreamableHttpTransport: handle HTTP 405
- Fixes #877 Signed-off-by: Daniel Garnier-Moiroux <git@garnier.wf>
This commit is contained in:
parent
7e4c179b77
commit
8e1c5d0cd4
2 changed files with 25 additions and 5 deletions
|
|
@ -298,6 +298,10 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
|
||||||
"Authorization error connecting to SSE stream",
|
"Authorization error connecting to SSE stream",
|
||||||
responseEvent.responseInfo()));
|
responseEvent.responseInfo()));
|
||||||
}
|
}
|
||||||
|
else if (statusCode == METHOD_NOT_ALLOWED) {
|
||||||
|
logger.debug("The server does not support SSE streams, using request-response mode.");
|
||||||
|
return Flux.empty();
|
||||||
|
}
|
||||||
|
|
||||||
if (!(responseEvent instanceof ResponseSubscribers.SseResponseEvent sseResponseEvent)) {
|
if (!(responseEvent instanceof ResponseSubscribers.SseResponseEvent sseResponseEvent)) {
|
||||||
return Flux.<McpSchema.JSONRPCMessage>error(new McpTransportException(
|
return Flux.<McpSchema.JSONRPCMessage>error(new McpTransportException(
|
||||||
|
|
@ -344,10 +348,6 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
|
||||||
return Flux.empty();
|
return Flux.empty();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
else if (statusCode == METHOD_NOT_ALLOWED) { // NotAllowed
|
|
||||||
logger.debug("The server does not support SSE streams, using request-response mode.");
|
|
||||||
return Flux.empty();
|
|
||||||
}
|
|
||||||
else if (statusCode == NOT_FOUND) {
|
else if (statusCode == NOT_FOUND) {
|
||||||
|
|
||||||
if (transportSession != null && transportSession.sessionId().isPresent()) {
|
if (transportSession != null && transportSession.sessionId().isPresent()) {
|
||||||
|
|
|
||||||
|
|
@ -18,7 +18,6 @@ import java.util.function.Predicate;
|
||||||
import com.sun.net.httpserver.HttpServer;
|
import com.sun.net.httpserver.HttpServer;
|
||||||
import io.modelcontextprotocol.client.transport.customizer.McpHttpClientAuthorizationErrorHandler;
|
import io.modelcontextprotocol.client.transport.customizer.McpHttpClientAuthorizationErrorHandler;
|
||||||
import io.modelcontextprotocol.common.McpTransportContext;
|
import io.modelcontextprotocol.common.McpTransportContext;
|
||||||
import org.reactivestreams.Publisher;
|
|
||||||
import io.modelcontextprotocol.server.transport.TomcatTestUtil;
|
import io.modelcontextprotocol.server.transport.TomcatTestUtil;
|
||||||
import io.modelcontextprotocol.spec.HttpHeaders;
|
import io.modelcontextprotocol.spec.HttpHeaders;
|
||||||
import io.modelcontextprotocol.spec.McpClientTransport;
|
import io.modelcontextprotocol.spec.McpClientTransport;
|
||||||
|
|
@ -34,6 +33,7 @@ import org.junit.jupiter.api.Test;
|
||||||
import org.junit.jupiter.api.Timeout;
|
import org.junit.jupiter.api.Timeout;
|
||||||
import org.junit.jupiter.params.ParameterizedTest;
|
import org.junit.jupiter.params.ParameterizedTest;
|
||||||
import org.junit.jupiter.params.provider.ValueSource;
|
import org.junit.jupiter.params.provider.ValueSource;
|
||||||
|
import org.reactivestreams.Publisher;
|
||||||
import reactor.core.publisher.Mono;
|
import reactor.core.publisher.Mono;
|
||||||
import reactor.test.StepVerifier;
|
import reactor.test.StepVerifier;
|
||||||
|
|
||||||
|
|
@ -369,6 +369,26 @@ public class HttpClientStreamableHttpTransportErrorHandlingTest {
|
||||||
StepVerifier.create(transport.closeGracefully()).verifyComplete();
|
StepVerifier.create(transport.closeGracefully()).verifyComplete();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void test405OnConnectReturnsEmptyFlux() {
|
||||||
|
serverSseResponseStatus.set(405);
|
||||||
|
AtomicReference<Throwable> capturedException = new AtomicReference<>();
|
||||||
|
var transport = HttpClientStreamableHttpTransport.builder(HOST).openConnectionOnStartup(true).build();
|
||||||
|
transport.setExceptionHandler(capturedException::set);
|
||||||
|
|
||||||
|
var messages = new ArrayList<McpSchema.JSONRPCMessage>();
|
||||||
|
StepVerifier.create(transport.connect(msg -> msg.doOnNext(messages::add))).verifyComplete();
|
||||||
|
|
||||||
|
Awaitility.await()
|
||||||
|
.atMost(Duration.ofSeconds(1))
|
||||||
|
.untilAsserted(() -> assertThat(processedSseConnectCount.get()).isEqualTo(1));
|
||||||
|
|
||||||
|
assertThat(messages).isEmpty();
|
||||||
|
assertThat(capturedException.get()).isNull();
|
||||||
|
|
||||||
|
StepVerifier.create(transport.closeGracefully()).verifyComplete();
|
||||||
|
}
|
||||||
|
|
||||||
@Nested
|
@Nested
|
||||||
class AuthorizationError {
|
class AuthorizationError {
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue