From 248faeb127eed0e848ac3250fe6066e8d6443a5f Mon Sep 17 00:00:00 2001 From: Arjun Prabhulal Date: Sat, 31 Oct 2026 14:56:49 -0400 Subject: [PATCH] fix(client): close the SSE transport when its stream ends for good After start() resolved, the EventSource onerror handler only reported the error. When the stream dropped and the reconnect failed with a non-200 status, EventSource stopped retrying but the transport never called close(), so onclose did not fire and pending requests waited for their timeout. Close the transport once the EventSource is CLOSED, unless an onerror handler already closed it, so onclose fires once. --- .changeset/sse-client-close-on-dead-stream.md | 5 + packages/client/src/client/sse.ts | 14 +++ .../client/test/client/sseStreamEnd.test.ts | 93 +++++++++++++++++++ 3 files changed, 112 insertions(+) create mode 100644 .changeset/sse-client-close-on-dead-stream.md create mode 100644 packages/client/test/client/sseStreamEnd.test.ts diff --git a/.changeset/sse-client-close-on-dead-stream.md b/.changeset/sse-client-close-on-dead-stream.md new file mode 100644 index 0000000000..9ff00ab32a --- /dev/null +++ b/.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. diff --git a/packages/client/src/client/sse.ts b/packages/client/src/client/sse.ts index 55c346ac85..46c2c48e9b 100644 --- a/packages/client/src/client/sse.ts +++ b/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.'); diff --git a/packages/client/test/client/sseStreamEnd.test.ts b/packages/client/test/client/sseStreamEnd.test.ts new file mode 100644 index 0000000000..3147c62bfc --- /dev/null +++ b/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): Promise { + 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(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(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(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);