Fix streamable HTTP SSE exhaustion

.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.

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<void> {
+    private async _startOrAuthSse(options: InternalStartSSEOptions, isAuthRetry = false, stepUpRetries = 0): Promise<void> {
         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<Uint8Array> | null, options: StartSSEOptions, isReconnectable: boolean): void {
+    private _handleSseStream(stream: ReadableStream<Uint8Array> | 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<void> {
         await this._startOrAuthSse({
             resumptionToken: lastEventId,
-            onresumptiontoken: options?.onresumptiontoken
+            onresumptiontoken: options?.onresumptiontoken,
+            isStandalone: true
         });
     }
 }

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'), {