feat(server): add keepAliveInterval for standalone GET SSE stream

.changeset/add-sse-keepalive.md

@@ -0,0 +1,5 @@
+---
+'@modelcontextprotocol/server': minor
+---
+
+Add optional `keepAliveInterval` to `WebStandardStreamableHTTPServerTransportOptions` that sends periodic SSE comments on the standalone GET stream to prevent reverse proxy idle timeout disconnections.

packages/server/src/server/streamableHttp.ts

@@ -72,6 +72,8 @@ interface StreamMapping {
     replayedEventIds?: Set<string>;
     /** Cleanup function to close stream and remove mapping */
     cleanup: () => void;
+    /** Per-stream keepalive timer; cleared by this stream's cleanup/cancel */
+    keepAliveTimer?: ReturnType<typeof setInterval>;
 }
 
 /**
@@ -148,6 +150,15 @@ export interface WebStandardStreamableHTTPServerTransportOptions {
      */
     retryInterval?: number;
 
+    /**
+     * Interval in milliseconds for sending SSE keepalive comments on the standalone
+     * GET SSE stream. When set, the transport sends periodic SSE comments
+     * (`: keepalive`) to prevent reverse proxies from closing idle connections.
+     *
+     * Disabled by default (no keepalive comments are sent).
+     */
+    keepAliveInterval?: number;
+
     /**
      * List of protocol versions that this transport will accept.
      * Used to validate the `mcp-protocol-version` header in incoming requests.
@@ -246,6 +257,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
     private _allowedOrigins?: string[];
     private _enableDnsRebindingProtection: boolean;
     private _retryInterval?: number;
+    private _keepAliveInterval?: number;
     private _supportedProtocolVersions: string[];
 
     sessionId?: string;
@@ -263,6 +275,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
         this._allowedOrigins = options.allowedOrigins;
         this._enableDnsRebindingProtection = options.enableDnsRebindingProtection ?? false;
         this._retryInterval = options.retryInterval;
+        this._keepAliveInterval = options.keepAliveInterval;
         this._supportedProtocolVersions = options.supportedProtocolVersions ?? SUPPORTED_PROTOCOL_VERSIONS;
     }
 
@@ -461,18 +474,33 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
         }
 
         const encoder = new TextEncoder();
-        let streamController: ReadableStreamDefaultController<Uint8Array>;
+        let streamController!: ReadableStreamDefaultController<Uint8Array>;
+
+        const mapping: StreamMapping = {
+            encoder,
+            cleanup: () => {
+                if (mapping.keepAliveTimer) clearInterval(mapping.keepAliveTimer);
+                this._streamMapping.delete(this._standaloneSseStreamId);
+                try {
+                    streamController.close();
+                } catch {
+                    // Controller might already be closed
+                }
+            }
+        };
 
         // Create a ReadableStream with a controller we can use to push SSE events
         const readable = new ReadableStream<Uint8Array>({
             start: controller => {
                 streamController = controller;
+                mapping.controller = controller;
             },
             cancel: () => {
                 // Stream was cancelled by client. Only drop the mapping when
                 // 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) {
+                    if (mapping.keepAliveTimer) clearInterval(mapping.keepAliveTimer);
                     this._streamMapping.delete(this._standaloneSseStreamId);
                 }
             }
@@ -489,19 +517,20 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
             headers['mcp-session-id'] = this.sessionId;
         }
 
-        // Store the stream mapping with the controller for pushing data
-        this._streamMapping.set(this._standaloneSseStreamId, {
-            controller: streamController!,
-            encoder,
-            cleanup: () => {
-                this._streamMapping.delete(this._standaloneSseStreamId);
+        this._streamMapping.set(this._standaloneSseStreamId, mapping);
+
+        // Start keepalive timer to send periodic SSE comments that prevent
+        // reverse proxies from closing the connection due to idle timeouts
+        if (this._keepAliveInterval !== undefined) {
+            mapping.keepAliveTimer = setInterval(() => {
                 try {
-                    streamController!.close();
+                    streamController.enqueue(encoder.encode(': keepalive\n\n'));
                 } catch {
-                    // Controller might already be closed
+                    // Controller is closed or errored, stop sending keepalives
+                    if (mapping.keepAliveTimer) clearInterval(mapping.keepAliveTimer);
                 }
-            }
-        });
+            }, this._keepAliveInterval);
+        }
 
         return new Response(readable, { headers });
     }
@@ -551,18 +580,21 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
             // replayEventsAfter resolves) — must be `let`.
             // eslint-disable-next-line prefer-const
             let replayedStreamId: string | undefined;
+            let keepAliveTimer: ReturnType<typeof setInterval> | undefined;
 
             const readable = new ReadableStream<Uint8Array>({
                 start: controller => {
                     streamController = controller;
                 },
                 cancel: () => {
-                    // Stream was cancelled by client — drop the mapping so a
-                    // subsequent reconnect with the same Last-Event-ID is not
-                    // refused with 409 by the conflict check above. Only delete
-                    // when the mapped entry is still THIS closure's controller:
-                    // a stale cancel from an earlier resume must not delete a
-                    // successor resumed stream a re-poll has since registered.
+                    // Always clear the closure-local keepalive timer; the
+                    // mapping may already have been removed by the early-close
+                    // block for completed requests, but the timer is still ours.
+                    if (keepAliveTimer) clearInterval(keepAliveTimer);
+                    // Drop the mapping so a subsequent reconnect with the same
+                    // Last-Event-ID is not refused with 409. Only delete when
+                    // the mapped entry is still THIS closure's controller: a
+                    // stale cancel must not delete a successor resumed stream.
                     if (replayedStreamId !== undefined && this._streamMapping.get(replayedStreamId)?.controller === streamController) {
                         this._streamMapping.delete(replayedStreamId);
                     }
@@ -577,27 +609,29 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
                     const success = this.writeSSEEvent(streamController!, encoder, message, eventId);
                     if (!success) {
                         try {
-                            streamController!.close();
+                            streamController.close();
                         } catch {
                             // Controller might already be closed
                         }
                     }
                 }
             });
 
-            this._streamMapping.set(replayedStreamId, {
+            const mapping: StreamMapping = {
                 controller: streamController!,
                 encoder,
                 replayedEventIds,
                 cleanup: () => {
+                    if (keepAliveTimer) clearInterval(keepAliveTimer);
                     this._streamMapping.delete(replayedStreamId!);
                     try {
                         streamController!.close();
                     } catch {
                         // Controller might already be closed
                     }
                 }
-            });
+            };
+            this._streamMapping.set(replayedStreamId!, mapping);
 
             // If this is a per-request stream and no in-flight request still
             // targets this streamId, the request was already retired by the
@@ -618,6 +652,20 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
                 }
             }
 
+            // Start keepalive timer for the replayed stream so reconnecting
+            // clients remain protected from proxy idle timeouts.
+            // Skip if the early-close block above already removed the mapping.
+            if (this._keepAliveInterval !== undefined && this._streamMapping.has(replayedStreamId!)) {
+                keepAliveTimer = setInterval(() => {
+                    try {
+                        streamController!.enqueue(encoder.encode(': keepalive\n\n'));
+                    } catch {
+                        if (keepAliveTimer) clearInterval(keepAliveTimer);
+                    }
+                }, this._keepAliveInterval);
+                mapping.keepAliveTimer = keepAliveTimer;
+            }
+
             return new Response(readable, { headers });
         } catch (error) {
             this.onerror?.(error as Error);
@@ -980,7 +1028,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
         }
         this._closed = true;
 
-        // Close all SSE connections
+        // Close all SSE connections (each cleanup() also clears its own keepAliveTimer)
         for (const { cleanup } of this._streamMapping.values()) {
             cleanup();
         }

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

@@ -1406,4 +1406,148 @@ describe('Zod v4', () => {
             expect(cleanupCalls).toEqual(['stream-1']);
         });
     });
+
+    describe('HTTPServerTransport - keepAliveInterval', () => {
+        let transport: WebStandardStreamableHTTPServerTransport;
+        let mcpServer: McpServer;
+
+        beforeEach(() => {
+            vi.useFakeTimers();
+        });
+
+        afterEach(async () => {
+            vi.useRealTimers();
+            await transport.close();
+        });
+
+        async function setupTransport(keepAliveInterval?: number): Promise<string> {
+            mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' });
+
+            transport = new WebStandardStreamableHTTPServerTransport({
+                sessionIdGenerator: () => randomUUID(),
+                keepAliveInterval
+            });
+
+            await mcpServer.connect(transport);
+
+            const initReq = createRequest('POST', TEST_MESSAGES.initialize);
+            const initRes = await transport.handleRequest(initReq);
+            return initRes.headers.get('mcp-session-id') as string;
+        }
+
+        it('should send SSE keepalive comments periodically when keepAliveInterval is set', async () => {
+            const sessionId = await setupTransport(50);
+
+            const getReq = createRequest('GET', undefined, { sessionId });
+            const getRes = await transport.handleRequest(getReq);
+
+            expect(getRes.status).toBe(200);
+            expect(getRes.body).not.toBeNull();
+
+            const reader = getRes.body!.getReader();
+
+            // Advance past two intervals to accumulate keepalive comments
+            vi.advanceTimersByTime(120);
+
+            const { value } = await reader.read();
+            const text = new TextDecoder().decode(value);
+            expect(text).toContain(': keepalive');
+        });
+
+        it('should not send SSE comments when keepAliveInterval is not set', async () => {
+            const sessionId = await setupTransport(undefined);
+
+            const getReq = createRequest('GET', undefined, { sessionId });
+            const getRes = await transport.handleRequest(getReq);
+
+            expect(getRes.status).toBe(200);
+            expect(getRes.body).not.toBeNull();
+
+            const reader = getRes.body!.getReader();
+
+            // Advance time; no keepalive should be enqueued
+            vi.advanceTimersByTime(200);
+
+            // Close the transport to end the stream, then read whatever was buffered
+            await transport.close();
+
+            const chunks: string[] = [];
+            for (let result = await reader.read(); !result.done; result = await reader.read()) {
+                chunks.push(new TextDecoder().decode(result.value));
+            }
+
+            const allText = chunks.join('');
+            expect(allText).not.toContain(': keepalive');
+        });
+
+        it('should clear the keepalive interval when the transport is closed', async () => {
+            const sessionId = await setupTransport(50);
+
+            const getReq = createRequest('GET', undefined, { sessionId });
+            const getRes = await transport.handleRequest(getReq);
+
+            expect(getRes.status).toBe(200);
+            expect(vi.getTimerCount()).toBe(1);
+
+            // Close the transport, which should clear the interval
+            await transport.close();
+
+            expect(vi.getTimerCount()).toBe(0);
+        });
+
+        it('should maintain independent keepalive timers per concurrent stream', async () => {
+            // Minimal event store so a GET with last-event-id triggers replayEvents()
+            // and creates a second StreamMapping entry concurrently with the standalone GET.
+            const eventStore: EventStore = {
+                async storeEvent(streamId) {
+                    return `${streamId}_evt`;
+                },
+                async getStreamIdForEventId() {
+                    return 'replay-stream';
+                },
+                async replayEventsAfter() {
+                    return 'replay-stream';
+                }
+            };
+
+            mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' });
+            transport = new WebStandardStreamableHTTPServerTransport({
+                sessionIdGenerator: () => randomUUID(),
+                keepAliveInterval: 50,
+                eventStore
+            });
+            await mcpServer.connect(transport);
+            const initRes = await transport.handleRequest(createRequest('POST', TEST_MESSAGES.initialize));
+            const sessionId = initRes.headers.get('mcp-session-id') as string;
+
+            // Stream A: standalone GET
+            const resA = await transport.handleRequest(createRequest('GET', undefined, { sessionId }));
+            expect(resA.status).toBe(200);
+            const readerA = resA.body!.getReader();
+            expect(vi.getTimerCount()).toBe(1);
+
+            // Stream B: GET with last-event-id -> replayEvents path, separate mapping key.
+            // Simulate an in-flight request for the replay stream so the
+            // early-close-for-completed-requests block keeps the stream open.
+            (transport as any)._requestToStreamMapping.set('fake-req', 'replay-stream');
+            const resB = await transport.handleRequest(
+                createRequest('GET', undefined, { sessionId, extraHeaders: { 'last-event-id': 'evt-1' } })
+            );
+            expect(resB.status).toBe(200);
+            const readerB = resB.body!.getReader();
+            expect(vi.getTimerCount()).toBe(2);
+
+            // Cancel stream B; its keepalive timer must be cleared without affecting A's
+            await readerB.cancel();
+            expect(vi.getTimerCount()).toBe(1);
+
+            // Stream A still receives keepalives after B is cancelled
+            vi.advanceTimersByTime(60);
+            const { value } = await readerA.read();
+            expect(new TextDecoder().decode(value)).toContain(': keepalive');
+
+            await readerA.cancel();
+            expect(vi.getTimerCount()).toBe(0);
+        });
+    });
 });