HttpClientStreamableHttpTransport: handle HTTP 405
- Forward-port of #900 Signed-off-by: Daniel Garnier-Moiroux <git@garnier.wf>
This commit is contained in:
parent
8c7774ad5f
commit
5e77762eeb
2 changed files with 25 additions and 5 deletions
|
|
@ -298,6 +298,10 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
|
|||
"Authorization error connecting to SSE stream",
|
||||
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)) {
|
||||
return Flux.<McpSchema.JSONRPCMessage>error(new McpTransportException(
|
||||
|
|
@ -344,10 +348,6 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
|
|||
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) {
|
||||
|
||||
if (transportSession != null && transportSession.sessionId().isPresent()) {
|
||||
|
|
|
|||
|
|
@ -18,7 +18,6 @@ import java.util.function.Predicate;
|
|||
import com.sun.net.httpserver.HttpServer;
|
||||
import io.modelcontextprotocol.client.transport.customizer.McpHttpClientAuthorizationErrorHandler;
|
||||
import io.modelcontextprotocol.common.McpTransportContext;
|
||||
import org.reactivestreams.Publisher;
|
||||
import io.modelcontextprotocol.server.transport.TomcatTestUtil;
|
||||
import io.modelcontextprotocol.spec.HttpHeaders;
|
||||
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.params.ParameterizedTest;
|
||||
import org.junit.jupiter.params.provider.ValueSource;
|
||||
import org.reactivestreams.Publisher;
|
||||
import reactor.core.publisher.Mono;
|
||||
import reactor.test.StepVerifier;
|
||||
|
||||
|
|
@ -369,6 +369,26 @@ public class HttpClientStreamableHttpTransportErrorHandlingTest {
|
|||
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
|
||||
class AuthorizationError {
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue