fix: send SSE keep-alive comment frames from Streamable HTTP server transport (v1.x)

.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).

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<string, ReturnType<typeof setInterval>> = 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<Uint8Array>, 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?.();

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<string, string> }): Request {
+        const headers: Record<string, string> = { ...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<void>(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<EventId> {
+                return 'evt-1';
+            },
+            async replayEventsAfter(): Promise<StreamId> {
+                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();
+    });
+});