fix: send SSE keep-alive comment frames from WebStandardStreamableHTTPServerTransport

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, which clients observe as
"SSE stream disconnected: TypeError: terminated" followed by a
reconnect loop.

The 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. Naming matches the existing
keepAliveMs on createMcpHandler's subscriptions/listen streams.

v2 port of #2538 (v1.x); refs #1211

Author: mattzcarey

.changeset/streamable-http-sse-keepalive.md

@@ -0,0 +1,5 @@
+---
+'@modelcontextprotocol/server': patch
+---
+
+`WebStandardStreamableHTTPServerTransport` now writes 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).

packages/server/src/server/streamableHttp.ts

@@ -148,6 +148,19 @@ export interface WebStandardStreamableHTTPServerTransportOptions {
      */
     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;
+
     /**
      * List of protocol versions that this transport will accept.
      * Used to validate the `mcp-protocol-version` header in incoming requests.
@@ -161,6 +174,9 @@ export interface WebStandardStreamableHTTPServerTransportOptions {
     supportedProtocolVersions?: string[];
 }
 
+/** Default interval between SSE keep-alive comment frames. */
+const DEFAULT_KEEP_ALIVE_MS = 15_000;
+
 /**
  * Options for handling a request
  */
@@ -247,6 +263,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
     private _enableDnsRebindingProtection: boolean;
     private _retryInterval?: number;
     private _supportedProtocolVersions: string[];
+    private _keepAliveMs: number;
+    private _keepAliveTimers: Map<string, ReturnType<typeof setInterval>> = new Map();
 
     sessionId?: string;
     onclose?: () => void;
@@ -264,6 +282,47 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
         this._enableDnsRebindingProtection = options.enableDnsRebindingProtection ?? false;
         this._retryInterval = options.retryInterval;
         this._supportedProtocolVersions = options.supportedProtocolVersions ?? SUPPORTED_PROTOCOL_VERSIONS;
+        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 {@linkcode 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: InstanceType<typeof 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);
+        }
     }
 
     /**
@@ -473,6 +532,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
                 // it still points at THIS controller — a stale cancel must not
                 // delete a successor stream registered by a later GET/resume.
                 if (this._streamMapping.get(this._standaloneSseStreamId)?.controller === streamController) {
+                    this.stopKeepAlive(this._standaloneSseStreamId);
                     this._streamMapping.delete(this._standaloneSseStreamId);
                 }
             }
@@ -494,6 +554,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
             controller: streamController!,
             encoder,
             cleanup: () => {
+                this.stopKeepAlive(this._standaloneSseStreamId);
                 this._streamMapping.delete(this._standaloneSseStreamId);
                 try {
                     streamController!.close();
@@ -503,6 +564,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
             }
         });
 
+        this.startKeepAlive(this._standaloneSseStreamId, streamController!, encoder);
+
         return new Response(readable, { headers });
     }
 
@@ -564,6 +627,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
                     // a stale cancel from an earlier resume must not delete a
                     // successor resumed stream a re-poll has since registered.
                     if (replayedStreamId !== undefined && this._streamMapping.get(replayedStreamId)?.controller === streamController) {
+                        this.stopKeepAlive(replayedStreamId);
                         this._streamMapping.delete(replayedStreamId);
                     }
                 }
@@ -590,6 +654,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
                 encoder,
                 replayedEventIds,
                 cleanup: () => {
+                    this.stopKeepAlive(replayedStreamId!);
                     this._streamMapping.delete(replayedStreamId!);
                     try {
                         streamController!.close();
@@ -618,6 +683,12 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
                 }
             }
 
+            // Only arm keep-alive if the stream is still registered — the
+            // no-in-flight-request path above may have already closed it.
+            if (this._streamMapping.get(replayedStreamId)?.controller === streamController!) {
+                this.startKeepAlive(replayedStreamId, streamController!, encoder);
+            }
+
             return new Response(readable, { headers });
         } catch (error) {
             this.onerror?.(error as Error);
@@ -830,6 +901,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
                     // resumed stream under the same streamId) must not delete
                     // the successor.
                     if (this._streamMapping.get(streamId)?.controller === streamController) {
+                        this.stopKeepAlive(streamId);
                         this._streamMapping.delete(streamId);
                     }
                 }
@@ -854,6 +926,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
                         controller: streamController!,
                         encoder,
                         cleanup: () => {
+                            this.stopKeepAlive(streamId);
                             this._streamMapping.delete(streamId);
                             try {
                                 streamController!.close();
@@ -866,6 +939,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);
 
@@ -986,6 +1061,12 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
         }
         this._streamMapping.clear();
 
+        // Clear any keep-alive timers not already cleared by stream cleanup
+        for (const timer of this._keepAliveTimers.values()) {
+            clearInterval(timer);
+        }
+        this._keepAliveTimers.clear();
+
         // Clear any pending responses
         this._requestResponseMap.clear();
         this.onclose?.();

packages/server/test/server/streamableHttp.test.ts

@@ -1407,3 +1407,118 @@ describe('Zod v4', () => {
         });
     });
 });
+
+describe('WebStandardStreamableHTTPServerTransport SSE keep-alive', () => {
+    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(createRequest('POST', 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(createRequest('GET', undefined, { sessionId }));
+        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(createRequest('GET', undefined, { sessionId }));
+        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(createRequest('GET', undefined, { sessionId }));
+        const reader = response.body!.getReader();
+
+        await vi.advanceTimersByTimeAsync(60000);
+        const raced = await Promise.race([reader.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(createRequest('GET', undefined, { sessionId }));
+        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.registerTool('slow', { description: 'never resolves until released' }, async (): Promise<CallToolResult> => {
+            await new Promise<void>(resolve => {
+                resolveTool = resolve;
+            });
+            return { content: [{ type: 'text', text: 'done' }] };
+        });
+        await mcpServer.connect(transport);
+
+        const initResponse = await transport.handleRequest(createRequest('POST', TEST_MESSAGES.initialize));
+        const sessionId = initResponse.headers.get('mcp-session-id') as string;
+
+        const response = await transport.handleRequest(
+            createRequest(
+                'POST',
+                { jsonrpc: '2.0', method: 'tools/call', params: { name: 'slow', arguments: {} }, id: 'call-1' } as JSONRPCMessage,
+                {
+                    sessionId
+                }
+            )
+        );
+        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();
+    });
+});