From e5f42d6b1d7f550fd081905db5803c34e45d1c8e Mon Sep 17 00:00:00 2001 From: Felix Weinberger <3823880+felixweinberger@users.noreply.github.com> Date: Wed, 30 Sep 2026 13:33:29 +0000 Subject: [PATCH] fix(client): retry the SSE connection once after onUnauthorized SSEClientTransport reconnected after every 401 for as long as onUnauthorized() kept resolving, so start() never settled against a server that always answers 401. It now retries once, as documented and as the POST path and the Streamable HTTP transport do, and then rejects with SdkHttpError. The guard resets when the stream opens, so a 401 on a later reconnect still gets one refresh. --- .changeset/sse-connect-retry-once.md | 5 ++ packages/client/src/client/sse.ts | 19 +++++- packages/client/test/client/sse.test.ts | 82 +++++++++++++++++++++---- 3 files changed, 92 insertions(+), 14 deletions(-) create mode 100644 .changeset/sse-connect-retry-once.md diff --git a/.changeset/sse-connect-retry-once.md b/.changeset/sse-connect-retry-once.md new file mode 100644 index 0000000000..cb90c3aa39 --- /dev/null +++ b/.changeset/sse-connect-retry-once.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/client': patch +--- + +`SSEClientTransport` now retries the SSE connection once after `onUnauthorized()` resolves, as documented. If the retry is also answered with 401, `start()` rejects with `SdkHttpError` (`ClientHttpAuthentication`) instead of calling `onUnauthorized()` again. A 401 on a later reconnect of a stream that had opened still gets one refresh. diff --git a/packages/client/src/client/sse.ts b/packages/client/src/client/sse.ts index e0b320b107..e249636ed2 100644 --- a/packages/client/src/client/sse.ts +++ b/packages/client/src/client/sse.ts @@ -175,6 +175,8 @@ export class SSEClientTransport implements Transport { } private _last401Response?: Response; + // True between a 401-triggered reconnect and the next successful open. + private _connectAuthRetried = false; private async _commonHeaders(): Promise { // Start from the caller-supplied `requestInit.headers` and `set()` the @@ -239,9 +241,10 @@ export class SSEClientTransport implements Transport { this._eventSource.onerror = event => { if (event.code === 401 && this._authProvider) { - if (this._authProvider.onUnauthorized && this._last401Response) { + if (this._authProvider.onUnauthorized && this._last401Response && !this._connectAuthRetried) { const response = this._last401Response; this._last401Response = undefined; + this._connectAuthRetried = true; this._eventSource?.close(); this._authProvider.onUnauthorized({ response, serverUrl: this._url, fetchFn: this._fetchWithInit }).then( // onUnauthorized succeeded → retry fresh. Its onerror handles its own onerror?.() + reject. @@ -250,6 +253,7 @@ export class SSEClientTransport implements Transport { // stamp: covers the SDK's OAuth flow and custom // callbacks alike. (error: unknown) => { + this._connectAuthRetried = false; markAuthSeamEscape(error); this.onerror?.(error as Error); reject(error); @@ -257,7 +261,17 @@ export class SSEClientTransport implements Transport { ); return; } - const error = markAuthSeamEscape(new UnauthorizedError()); + const retried = this._connectAuthRetried; + this._connectAuthRetried = false; + const error = markAuthSeamEscape( + retried + ? new SdkHttpError(SdkErrorCode.ClientHttpAuthentication, 'Server returned 401 after re-authentication', { + status: 401, + statusText: this._last401Response?.statusText ?? '' + }) + : new UnauthorizedError() + ); + this._last401Response = undefined; reject(error); this.onerror?.(error); return; @@ -270,6 +284,7 @@ export class SSEClientTransport implements Transport { this._eventSource.onopen = () => { // The connection is open, but we need to wait for the endpoint to be received. + this._connectAuthRetried = false; }; this._eventSource.addEventListener('endpoint', (event: Event) => { diff --git a/packages/client/test/client/sse.test.ts b/packages/client/test/client/sse.test.ts index 6952af8f76..8b4a8edae8 100644 --- a/packages/client/test/client/sse.test.ts +++ b/packages/client/test/client/sse.test.ts @@ -1783,10 +1783,43 @@ describe('SSEClientTransport', () => { await expect(transport.finishAuth('auth-code')).rejects.toThrow('finishAuth requires an OAuthClientProvider'); }); - it('SSE connect 401 retry does not poison future 401s — onUnauthorized called on each attempt', async () => { + it('SSE connect: a second 401 right after a refresh rejects, onUnauthorized called once', async () => { + await resourceServer.close(); + + let getAttempt = 0; + resourceServer = createServer((req, res) => { + if (req.method === 'GET') { + getAttempt++; + res.writeHead(401).end(); // always 401 + } + }); + resourceBaseUrl = await listenOnRandomPort(resourceServer); + + // Backstop so an unbounded retry ends the test with a clear failure instead of a hang. + let calls = 0; + const authProvider: AuthProvider = { + token: vi.fn(async () => 'still-bad'), + onUnauthorized: vi.fn(async () => { + if (++calls >= 10) throw new Error('backstop: onUnauthorized called 10 times'); + }) + }; + transport = new SSEClientTransport(resourceBaseUrl, { authProvider }); + const onerror = vi.fn(); + transport.onerror = onerror; + + const error = await transport.start().catch(e => e); + expect(error).toBeInstanceOf(SdkHttpError); + expect((error as SdkHttpError).code).toBe(SdkErrorCode.ClientHttpAuthentication); + expect((error as SdkHttpError).status).toBe(401); + expect(authProvider.onUnauthorized).toHaveBeenCalledTimes(1); + expect(getAttempt).toBe(2); + expect(onerror).toHaveBeenCalledTimes(1); + }); + + it('SSE connect 401 retry does not poison future 401s — a 401 on a later reconnect refreshes again', async () => { // Regression: _startOrAuth(true) baked isAuthRetry=true into the retry EventSource's // onerror closure, so a subsequent 401 (token expiry on reconnect) would throw - // instead of refreshing. Fix: retry always calls _startOrAuth() fresh. + // instead of refreshing. The retry guard resets once the stream opens. await resourceServer.close(); let getAttempt = 0; @@ -1796,7 +1829,8 @@ describe('SSEClientTransport', () => { return; } getAttempt++; - if (getAttempt < 3) { + // 1: 401, 2: opens then drops, 3: 401 on the automatic reconnect, 4: opens and stays. + if (getAttempt === 1 || getAttempt === 3) { res.writeHead(401).end(); return; } @@ -1805,8 +1839,12 @@ describe('SSEClientTransport', () => { 'Cache-Control': 'no-cache, no-transform', Connection: 'keep-alive' }); + res.write('retry: 10\n'); res.write('event: endpoint\n'); res.write(`data: ${resourceBaseUrl.href}post\n\n`); + if (getAttempt === 2) { + res.end(); + } }); resourceBaseUrl = await listenOnRandomPort(resourceServer); @@ -1816,17 +1854,17 @@ describe('SSEClientTransport', () => { }; transport = new SSEClientTransport(resourceBaseUrl, { authProvider }); - await transport.start(); // should resolve on attempt 3 + await transport.start(); // resolves on attempt 2 + expect(authProvider.onUnauthorized).toHaveBeenCalledTimes(1); + await vi.waitFor(() => expect(getAttempt).toBe(4), { timeout: 3000 }); expect(authProvider.onUnauthorized).toHaveBeenCalledTimes(2); - expect(getAttempt).toBe(3); }); it('retry failure during SSE connect fires onerror exactly once', async () => { // Regression: when the retry EventSource rejected, its onerror fired inside, then // the outer .then() rejection handler fired onerror AGAIN for the same error. // Fix: inner retry chains to .then(resolve, reject) — no outer onerror call. - // onUnauthorized's own failure is handled separately and fires onerror once. await resourceServer.close(); resourceServer = createServer((req, res) => { @@ -1836,20 +1874,40 @@ describe('SSEClientTransport', () => { }); resourceBaseUrl = await listenOnRandomPort(resourceServer); - const onUnauthorized: AuthProvider['onUnauthorized'] = vi - .fn() - .mockResolvedValueOnce(undefined) // first call succeeds → triggers retry - .mockRejectedValueOnce(new Error('refresh failed')); // second call (in retry) throws const authProvider: AuthProvider = { token: vi.fn(async () => 'token'), - onUnauthorized + onUnauthorized: vi.fn(async () => {}) + }; + transport = new SSEClientTransport(resourceBaseUrl, { authProvider }); + const onerror = vi.fn(); + transport.onerror = onerror; + + await expect(transport.start()).rejects.toThrow('Server returned 401 after re-authentication'); + expect(authProvider.onUnauthorized).toHaveBeenCalledTimes(1); + expect(onerror).toHaveBeenCalledTimes(1); + expect(onerror.mock.calls[0]![0].message).toBe('Server returned 401 after re-authentication'); + }); + + it('a failing onUnauthorized during SSE connect fires onerror exactly once', async () => { + await resourceServer.close(); + + resourceServer = createServer((req, res) => { + if (req.method === 'GET') { + res.writeHead(401).end(); // always 401 + } + }); + resourceBaseUrl = await listenOnRandomPort(resourceServer); + + const authProvider: AuthProvider = { + token: vi.fn(async () => 'token'), + onUnauthorized: vi.fn().mockRejectedValueOnce(new Error('refresh failed')) }; transport = new SSEClientTransport(resourceBaseUrl, { authProvider }); const onerror = vi.fn(); transport.onerror = onerror; await expect(transport.start()).rejects.toThrow('refresh failed'); - expect(authProvider.onUnauthorized).toHaveBeenCalledTimes(2); + expect(authProvider.onUnauthorized).toHaveBeenCalledTimes(1); expect(onerror).toHaveBeenCalledTimes(1); expect(onerror.mock.calls[0]![0].message).toBe('refresh failed'); });