Handle 401 Unauthorized on WebFluxSseClientTransport with client.initialize

[jaketothepast]

Please do a quick search on GitHub issues first, the feature you are about to request might have already been requested.

Expected Behavior

When calling client.initialize(), I'd like to know if a 401 happened. That way, I can initiate an oauth login for the current user.

Current Behavior

In the current code, 401 is captured as a fatal sse error, with no way to attach an error callback.

	/**
	 * Initializes and starts the inbound SSE event processing. Establishes the SSE
	 * connection and sets up event handling for both message and endpoint events.
	 * Includes automatic retry logic for handling transient connection failures.
	 */
	// visible for tests
	protected Flux<ServerSentEvent<String>> eventStream() {// @formatter:off
		return this.webClient
			.get()
			.uri(this.sseEndpoint)
			.accept(MediaType.TEXT_EVENT_STREAM)
			.retrieve()
			.bodyToFlux(SSE_TYPE)
			.retryWhen(Retry.from(retrySignal -> retrySignal.handle(inboundRetryHandler)));
	} // @formatter:on

	/**
	 * Retry handler for the inbound SSE stream. Implements the retry logic for handling
	 * connection failures and other errors.
	 */
	private BiConsumer<RetrySignal, SynchronousSink<Object>> inboundRetryHandler = (retrySpec, sink) -> {
		if (isClosing) {
			logger.debug("SSE connection closed during shutdown");
			sink.error(retrySpec.failure());
			return;
		}
		if (retrySpec.failure() instanceof IOException) {
			logger.debug("Retrying SSE connection after IO error");
			sink.next(retrySpec);
			return;
		}
		logger.error("Fatal SSE error, not retrying: {}", retrySpec.failure().getMessage());
		sink.error(retrySpec.failure());
	};

I want to capture the error, so that I can initiate the oauth flow to log the user in.

Context

I am building a way for users of our app to connect to MCP servers, and some servers (most really) will be protected via oauth

[jaketothepast]

Just as a hack to get around this, I added errorSink to the Client specs, so that I could report on an error within initialization and block on it in my code.