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'); });