diff --git a/.changeset/streamable-http-sse-keepalive.md b/.changeset/streamable-http-sse-keepalive.md new file mode 100644 index 0000000000..5611a547e5 --- /dev/null +++ b/.changeset/streamable-http-sse-keepalive.md @@ -0,0 +1,6 @@ +--- +'@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). diff --git a/src/server/webStandardStreamableHttp.ts b/src/server/webStandardStreamableHttp.ts index 5721d3e370..932ad56600 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,43 @@ 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. + * 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')); + } 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 +480,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { }, cancel: () => { // Stream was cancelled by client + this.stopKeepAlive(this._standaloneSseStreamId); this._streamMapping.delete(this._standaloneSseStreamId); } }); @@ -445,6 +501,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { controller: streamController!, encoder, cleanup: () => { + this.stopKeepAlive(this._standaloneSseStreamId); this._streamMapping.delete(this._standaloneSseStreamId); try { streamController!.close(); @@ -454,6 +511,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } }); + this.startKeepAlive(this._standaloneSseStreamId, streamController!, encoder); + return new Response(readable, { headers }); } @@ -528,6 +587,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { controller: streamController!, encoder, cleanup: () => { + this.stopKeepAlive(replayedStreamId); this._streamMapping.delete(replayedStreamId); try { streamController!.close(); @@ -537,6 +597,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 +805,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { }, cancel: () => { // Stream was cancelled by client + this.stopKeepAlive(streamId); this._streamMapping.delete(streamId); } }); @@ -766,6 +829,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { controller: streamController!, encoder, cleanup: () => { + this.stopKeepAlive(streamId); this._streamMapping.delete(streamId); try { streamController!.close(); @@ -778,6 +842,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 +967,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..2046a71697 100644 --- a/test/server/streamableHttp.test.ts +++ b/test/server/streamableHttp.test.ts @@ -3272,3 +3272,179 @@ 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(); + }); + + 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(); + }); +});