From 0ace7f697ff1eef0c140f8002701e0f8f2daff6b Mon Sep 17 00:00:00 2001 From: Pradeep Ramola Date: Sun, 19 Jul 2026 02:02:32 -0400 Subject: [PATCH 1/3] Fix streamable HTTP SSE exhaustion --- packages/client/src/client/streamableHttp.ts | 63 +++++--- .../client/test/client/streamableHttp.test.ts | 146 ++++++++++++++++++ 2 files changed, 191 insertions(+), 18 deletions(-) diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index 9067fd1ec4..7d881a7f28 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -43,7 +43,7 @@ const DEFAULT_STREAMABLE_HTTP_RECONNECTION_OPTIONS: StreamableHTTPReconnectionOp initialReconnectionDelay: 1000, maxReconnectionDelay: 30_000, reconnectionDelayGrowFactor: 1.5, - maxRetries: 2 + maxRetries: 10 }; /** @@ -113,11 +113,20 @@ export interface StreamableHTTPReconnectionOptions { /** * Maximum number of reconnection attempts before giving up. - * Default is 2. + * Default is 10. */ maxRetries: number; } +type InternalStartSSEOptions = StartSSEOptions & { + /** + * True for the transport-lifetime GET SSE stream that receives responses + * to POSTs accepted with 202. Exhausting this stream means future sends + * would otherwise be delivered into a dead response channel. + */ + isStandalone?: boolean; +}; + /** * Custom scheduler for SSE stream reconnection attempts. * @@ -323,6 +332,7 @@ export class StreamableHTTPClientTransport implements Transport { private _serverRetryMs?: number; // Server-provided retry delay from SSE retry field private readonly _reconnectionScheduler?: ReconnectionScheduler; private _cancelReconnection?: () => void; + private _standaloneSseReconnectionError?: Error; onclose?: () => void; onerror?: (error: Error) => void; @@ -498,7 +508,7 @@ export class StreamableHTTPClientTransport implements Transport { return typeof v === 'string' && isModernProtocolVersion(v); } - private async _startOrAuthSse(options: StartSSEOptions, isAuthRetry = false, stepUpRetries = 0): Promise { + private async _startOrAuthSse(options: InternalStartSSEOptions, 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 @@ -577,7 +587,7 @@ export class StreamableHTTPClientTransport implements Transport { } } - await response.text?.().catch(() => {}); + const text = await response.text?.().catch(() => null); // 405 indicates that the server does not offer an SSE stream at GET endpoint // This is an expected case that should not trigger an error @@ -594,12 +604,17 @@ export class StreamableHTTPClientTransport implements Transport { return; } - throw new SdkHttpError(SdkErrorCode.ClientHttpFailedToOpenStream, `Failed to open SSE stream: ${response.statusText}`, { + const detail = response.statusText || text || `HTTP ${response.status}`; + throw new SdkHttpError(SdkErrorCode.ClientHttpFailedToOpenStream, `Failed to open SSE stream: ${detail}`, { status: response.status, - statusText: response.statusText + statusText: response.statusText, + text }); } + if (options.isStandalone) { + this._standaloneSseReconnectionError = undefined; + } this._handleSseStream(response.body, options, true); } catch (error) { if (!isIntentionalAbort()) { @@ -636,13 +651,17 @@ export class StreamableHTTPClientTransport implements Transport { * @param lastEventId The ID of the last received event for resumability * @param attemptCount Current reconnection attempt count for this specific stream */ - private _scheduleReconnection(options: StartSSEOptions, attemptCount = 0): void { + private _scheduleReconnection(options: InternalStartSSEOptions, attemptCount = 0): void { // Use provided options or default options const maxRetries = this._reconnectionOptions.maxRetries; // Check if we've exceeded maximum retry attempts if (attemptCount >= maxRetries) { - this.onerror?.(new Error(`Maximum reconnection attempts (${maxRetries}) exceeded.`)); + const error = new Error(`Maximum reconnection attempts (${maxRetries}) exceeded.`); + this.onerror?.(error); + if (options.isStandalone) { + this._standaloneSseReconnectionError = error; + } // The per-request stream is now definitively gone. options.onRequestStreamEnd?.(); return; @@ -677,7 +696,7 @@ export class StreamableHTTPClientTransport implements Transport { } } - private _handleSseStream(stream: ReadableStream | null, options: StartSSEOptions, isReconnectable: boolean): void { + private _handleSseStream(stream: ReadableStream | null, options: InternalStartSSEOptions, isReconnectable: boolean): void { if (!stream) { // A null body on a per-request stream (or its GET resume) is the // same terminal non-resumable outcome as a 405 — fire the @@ -768,7 +787,8 @@ export class StreamableHTTPClientTransport implements Transport { onresumptiontoken, replayMessageId, requestSignal, - onRequestStreamEnd + onRequestStreamEnd, + isStandalone: options.isStandalone }, 0 ); @@ -801,7 +821,8 @@ export class StreamableHTTPClientTransport implements Transport { onresumptiontoken, replayMessageId, requestSignal, - onRequestStreamEnd + onRequestStreamEnd, + isStandalone: options.isStandalone }, 0 ); @@ -960,6 +981,11 @@ export class StreamableHTTPClientTransport implements Transport { const types = [...(userAccept?.split(',').map(s => s.trim().toLowerCase()) ?? []), 'application/json', 'text/event-stream']; headers.set('accept', [...new Set(types)].join(', ')); + // Get original message(s) for detecting request IDs + const messages = Array.isArray(message) ? message : [message]; + + const hasRequests = messages.some(msg => 'method' in msg && 'id' in msg && msg.id !== undefined); + // Per-request abort: when the caller supplies a request-scoped // signal (the `subscriptions/listen` driver), aborting it cancels // this POST and its SSE response stream without closing the @@ -1073,20 +1099,20 @@ export class StreamableHTTPClientTransport implements Transport { // If the response is 202 Accepted, there's no body to process if (response.status === 202) { await response.text?.().catch(() => {}); + if (hasRequests && this._standaloneSseReconnectionError) { + throw new Error( + `Cannot receive a response for the accepted request because the Streamable HTTP SSE stream is disconnected and reconnection attempts have been exhausted: ${this._standaloneSseReconnectionError.message}` + ); + } // if the accepted notification is initialized, we start the SSE stream // if it's supported by the server if (isInitializedNotification(message)) { // Start without a lastEventId since this is a fresh connection - this._startOrAuthSse({ resumptionToken: undefined }).catch(error => this.onerror?.(error)); + this._startOrAuthSse({ resumptionToken: undefined, isStandalone: true }).catch(error => this.onerror?.(error)); } return; } - // Get original message(s) for detecting request IDs - const messages = Array.isArray(message) ? message : [message]; - - const hasRequests = messages.some(msg => 'method' in msg && 'id' in msg && msg.id !== undefined); - // Check the response type (parsed media type — see mediaTypeEssence) const contentType = response.headers.get('content-type'); const responseMediaType = mediaTypeEssence(contentType); @@ -1209,7 +1235,8 @@ export class StreamableHTTPClientTransport implements Transport { async resumeStream(lastEventId: string, options?: { onresumptiontoken?: (token: string) => void }): Promise { await this._startOrAuthSse({ resumptionToken: lastEventId, - onresumptiontoken: options?.onresumptiontoken + onresumptiontoken: options?.onresumptiontoken, + isStandalone: true }); } } diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index a36bbc0ad3..4a9be53b14 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -430,6 +430,19 @@ describe('StreamableHTTPClientTransport', () => { expect(globalThis.fetch).toHaveBeenCalledTimes(2); }); + it('should include response details when opening an SSE stream fails', async () => { + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: false, + status: 503, + statusText: '', + text: () => Promise.resolve('upstream reset'), + headers: new Headers() + }); + + await transport.start(); + await expect(transport['_startOrAuthSse']({})).rejects.toThrow('Failed to open SSE stream: upstream reset'); + }); + it('should handle successful initial GET connection for SSE', async () => { // Set up readable stream for SSE events const encoder = new TextEncoder(); @@ -553,6 +566,13 @@ describe('StreamableHTTPClientTransport', () => { expect(transportInstance._reconnectionOptions.maxRetries).toBe(5); }); + it('should use a production-friendly default reconnection retry count', () => { + const transportInstance = transport as unknown as { + _reconnectionOptions: StreamableHTTPReconnectionOptions; + }; + expect(transportInstance._reconnectionOptions.maxRetries).toBe(10); + }); + it('should pass lastEventId when reconnecting', async () => { // Create a fresh transport transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp')); @@ -2500,6 +2520,132 @@ describe('StreamableHTTPClientTransport', () => { expect(transport['_cancelReconnection']).toBeUndefined(); }); + it('should reject accepted requests after standalone SSE reconnection is exhausted', async () => { + // ARRANGE + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxRetries: 0, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1 + } + }); + + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + (globalThis.fetch as Mock).mockClear(); + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: true, + status: 202, + text: () => Promise.resolve(''), + headers: new Headers() + }); + + // ACT - exhaust the transport-lifetime GET SSE stream. + transport['_scheduleReconnection']({ isStandalone: true }); + + const message: JSONRPCMessage = { + jsonrpc: '2.0', + method: 'tools/call', + params: {}, + id: 'call-1' + }; + + await expect(transport.send(message)).rejects.toThrow( + 'Cannot receive a response for the accepted request because the Streamable HTTP SSE stream is disconnected and reconnection attempts have been exhausted' + ); + + // ASSERT - the 202 response was rejected instead of leaving the request pending forever. + expect(globalThis.fetch).toHaveBeenCalledTimes(1); + expect(errorSpy).toHaveBeenNthCalledWith( + 1, + expect.objectContaining({ + message: 'Maximum reconnection attempts (0) exceeded.' + }) + ); + expect(errorSpy).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ + message: expect.stringContaining('Cannot receive a response for the accepted request') + }) + ); + }); + + it('should allow direct responses after standalone SSE reconnection is exhausted', async () => { + // ARRANGE + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxRetries: 0, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1 + } + }); + + const responseMessage: JSONRPCMessage = { + jsonrpc: '2.0', + result: { content: [{ type: 'text', text: 'still works' }] }, + id: 'call-1' + }; + const messageSpy = vi.fn(); + transport.onmessage = messageSpy; + (globalThis.fetch as Mock).mockClear(); + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'application/json' }), + json: () => Promise.resolve(responseMessage) + }); + + // ACT - exhaust the transport-lifetime GET SSE stream. + transport['_scheduleReconnection']({ isStandalone: true }); + + await expect( + transport.send({ + jsonrpc: '2.0', + method: 'tools/call', + params: {}, + id: 'call-1' + }) + ).resolves.toBeUndefined(); + + // ASSERT + expect(messageSpy).toHaveBeenCalledWith(responseMessage); + expect(globalThis.fetch).toHaveBeenCalledTimes(1); + }); + + it('should not fail subsequent sends after per-request stream reconnection is exhausted', async () => { + // ARRANGE + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxRetries: 0, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1 + } + }); + + const onRequestStreamEnd = vi.fn(); + + // ACT - exhaust a per-request stream without marking the standalone channel dead. + transport['_scheduleReconnection']({ onRequestStreamEnd }); + + (globalThis.fetch as Mock).mockClear(); + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: true, + status: 202, + text: () => Promise.resolve(''), + headers: new Headers() + }); + + await expect(transport.send({ jsonrpc: '2.0', method: 'notifications/ping', params: {} })).resolves.toBeUndefined(); + + // ASSERT + expect(onRequestStreamEnd).toHaveBeenCalledTimes(1); + expect(transport['_standaloneSseReconnectionError']).toBeUndefined(); + expect(globalThis.fetch).toHaveBeenCalledTimes(1); + }); + it('should schedule reconnection when maxRetries is greater than 0', async () => { // ARRANGE transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { From ce7a5beefc1ed7eaa0db56e4b5186398319e620a Mon Sep 17 00:00:00 2001 From: Pradeep Ramola Date: Sun, 19 Jul 2026 03:25:03 -0400 Subject: [PATCH 2/3] Add changeset for streamable HTTP SSE fix --- .changeset/fix-streamable-http-sse-exhaustion.md | 6 ++++++ 1 file changed, 6 insertions(+) create mode 100644 .changeset/fix-streamable-http-sse-exhaustion.md diff --git a/.changeset/fix-streamable-http-sse-exhaustion.md b/.changeset/fix-streamable-http-sse-exhaustion.md new file mode 100644 index 0000000000..6e7d71cb88 --- /dev/null +++ b/.changeset/fix-streamable-http-sse-exhaustion.md @@ -0,0 +1,6 @@ +--- +'@modelcontextprotocol/client': patch +--- + +Reject standalone SSE request sends after the streamable HTTP client stream has +been exhausted. From 024c34b735502960c98790c4fc598c3a0495033a Mon Sep 17 00:00:00 2001 From: Pradeep Ramola Date: Mon, 20 Jul 2026 03:35:42 -0400 Subject: [PATCH 3/3] chore: retry CI