fix(client): close the SSE transport when its stream ends for good

.changeset/sse-client-close-on-dead-stream.md

@@ -0,0 +1,5 @@
+---
+'@modelcontextprotocol/client': patch
+---
+
+`SSEClientTransport` now closes, firing `onclose` and rejecting pending requests with `ConnectionClosed`, when its SSE stream drops after connecting and the reconnect fails for good, for example with a 404 after the server restarts. Previously the transport stayed open with no stream, so pending requests waited for their timeout and the application was never told the connection had ended.

packages/client/src/client/sse.ts

@@ -315,6 +315,7 @@ export class SSEClientTransport implements Transport {
                     this._last401Response = undefined;
                     reject(error);
                     this.onerror?.(error);
+                    this._closeIfStreamEnded();
                     return;
                 }
 
@@ -323,6 +324,7 @@ export class SSEClientTransport implements Transport {
                 const error = new SseError(event.code, redirect ?? event.message, event);
                 reject(error);
                 this.onerror?.(error);
+                this._closeIfStreamEnded();
             };
 
             this._eventSource.onopen = () => {
@@ -364,6 +366,18 @@ export class SSEClientTransport implements Transport {
         });
     }
 
+    // After start(), an EventSource that stopped reconnecting (readyState CLOSED) ends the connection.
+    // Skip it when an onerror handler already closed the transport, so onclose does not fire twice.
+    private _closeIfStreamEnded(): void {
+        if (
+            this._endpoint !== undefined &&
+            this._eventSource?.readyState === EventSource.CLOSED &&
+            !this._abortController?.signal.aborted
+        ) {
+            void this.close();
+        }
+    }
+
     async start() {
         if (this._eventSource) {
             throw new Error('SSEClientTransport already started! If using Client class, note that connect() calls start() automatically.');

packages/client/test/client/sseStreamEnd.test.ts

@@ -0,0 +1,93 @@
+import { createServer } from 'node:http';
+import type { AddressInfo } from 'node:net';
+
+import { LATEST_PROTOCOL_VERSION, SdkErrorCode } from '@modelcontextprotocol/core-internal';
+
+import { Client } from '../../src/client/client';
+import { SSEClientTransport } from '../../src/client/sse';
+
+// Serves one SSE stream, ends it while tools/list is pending, and answers the reconnect with 404.
+async function withServerThatDropsTheStream(run: (url: URL, gets: () => number) => Promise<void>): Promise<void> {
+    let gets = 0;
+    let write: ((data: string) => void) | undefined;
+    let end: (() => void) | undefined;
+    const server = createServer((req, res) => {
+        if (req.method === 'GET') {
+            gets++;
+            if (gets > 1) {
+                res.writeHead(404).end();
+                return;
+            }
+            res.writeHead(200, { 'Content-Type': 'text/event-stream' });
+            res.write('event: endpoint\ndata: /messages\n\n');
+            write = data => res.write(`event: message\ndata: ${data}\n\n`);
+            end = () => res.end();
+            return;
+        }
+        let body = '';
+        req.on('data', chunk => (body += chunk));
+        req.on('end', () => {
+            res.writeHead(202).end();
+            const message = JSON.parse(body);
+            if (message.method === 'initialize') {
+                write?.(
+                    JSON.stringify({
+                        jsonrpc: '2.0',
+                        id: message.id,
+                        result: {
+                            protocolVersion: LATEST_PROTOCOL_VERSION,
+                            capabilities: { tools: {} },
+                            serverInfo: { name: 's', version: '1' }
+                        }
+                    })
+                );
+            } else if (message.method === 'tools/list') {
+                end?.(); // the server goes away before answering
+            }
+        });
+    });
+    await new Promise<void>(resolve => server.listen(0, '127.0.0.1', () => resolve()));
+    try {
+        await run(new URL(`http://127.0.0.1:${(server.address() as AddressInfo).port}/sse`), () => gets);
+    } finally {
+        server.close();
+    }
+}
+
+test('a pending request rejects and onclose fires when the SSE stream drops and cannot be reopened', async () => {
+    await withServerThatDropsTheStream(async (url, gets) => {
+        const client = new Client({ name: 'c', version: '1' });
+        client.onerror = () => {};
+        const closed = new Promise<void>(resolve => (client.onclose = resolve));
+        await client.connect(new SSEClientTransport(url));
+
+        const pending = client.listTools(undefined, { timeout: 30_000 });
+        await expect(pending).rejects.toMatchObject({ code: SdkErrorCode.ConnectionClosed });
+        await closed;
+        expect(gets()).toBe(2);
+    });
+}, 15_000);
+
+test('onclose fires once when an onerror handler already closed the client', async () => {
+    await withServerThatDropsTheStream(async url => {
+        const client = new Client({ name: 'c', version: '1' });
+        let closes = 0;
+        const closed = new Promise<void>(resolve => {
+            client.onclose = () => {
+                closes++;
+                resolve();
+            };
+        });
+        const transport = new SSEClientTransport(url);
+        await client.connect(transport);
+        client.onerror = () => {
+            void client.close();
+        };
+
+        const pending = client.listTools(undefined, { timeout: 30_000 });
+        await expect(pending).rejects.toMatchObject({ code: SdkErrorCode.ConnectionClosed });
+        await closed;
+        await new Promise(resolve => setTimeout(resolve, 50));
+        expect(closes).toBe(1);
+    });
+}, 15_000);