diff --git a/.changeset/single-onerror-standalone-get.md b/.changeset/single-onerror-standalone-get.md new file mode 100644 index 0000000000..9d6228ce6b --- /dev/null +++ b/.changeset/single-onerror-standalone-get.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/client': patch +--- + +`StreamableHTTPClientTransport` no longer reports the same standalone GET failure to `onerror` twice, including when `close()` aborts an in-flight stream. diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index ace0663158..cd0120168c 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -519,12 +519,6 @@ export class StreamableHTTPClientTransport implements Transport { private async _startOrAuthSse(options: StartSSEOptions, isAuthRetry = false, stepUpRetries = 0): Promise { const { resumptionToken, requestSignal } = options; - // Same guard as `_handleSseStream`: a resurrected listen stream (the - // POST-SSE → GET reconnect path threads `requestSignal` through - // `StartSSEOptions`) must honour the per-request abort exactly as the - // original POST did — both as a fetch signal and as a "do not surface - // onerror" gate. - const isIntentionalAbort = (): boolean => this._abortController?.signal.aborted === true || requestSignal?.aborted === true; try { // Try to open an initial SSE stream with GET to listen for server messages @@ -629,10 +623,7 @@ export class StreamableHTTPClientTransport implements Transport { this._handleSseStream(response.body, options, true); } catch (error) { - if (!isIntentionalAbort()) { - this.onerror?.(error as Error); - } - throw error; + throw error as Error; } } @@ -1242,9 +1233,14 @@ export class StreamableHTTPClientTransport implements Transport { * @param options Optional callback to receive new resumption tokens */ async resumeStream(lastEventId: string, options?: { onresumptiontoken?: (token: string) => void }): Promise { - await this._startOrAuthSse({ - resumptionToken: lastEventId, - onresumptiontoken: options?.onresumptiontoken - }); + try { + await this._startOrAuthSse({ + resumptionToken: lastEventId, + onresumptiontoken: options?.onresumptiontoken + }); + } catch (error) { + this.onerror?.(error as Error); + throw error; + } } } diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index a36bbc0ad3..6d6fd121a5 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -7,6 +7,23 @@ import { UnauthorizedError } from '../../src/client/auth'; import type { ReconnectionScheduler, StartSSEOptions, StreamableHTTPReconnectionOptions } from '../../src/client/streamableHttp'; import { StreamableHTTPClientTransport } from '../../src/client/streamableHttp'; +function captureTransportErrors(transport: StreamableHTTPClientTransport): { errors: Error[]; firstError: Promise } { + const errors: Error[] = []; + let resolveFirstError!: (error: Error) => void; + const firstError = new Promise(resolve => { + resolveFirstError = resolve; + }); + + transport.onerror = error => { + errors.push(error); + if (errors.length === 1) { + resolveFirstError(error); + } + }; + + return { errors, firstError }; +} + describe('StreamableHTTPClientTransport', () => { let transport: StreamableHTTPClientTransport; let mockAuthProvider: Mocked; @@ -467,6 +484,110 @@ describe('StreamableHTTPClientTransport', () => { ); }); + it('should fire onerror only once when the standalone GET stream fails to open', async () => { + const { errors, firstError } = captureTransportErrors(transport); + + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 202, + headers: new Headers() + }); + fetchMock.mockResolvedValueOnce({ + ok: false, + status: 500, + statusText: 'Internal Server Error', + text: () => Promise.resolve(''), + headers: new Headers() + }); + + await transport.start(); + await transport.send({ jsonrpc: '2.0', method: 'notifications/initialized', params: {} } as JSONRPCMessage); + + const error = await firstError; + await Promise.resolve(); + + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(errors).toHaveLength(1); + expect(error.message).toContain('Failed to open SSE stream'); + }); + + it('should fire onerror only once when close() aborts an in-flight standalone GET request', async () => { + const { errors, firstError } = captureTransportErrors(transport); + let markGetStarted!: () => void; + const getStarted = new Promise(resolve => { + markGetStarted = resolve; + }); + + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 202, + headers: new Headers() + }); + fetchMock.mockImplementationOnce( + (_url, init: RequestInit) => + new Promise((_resolve, reject) => { + markGetStarted(); + init.signal?.addEventListener('abort', () => reject(init.signal?.reason), { once: true }); + }) + ); + + await transport.start(); + await transport.send({ jsonrpc: '2.0', method: 'notifications/initialized', params: {} } as JSONRPCMessage); + + await getStarted; + expect(fetchMock).toHaveBeenCalledTimes(2); + + await transport.close(); + await firstError; + await Promise.resolve(); + + expect(errors).toHaveLength(1); + }); + + it('should fire onerror once and reject when resumeStream fails', async () => { + const { errors } = captureTransportErrors(transport); + + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockResolvedValueOnce({ + ok: false, + status: 500, + statusText: 'Internal Server Error', + text: () => Promise.resolve(''), + headers: new Headers() + }); + + await transport.start(); + await expect(transport.resumeStream('event-123')).rejects.toThrow('Failed to open SSE stream'); + + expect(errors).toHaveLength(1); + expect(errors[0]?.message).toContain('Failed to open SSE stream'); + }); + + it('should fire onerror only once when a resumption GET fails', async () => { + const { errors, firstError } = captureTransportErrors(transport); + + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockResolvedValueOnce({ + ok: false, + status: 500, + statusText: 'Internal Server Error', + text: () => Promise.resolve(''), + headers: new Headers() + }); + + await transport.start(); + await transport.send({ jsonrpc: '2.0', method: 'test', params: {}, id: 'request-1' }, { resumptionToken: 'event-123' }); + + const error = await firstError; + await Promise.resolve(); + + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(errors).toHaveLength(1); + expect(error.message).toContain('Failed to open SSE stream'); + }); + it('should handle multiple concurrent SSE streams', async () => { // Mock two POST requests that return SSE streams const makeStream = (id: string) => {