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,13 @@ export class SSEClientTransport implements Transport {
         });
     }
 
+    // After start(), an EventSource that stopped reconnecting (readyState CLOSED) ends the connection.
+    private _closeIfStreamEnded(): void {
+        if (this._endpoint !== undefined && this._eventSource?.readyState === EventSource.CLOSED) {
+            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,63 @@
+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';
+
+test('a pending request rejects and onclose fires when the SSE stream drops and cannot be reopened', async () => {
+    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 {
+        const url = new URL(`http://127.0.0.1:${(server.address() as AddressInfo).port}/sse`);
+        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);
+    } finally {
+        server.close();
+    }
+}, 15_000);