HttpClientStreamHttpTransport: add authorization error handler

Author: Kehrlann

mcp-core/src/main/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransport.java

@@ -23,6 +23,7 @@
 import io.modelcontextprotocol.client.McpAsyncClient;
 import io.modelcontextprotocol.client.transport.ResponseSubscribers.ResponseEvent;
 import io.modelcontextprotocol.client.transport.customizer.McpAsyncHttpClientRequestCustomizer;
+import io.modelcontextprotocol.client.transport.customizer.McpHttpClientAuthorizationErrorHandler;
 import io.modelcontextprotocol.client.transport.customizer.McpSyncHttpClientRequestCustomizer;
 import io.modelcontextprotocol.common.McpTransportContext;
 import io.modelcontextprotocol.json.McpJsonDefaults;
@@ -115,6 +116,8 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
 
 	private final boolean openConnectionOnStartup;
 
+	private final McpHttpClientAuthorizationErrorHandler authorizationErrorHandler;
+
 	private final boolean resumableStreams;
 
 	private final McpAsyncHttpClientRequestCustomizer httpRequestCustomizer;
@@ -132,14 +135,15 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
 	private HttpClientStreamableHttpTransport(McpJsonMapper jsonMapper, HttpClient httpClient,
 			HttpRequest.Builder requestBuilder, String baseUri, String endpoint, boolean resumableStreams,
 			boolean openConnectionOnStartup, McpAsyncHttpClientRequestCustomizer httpRequestCustomizer,
-			List<String> supportedProtocolVersions) {
+			McpHttpClientAuthorizationErrorHandler authorizationErrorHandler, List<String> supportedProtocolVersions) {
 		this.jsonMapper = jsonMapper;
 		this.httpClient = httpClient;
 		this.requestBuilder = requestBuilder;
 		this.baseUri = URI.create(baseUri);
 		this.endpoint = endpoint;
 		this.resumableStreams = resumableStreams;
 		this.openConnectionOnStartup = openConnectionOnStartup;
+		this.authorizationErrorHandler = authorizationErrorHandler;
 		this.activeSession.set(createTransportSession());
 		this.httpRequestCustomizer = httpRequestCustomizer;
 		this.supportedProtocolVersions = Collections.unmodifiableList(supportedProtocolVersions);
@@ -478,6 +482,17 @@ public Mono<Void> sendMessage(McpSchema.JSONRPCMessage sentMessage) {
 					})).onErrorMap(CompletionException.class, t -> t.getCause()).onErrorComplete().subscribe();
 
 			})).flatMap(responseEvent -> {
+				int statusCode = responseEvent.responseInfo().statusCode();
+				if (statusCode == 401 || statusCode == 403) {
+					return Mono.deferContextual(ctx -> {
+						var transportContext = ctx.getOrDefault(McpTransportContext.KEY, McpTransportContext.EMPTY);
+						return Mono.from(this.authorizationErrorHandler.handle(responseEvent.responseInfo(),
+								transportContext, Mono.defer(() -> this.sendMessage(sentMessage))));
+					})
+						.then(Mono.error(new McpHttpClientTransportException("Authorization error when sending message",
+								responseEvent.responseInfo())));
+				}
+
 				if (transportSession.markInitialized(
 						responseEvent.responseInfo().headers().firstValue("mcp-session-id").orElseGet(() -> null))) {
 					// Once we have a session, we try to open an async stream for
@@ -488,8 +503,6 @@ public Mono<Void> sendMessage(McpSchema.JSONRPCMessage sentMessage) {
 
 				String sessionRepresentation = sessionIdOrPlaceholder(transportSession);
 
-				int statusCode = responseEvent.responseInfo().statusCode();
-
 				if (statusCode >= 200 && statusCode < 300) {
 
 					String contentType = responseEvent.responseInfo()
@@ -664,6 +677,8 @@ public static class Builder {
 		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);
 
+		private McpHttpClientAuthorizationErrorHandler authorizationErrorHandler = McpHttpClientAuthorizationErrorHandler.NOOP;
+
 		/**
 		 * Creates a new builder with the specified base URI.
 		 * @param baseUri the base URI of the MCP server
@@ -801,6 +816,16 @@ public Builder asyncHttpRequestCustomizer(McpAsyncHttpClientRequestCustomizer as
 			return this;
 		}
 
+		/**
+		 * Sets the handler
+		 * @param authorizationErrorHandler
+		 * @return
+		 */
+		public Builder authorizationErrorHandler(McpHttpClientAuthorizationErrorHandler authorizationErrorHandler) {
+			this.authorizationErrorHandler = authorizationErrorHandler;
+			return this;
+		}
+
 		/**
 		 * Sets the connection timeout for the HTTP client.
 		 * @param connectTimeout the connection timeout duration
@@ -845,7 +870,7 @@ public HttpClientStreamableHttpTransport build() {
 			HttpClient httpClient = this.clientBuilder.connectTimeout(this.connectTimeout).build();
 			return new HttpClientStreamableHttpTransport(jsonMapper == null ? McpJsonDefaults.getMapper() : jsonMapper,
 					httpClient, requestBuilder, baseUri, endpoint, resumableStreams, openConnectionOnStartup,
-					httpRequestCustomizer, supportedProtocolVersions);
+					httpRequestCustomizer, authorizationErrorHandler, supportedProtocolVersions);
 		}
 
 	}

mcp-core/src/main/java/io/modelcontextprotocol/client/transport/McpHttpClientTransportException.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2024-2025 the original author or authors.
+ */
+
+package io.modelcontextprotocol.client.transport;
+
+import java.net.http.HttpResponse;
+
+import io.modelcontextprotocol.spec.McpTransportException;
+
+/**
+ * Authorization-related exception for {@link java.net.http.HttpClient}-based client
+ * transport.
+ *
+ * @see <a href=
+ * "https://modelcontextprotocol.io/specification/2025-11-25/basic/authorization">MCP
+ * Specification: Authorization</a>
+ * @author Daniel Garnier-Moiroux
+ */
+public class McpHttpClientTransportException extends McpTransportException {
+
+	private final HttpResponse.ResponseInfo responseInfo;
+
+	public McpHttpClientTransportException(String message, HttpResponse.ResponseInfo responseInfo) {
+		super(message);
+		this.responseInfo = responseInfo;
+	}
+
+	public HttpResponse.ResponseInfo getResponseInfo() {
+		return responseInfo;
+	}
+
+}

mcp-core/src/main/java/io/modelcontextprotocol/client/transport/customizer/McpHttpClientAuthorizationErrorHandler.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2024-2025 the original author or authors.
+ */
+
+package io.modelcontextprotocol.client.transport.customizer;
+
+import java.net.http.HttpResponse;
+
+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 McpHttpClientAuthorizationErrorHandler {
+
+	/**
+	 * Handle HTTP error, and signal whether the HTTP request should be retried or not.
+	 * @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(HttpResponse.ResponseInfo responseInfo, McpTransportContext context);
+
+	McpHttpClientAuthorizationErrorHandler NOOP = new Noop();
+
+	/**
+	 * Handle HTTP error, and optionally retry the HTTP request.
+	 * @param responseInfo the HTTP response information
+	 * @param context the MCP client transport context
+	 * @param retryHandler the handler to use to retry the HTTP request.
+	 * @return a {@link Publisher} to signal either an error or a retry
+	 */
+	default Publisher<Void> handle(HttpResponse.ResponseInfo responseInfo, McpTransportContext context,
+			Publisher<Void> retryHandler) {
+		return Mono.from(this.handle(responseInfo, context))
+			.flatMap(shouldRetry -> shouldRetry != null && shouldRetry ? Mono.from(retryHandler) : Mono.empty());
+	}
+
+	/**
+	 * Create a {@link McpHttpClientAuthorizationErrorHandler} 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 McpHttpClientAuthorizationErrorHandler fromSync(Sync handler) {
+		return (info, context) -> {
+			try {
+				var shouldRetry = handler.handle(info, context);
+				return Mono.just(shouldRetry).subscribeOn(Schedulers.boundedElastic());
+			}
+			catch (Exception e) {
+				return Mono.error(e);
+			}
+		};
+	}
+
+	/**
+	 * Synchronous authorization error handler.
+	 */
+	interface Sync {
+
+		/**
+		 * Handle HTTP error, and signal whether the HTTP request should be retried or
+		 * not.
+		 * @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(HttpResponse.ResponseInfo responseInfo, McpTransportContext context);
+
+	}
+
+	class Noop implements McpHttpClientAuthorizationErrorHandler {
+
+		@Override
+		public Publisher<Boolean> handle(HttpResponse.ResponseInfo responseInfo, McpTransportContext context) {
+			return Mono.just(false);
+		}
+
+	}
+
+}