diff --git a/.changeset/add-sse-keepalive.md b/.changeset/add-sse-keepalive.md new file mode 100644 index 0000000000..78654d9d9a --- /dev/null +++ b/.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. diff --git a/packages/server/src/server/streamableHttp.ts b/packages/server/src/server/streamableHttp.ts index 7da5fb853c..4f1c4d51f6 100644 --- a/packages/server/src/server/streamableHttp.ts +++ b/packages/server/src/server/streamableHttp.ts @@ -72,6 +72,8 @@ interface StreamMapping { replayedEventIds?: Set; /** Cleanup function to close stream and remove mapping */ cleanup: () => void; + /** Per-stream keepalive timer; cleared by this stream's cleanup/cancel */ + keepAliveTimer?: ReturnType; } /** @@ -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; + let streamController!: ReadableStreamDefaultController; + + 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({ 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 | undefined; const readable = new ReadableStream({ 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,7 +609,7 @@ 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 } @@ -585,11 +617,12 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } }); - this._streamMapping.set(replayedStreamId, { + const mapping: StreamMapping = { controller: streamController!, encoder, replayedEventIds, cleanup: () => { + if (keepAliveTimer) clearInterval(keepAliveTimer); this._streamMapping.delete(replayedStreamId!); try { streamController!.close(); @@ -597,7 +630,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { // 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(); } diff --git a/packages/server/test/server/streamableHttp.test.ts b/packages/server/test/server/streamableHttp.test.ts index beca451113..01e0c1972c 100644 --- a/packages/server/test/server/streamableHttp.test.ts +++ b/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 { + 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); + }); + }); });