fix(client): allow transport restart after close()

.changeset/fix-transport-restart-after-close.md

@@ -0,0 +1,5 @@
+---
+'@modelcontextprotocol/client': patch
+---
+
+Allow `StreamableHTTPClientTransport` and `SSEClientTransport` to restart after `close()`. `close()` now clears `_abortController` (previously aborted but not unset, blocking the start guard) and `_sessionId` (previously leaked into post-restart requests, causing 404s).

packages/client/src/client/sse.ts

@@ -238,12 +238,15 @@
             throw new UnauthorizedError('Failed to authorize');
         }
     }
 
     async close(): Promise<void> {
         this._abortController?.abort();
+        this._abortController = undefined;
         this._eventSource?.close();
+        this._eventSource = undefined;
+        this._endpoint = undefined;
         this.onclose?.();
     }
 
     async send(message: JSONRPCMessage): Promise<void> {
         return this._send(message, false);

packages/client/src/client/streamableHttp.ts

@@ -230,16 +230,21 @@
         });
     }
 
     private async _startOrAuthSse(options: StartSSEOptions, isAuthRetry = false): Promise<void> {
         const { resumptionToken } = options;
 
+        // Capture the signal active when this call started so a close()/start() during the awaits
+        // below doesn't bind this stale GET to the restarted transport's controller.
+        const signal = this._abortController?.signal;
+
         try {
             // Try to open an initial SSE stream with GET to listen for server messages
             // This is optional according to the spec - server may not support it
             const headers = await this._commonHeaders();
+            if (signal?.aborted) return;
             const userAccept = headers.get('accept');
             const types = [...(userAccept?.split(',').map(s => s.trim().toLowerCase()) ?? []), 'text/event-stream'];
             headers.set('accept', [...new Set(types)].join(', '));
 
             // Include Last-Event-ID header for resumable streams if provided
             if (resumptionToken) {
@@ -250,7 +255,7 @@
                 ...this._requestInit,
                 method: 'GET',
                 headers,
-                signal: this._abortController?.signal
+                signal
             });
 
             if (!response.ok) {
@@ -341,11 +346,18 @@
         // Calculate next delay based on current attempt count
         const delay = this._getNextReconnectionDelay(attemptCount);
 
+        // Capture the signal active when this reconnection was scheduled. close() + start()
+        // replaces this._abortController, so re-reading it later would see the new session's
+        // controller and allow a stale reconnect to fire into the restarted transport.
+        const signal = this._abortController?.signal;
+
         const reconnect = (): void => {
             this._cancelReconnection = undefined;
-            if (this._abortController?.signal.aborted) return;
+            if (signal?.aborted) return;
             this._startOrAuthSse(options).catch(error => {
+                if (signal?.aborted) return;
                 this.onerror?.(new Error(`Failed to reconnect SSE stream: ${error instanceof Error ? error.message : String(error)}`));
+                if (signal?.aborted) return;
                 try {
                     this._scheduleReconnection(options, attemptCount + 1);
                 } catch (scheduleError) {
@@ -369,6 +381,9 @@
         }
         const { onresumptiontoken, replayMessageId } = options;
 
+        // Capture the signal this stream is bound to so we don't reconnect into a restarted transport.
+        const signal = this._abortController?.signal;
+
         let lastEventId: string | undefined;
         // Track whether we've received a priming event (event with ID)
         // Per spec, server SHOULD send a priming event with ID before closing
@@ -436,7 +451,7 @@
                 // BUT don't reconnect if we already received a response - the request is complete
                 const canResume = isReconnectable || hasPrimingEvent;
                 const needsReconnect = canResume && !receivedResponse;
-                if (needsReconnect && this._abortController && !this._abortController.signal.aborted) {
+                if (needsReconnect && signal && !signal.aborted) {
                     this._scheduleReconnection(
                         {
                             resumptionToken: lastEventId,
@@ -455,7 +470,7 @@
                 // BUT don't reconnect if we already received a response - the request is complete
                 const canResume = isReconnectable || hasPrimingEvent;
                 const needsReconnect = canResume && !receivedResponse;
-                if (needsReconnect && this._abortController && !this._abortController.signal.aborted) {
+                if (needsReconnect && signal && !signal.aborted) {
                     // Use the exponential backoff reconnection strategy
                     try {
                         this._scheduleReconnection(
@@ -476,7 +491,7 @@
     }
 
     async start() {
-        if (this._abortController) {
+        if (this._abortController && !this._abortController.signal.aborted) {
             throw new Error(
                 'StreamableHTTPClientTransport already started! If using Client class, note that connect() calls start() automatically.'
             );
@@ -511,6 +526,9 @@
         } finally {
             this._cancelReconnection = undefined;
             this._abortController?.abort();
+            this._sessionId = undefined;
+            this._lastUpscopingHeader = undefined;
+            this._serverRetryMs = undefined;
             this.onclose?.();
         }
     }

packages/client/test/client/streamableHttp.test.ts

@@ -1998,6 +1998,31 @@ describe('StreamableHTTPClientTransport', () => {
             expect(onerror).not.toHaveBeenCalled();
         });
 
+        it('ignores a late-firing reconnect after close() + start()', async () => {
+            let capturedReconnect: (() => void) | undefined;
+            transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), {
+                reconnectionOptions,
+                reconnectionScheduler: reconnect => {
+                    capturedReconnect = reconnect;
+                }
+            });
+            const onerror = vi.fn();
+            transport.onerror = onerror;
+            const fetchMock = globalThis.fetch as Mock;
+
+            await transport.start();
+            triggerReconnection(transport);
+            await transport.close();
+            await transport.start();
+
+            fetchMock.mockClear();
+            capturedReconnect?.();
+            await vi.runAllTimersAsync();
+
+            expect(fetchMock).not.toHaveBeenCalled();
+            expect(onerror).not.toHaveBeenCalled();
+        });
+
         it('still aborts and fires onclose if the cancel function throws', async () => {
             transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), {
                 reconnectionOptions,
@@ -2017,4 +2042,61 @@ describe('StreamableHTTPClientTransport', () => {
             expect(onclose).toHaveBeenCalledTimes(1);
         });
     });
+
+    describe('Transport restart after close()', () => {
+        it('should allow start() after close() and not send stale session ID', async () => {
+            transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'));
+
+            const fetchMock = globalThis.fetch as Mock;
+
+            // First lifecycle: start, receive a session ID, close
+            await transport.start();
+
+            fetchMock.mockResolvedValueOnce({
+                ok: true,
+                status: 200,
+                headers: new Headers({
+                    'content-type': 'application/json',
+                    'mcp-session-id': 'stale-session-abc'
+                }),
+                json: async () => ({ jsonrpc: '2.0', result: {}, id: 'init-1' })
+            });
+
+            await transport.send({ jsonrpc: '2.0', method: 'initialize', params: {}, id: 'init-1' });
+            expect(transport.sessionId).toBe('stale-session-abc');
+
+            await transport.close();
+            expect(transport.sessionId).toBeUndefined();
+
+            // Second lifecycle: start() should not throw
+            await transport.start();
+
+            fetchMock.mockResolvedValueOnce({
+                ok: true,
+                status: 202,
+                headers: new Headers(),
+                text: async () => ''
+            });
+
+            await transport.send({ jsonrpc: '2.0', method: 'notifications/initialized' });
+
+            // The post-restart request must NOT include the stale session ID
+            const postRestartHeaders = fetchMock.mock.calls[1]![1]?.headers as Headers;
+            expect(postRestartHeaders.get('mcp-session-id')).toBeNull();
+        });
+
+        it('should reset server-provided retry delay and upscoping header on close()', async () => {
+            transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'));
+            await transport.start();
+
+            const internal = transport as unknown as { _serverRetryMs?: number; _lastUpscopingHeader?: string };
+            internal._serverRetryMs = 3000;
+            internal._lastUpscopingHeader = 'Bearer realm="x"';
+
+            await transport.close();
+
+            expect(internal._serverRetryMs).toBeUndefined();
+            expect(internal._lastUpscopingHeader).toBeUndefined();
+        });
+    });
 });