From b0b70001c182fca7541322a21d1dbe4f24db248e Mon Sep 17 00:00:00 2001 From: Matt Carey Date: Thu, 23 Jul 2026 16:05:37 +0100 Subject: [PATCH 1/3] fix: send SSE keep-alive comment frames from Streamable HTTP server transport Idle SSE streams (the standalone GET stream in particular, but also POST response streams during long-running tool calls) are killed by intermediaries and server idle timeouts (Node's requestTimeout defaults to 300s), which clients observe as "SSE stream disconnected: TypeError: terminated" roughly every 5 minutes, followed by a reconnect loop. The server transport now writes an SSE comment frame (`: keepalive`) to every open SSE stream every keepAliveMs milliseconds (default 15000, per the WHATWG SSE spec recommendation; set 0 to disable). Comment frames are dropped by SSE parsers and never surface as protocol messages. The timer is unref'd so it never holds the process open, and is cleared on stream cleanup/cancel and transport close. Fixes #1211 --- src/server/webStandardStreamableHttp.ts | 67 ++++++++++++ test/server/streamableHttp.test.ts | 137 ++++++++++++++++++++++++ 2 files changed, 204 insertions(+) diff --git a/src/server/webStandardStreamableHttp.ts b/src/server/webStandardStreamableHttp.ts index 5721d3e370..b2172a9284 100644 --- a/src/server/webStandardStreamableHttp.ts +++ b/src/server/webStandardStreamableHttp.ts @@ -146,8 +146,24 @@ export interface WebStandardStreamableHTTPServerTransportOptions { * client reconnection timing for polling behavior. */ retryInterval?: number; + + /** + * Interval in milliseconds between SSE keep-alive comment frames (`: keepalive`) + * written to open SSE streams. Keep-alive frames prevent idle streams (e.g. the + * standalone GET stream, or a POST stream during a long-running tool call) from + * being killed by intermediaries and server idle timeouts, which clients observe + * as `SSE stream disconnected: TypeError: terminated`. + * + * Comment frames are ignored by SSE parsers and never surface as messages. + * Defaults to 15000 (per the WHATWG SSE spec recommendation of roughly every + * 15 seconds). Set to 0 to disable keep-alive frames. + */ + keepAliveMs?: number; } +/** Default interval between SSE keep-alive comment frames. */ +const DEFAULT_KEEP_ALIVE_MS = 15_000; + /** * Options for handling a request */ @@ -225,6 +241,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { private _allowedOrigins?: string[]; private _enableDnsRebindingProtection: boolean; private _retryInterval?: number; + private _keepAliveMs: number; + private _keepAliveTimers: Map> = new Map(); sessionId?: string; onclose?: () => void; @@ -241,6 +259,40 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { this._allowedOrigins = options.allowedOrigins; this._enableDnsRebindingProtection = options.enableDnsRebindingProtection ?? false; this._retryInterval = options.retryInterval; + this._keepAliveMs = options.keepAliveMs ?? DEFAULT_KEEP_ALIVE_MS; + } + + /** + * Arms a keep-alive interval for an SSE stream that periodically writes an SSE + * comment frame so intermediaries and idle timeouts don't kill the connection. + * The timer is cleared via stopKeepAlive when the stream is cleaned up, and + * clears itself if a write fails (stream already closed/cancelled). + */ + private startKeepAlive(streamId: string, controller: ReadableStreamDefaultController, encoder: TextEncoder): void { + if (this._keepAliveMs <= 0) { + return; + } + const timer = setInterval(() => { + try { + controller.enqueue(encoder.encode(': keepalive\n\n')); + } catch { + this.stopKeepAlive(streamId); + } + }, this._keepAliveMs); + // Don't let the keep-alive timer hold the process open (Node.js only) + (timer as { unref?: () => void }).unref?.(); + this._keepAliveTimers.set(streamId, timer); + } + + /** + * Clears the keep-alive interval for a stream, if one is armed. + */ + private stopKeepAlive(streamId: string): void { + const timer = this._keepAliveTimers.get(streamId); + if (timer !== undefined) { + clearInterval(timer); + this._keepAliveTimers.delete(streamId); + } } /** @@ -425,6 +477,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { }, cancel: () => { // Stream was cancelled by client + this.stopKeepAlive(this._standaloneSseStreamId); this._streamMapping.delete(this._standaloneSseStreamId); } }); @@ -445,6 +498,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { controller: streamController!, encoder, cleanup: () => { + this.stopKeepAlive(this._standaloneSseStreamId); this._streamMapping.delete(this._standaloneSseStreamId); try { streamController!.close(); @@ -454,6 +508,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } }); + this.startKeepAlive(this._standaloneSseStreamId, streamController!, encoder); + return new Response(readable, { headers }); } @@ -528,6 +584,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { controller: streamController!, encoder, cleanup: () => { + this.stopKeepAlive(replayedStreamId); this._streamMapping.delete(replayedStreamId); try { streamController!.close(); @@ -537,6 +594,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } }); + this.startKeepAlive(replayedStreamId, streamController!, encoder); + return new Response(readable, { headers }); } catch (error) { this.onerror?.(error as Error); @@ -743,6 +802,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { }, cancel: () => { // Stream was cancelled by client + this.stopKeepAlive(streamId); this._streamMapping.delete(streamId); } }); @@ -766,6 +826,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { controller: streamController!, encoder, cleanup: () => { + this.stopKeepAlive(streamId); this._streamMapping.delete(streamId); try { streamController!.close(); @@ -778,6 +839,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } } + this.startKeepAlive(streamId, streamController!, encoder); + // Write priming event if event store is configured (after mapping is set up) await this.writePrimingEvent(streamController!, encoder, streamId, clientProtocolVersion); @@ -901,6 +964,10 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { }); this._streamMapping.clear(); + // Clear any keep-alive timers not already cleared by stream cleanup + this._keepAliveTimers.forEach(timer => clearInterval(timer)); + this._keepAliveTimers.clear(); + // Clear any pending responses this._requestResponseMap.clear(); this.onclose?.(); diff --git a/test/server/streamableHttp.test.ts b/test/server/streamableHttp.test.ts index 4a4f7d8248..5671a450a7 100644 --- a/test/server/streamableHttp.test.ts +++ b/test/server/streamableHttp.test.ts @@ -3272,3 +3272,140 @@ describe('WebStandardStreamableHTTPServerTransport - onerror callback', () => { await storeTransport.close(); }); }); + +describe('WebStandardStreamableHTTPServerTransport SSE keep-alive', () => { + /** Shorthand to build a Web Standard Request for direct transport testing. */ + function req(method: string, opts?: { body?: unknown; headers?: Record }): Request { + const headers: Record = { ...opts?.headers }; + if (method === 'POST') { + headers['Accept'] ??= 'application/json, text/event-stream'; + headers['Content-Type'] ??= 'application/json'; + } else if (method === 'GET') { + headers['Accept'] ??= 'text/event-stream'; + } + return new Request('http://localhost/mcp', { + method, + headers, + body: opts?.body !== undefined ? JSON.stringify(opts.body) : undefined + }); + } + + async function createTransport(options?: { keepAliveMs?: number }): Promise<{ + transport: WebStandardStreamableHTTPServerTransport; + sessionId: string; + }> { + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), ...options }); + await new McpServer({ name: 'test-server', version: '1.0.0' }).connect(transport); + const initResponse = await transport.handleRequest(req('POST', { body: TEST_MESSAGES.initialize })); + expect(initResponse.status).toBe(200); + return { transport, sessionId: initResponse.headers.get('mcp-session-id') as string }; + } + + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it('should write keep-alive comment frames to an idle standalone GET stream', async () => { + const { transport, sessionId } = await createTransport(); + + const response = await transport.handleRequest( + req('GET', { headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' } }) + ); + expect(response.status).toBe(200); + + const reader = response.body!.getReader(); + await vi.advanceTimersByTimeAsync(15000); + const { value } = await reader.read(); + expect(new TextDecoder().decode(value)).toBe(': keepalive\n\n'); + + await transport.close(); + }); + + it('should honor a custom keepAliveMs interval', async () => { + const { transport, sessionId } = await createTransport({ keepAliveMs: 1000 }); + + const response = await transport.handleRequest( + req('GET', { headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' } }) + ); + const reader = response.body!.getReader(); + + await vi.advanceTimersByTimeAsync(3000); + let received = ''; + for (let i = 0; i < 3; i++) { + const { value } = await reader.read(); + received += new TextDecoder().decode(value); + } + expect(received).toBe(': keepalive\n\n'.repeat(3)); + + await transport.close(); + }); + + it('should not write keep-alive frames when keepAliveMs is 0', async () => { + const { transport, sessionId } = await createTransport({ keepAliveMs: 0 }); + + const response = await transport.handleRequest( + req('GET', { headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' } }) + ); + const reader = response.body!.getReader(); + + await vi.advanceTimersByTimeAsync(60000); + const read = reader.read(); + const raced = await Promise.race([read, Promise.resolve('pending')]); + expect(raced).toBe('pending'); + + await transport.close(); + }); + + it('should stop keep-alive frames after the stream is closed', async () => { + const { transport, sessionId } = await createTransport(); + + const response = await transport.handleRequest( + req('GET', { headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' } }) + ); + const reader = response.body!.getReader(); + + await transport.close(); + const { done } = await reader.read(); + expect(done).toBe(true); + + // Advancing time after close must not throw or fire further writes + expect(vi.getTimerCount()).toBe(0); + await vi.advanceTimersByTimeAsync(60000); + }); + + it('should write keep-alive frames on a POST SSE stream while a request is pending', async () => { + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID() }); + const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }); + let resolveTool: (() => void) | undefined; + mcpServer.tool('slow', async () => { + await new Promise(resolve => { + resolveTool = resolve; + }); + return { content: [{ type: 'text', text: 'done' }] }; + }); + await mcpServer.connect(transport); + + const initResponse = await transport.handleRequest(req('POST', { body: TEST_MESSAGES.initialize })); + const sessionId = initResponse.headers.get('mcp-session-id') as string; + + const response = await transport.handleRequest( + req('POST', { + body: { jsonrpc: '2.0', method: 'tools/call', params: { name: 'slow', arguments: {} }, id: 'call-1' }, + headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' } + }) + ); + expect(response.status).toBe(200); + const reader = response.body!.getReader(); + + await vi.advanceTimersByTimeAsync(15000); + const { value } = await reader.read(); + expect(new TextDecoder().decode(value)).toBe(': keepalive\n\n'); + + resolveTool?.(); + await transport.close(); + }); +}); From c1edde214c68b5855060b53ad42f55517cfc1f26 Mon Sep 17 00:00:00 2001 From: Matt Carey Date: Thu, 23 Jul 2026 17:07:40 +0100 Subject: [PATCH 2/3] chore: add changeset --- .changeset/streamable-http-sse-keepalive.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changeset/streamable-http-sse-keepalive.md diff --git a/.changeset/streamable-http-sse-keepalive.md b/.changeset/streamable-http-sse-keepalive.md new file mode 100644 index 0000000000..6f7b3aee31 --- /dev/null +++ b/.changeset/streamable-http-sse-keepalive.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/sdk': patch +--- + +`StreamableHTTPServerTransport` and `WebStandardStreamableHTTPServerTransport` now write SSE keep-alive comment frames (`: keepalive`) to open SSE streams so idle connections (e.g. the standalone GET stream, or a POST stream during a long-running tool call) are not killed by intermediaries or server idle timeouts. Configurable via the new `keepAliveMs` option (default 15000; set 0 to disable). From 2bdf8c5402aca897e6e555b550724e6dc992c6d4 Mon Sep 17 00:00:00 2001 From: Matt Carey Date: Thu, 23 Jul 2026 17:19:57 +0100 Subject: [PATCH 3/3] fix: replace existing keep-alive timer when a stream re-registers under the same id MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A replayed stream can re-register under the same stream id when the event store doesn't implement the optional getStreamIdForEventId (the 409 conflict check is skipped). startKeepAlive now clears any timer already armed for the stream id before arming a new one, so the predecessor's timer can't leak or — via its failing write's stopKeepAlive on the shared id — silently disable the live stream's keep-alive. Also formats the changeset. --- .changeset/streamable-http-sse-keepalive.md | 3 +- src/server/webStandardStreamableHttp.ts | 5 ++- test/server/streamableHttp.test.ts | 39 +++++++++++++++++++++ 3 files changed, 45 insertions(+), 2 deletions(-) diff --git a/.changeset/streamable-http-sse-keepalive.md b/.changeset/streamable-http-sse-keepalive.md index 6f7b3aee31..5611a547e5 100644 --- a/.changeset/streamable-http-sse-keepalive.md +++ b/.changeset/streamable-http-sse-keepalive.md @@ -2,4 +2,5 @@ '@modelcontextprotocol/sdk': patch --- -`StreamableHTTPServerTransport` and `WebStandardStreamableHTTPServerTransport` now write SSE keep-alive comment frames (`: keepalive`) to open SSE streams so idle connections (e.g. the standalone GET stream, or a POST stream during a long-running tool call) are not killed by intermediaries or server idle timeouts. Configurable via the new `keepAliveMs` option (default 15000; set 0 to disable). +`StreamableHTTPServerTransport` and `WebStandardStreamableHTTPServerTransport` now write SSE keep-alive comment frames (`: keepalive`) to open SSE streams so idle connections (e.g. the standalone GET stream, or a POST stream during a long-running tool call) are not killed by +intermediaries or server idle timeouts. Configurable via the new `keepAliveMs` option (default 15000; set 0 to disable). diff --git a/src/server/webStandardStreamableHttp.ts b/src/server/webStandardStreamableHttp.ts index b2172a9284..932ad56600 100644 --- a/src/server/webStandardStreamableHttp.ts +++ b/src/server/webStandardStreamableHttp.ts @@ -265,13 +265,16 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { /** * Arms a keep-alive interval for an SSE stream that periodically writes an SSE * comment frame so intermediaries and idle timeouts don't kill the connection. - * The timer is cleared via stopKeepAlive when the stream is cleaned up, and + * Replaces any timer already armed for the same stream id (a resumed stream + * re-registered under the same id supersedes its predecessor's timer). The + * timer is cleared via stopKeepAlive when the stream is cleaned up, and * clears itself if a write fails (stream already closed/cancelled). */ private startKeepAlive(streamId: string, controller: ReadableStreamDefaultController, encoder: TextEncoder): void { if (this._keepAliveMs <= 0) { return; } + this.stopKeepAlive(streamId); const timer = setInterval(() => { try { controller.enqueue(encoder.encode(': keepalive\n\n')); diff --git a/test/server/streamableHttp.test.ts b/test/server/streamableHttp.test.ts index 5671a450a7..2046a71697 100644 --- a/test/server/streamableHttp.test.ts +++ b/test/server/streamableHttp.test.ts @@ -3408,4 +3408,43 @@ describe('WebStandardStreamableHTTPServerTransport SSE keep-alive', () => { resolveTool?.(); await transport.close(); }); + + it('should supersede the previous keep-alive timer when a replayed stream re-registers under the same stream id', async () => { + // Event store WITHOUT the optional getStreamIdForEventId — the replay + // path then skips its 409 conflict check, so a reconnect re-registers + // the same stream id. The predecessor's timer must be replaced, not + // orphaned (an orphaned timer's failing write would clear the live + // stream's keep-alive via stopKeepAlive on the shared stream id). + const eventStore: EventStore = { + async storeEvent(): Promise { + return 'evt-1'; + }, + async replayEventsAfter(): Promise { + return 'stream-1'; + } + }; + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), eventStore }); + await new McpServer({ name: 'test-server', version: '1.0.0' }).connect(transport); + const initResponse = await transport.handleRequest(req('POST', { body: TEST_MESSAGES.initialize })); + const sessionId = initResponse.headers.get('mcp-session-id') as string; + + const replayHeaders = { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25', 'Last-Event-ID': 'evt-1' }; + const first = await transport.handleRequest(req('GET', { headers: replayHeaders })); + expect(first.status).toBe(200); + + // Reconnect with the same Last-Event-ID — re-registers 'stream-1' + const second = await transport.handleRequest(req('GET', { headers: replayHeaders })); + expect(second.status).toBe(200); + + // Exactly one keep-alive timer must remain armed (plus none orphaned) + expect(vi.getTimerCount()).toBe(1); + + // The live (second) stream still receives keep-alive frames + const reader = second.body!.getReader(); + await vi.advanceTimersByTimeAsync(15000); + const { value } = await reader.read(); + expect(new TextDecoder().decode(value)).toBe(': keepalive\n\n'); + + await transport.close(); + }); });