Expose request URI in McpHttpClientAuthorizationErrorHandler

Breaking change

Signed-off-by: Daniel Garnier-Moiroux <git@garnier.wf>
This commit is contained in:
Daniel Garnier-Moiroux 2026-05-22 09:38:18 +02:00 • committed by Daniel Garnier-Moiroux
commit 4f3f7d9fa4
8 changed files with 282 additions and 36 deletions

View file

@ -24,6 +24,7 @@ import io.modelcontextprotocol.client.McpAsyncClient;
import io.modelcontextprotocol.client.transport.ResponseSubscribers.ResponseEvent; import io.modelcontextprotocol.client.transport.ResponseSubscribers.ResponseEvent;
import io.modelcontextprotocol.client.transport.customizer.McpAsyncHttpClientRequestCustomizer; import io.modelcontextprotocol.client.transport.customizer.McpAsyncHttpClientRequestCustomizer;
import io.modelcontextprotocol.client.transport.customizer.McpHttpClientAuthorizationErrorHandler; import io.modelcontextprotocol.client.transport.customizer.McpHttpClientAuthorizationErrorHandler;
import io.modelcontextprotocol.client.transport.customizer.McpHttpClientTransportAuthorizationErrorHandler;
import io.modelcontextprotocol.client.transport.customizer.McpSyncHttpClientRequestCustomizer; import io.modelcontextprotocol.client.transport.customizer.McpSyncHttpClientRequestCustomizer;
import io.modelcontextprotocol.common.McpTransportContext; import io.modelcontextprotocol.common.McpTransportContext;
import io.modelcontextprotocol.json.McpJsonDefaults; import io.modelcontextprotocol.json.McpJsonDefaults;
@ -120,7 +121,7 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
private final boolean openConnectionOnStartup; private final boolean openConnectionOnStartup;
private final McpHttpClientAuthorizationErrorHandler authorizationErrorHandler; private final McpHttpClientTransportAuthorizationErrorHandler authorizationErrorHandler;
private final boolean resumableStreams; private final boolean resumableStreams;
@ -139,7 +140,8 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
private HttpClientStreamableHttpTransport(McpJsonMapper jsonMapper, HttpClient httpClient, private HttpClientStreamableHttpTransport(McpJsonMapper jsonMapper, HttpClient httpClient,
HttpRequest.Builder requestBuilder, String baseUri, String endpoint, boolean resumableStreams, HttpRequest.Builder requestBuilder, String baseUri, String endpoint, boolean resumableStreams,
boolean openConnectionOnStartup, McpAsyncHttpClientRequestCustomizer httpRequestCustomizer, boolean openConnectionOnStartup, McpAsyncHttpClientRequestCustomizer httpRequestCustomizer,
McpHttpClientAuthorizationErrorHandler authorizationErrorHandler, List<String> supportedProtocolVersions) { McpHttpClientTransportAuthorizationErrorHandler authorizationErrorHandler,
List<String> supportedProtocolVersions) {
this.jsonMapper = jsonMapper; this.jsonMapper = jsonMapper;
this.httpClient = httpClient; this.httpClient = httpClient;
this.requestBuilder = requestBuilder; this.requestBuilder = requestBuilder;
@ -295,9 +297,12 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
int statusCode = responseEvent.responseInfo().statusCode(); int statusCode = responseEvent.responseInfo().statusCode();
if (statusCode == 401 || statusCode == 403) { if (statusCode == 401 || statusCode == 403) {
logger.debug("Authorization error in reconnect with code {}", statusCode); logger.debug("Authorization error in reconnect with code {}", statusCode);
var request = requestBuilder.build();
var requestSnapshot = new HttpRequestSnapshot(request.uri(), request.method(),
request.headers());
return Mono.<McpSchema.JSONRPCMessage>error( return Mono.<McpSchema.JSONRPCMessage>error(
new McpHttpClientTransportAuthorizationException( new McpHttpClientTransportAuthorizationException(
"Authorization error connecting to SSE stream", "Authorization error connecting to SSE stream", requestSnapshot,
responseEvent.responseInfo())); responseEvent.responseInfo()));
} }
else if (statusCode == METHOD_NOT_ALLOWED) { else if (statusCode == METHOD_NOT_ALLOWED) {
@ -417,7 +422,8 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
return Mono.deferContextual(ctx -> { return Mono.deferContextual(ctx -> {
var transportContext = ctx.getOrDefault(McpTransportContext.KEY, McpTransportContext.EMPTY); var transportContext = ctx.getOrDefault(McpTransportContext.KEY, McpTransportContext.EMPTY);
return Mono return Mono
.from(this.authorizationErrorHandler.handle(authException.getResponseInfo(), transportContext)) .from(this.authorizationErrorHandler.handle(authException.getRequestSnapshot(),
authException.getResponseInfo(), transportContext))
.switchIfEmpty(Mono.just(false)) .switchIfEmpty(Mono.just(false))
.flatMap(shouldRetry -> shouldRetry ? Mono.just(retrySignal.totalRetries()) .flatMap(shouldRetry -> shouldRetry ? Mono.just(retrySignal.totalRetries())
: Mono.error(retrySignal.failure())); : Mono.error(retrySignal.failure()));
@ -489,7 +495,6 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
return Mono return Mono
.from(this.httpRequestCustomizer.customize(builder, "POST", uri, jsonBody, transportContext)); .from(this.httpRequestCustomizer.customize(builder, "POST", uri, jsonBody, transportContext));
}).flatMapMany(requestBuilder -> Flux.<ResponseEvent>create(responseEventSink -> { }).flatMapMany(requestBuilder -> Flux.<ResponseEvent>create(responseEventSink -> {
// Create the async request with proper body subscriber selection // Create the async request with proper body subscriber selection
Mono.fromFuture(this.httpClient Mono.fromFuture(this.httpClient
.sendAsync(requestBuilder.build(), this.toSendMessageBodySubscriber(responseEventSink)) .sendAsync(requestBuilder.build(), this.toSendMessageBodySubscriber(responseEventSink))
@ -502,12 +507,14 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
} }
})).onErrorMap(CompletionException.class, t -> t.getCause()).onErrorComplete().subscribe(); })).onErrorMap(CompletionException.class, t -> t.getCause()).onErrorComplete().subscribe();
})).flatMap(responseEvent -> { }).flatMap(responseEvent -> {
int statusCode = responseEvent.responseInfo().statusCode(); int statusCode = responseEvent.responseInfo().statusCode();
if (statusCode == 401 || statusCode == 403) { if (statusCode == 401 || statusCode == 403) {
var request = requestBuilder.build();
var requestSnapshot = new HttpRequestSnapshot(request.uri(), request.method(), request.headers());
logger.debug("Authorization error in sendMessage with code {}", statusCode); logger.debug("Authorization error in sendMessage with code {}", statusCode);
return Mono.<McpSchema.JSONRPCMessage>error(new McpHttpClientTransportAuthorizationException( return Mono.<McpSchema.JSONRPCMessage>error(new McpHttpClientTransportAuthorizationException(
"Authorization error when sending message", responseEvent.responseInfo())); "Authorization error when sending message", requestSnapshot, responseEvent.responseInfo()));
} }
if (transportSession.markInitialized( if (transportSession.markInitialized(
@ -651,13 +658,12 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
if (ref != null) { if (ref != null) {
transportSession.removeConnection(ref); transportSession.removeConnection(ref);
} }
}) })).contextWrite(deliveredSink.contextView()).subscribe();
.contextWrite(deliveredSink.contextView())
.subscribe();
disposableRef.set(connection); disposableRef.set(connection);
transportSession.addConnection(connection); transportSession.addConnection(connection);
}); });
} }
private static String sessionIdOrPlaceholder(McpTransportSession<?> transportSession) { private static String sessionIdOrPlaceholder(McpTransportSession<?> transportSession) {
@ -695,7 +701,7 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
private List<String> supportedProtocolVersions = List.of(ProtocolVersions.MCP_2024_11_05, private List<String> supportedProtocolVersions = List.of(ProtocolVersions.MCP_2024_11_05,
ProtocolVersions.MCP_2025_03_26, ProtocolVersions.MCP_2025_06_18, ProtocolVersions.MCP_2025_11_25); ProtocolVersions.MCP_2025_03_26, ProtocolVersions.MCP_2025_06_18, ProtocolVersions.MCP_2025_11_25);
private McpHttpClientAuthorizationErrorHandler authorizationErrorHandler = McpHttpClientAuthorizationErrorHandler.NOOP; private McpHttpClientTransportAuthorizationErrorHandler authorizationErrorHandler = McpHttpClientTransportAuthorizationErrorHandler.NOOP;
/** /**
* Creates a new builder with the specified base URI. * Creates a new builder with the specified base URI.
@ -828,8 +834,34 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
* when sending a message. * when sending a message.
* @param authorizationErrorHandler the handler * @param authorizationErrorHandler the handler
* @return this builder * @return this builder
* @deprecated in favor of
* {@link #authorizationErrorHandler(McpHttpClientTransportAuthorizationErrorHandler)}
*/ */
@Deprecated(forRemoval = true, since = "2.0.0")
public Builder authorizationErrorHandler(McpHttpClientAuthorizationErrorHandler authorizationErrorHandler) { public Builder authorizationErrorHandler(McpHttpClientAuthorizationErrorHandler authorizationErrorHandler) {
this.authorizationErrorHandler = new McpHttpClientTransportAuthorizationErrorHandler() {
@Override
public Publisher<Boolean> handle(HttpRequestSnapshot requestSnapshot,
HttpResponse.ResponseInfo responseInfo, McpTransportContext context) {
return authorizationErrorHandler.handle(responseInfo, context);
}
@Override
public int maxRetries() {
return authorizationErrorHandler.maxRetries();
}
};
return this;
}
/**
* Sets the handler to be used when the server responds with HTTP 401 or HTTP 403
* when sending a message.
* @param authorizationErrorHandler the handler
* @return this builder
*/
public Builder authorizationErrorHandler(
McpHttpClientTransportAuthorizationErrorHandler authorizationErrorHandler) {
this.authorizationErrorHandler = authorizationErrorHandler; this.authorizationErrorHandler = authorizationErrorHandler;
return this; return this;
} }

View file

@ -0,0 +1,23 @@
/*
* Copyright 2026-2026 the original author or authors.
*/
package io.modelcontextprotocol.client.transport;
import java.net.URI;
import java.net.http.HttpHeaders;
import java.net.http.HttpRequest;
import java.net.http.HttpRequest.BodyPublisher;
/**
* Captures information about an HTTP request. We use this instead of passing the plain
* {@link HttpRequest} object because we want to avoid retaining a reference to the
* request's {@link BodyPublisher}.
*
* @param requestUri the HTTP request URI
* @param method the HTTP method
* @param headers the HTTP request headers
* @author Daniel Garnier-Moiroux
*/
public record HttpRequestSnapshot(URI requestUri, String method, HttpHeaders headers) {
}

View file

@ -19,13 +19,21 @@ public class McpHttpClientTransportAuthorizationException extends McpTransportEx
private final HttpResponse.ResponseInfo responseInfo; private final HttpResponse.ResponseInfo responseInfo;
public McpHttpClientTransportAuthorizationException(String message, HttpResponse.ResponseInfo responseInfo) { private final HttpRequestSnapshot requestSnapshot;
public McpHttpClientTransportAuthorizationException(String message, HttpRequestSnapshot requestSnapshot,
HttpResponse.ResponseInfo responseInfo) {
super(message); super(message);
this.responseInfo = responseInfo; this.responseInfo = responseInfo;
this.requestSnapshot = requestSnapshot;
} }
public HttpResponse.ResponseInfo getResponseInfo() { public HttpResponse.ResponseInfo getResponseInfo() {
return responseInfo; return responseInfo;
} }
public HttpRequestSnapshot getRequestSnapshot() {
return requestSnapshot;
}
} }

View file

@ -6,6 +6,7 @@ package io.modelcontextprotocol.client.transport.customizer;
import java.net.http.HttpResponse; import java.net.http.HttpResponse;
import io.modelcontextprotocol.client.transport.HttpRequestSnapshot;
import io.modelcontextprotocol.client.transport.McpHttpClientTransportAuthorizationException; import io.modelcontextprotocol.client.transport.McpHttpClientTransportAuthorizationException;
import io.modelcontextprotocol.common.McpTransportContext; import io.modelcontextprotocol.common.McpTransportContext;
import org.reactivestreams.Publisher; import org.reactivestreams.Publisher;
@ -20,7 +21,9 @@ import reactor.core.scheduler.Schedulers;
* "https://modelcontextprotocol.io/specification/2025-11-25/basic/authorization">MCP * "https://modelcontextprotocol.io/specification/2025-11-25/basic/authorization">MCP
* Specification: Authorization</a> * Specification: Authorization</a>
* @author Daniel Garnier-Moiroux * @author Daniel Garnier-Moiroux
* @deprecated in favor of {@link McpHttpClientTransportAuthorizationErrorHandler}
*/ */
@Deprecated(forRemoval = true, since = "2.0.0")
public interface McpHttpClientAuthorizationErrorHandler { public interface McpHttpClientAuthorizationErrorHandler {
/** /**
@ -38,7 +41,10 @@ public interface McpHttpClientAuthorizationErrorHandler {
* @param context the MCP client transport context * @param context the MCP client transport context
* @return {@link Publisher} emitting true if the original request should be replayed, * @return {@link Publisher} emitting true if the original request should be replayed,
* false otherwise. * false otherwise.
* @deprecated in favor of
* {@link McpHttpClientTransportAuthorizationErrorHandler#handle(HttpRequestSnapshot, HttpResponse.ResponseInfo, McpTransportContext)}
*/ */
@Deprecated(forRemoval = true, since = "2.0.0")
Publisher<Boolean> handle(HttpResponse.ResponseInfo responseInfo, McpTransportContext context); Publisher<Boolean> handle(HttpResponse.ResponseInfo responseInfo, McpTransportContext context);
/** /**
@ -87,7 +93,10 @@ public interface McpHttpClientAuthorizationErrorHandler {
* @param responseInfo the HTTP response information * @param responseInfo the HTTP response information
* @param context the MCP client transport context * @param context the MCP client transport context
* @return true if the original request should be replayed, false otherwise. * @return true if the original request should be replayed, false otherwise.
* @deprecated in favor of
* {@link McpHttpClientTransportAuthorizationErrorHandler.Sync#handle(HttpRequestSnapshot, HttpResponse.ResponseInfo, McpTransportContext)}
*/ */
@Deprecated(forRemoval = true, since = "2.0.0")
boolean handle(HttpResponse.ResponseInfo responseInfo, McpTransportContext context); boolean handle(HttpResponse.ResponseInfo responseInfo, McpTransportContext context);
} }

View file

@ -0,0 +1,110 @@
/*
* Copyright 2026-2026 the original author or authors.
*/
package io.modelcontextprotocol.client.transport.customizer;
import java.net.http.HttpResponse;
import io.modelcontextprotocol.client.transport.HttpRequestSnapshot;
import io.modelcontextprotocol.client.transport.McpHttpClientTransportAuthorizationException;
import io.modelcontextprotocol.common.McpTransportContext;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
/**
* Handle security-related errors in HTTP-client based transports. This class handles MCP
* server responses with status code 401 and 403.
*
* @see <a href=
* "https://modelcontextprotocol.io/specification/2025-11-25/basic/authorization">MCP
* Specification: Authorization</a>
* @author Daniel Garnier-Moiroux
*/
public interface McpHttpClientTransportAuthorizationErrorHandler {
/**
* Handle authorization error (HTTP 401 or 403), and signal whether the HTTP request
* should be retried or not. If the publisher returns true, the original transport
* method (connect, sendMessage) will be replayed with the original arguments.
* Otherwise, the transport will throw an
* {@link McpHttpClientTransportAuthorizationException}, indicating the error status.
* <p>
* If the returned {@link Publisher} errors, the error will be propagated to the
* calling method, to be handled by the caller.
* <p>
* The number of retries is bounded by {@link #maxRetries()}.
* @param requestSnapshot the HTTP request snapshot that failed authorization
* @param responseInfo the HTTP response information
* @param context the MCP client transport context
* @return {@link Publisher} emitting true if the original request should be replayed,
* false otherwise.
*/
Publisher<Boolean> handle(HttpRequestSnapshot requestSnapshot, HttpResponse.ResponseInfo responseInfo,
McpTransportContext context);
/**
* Maximum number of authorization error retries the transport will attempt. When the
* handler signals a retry via {@link #handle}, the transport will replay the original
* request at most this many times. If the authorization error persists after
* exhausting all retries, the transport will propagate the
* {@link McpHttpClientTransportAuthorizationException}.
* <p>
* Defaults to {@code 1}.
* @return the maximum number of retries
*/
default int maxRetries() {
return 1;
}
/**
* A no-op handler, used in the default use-case.
*/
McpHttpClientTransportAuthorizationErrorHandler NOOP = new Noop();
/**
* Create a {@link McpHttpClientTransportAuthorizationErrorHandler} from a synchronous
* handler. Will be subscribed on {@link Schedulers#boundedElastic()}. The handler may
* be blocking.
* @param handler the synchronous handler
* @return an async handler
*/
static McpHttpClientTransportAuthorizationErrorHandler fromSync(Sync handler) {
return (snapshot, info, context) -> Mono.fromCallable(() -> handler.handle(snapshot, info, context))
.subscribeOn(Schedulers.boundedElastic());
}
/**
* Synchronous authorization error handler.
*/
interface Sync {
/**
* Handle authorization error (HTTP 401 or 403), and signal whether the HTTP
* request should be retried or not. If the return value is true, the original
* transport method (connect, sendMessage) will be replayed with the original
* arguments. Otherwise, the transport will throw an
* {@link McpHttpClientTransportAuthorizationException}, indicating the error
* status.
* @param requestSnapshot the HTTP request snapshot that failed authorization
* @param responseInfo the HTTP response information
* @param context the MCP client transport context
* @return true if the original request should be replayed, false otherwise.
*/
boolean handle(HttpRequestSnapshot requestSnapshot, HttpResponse.ResponseInfo responseInfo,
McpTransportContext context);
}
class Noop implements McpHttpClientTransportAuthorizationErrorHandler {
@Override
public Publisher<Boolean> handle(HttpRequestSnapshot requestSnapshot, HttpResponse.ResponseInfo responseInfo,
McpTransportContext context) {
return Mono.just(false);
}
}
}

View file

@ -13,7 +13,9 @@ import static org.mockito.Mockito.mock;
/** /**
* @author Daniel Garnier-Moiroux * @author Daniel Garnier-Moiroux
* @deprecated use {@link McpHttpClientTransportAuthorizationErrorHandlerTest}
*/ */
@Deprecated
class McpHttpClientAuthorizationErrorHandlerTest { class McpHttpClientAuthorizationErrorHandlerTest {
private final HttpResponse.ResponseInfo responseInfo = mock(HttpResponse.ResponseInfo.class); private final HttpResponse.ResponseInfo responseInfo = mock(HttpResponse.ResponseInfo.class);
@ -21,21 +23,21 @@ class McpHttpClientAuthorizationErrorHandlerTest {
private final McpTransportContext context = McpTransportContext.EMPTY; private final McpTransportContext context = McpTransportContext.EMPTY;
@Test @Test
void whenTrueThenRetry() { void returnsTrue() {
McpHttpClientAuthorizationErrorHandler handler = McpHttpClientAuthorizationErrorHandler McpHttpClientAuthorizationErrorHandler handler = McpHttpClientAuthorizationErrorHandler
.fromSync((info, ctx) -> true); .fromSync((info, ctx) -> true);
StepVerifier.create(handler.handle(responseInfo, context)).expectNext(true).verifyComplete(); StepVerifier.create(handler.handle(responseInfo, context)).expectNext(true).verifyComplete();
} }
@Test @Test
void whenFalseThenError() { void returnsFalse() {
McpHttpClientAuthorizationErrorHandler handler = McpHttpClientAuthorizationErrorHandler McpHttpClientAuthorizationErrorHandler handler = McpHttpClientAuthorizationErrorHandler
.fromSync((info, ctx) -> false); .fromSync((info, ctx) -> false);
StepVerifier.create(handler.handle(responseInfo, context)).expectNext(false).verifyComplete(); StepVerifier.create(handler.handle(responseInfo, context)).expectNext(false).verifyComplete();
} }
@Test @Test
void whenExceptionThenPropagate() { void propragateExceptions() {
McpHttpClientAuthorizationErrorHandler handler = McpHttpClientAuthorizationErrorHandler McpHttpClientAuthorizationErrorHandler handler = McpHttpClientAuthorizationErrorHandler
.fromSync((info, ctx) -> { .fromSync((info, ctx) -> {
throw new IllegalStateException("sync handler error"); throw new IllegalStateException("sync handler error");

View file

@ -0,0 +1,53 @@
/*
* Copyright 2026-2026 the original author or authors.
*/
package io.modelcontextprotocol.client.transport.customizer;
import java.net.URI;
import java.net.http.HttpResponse;
import io.modelcontextprotocol.client.transport.HttpRequestSnapshot;
import io.modelcontextprotocol.common.McpTransportContext;
import org.junit.jupiter.api.Test;
import reactor.test.StepVerifier;
import static org.mockito.Mockito.mock;
/**
* @author Daniel Garnier-Moiroux
*/
class McpHttpClientTransportAuthorizationErrorHandlerTest {
private final HttpResponse.ResponseInfo responseInfo = mock(HttpResponse.ResponseInfo.class);
private final HttpRequestSnapshot requestSnapshot = new HttpRequestSnapshot(URI.create("http://localhost/mcp"),
"GET", java.net.http.HttpHeaders.of(java.util.Map.of(), (a, b) -> true));
private final McpTransportContext context = McpTransportContext.EMPTY;
@Test
void returnsTrue() {
McpHttpClientTransportAuthorizationErrorHandler handler = McpHttpClientTransportAuthorizationErrorHandler
.fromSync((snapshot, info, ctx) -> true);
StepVerifier.create(handler.handle(requestSnapshot, responseInfo, context)).expectNext(true).verifyComplete();
}
@Test
void returnsFalse() {
McpHttpClientTransportAuthorizationErrorHandler handler = McpHttpClientTransportAuthorizationErrorHandler
.fromSync((snapshot, info, ctx) -> false);
StepVerifier.create(handler.handle(requestSnapshot, responseInfo, context)).expectNext(false).verifyComplete();
}
@Test
void propagateExceptions() {
McpHttpClientTransportAuthorizationErrorHandler handler = McpHttpClientTransportAuthorizationErrorHandler
.fromSync((snapshot, info, ctx) -> {
throw new IllegalStateException("sync handler error");
});
StepVerifier.create(handler.handle(requestSnapshot, responseInfo, context))
.expectErrorMatches(t -> t instanceof IllegalStateException && t.getMessage().equals("sync handler error"))
.verify();
}
}

View file

@ -16,7 +16,7 @@ import java.util.function.Consumer;
import java.util.function.Predicate; 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.McpHttpClientTransportAuthorizationErrorHandler;
import io.modelcontextprotocol.common.McpTransportContext; import io.modelcontextprotocol.common.McpTransportContext;
import io.modelcontextprotocol.server.transport.TomcatTestUtil; import io.modelcontextprotocol.server.transport.TomcatTestUtil;
import io.modelcontextprotocol.spec.HttpHeaders; import io.modelcontextprotocol.spec.HttpHeaders;
@ -403,9 +403,12 @@ public class HttpClientStreamableHttpTransportErrorHandlingTest {
AtomicReference<HttpResponse.ResponseInfo> capturedResponseInfo = new AtomicReference<>(); AtomicReference<HttpResponse.ResponseInfo> capturedResponseInfo = new AtomicReference<>();
AtomicReference<McpTransportContext> capturedContext = new AtomicReference<>(); AtomicReference<McpTransportContext> capturedContext = new AtomicReference<>();
AtomicReference<HttpRequestSnapshot> capturedSnapshot = new AtomicReference<>();
var authTransport = HttpClientStreamableHttpTransport.builder(HOST) var authTransport = HttpClientStreamableHttpTransport.builder(HOST)
.authorizationErrorHandler((responseInfo, context) -> { .authorizationErrorHandler((requestSnapshot, responseInfo, context) -> {
capturedResponseInfo.set(responseInfo); capturedResponseInfo.set(responseInfo);
capturedSnapshot.set(requestSnapshot);
capturedContext.set(context); capturedContext.set(context);
return Mono.just(false); return Mono.just(false);
}) })
@ -417,6 +420,8 @@ public class HttpClientStreamableHttpTransportErrorHandlingTest {
assertThat(processedMessagesCount.get()).isEqualTo(1); assertThat(processedMessagesCount.get()).isEqualTo(1);
assertThat(capturedResponseInfo.get()).isNotNull(); assertThat(capturedResponseInfo.get()).isNotNull();
assertThat(capturedResponseInfo.get().statusCode()).isEqualTo(httpStatus); assertThat(capturedResponseInfo.get().statusCode()).isEqualTo(httpStatus);
assertThat(capturedSnapshot.get()).isNotNull();
assertThat(capturedSnapshot.get().requestUri().toString()).isEqualTo(HOST + "/mcp");
assertThat(capturedContext.get()).isNotNull(); assertThat(capturedContext.get()).isNotNull();
StepVerifier.create(authTransport.closeGracefully()).verifyComplete(); StepVerifier.create(authTransport.closeGracefully()).verifyComplete();
@ -440,7 +445,7 @@ public class HttpClientStreamableHttpTransportErrorHandlingTest {
void retry() { void retry() {
serverResponseStatus.set(401); serverResponseStatus.set(401);
var authTransport = HttpClientStreamableHttpTransport.builder(HOST) var authTransport = HttpClientStreamableHttpTransport.builder(HOST)
.authorizationErrorHandler((responseInfo, context) -> { .authorizationErrorHandler((requestSnapshot, responseInfo, context) -> {
serverResponseStatus.set(200); serverResponseStatus.set(200);
return Mono.just(true); return Mono.just(true);
}) })
@ -456,7 +461,7 @@ public class HttpClientStreamableHttpTransportErrorHandlingTest {
void retryAtMostOnce() { void retryAtMostOnce() {
serverResponseStatus.set(401); serverResponseStatus.set(401);
var authTransport = HttpClientStreamableHttpTransport.builder(HOST) var authTransport = HttpClientStreamableHttpTransport.builder(HOST)
.authorizationErrorHandler((responseInfo, context) -> Mono.just(true)) .authorizationErrorHandler((requestSnapshot, responseInfo, context) -> Mono.just(true))
.build(); .build();
StepVerifier.create(authTransport.sendMessage(createTestRequestMessage())) StepVerifier.create(authTransport.sendMessage(createTestRequestMessage()))
.expectErrorMatches(authorizationError(401)) .expectErrorMatches(authorizationError(401))
@ -471,10 +476,10 @@ public class HttpClientStreamableHttpTransportErrorHandlingTest {
void customMaxRetries() { void customMaxRetries() {
serverResponseStatus.set(401); serverResponseStatus.set(401);
var authTransport = HttpClientStreamableHttpTransport.builder(HOST) var authTransport = HttpClientStreamableHttpTransport.builder(HOST)
.authorizationErrorHandler(new McpHttpClientAuthorizationErrorHandler() { .authorizationErrorHandler(new McpHttpClientTransportAuthorizationErrorHandler() {
@Override @Override
public Publisher<Boolean> handle(HttpResponse.ResponseInfo responseInfo, public Publisher<Boolean> handle(HttpRequestSnapshot requestSnapshot,
McpTransportContext context) { HttpResponse.ResponseInfo responseInfo, McpTransportContext context) {
return Mono.just(true); return Mono.just(true);
} }
@ -498,7 +503,7 @@ public class HttpClientStreamableHttpTransportErrorHandlingTest {
serverResponseStatus.set(401); serverResponseStatus.set(401);
var authTransport = HttpClientStreamableHttpTransport.builder(HOST) var authTransport = HttpClientStreamableHttpTransport.builder(HOST)
.authorizationErrorHandler((responseInfo, context) -> Mono.just(false)) .authorizationErrorHandler((requestSnapshot, responseInfo, context) -> Mono.just(false))
.build(); .build();
StepVerifier.create(authTransport.sendMessage(createTestRequestMessage())) StepVerifier.create(authTransport.sendMessage(createTestRequestMessage()))
@ -513,8 +518,8 @@ public class HttpClientStreamableHttpTransportErrorHandlingTest {
void propagateHandlerError() { void propagateHandlerError() {
serverResponseStatus.set(401); serverResponseStatus.set(401);
var authTransport = HttpClientStreamableHttpTransport.builder(HOST) var authTransport = HttpClientStreamableHttpTransport.builder(HOST)
.authorizationErrorHandler( .authorizationErrorHandler((requestUri, responseInfo, context) -> Mono
(responseInfo, context) -> Mono.error(new IllegalStateException("handler error"))) .error(new IllegalStateException("handler error")))
.build(); .build();
StepVerifier.create(authTransport.sendMessage(createTestRequestMessage())) StepVerifier.create(authTransport.sendMessage(createTestRequestMessage()))
@ -529,7 +534,7 @@ public class HttpClientStreamableHttpTransportErrorHandlingTest {
void emptyHandler() { void emptyHandler() {
serverResponseStatus.set(401); serverResponseStatus.set(401);
var authTransport = HttpClientStreamableHttpTransport.builder(HOST) var authTransport = HttpClientStreamableHttpTransport.builder(HOST)
.authorizationErrorHandler((responseInfo, context) -> Mono.empty()) .authorizationErrorHandler((requestSnapshot, responseInfo, context) -> Mono.empty())
.build(); .build();
StepVerifier.create(authTransport.sendMessage(createTestRequestMessage())) StepVerifier.create(authTransport.sendMessage(createTestRequestMessage()))
@ -552,11 +557,13 @@ public class HttpClientStreamableHttpTransportErrorHandlingTest {
AtomicReference<Throwable> capturedException = new AtomicReference<>(); AtomicReference<Throwable> capturedException = new AtomicReference<>();
AtomicReference<HttpResponse.ResponseInfo> capturedResponseInfo = new AtomicReference<>(); AtomicReference<HttpResponse.ResponseInfo> capturedResponseInfo = new AtomicReference<>();
AtomicReference<HttpRequestSnapshot> capturedSnapshot = new AtomicReference<>();
AtomicReference<McpTransportContext> capturedContext = new AtomicReference<>(); AtomicReference<McpTransportContext> capturedContext = new AtomicReference<>();
var authTransport = HttpClientStreamableHttpTransport.builder(HOST) var authTransport = HttpClientStreamableHttpTransport.builder(HOST)
.authorizationErrorHandler((responseInfo, context) -> { .authorizationErrorHandler((requestSnapshot, responseInfo, context) -> {
capturedResponseInfo.set(responseInfo); capturedResponseInfo.set(responseInfo);
capturedSnapshot.set(requestSnapshot);
capturedContext.set(context); capturedContext.set(context);
return Mono.just(false); return Mono.just(false);
}) })
@ -572,6 +579,8 @@ public class HttpClientStreamableHttpTransportErrorHandlingTest {
assertThat(messages).isEmpty(); assertThat(messages).isEmpty();
assertThat(capturedResponseInfo.get()).isNotNull(); assertThat(capturedResponseInfo.get()).isNotNull();
assertThat(capturedResponseInfo.get().statusCode()).isEqualTo(httpStatus); assertThat(capturedResponseInfo.get().statusCode()).isEqualTo(httpStatus);
assertThat(capturedSnapshot.get()).isNotNull();
assertThat(capturedSnapshot.get().requestUri().toString()).isEqualTo(HOST + "/mcp");
assertThat(capturedContext.get()).isNotNull(); assertThat(capturedContext.get()).isNotNull();
assertThat(capturedException.get()).hasMessage("Authorization error connecting to SSE stream") assertThat(capturedException.get()).hasMessage("Authorization error connecting to SSE stream")
.asInstanceOf(type(McpHttpClientTransportAuthorizationException.class)) .asInstanceOf(type(McpHttpClientTransportAuthorizationException.class))
@ -606,7 +615,7 @@ public class HttpClientStreamableHttpTransportErrorHandlingTest {
AtomicReference<Throwable> capturedException = new AtomicReference<>(); AtomicReference<Throwable> capturedException = new AtomicReference<>();
var authTransport = HttpClientStreamableHttpTransport.builder(HOST) var authTransport = HttpClientStreamableHttpTransport.builder(HOST)
.openConnectionOnStartup(true) .openConnectionOnStartup(true)
.authorizationErrorHandler((responseInfo, context) -> { .authorizationErrorHandler((requestSnapshot, responseInfo, context) -> {
serverSseResponseStatus.set(200); serverSseResponseStatus.set(200);
return Mono.just(true); return Mono.just(true);
}) })
@ -636,7 +645,7 @@ public class HttpClientStreamableHttpTransportErrorHandlingTest {
AtomicReference<Throwable> capturedException = new AtomicReference<>(); AtomicReference<Throwable> capturedException = new AtomicReference<>();
var authTransport = HttpClientStreamableHttpTransport.builder(HOST) var authTransport = HttpClientStreamableHttpTransport.builder(HOST)
.openConnectionOnStartup(true) .openConnectionOnStartup(true)
.authorizationErrorHandler((responseInfo, context) -> { .authorizationErrorHandler((requestSnapshot, responseInfo, context) -> {
return Mono.just(true); return Mono.just(true);
}) })
.build(); .build();
@ -661,10 +670,10 @@ public class HttpClientStreamableHttpTransportErrorHandlingTest {
AtomicReference<Throwable> capturedException = new AtomicReference<>(); AtomicReference<Throwable> capturedException = new AtomicReference<>();
var authTransport = HttpClientStreamableHttpTransport.builder(HOST) var authTransport = HttpClientStreamableHttpTransport.builder(HOST)
.openConnectionOnStartup(true) .openConnectionOnStartup(true)
.authorizationErrorHandler(new McpHttpClientAuthorizationErrorHandler() { .authorizationErrorHandler(new McpHttpClientTransportAuthorizationErrorHandler() {
@Override @Override
public Publisher<Boolean> handle(HttpResponse.ResponseInfo responseInfo, public Publisher<Boolean> handle(HttpRequestSnapshot requestSnapshot,
McpTransportContext context) { HttpResponse.ResponseInfo responseInfo, McpTransportContext context) {
return Mono.just(true); return Mono.just(true);
} }
@ -695,7 +704,7 @@ public class HttpClientStreamableHttpTransportErrorHandlingTest {
AtomicReference<Throwable> capturedException = new AtomicReference<>(); AtomicReference<Throwable> capturedException = new AtomicReference<>();
var authTransport = HttpClientStreamableHttpTransport.builder(HOST) var authTransport = HttpClientStreamableHttpTransport.builder(HOST)
.openConnectionOnStartup(true) .openConnectionOnStartup(true)
.authorizationErrorHandler((responseInfo, context) -> { .authorizationErrorHandler((requestUri, responseInfo, context) -> {
// if there was a retry, the request would succeed. // if there was a retry, the request would succeed.
serverSseResponseStatus.set(200); serverSseResponseStatus.set(200);
return Mono.just(false); return Mono.just(false);
@ -720,7 +729,7 @@ public class HttpClientStreamableHttpTransportErrorHandlingTest {
AtomicReference<Throwable> capturedException = new AtomicReference<>(); AtomicReference<Throwable> capturedException = new AtomicReference<>();
var authTransport = HttpClientStreamableHttpTransport.builder(HOST) var authTransport = HttpClientStreamableHttpTransport.builder(HOST)
.openConnectionOnStartup(true) .openConnectionOnStartup(true)
.authorizationErrorHandler((responseInfo, context) -> Mono.empty()) .authorizationErrorHandler((requestUri, responseInfo, context) -> Mono.empty())
.build(); .build();
authTransport.setExceptionHandler(capturedException::set); authTransport.setExceptionHandler(capturedException::set);
@ -741,8 +750,8 @@ public class HttpClientStreamableHttpTransportErrorHandlingTest {
AtomicReference<Throwable> capturedException = new AtomicReference<>(); AtomicReference<Throwable> capturedException = new AtomicReference<>();
var authTransport = HttpClientStreamableHttpTransport.builder(HOST) var authTransport = HttpClientStreamableHttpTransport.builder(HOST)
.openConnectionOnStartup(true) .openConnectionOnStartup(true)
.authorizationErrorHandler( .authorizationErrorHandler((requestUri, responseInfo, context) -> Mono
(responseInfo, context) -> Mono.error(new IllegalStateException("handler error"))) .error(new IllegalStateException("handler error")))
.build(); .build();
authTransport.setExceptionHandler(capturedException::set); authTransport.setExceptionHandler(capturedException::set);