From 9505f9339d32b398cd4bd2fdea7854d4c416c729 Mon Sep 17 00:00:00 2001 From: Matt Carey Date: Thu, 23 Jul 2026 16:58:44 +0100 Subject: [PATCH 1/7] 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 --- .changeset/streamable-http-sse-keepalive.md | 5 + packages/server/src/server/streamableHttp.ts | 81 ++++++++++++ .../server/test/server/streamableHttp.test.ts | 115 ++++++++++++++++++ 3 files changed, 201 insertions(+) create mode 100644 .changeset/streamable-http-sse-keepalive.md diff --git a/.changeset/streamable-http-sse-keepalive.md b/.changeset/streamable-http-sse-keepalive.md new file mode 100644 index 0000000000..b4ba37f06e --- /dev/null +++ b/.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). diff --git a/packages/server/src/server/streamableHttp.ts b/packages/server/src/server/streamableHttp.ts index 7da5fb853c..709e802abe 100644 --- a/packages/server/src/server/streamableHttp.ts +++ b/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> = 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, + encoder: InstanceType + ): 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?.(); diff --git a/packages/server/test/server/streamableHttp.test.ts b/packages/server/test/server/streamableHttp.test.ts index beca451113..7cb4456010 100644 --- a/packages/server/test/server/streamableHttp.test.ts +++ b/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 => { + await new Promise(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(); + }); +}); From 6d20a1f19bdce5c146cb2a83fb7369c814b95890 Mon Sep 17 00:00:00 2001 From: Matt Carey Date: Thu, 23 Jul 2026 17:46:56 +0100 Subject: [PATCH 2/7] fix: address keep-alive lifecycle review findings - startKeepAlive is a no-op after transport close: a deferred arm (e.g. resuming after an event-store replay await that straddled close()) would otherwise create a timer that close()'s sweep can never clear. - The POST SSE path arms keep-alive after the fallible awaits (priming event write, message dispatch) instead of before them, so an error path that discards the Response cannot leak a permanently-firing timer against a stream nothing can cancel. - createMcpHandler's keepAliveMs now also reaches the legacy stateless fallback's per-request transport, instead of governing only subscriptions/listen streams while the legacy leg silently used the transport default. - Documents the keep-alive behavior and the 'SSE stream disconnected: TypeError: terminated' symptom in docs/troubleshooting.md. --- docs/troubleshooting.md | 8 ++ .../server/src/server/createMcpHandler.ts | 23 +++-- packages/server/src/server/streamableHttp.ts | 14 ++- .../server/test/server/streamableHttp.test.ts | 90 +++++++++++++++++++ 4 files changed, 127 insertions(+), 8 deletions(-) diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 1fc325b7e4..533eaf3927 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -154,6 +154,14 @@ Rewrite the imports: The Resource Server helpers did not move there: `requireBearerAuth`, `mcpAuthMetadataRouter` and `OAuthTokenVerifier` are first-class in `@modelcontextprotocol/express` — see [Authorization](./serving/authorization.md). `@modelcontextprotocol/server-legacy` is frozen and receives no new features; serve new code over [Streamable HTTP](./serving/http.md), which still reaches 2025-era clients through [legacy client support](./serving/legacy-clients.md). A client limited to the HTTP+SSE transport is the one case that still needs the frozen `@modelcontextprotocol/server-legacy/sse` import above. +## `SSE stream disconnected: TypeError: terminated` + +An idle SSE stream was killed by an intermediary or an idle-connection timeout — Node's `server.requestTimeout` defaults to 300 seconds, and reverse proxies and cloud load balancers have similar watchdogs. The client observes the dropped socket as this error (typically every ~5 minutes) and reconnects in a loop. + +`WebStandardStreamableHTTPServerTransport` prevents this by writing an SSE comment frame (`: keepalive`) to every open SSE stream every 15 seconds by default. Comment frames are dropped by SSE parsers before event dispatch, so they never surface as protocol messages. Tune or disable the interval with the transport's `keepAliveMs` option (`0` disables); `createMcpHandler`'s `keepAliveMs` option covers both its `subscriptions/listen` streams and the legacy fallback's per-request transport. + +If you still see this error, either keep-alive is disabled (`keepAliveMs: 0`) or an intermediary between client and server buffers or strips SSE data — check for proxies that buffer streaming responses (e.g. nginx without `proxy_buffering off`). + ## Recap - Every heading on this page is the exact message you searched for. diff --git a/packages/server/src/server/createMcpHandler.ts b/packages/server/src/server/createMcpHandler.ts index f17deaa860..89841231a3 100644 --- a/packages/server/src/server/createMcpHandler.ts +++ b/packages/server/src/server/createMcpHandler.ts @@ -194,8 +194,9 @@ export interface CreateMcpHandlerOptions { */ maxSubscriptions?: number; /** - * SSE comment-frame keepalive interval for `subscriptions/listen` streams, - * in milliseconds. Set to `0` to disable. + * SSE comment-frame keepalive interval, in milliseconds, applied to + * `subscriptions/listen` streams and to the SSE streams of the legacy + * stateless fallback's per-request transport. Set to `0` to disable. * @default 15000 */ keepAliveMs?: number; @@ -306,7 +307,11 @@ function internalServerErrorResponse(id: RequestId | null = null): Response { * The entry passes its own `onerror` here when expanding the default, so * legacy-leg failures are never silently swallowed. */ -export function legacyStatelessFallback(factory: McpServerFactory, onerror?: (error: Error) => void): LegacyHttpHandler { +export function legacyStatelessFallback( + factory: McpServerFactory, + onerror?: (error: Error) => void, + transportOptions?: { keepAliveMs?: number } +): LegacyHttpHandler { return async (request, options) => { if (request.method.toUpperCase() !== 'POST') { return jsonRpcErrorResponse(405, -32_000, 'Method not allowed.'); @@ -317,7 +322,10 @@ export function legacyStatelessFallback(factory: McpServerFactory, onerror?: (er ...(options?.authInfo !== undefined && { authInfo: options.authInfo }), requestInfo: request }); - const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); + const transport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: undefined, + ...(transportOptions?.keepAliveMs !== undefined && { keepAliveMs: transportOptions.keepAliveMs }) + }); await product.connect(transport); const teardown = () => { @@ -632,7 +640,12 @@ export function createMcpHandler(factory: McpServerFactory, options: CreateMcpHa // The default posture is the stateless fallback; 'reject' is the only way // to turn legacy serving off (modern-only strict). - const legacyHandler: LegacyHttpHandler | undefined = legacy === 'reject' ? undefined : legacyStatelessFallback(factory, reportError); + const legacyHandler: LegacyHttpHandler | undefined = + legacy === 'reject' + ? undefined + : legacyStatelessFallback(factory, reportError, { + ...(options.keepAliveMs !== undefined && { keepAliveMs: options.keepAliveMs }) + }); async function serveModern(route: InboundModernRoute, request: Request, authInfo: AuthInfo | undefined): Promise { const claimedRevision = route.classification.revision; diff --git a/packages/server/src/server/streamableHttp.ts b/packages/server/src/server/streamableHttp.ts index 709e802abe..de3760130b 100644 --- a/packages/server/src/server/streamableHttp.ts +++ b/packages/server/src/server/streamableHttp.ts @@ -298,7 +298,9 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { controller: ReadableStreamDefaultController, encoder: InstanceType ): void { - if (this._keepAliveMs <= 0) { + // A deferred arm (e.g. after an event-store await) must not outlive the + // transport: close()'s timer sweep has already run and never runs again. + if (this._keepAliveMs <= 0 || this._closed) { return; } this.stopKeepAlive(streamId); @@ -939,8 +941,6 @@ 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); @@ -966,6 +966,14 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { // The server SHOULD NOT close the SSE stream before sending all JSON-RPC responses // This will be handled by the send() method when responses are ready + // Arm keep-alive only after the fallible awaits above — an error + // path returning 400 discards the Response, so nothing could ever + // cancel the stream and clear an already-armed timer. Skip if the + // responses already completed and cleaned the stream up. + if (this._streamMapping.get(streamId)?.controller === streamController!) { + this.startKeepAlive(streamId, streamController!, encoder); + } + return new Response(readable, { status: 200, headers }); } catch (error) { // return JSON-RPC formatted error diff --git a/packages/server/test/server/streamableHttp.test.ts b/packages/server/test/server/streamableHttp.test.ts index 7cb4456010..f933a947a5 100644 --- a/packages/server/test/server/streamableHttp.test.ts +++ b/packages/server/test/server/streamableHttp.test.ts @@ -1522,3 +1522,93 @@ describe('WebStandardStreamableHTTPServerTransport SSE keep-alive', () => { await transport.close(); }); }); + +describe('WebStandardStreamableHTTPServerTransport SSE keep-alive lifecycle', () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it('should not arm keep-alive when the transport closes during an event-store replay await', async () => { + let releaseReplay: (() => void) | undefined; + const eventStore: EventStore = { + async storeEvent(): Promise { + return 'evt-1'; + }, + async replayEventsAfter(): Promise { + await new Promise(resolve => { + releaseReplay = resolve; + }); + 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(createRequest('POST', TEST_MESSAGES.initialize)); + const sessionId = initResponse.headers.get('mcp-session-id') as string; + + // Enter replayEvents and park on the replayEventsAfter await + const pendingGet = transport.handleRequest( + createRequest('GET', undefined, { sessionId, extraHeaders: { 'Last-Event-ID': 'evt-1' } }) + ); + await vi.advanceTimersByTimeAsync(0); + expect(releaseReplay).toBeDefined(); + + // Close the transport mid-await, then let the replay continuation run + await transport.close(); + releaseReplay?.(); + await pendingGet; + + // The deferred continuation must not have armed a timer close() can never sweep + expect(vi.getTimerCount()).toBe(0); + }); + + it('should not leak a keep-alive timer when the priming event write fails on a POST SSE stream', async () => { + // Healthy during initialization, then the store starts failing — the + // tool call's priming event write must reject inside handlePostRequest + let storeFails = false; + const eventStore: EventStore = { + async storeEvent(): Promise { + if (storeFails) { + throw new Error('event store unavailable'); + } + return `evt-${randomUUID()}`; + }, + async replayEventsAfter(): Promise { + return 'stream-1'; + } + }; + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), eventStore }); + const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }); + mcpServer.registerTool('noop', { description: 'noop' }, async (): Promise => ({ content: [] })); + await mcpServer.connect(transport); + + const initResponse = await transport.handleRequest(createRequest('POST', TEST_MESSAGES.initialize)); + expect(initResponse.status).toBe(200); + const sessionId = initResponse.headers.get('mcp-session-id') as string; + // Let the init response finish sending (its send() stores an event and + // then cleans up the init stream's keep-alive) before failing the store + await vi.advanceTimersByTimeAsync(0); + expect(vi.getTimerCount()).toBe(0); + storeFails = true; + + const response = await transport.handleRequest( + createRequest( + 'POST', + { jsonrpc: '2.0', method: 'tools/call', params: { name: 'noop', arguments: {} }, id: 'call-1' } as JSONRPCMessage, + { + sessionId + } + ) + ); + expect(response.status).toBe(400); + + // The discarded stream must not carry a permanently-firing timer + expect(vi.getTimerCount()).toBe(0); + + await transport.close(); + }); +}); From 3ec1fd1efaf239f9c6667d23f4658b0e4e4dc396 Mon Sep 17 00:00:00 2001 From: Matt Carey Date: Fri, 24 Jul 2026 10:56:07 +0100 Subject: [PATCH 3/7] fix: extend SSE keep-alive to the modern per-request leg; address review round 2 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - PerRequestHTTPServerTransport gains keepAliveMs (default 15000, 0 disables): while an exchange's SSE stream is open, an interval drives writeCommentFrame so a long-running handler with no mid-call output doesn't idle past intermediary/server timeouts — the same failure the session transport fix targets, previously unaddressed on the modern serving path. Threaded from createMcpHandler through invoke(), so the handler's keepAliveMs now uniformly covers listen streams, modern per-request exchanges, and the legacy stateless fallback. - Keep-alive guards use the > 0 polarity (matching listenRouter) so a non-finite keepAliveMs disables keep-alive instead of arming a Node- clamped ~1ms interval. - The close-during-replay regression test resumes the standalone GET stream so the continuation genuinely reaches the keep-alive arm (mutation-verified: removing the _closed guard now fails the test). - Reworded the two stale legacy-fallback doc blocks that still claimed the transport is constructed 'with only sessionIdGenerator: undefined', documented legacyStatelessFallback's transportOptions parameter, and scoped the troubleshooting entry to match actual coverage. --- .changeset/streamable-http-sse-keepalive.md | 2 +- docs/troubleshooting.md | 2 +- .../server/src/server/createMcpHandler.ts | 29 +++-- packages/server/src/server/invoke.ts | 9 +- .../server/src/server/perRequestTransport.ts | 43 +++++++ .../test/server/perRequestStreaming.test.ts | 107 +++++++++++++++++- .../server/test/server/streamableHttp.test.ts | 6 +- 7 files changed, 180 insertions(+), 18 deletions(-) diff --git a/.changeset/streamable-http-sse-keepalive.md b/.changeset/streamable-http-sse-keepalive.md index b4ba37f06e..77347b39fb 100644 --- a/.changeset/streamable-http-sse-keepalive.md +++ b/.changeset/streamable-http-sse-keepalive.md @@ -2,4 +2,4 @@ '@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). +SSE streams served by the SDK now emit keep-alive comment frames (`: keepalive`) so idle connections (e.g. the standalone GET stream, or a stream during a long-running tool call) are not killed by intermediaries or server idle timeouts. `WebStandardStreamableHTTPServerTransport` and `PerRequestHTTPServerTransport` gain a `keepAliveMs` option (default 15000; set 0 to disable), and `createMcpHandler`'s existing `keepAliveMs` now covers modern per-request exchange streams and the legacy stateless fallback in addition to `subscriptions/listen` streams. diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 533eaf3927..543828747c 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -158,7 +158,7 @@ The Resource Server helpers did not move there: `requireBearerAuth`, `mcpAuthMet An idle SSE stream was killed by an intermediary or an idle-connection timeout — Node's `server.requestTimeout` defaults to 300 seconds, and reverse proxies and cloud load balancers have similar watchdogs. The client observes the dropped socket as this error (typically every ~5 minutes) and reconnects in a loop. -`WebStandardStreamableHTTPServerTransport` prevents this by writing an SSE comment frame (`: keepalive`) to every open SSE stream every 15 seconds by default. Comment frames are dropped by SSE parsers before event dispatch, so they never surface as protocol messages. Tune or disable the interval with the transport's `keepAliveMs` option (`0` disables); `createMcpHandler`'s `keepAliveMs` option covers both its `subscriptions/listen` streams and the legacy fallback's per-request transport. +The SDK's HTTP serving prevents this by writing an SSE comment frame (`: keepalive`) to every open SSE stream every 15 seconds by default — `WebStandardStreamableHTTPServerTransport` on all of its streams, and `createMcpHandler` on `subscriptions/listen` streams, modern per-request exchange streams, and the legacy fallback's per-request transport. Comment frames are dropped by SSE parsers before event dispatch, so they never surface as protocol messages. Tune or disable the interval with the `keepAliveMs` option on the transport or handler (`0` disables). If you still see this error, either keep-alive is disabled (`keepAliveMs: 0`) or an intermediary between client and server buffers or strips SSE data — check for proxies that buffer streaming responses (e.g. nginx without `proxy_buffering off`). diff --git a/packages/server/src/server/createMcpHandler.ts b/packages/server/src/server/createMcpHandler.ts index 89841231a3..fb924c101e 100644 --- a/packages/server/src/server/createMcpHandler.ts +++ b/packages/server/src/server/createMcpHandler.ts @@ -145,8 +145,9 @@ export interface CreateMcpHandlerOptions { * - `'stateless'` (the default, also when the option is omitted) — * old-school stateless serving: each legacy request is answered by a * fresh instance from the same factory over a streamable HTTP transport - * constructed with only `sessionIdGenerator: undefined` (the established - * stateless idiom). Because serving is per-request and stateless, GET and + * constructed with `sessionIdGenerator: undefined` (the established + * stateless idiom), plus the handler's `keepAliveMs` when provided. + * Because serving is per-request and stateless, GET and * DELETE (2025 session operations) are answered with `405` / * `Method not allowed.`. * - `'reject'` — modern-only strict: legacy-classified requests are @@ -194,9 +195,10 @@ export interface CreateMcpHandlerOptions { */ maxSubscriptions?: number; /** - * SSE comment-frame keepalive interval, in milliseconds, applied to - * `subscriptions/listen` streams and to the SSE streams of the legacy - * stateless fallback's per-request transport. Set to `0` to disable. + * SSE comment-frame keepalive interval, in milliseconds, applied to every + * SSE stream this handler serves: `subscriptions/listen` streams, modern + * per-request exchange streams, and the legacy stateless fallback's + * per-request transport. Set to `0` to disable. * @default 15000 */ keepAliveMs?: number; @@ -296,16 +298,20 @@ function internalServerErrorResponse(id: RequestId | null = null): Response { * strict modern endpoint). * * Each POST is served by a fresh instance from the factory connected to a - * fresh streamable HTTP transport constructed with only - * `sessionIdGenerator: undefined` — the established stateless idiom, unchanged. - * Because serving is per-request and stateless, GET and DELETE (2025 session - * operations) are answered with `405` / `Method not allowed.`, exactly like the - * canonical stateless example. + * fresh streamable HTTP transport constructed with + * `sessionIdGenerator: undefined` (the established stateless idiom) plus any + * `transportOptions`. Because serving is per-request and stateless, GET and + * DELETE (2025 session operations) are answered with `405` / + * `Method not allowed.`, exactly like the canonical stateless example. * * The optional `onerror` callback receives factory and serving failures on * this leg (reporting only — the response stays the 500 internal-error body). * The entry passes its own `onerror` here when expanding the default, so * legacy-leg failures are never silently swallowed. + * + * The optional `transportOptions` are threaded into each per-request + * transport; currently just `keepAliveMs`, the SSE keep-alive comment-frame + * interval (the entry forwards its own `keepAliveMs` option here). */ export function legacyStatelessFallback( factory: McpServerFactory, @@ -791,7 +797,8 @@ export function createMcpHandler(factory: McpServerFactory, options: CreateMcpHa classification: route.classification, request, ...(authInfo !== undefined && { authInfo }), - ...(responseMode !== undefined && { responseMode }) + ...(responseMode !== undefined && { responseMode }), + ...(options.keepAliveMs !== undefined && { keepAliveMs: options.keepAliveMs }) }); if (route.messageKind === 'notification') { // Notification exchanges have no terminal response to ride the diff --git a/packages/server/src/server/invoke.ts b/packages/server/src/server/invoke.ts index 6966968604..97b5b17553 100644 --- a/packages/server/src/server/invoke.ts +++ b/packages/server/src/server/invoke.ts @@ -35,6 +35,12 @@ export interface InvokeContext { authInfo?: AuthInfo; /** Response shaping for the exchange; defaults to `auto` (lazy SSE upgrade). */ responseMode?: PerRequestResponseMode; + /** + * SSE keep-alive comment-frame interval for the exchange's stream, in + * milliseconds; passed through to the per-request transport. `0` disables. + * @default 15000 + */ + keepAliveMs?: number; } /** @@ -58,7 +64,8 @@ export async function invoke( ): Promise { const transport = new PerRequestHTTPServerTransport({ classification: ctx.classification, - ...(ctx.responseMode !== undefined && { responseMode: ctx.responseMode }) + ...(ctx.responseMode !== undefined && { responseMode: ctx.responseMode }), + ...(ctx.keepAliveMs !== undefined && { keepAliveMs: ctx.keepAliveMs }) }); await server.connect(transport); return transport.handleMessage(message, { diff --git a/packages/server/src/server/perRequestTransport.ts b/packages/server/src/server/perRequestTransport.ts index 5003946404..fae073c6e2 100644 --- a/packages/server/src/server/perRequestTransport.ts +++ b/packages/server/src/server/perRequestTransport.ts @@ -79,8 +79,19 @@ export interface PerRequestHTTPServerTransportOptions { classification: MessageClassification; /** Response shaping for the exchange; defaults to `auto`. */ responseMode?: PerRequestResponseMode; + /** + * Interval in milliseconds between SSE keep-alive comment frames + * (`: keepalive`) written while the exchange's SSE stream is open, so a + * long-running handler with no mid-call output doesn't idle past + * intermediary and server idle timeouts. Set to `0` to disable. + * @default 15000 + */ + keepAliveMs?: number; } +/** Default interval between SSE keep-alive comment frames. */ +const DEFAULT_KEEP_ALIVE_MS = 15_000; + /** Per-exchange context handed to {@linkcode PerRequestHTTPServerTransport.handleMessage}. */ export interface PerRequestMessageExtra { /** @@ -140,10 +151,13 @@ export class PerRequestHTTPServerTransport implements Transport { private _deferredResponse?: DeferredResponse; private _sse?: SseSink; private _abortCleanup?: () => void; + private readonly _keepAliveMs: number; + private _keepAliveTimer?: ReturnType; constructor(options: PerRequestHTTPServerTransportOptions) { this._classification = options.classification; this._responseMode = options.responseMode ?? 'auto'; + this._keepAliveMs = options.keepAliveMs ?? DEFAULT_KEEP_ALIVE_MS; } async start(): Promise { @@ -342,6 +356,7 @@ export class PerRequestHTTPServerTransport implements Transport { this._abortCleanup?.(); this._abortCleanup = undefined; + this.stopKeepAlive(); if (this._sse !== undefined && !this._sse.closed) { this._sse.closed = true; @@ -382,6 +397,7 @@ export class PerRequestHTTPServerTransport implements Transport { } }); this._sse = { controller, encoder: new TextEncoder(), closed: false }; + this.startKeepAlive(); this.settleResponse( new Response(readable, { @@ -398,7 +414,34 @@ export class PerRequestHTTPServerTransport implements Transport { ); } + /** + * Arms the exchange's keep-alive interval, writing an SSE comment frame + * every `keepAliveMs` while the stream is open. `writeCommentFrame` + * already drops frames once the exchange is closed or the stream is + * finalized, so the interval body needs no extra guards; the timer itself + * is cleared on stream finalization and transport close. + * Uses the `> 0` polarity so a non-finite value disables keep-alive + * instead of arming a clamped ~1ms interval. + */ + private startKeepAlive(): void { + if (!(this._keepAliveMs > 0) || this._closed) { + return; + } + const timer = setInterval(() => this.writeCommentFrame('keepalive'), this._keepAliveMs); + // Don't let the keep-alive timer hold the process open (Node.js only) + (timer as { unref?: () => void }).unref?.(); + this._keepAliveTimer = timer; + } + + private stopKeepAlive(): void { + if (this._keepAliveTimer !== undefined) { + clearInterval(this._keepAliveTimer); + this._keepAliveTimer = undefined; + } + } + private finalizeStream(): void { + this.stopKeepAlive(); if (this._sse !== undefined && !this._sse.closed) { this._sse.closed = true; try { diff --git a/packages/server/test/server/perRequestStreaming.test.ts b/packages/server/test/server/perRequestStreaming.test.ts index 6d350ed2ef..7dfc4578cb 100644 --- a/packages/server/test/server/perRequestStreaming.test.ts +++ b/packages/server/test/server/perRequestStreaming.test.ts @@ -11,7 +11,7 @@ import { PROTOCOL_VERSION_META_KEY, setNegotiatedProtocolVersion } from '@modelcontextprotocol/core-internal'; -import { describe, expect, it } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import type { PerRequestResponseMode } from '../../src/server/perRequestTransport'; import { PerRequestHTTPServerTransport } from '../../src/server/perRequestTransport'; @@ -46,14 +46,16 @@ interface StreamingSetup { async function setup( handler: (ctx: ServerContext) => Promise, - responseMode?: PerRequestResponseMode + responseMode?: PerRequestResponseMode, + keepAliveMs?: number ): Promise { const server = new Server({ name: 'streaming-test', version: '1.0.0' }, { capabilities: { tools: {} } }); server.setRequestHandler('tools/call', async (_request, ctx) => handler(ctx)); setNegotiatedProtocolVersion(server, MODERN_REVISION); const transport = new PerRequestHTTPServerTransport({ classification: MODERN, - ...(responseMode !== undefined && { responseMode }) + ...(responseMode !== undefined && { responseMode }), + ...(keepAliveMs !== undefined && { keepAliveMs }) }); await server.connect(transport); return { server, transport }; @@ -249,3 +251,102 @@ describe('disconnect is cancellation', () => { expect(observedSignal?.aborted).toBe(true); }); }); + +describe('keep-alive', () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it('writes keep-alive comment frames while a forced-sse exchange is streaming', async () => { + let release!: () => void; + const gate = new Promise(resolve => { + release = resolve; + }); + const { transport } = await setup(async () => { + await gate; + return { content: [] }; + }, 'sse'); + + const responsePromise = transport.handleMessage(toolsCall()); + // The stream opened at dispatch end; the handler now idles past the + // default interval with no mid-call output. + await vi.advanceTimersByTimeAsync(15_000); + release(); + const response = await responsePromise; + const frames = await sseFrames(response); + expect(frames[0]).toBe(': keepalive'); + + // The exchange completed and closed the transport: no timer survives. + expect(vi.getTimerCount()).toBe(0); + }); + + it('writes keep-alive frames after an auto exchange upgrades to SSE', async () => { + let release!: () => void; + const gate = new Promise(resolve => { + release = resolve; + }); + const { transport } = await setup(async ctx => { + await ctx.mcpReq.notify(progressNotification(1)); + await gate; + return { content: [] }; + }); + + const responsePromise = transport.handleMessage(toolsCall()); + // Let the handler run, emit the upgrading notification, then idle. + await vi.advanceTimersByTimeAsync(15_000); + release(); + const response = await responsePromise; + const frames = await sseFrames(response); + expect(frames).toContain(': keepalive'); + expect(vi.getTimerCount()).toBe(0); + }); + + it('does not write keep-alive frames when keepAliveMs is 0', async () => { + let release!: () => void; + const gate = new Promise(resolve => { + release = resolve; + }); + const { transport } = await setup( + async () => { + await gate; + return { content: [] }; + }, + 'sse', + 0 + ); + + const responsePromise = transport.handleMessage(toolsCall()); + await vi.advanceTimersByTimeAsync(60_000); + release(); + const response = await responsePromise; + const frames = await sseFrames(response); + expect(frames.some(frame => frame.startsWith(': keepalive'))).toBe(false); + }); + + it('disables keep-alive for a non-finite keepAliveMs instead of arming a clamped interval', async () => { + let release!: () => void; + const gate = new Promise(resolve => { + release = resolve; + }); + const { transport } = await setup( + async () => { + await gate; + return { content: [] }; + }, + 'sse', + Number.NaN + ); + + const responsePromise = transport.handleMessage(toolsCall()); + expect(vi.getTimerCount()).toBe(0); + await vi.advanceTimersByTimeAsync(1_000); + release(); + const response = await responsePromise; + const frames = await sseFrames(response); + expect(frames.some(frame => frame.startsWith(': keepalive'))).toBe(false); + }); +}); diff --git a/packages/server/test/server/streamableHttp.test.ts b/packages/server/test/server/streamableHttp.test.ts index f933a947a5..ea9533a8b5 100644 --- a/packages/server/test/server/streamableHttp.test.ts +++ b/packages/server/test/server/streamableHttp.test.ts @@ -1542,7 +1542,11 @@ describe('WebStandardStreamableHTTPServerTransport SSE keep-alive lifecycle', () await new Promise(resolve => { releaseReplay = resolve; }); - return 'stream-1'; + // Resume the standalone GET stream: it skips the + // no-in-flight-request early close unconditionally, so the + // continuation genuinely reaches the keep-alive arm and only + // the closed-transport guard keeps the timer count at zero. + return '_GET_stream'; } }; const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), eventStore }); From c9b02152d155fbedad271e7898cb2072a06d8d24 Mon Sep 17 00:00:00 2001 From: Matt Carey Date: Fri, 24 Jul 2026 11:18:08 +0100 Subject: [PATCH 4/7] fix: apply the NaN-safe keep-alive guard polarity to the session transport too The round-2 polarity fix was reverted from WebStandardStreamableHTTP- ServerTransport by an overzealous checkout during mutation testing and shipped only in PerRequestHTTPServerTransport. The guard now uses !(keepAliveMs > 0) at both sites, so a non-finite value disables keep-alive on every leg instead of arming a Node-clamped ~1ms interval on the session transport while the sibling guards disable. Mirrors the non-finite regression test into the streamableHttp suite. --- packages/server/src/server/streamableHttp.ts | 5 ++++- .../server/test/server/streamableHttp.test.ts | 17 +++++++++++++++++ 2 files changed, 21 insertions(+), 1 deletion(-) diff --git a/packages/server/src/server/streamableHttp.ts b/packages/server/src/server/streamableHttp.ts index de3760130b..c547507101 100644 --- a/packages/server/src/server/streamableHttp.ts +++ b/packages/server/src/server/streamableHttp.ts @@ -300,7 +300,10 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { ): void { // A deferred arm (e.g. after an event-store await) must not outlive the // transport: close()'s timer sweep has already run and never runs again. - if (this._keepAliveMs <= 0 || this._closed) { + // The `> 0` polarity disables keep-alive for non-finite values (NaN + // fails every comparison) instead of arming a Node-clamped ~1ms + // interval, matching listenRouter's guard for the same-named option. + if (!(this._keepAliveMs > 0) || this._closed) { return; } this.stopKeepAlive(streamId); diff --git a/packages/server/test/server/streamableHttp.test.ts b/packages/server/test/server/streamableHttp.test.ts index ea9533a8b5..16a5ae02c6 100644 --- a/packages/server/test/server/streamableHttp.test.ts +++ b/packages/server/test/server/streamableHttp.test.ts @@ -1472,6 +1472,23 @@ describe('WebStandardStreamableHTTPServerTransport SSE keep-alive', () => { await transport.close(); }); + it('should disable keep-alive for a non-finite keepAliveMs instead of arming a clamped interval', async () => { + const { transport, sessionId } = await createTransport({ keepAliveMs: Number.NaN }); + + const response = await transport.handleRequest(createRequest('GET', undefined, { sessionId })); + expect(response.status).toBe(200); + + // No timer may be armed: setInterval(fn, NaN) would be clamped by + // Node to ~1ms and flood the stream with keep-alive frames. + expect(vi.getTimerCount()).toBe(0); + 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(); From a606d7d78a7abeaefde2b065286e9fca49200f53 Mon Sep 17 00:00:00 2001 From: Matt Carey Date: Fri, 24 Jul 2026 12:53:33 +0100 Subject: [PATCH 5/7] fix: complete SSE keep-alive review hardening - reclaim stream bookkeeping and end replay streams after close races - reject invalid timer delays across all handler serving legs - disable proxy buffering and correct timeout troubleshooting guidance - pin createMcpHandler option forwarding and lifecycle behavior with tests --- docs/troubleshooting.md | 4 +- packages/server/src/server/listenRouter.ts | 4 +- .../server/src/server/perRequestTransport.ts | 6 +- packages/server/src/server/streamableHttp.ts | 47 +++++++++++++--- .../test/server/createMcpHandler.test.ts | 54 ++++++++++++++++++ .../server/createMcpHandlerListen.test.ts | 23 +++++++- .../test/server/perRequestStreaming.test.ts | 47 ++++++++-------- .../server/test/server/streamableHttp.test.ts | 56 +++++++++++++------ 8 files changed, 189 insertions(+), 52 deletions(-) diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 543828747c..f832246607 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -156,11 +156,11 @@ The Resource Server helpers did not move there: `requireBearerAuth`, `mcpAuthMet ## `SSE stream disconnected: TypeError: terminated` -An idle SSE stream was killed by an intermediary or an idle-connection timeout — Node's `server.requestTimeout` defaults to 300 seconds, and reverse proxies and cloud load balancers have similar watchdogs. The client observes the dropped socket as this error (typically every ~5 minutes) and reconnects in a loop. +An idle SSE response went too long without delivering body bytes. A client-side body-idle timeout (for example undici's `bodyTimeout`) or an intermediary such as a reverse proxy or cloud load balancer may then terminate the stream. The client observes this as `TypeError: terminated` and reconnects in a loop. The SDK's HTTP serving prevents this by writing an SSE comment frame (`: keepalive`) to every open SSE stream every 15 seconds by default — `WebStandardStreamableHTTPServerTransport` on all of its streams, and `createMcpHandler` on `subscriptions/listen` streams, modern per-request exchange streams, and the legacy fallback's per-request transport. Comment frames are dropped by SSE parsers before event dispatch, so they never surface as protocol messages. Tune or disable the interval with the `keepAliveMs` option on the transport or handler (`0` disables). -If you still see this error, either keep-alive is disabled (`keepAliveMs: 0`) or an intermediary between client and server buffers or strips SSE data — check for proxies that buffer streaming responses (e.g. nginx without `proxy_buffering off`). +If you still see this error, keep-alive may be disabled (`keepAliveMs: 0`), the client timeout may be shorter than the configured interval, or an intermediary may buffer or strip SSE data. Check for proxies that buffer streaming responses (for example nginx without `proxy_buffering off`). ## Recap diff --git a/packages/server/src/server/listenRouter.ts b/packages/server/src/server/listenRouter.ts index 96dcb16beb..c6e70c4bab 100644 --- a/packages/server/src/server/listenRouter.ts +++ b/packages/server/src/server/listenRouter.ts @@ -218,7 +218,9 @@ export function createListenRouter(options: ListenRouterOptions): ListenRouter { writeNotification(note.method, note.params); }); - if (keepAliveMs > 0) { + // Invalid timer delays disable keep-alive rather than letting + // setInterval clamp them to ~1ms and flood the stream. + if (Number.isFinite(keepAliveMs) && keepAliveMs > 0 && keepAliveMs <= 2_147_483_647) { keepAliveTimer = setInterval(() => writeFrame(': keepalive\n\n'), keepAliveMs); // Do not hold the event loop open on idle subscriptions. Node's // setInterval returns a Timeout with .unref(); browsers/Workers diff --git a/packages/server/src/server/perRequestTransport.ts b/packages/server/src/server/perRequestTransport.ts index fae073c6e2..cd88a8dc65 100644 --- a/packages/server/src/server/perRequestTransport.ts +++ b/packages/server/src/server/perRequestTransport.ts @@ -420,11 +420,11 @@ export class PerRequestHTTPServerTransport implements Transport { * already drops frames once the exchange is closed or the stream is * finalized, so the interval body needs no extra guards; the timer itself * is cleared on stream finalization and transport close. - * Uses the `> 0` polarity so a non-finite value disables keep-alive - * instead of arming a clamped ~1ms interval. + * Invalid timer delays disable keep-alive rather than letting setInterval + * clamp them to ~1ms and flood the stream. */ private startKeepAlive(): void { - if (!(this._keepAliveMs > 0) || this._closed) { + if (!Number.isFinite(this._keepAliveMs) || this._keepAliveMs <= 0 || this._keepAliveMs > 2_147_483_647 || this._closed) { return; } const timer = setInterval(() => this.writeCommentFrame('keepalive'), this._keepAliveMs); diff --git a/packages/server/src/server/streamableHttp.ts b/packages/server/src/server/streamableHttp.ts index c547507101..6ca26aa818 100644 --- a/packages/server/src/server/streamableHttp.ts +++ b/packages/server/src/server/streamableHttp.ts @@ -300,10 +300,9 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { ): void { // A deferred arm (e.g. after an event-store await) must not outlive the // transport: close()'s timer sweep has already run and never runs again. - // The `> 0` polarity disables keep-alive for non-finite values (NaN - // fails every comparison) instead of arming a Node-clamped ~1ms - // interval, matching listenRouter's guard for the same-named option. - if (!(this._keepAliveMs > 0) || this._closed) { + // Invalid timer delays disable keep-alive rather than letting + // setInterval clamp them to ~1ms and flood every stream. + if (!Number.isFinite(this._keepAliveMs) || this._keepAliveMs <= 0 || this._keepAliveMs > 2_147_483_647 || this._closed) { return; } this.stopKeepAlive(streamId); @@ -546,7 +545,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { const headers: Record = { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache, no-transform', - Connection: 'keep-alive' + Connection: 'keep-alive', + 'X-Accel-Buffering': 'no' }; // After initialization, always include the session ID if we have one @@ -605,7 +605,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { const headers: Record = { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache, no-transform', - Connection: 'keep-alive' + Connection: 'keep-alive', + 'X-Accel-Buffering': 'no' }; if (this.sessionId !== undefined) { @@ -654,6 +655,21 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } }); + // The transport may have closed while the replay await was parked: + // its cleanup sweep ran before this stream was registered, so + // registering now would strand a mapping (and an open controller) + // on a dead transport, hang the client on a stream that never + // ends, and 409-block a later resume of this stream id. End the + // stream instead so the client observes termination. + if (this._closed) { + try { + streamController!.close(); + } catch { + // Controller might already be closed + } + return new Response(readable, { headers }); + } + this._streamMapping.set(replayedStreamId, { controller: streamController!, encoder, @@ -753,6 +769,12 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { * Handles `POST` requests containing JSON-RPC messages */ private async handlePostRequest(req: Request, options?: HandleRequestOptions): Promise { + // Set once the SSE stream bookkeeping has been registered, so the + // catch below can reclaim it: an error after registration (a failed + // priming event write, a throwing message handler) returns an error + // response, leaving nothing that could ever cancel the discarded + // stream or retire the request mappings. + let reclaimSseBookkeeping: (() => void) | undefined; try { // Validate the Accept header const acceptHeader = req.headers.get('accept'); @@ -915,7 +937,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { const headers: Record = { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', - Connection: 'keep-alive' + Connection: 'keep-alive', + 'X-Accel-Buffering': 'no' }; // After initialization, always include the session ID if we have one @@ -944,6 +967,15 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } } + reclaimSseBookkeeping = () => { + this._streamMapping.get(streamId)?.cleanup(); + for (const message of messages) { + if (isJSONRPCRequest(message)) { + this._requestToStreamMapping.delete(message.id); + } + } + }; + // Write priming event if event store is configured (after mapping is set up) await this.writePrimingEvent(streamController!, encoder, streamId, clientProtocolVersion); @@ -981,6 +1013,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } catch (error) { // return JSON-RPC formatted error this.onerror?.(error as Error); + reclaimSseBookkeeping?.(); return this.createJsonErrorResponse(400, -32_700, 'Parse error', { data: String(error) }); } } diff --git a/packages/server/test/server/createMcpHandler.test.ts b/packages/server/test/server/createMcpHandler.test.ts index 232f781926..ded506e57c 100644 --- a/packages/server/test/server/createMcpHandler.test.ts +++ b/packages/server/test/server/createMcpHandler.test.ts @@ -820,3 +820,57 @@ describe('createMcpHandler — close()', () => { // Type-level pin: a zero-argument factory stays assignable to McpServerFactory unchanged. const zeroArgFactory = () => new McpServer({ name: 'zero-arg', version: '1.0.0' }); void createMcpHandler(zeroArgFactory); + +describe('createMcpHandler — keepAliveMs', () => { + function gatedFactory(): { factory: () => McpServer; release: () => void } { + let release!: () => void; + const gate = new Promise(resolve => { + release = resolve; + }); + const factory = (): McpServer => { + const s = new McpServer({ name: 'ka', version: '1.0.0' }); + s.registerTool('gated', { inputSchema: z.object({}) }, async () => { + await gate; + return { content: [{ type: 'text', text: 'done' }] }; + }); + return s; + }; + return { factory, release }; + } + + it('threads keepAliveMs into the modern per-request exchange stream', async () => { + vi.useFakeTimers(); + try { + const { factory, release } = gatedFactory(); + const handler = createMcpHandler(factory, { responseMode: 'sse', keepAliveMs: 1_000 }); + const responsePromise = handler.fetch(postRequest(modernToolsCall('gated', {}))); + await vi.advanceTimersByTimeAsync(1_000); + release(); + const response = await responsePromise; + expect(response.headers.get('content-type')).toContain('text/event-stream'); + const text = await response.text(); + expect(text).toContain(': keepalive'); + } finally { + vi.useRealTimers(); + } + }); + + it('threads keepAliveMs into the legacy stateless fallback per-request transport', async () => { + vi.useFakeTimers(); + try { + const { factory, release } = gatedFactory(); + const handler = createMcpHandler(factory, { keepAliveMs: 1_000 }); + const responsePromise = handler.fetch( + postRequest({ jsonrpc: '2.0', id: 9, method: 'tools/call', params: { name: 'gated', arguments: {} } }) + ); + await vi.advanceTimersByTimeAsync(1_000); + release(); + const response = await responsePromise; + expect(response.headers.get('content-type')).toContain('text/event-stream'); + const text = await response.text(); + expect(text).toContain(': keepalive'); + } finally { + vi.useRealTimers(); + } + }); +}); diff --git a/packages/server/test/server/createMcpHandlerListen.test.ts b/packages/server/test/server/createMcpHandlerListen.test.ts index 2fa5000742..29c0069061 100644 --- a/packages/server/test/server/createMcpHandlerListen.test.ts +++ b/packages/server/test/server/createMcpHandlerListen.test.ts @@ -13,7 +13,7 @@ import { PROTOCOL_VERSION_META_KEY, SUBSCRIPTION_ID_META_KEY } from '@modelcontextprotocol/core-internal'; -import { describe, expect, it } from 'vitest'; +import { describe, expect, it, vi } from 'vitest'; import { createMcpHandler } from '../../src/server/createMcpHandler'; import { McpServer } from '../../src/server/mcp'; @@ -116,6 +116,27 @@ describe('createMcpHandler — subscriptions/listen', () => { await handler.close(); }); + it.each([Number.NaN, Number.POSITIVE_INFINITY, 2_147_483_648])( + 'disables keep-alive for invalid keepAliveMs %s instead of arming a clamped interval', + async keepAliveMs => { + vi.useFakeTimers(); + try { + const handler = createMcpHandler(trivialFactory(), { keepAliveMs }); + const response = await handler.fetch(listenRequest(1, { toolsListChanged: true })); + const reader = response.body!.getReader(); + await reader.read(); // acknowledgement + expect(vi.getTimerCount()).toBe(0); + await vi.advanceTimersByTimeAsync(60_000); + const raced = await Promise.race([reader.read(), Promise.resolve('pending')]); + expect(raced).toBe('pending'); + await reader.cancel(); + await handler.close(); + } finally { + vi.useRealTimers(); + } + } + ); + it('ack is the first frame, stamped with the listen id verbatim, carrying the honored subset', async () => { const handler = createMcpHandler(trivialFactory(), { keepAliveMs: 0 }); const response = await handler.fetch(listenRequest('sub-42', { toolsListChanged: true, promptsListChanged: false })); diff --git a/packages/server/test/server/perRequestStreaming.test.ts b/packages/server/test/server/perRequestStreaming.test.ts index 7dfc4578cb..929b0efd39 100644 --- a/packages/server/test/server/perRequestStreaming.test.ts +++ b/packages/server/test/server/perRequestStreaming.test.ts @@ -327,26 +327,29 @@ describe('keep-alive', () => { expect(frames.some(frame => frame.startsWith(': keepalive'))).toBe(false); }); - it('disables keep-alive for a non-finite keepAliveMs instead of arming a clamped interval', async () => { - let release!: () => void; - const gate = new Promise(resolve => { - release = resolve; - }); - const { transport } = await setup( - async () => { - await gate; - return { content: [] }; - }, - 'sse', - Number.NaN - ); - - const responsePromise = transport.handleMessage(toolsCall()); - expect(vi.getTimerCount()).toBe(0); - await vi.advanceTimersByTimeAsync(1_000); - release(); - const response = await responsePromise; - const frames = await sseFrames(response); - expect(frames.some(frame => frame.startsWith(': keepalive'))).toBe(false); - }); + it.each([Number.NaN, Number.POSITIVE_INFINITY, 2_147_483_648])( + 'disables keep-alive for invalid keepAliveMs %s instead of arming a clamped interval', + async keepAliveMs => { + let release!: () => void; + const gate = new Promise(resolve => { + release = resolve; + }); + const { transport } = await setup( + async () => { + await gate; + return { content: [] }; + }, + 'sse', + keepAliveMs + ); + + const responsePromise = transport.handleMessage(toolsCall()); + expect(vi.getTimerCount()).toBe(0); + await vi.advanceTimersByTimeAsync(1_000); + release(); + const response = await responsePromise; + const frames = await sseFrames(response); + expect(frames.some(frame => frame.startsWith(': keepalive'))).toBe(false); + } + ); }); diff --git a/packages/server/test/server/streamableHttp.test.ts b/packages/server/test/server/streamableHttp.test.ts index 16a5ae02c6..7de000866b 100644 --- a/packages/server/test/server/streamableHttp.test.ts +++ b/packages/server/test/server/streamableHttp.test.ts @@ -165,6 +165,7 @@ describe('Zod v4', () => { expect(response.status).toBe(200); expect(response.headers.get('content-type')).toBe('text/event-stream'); + expect(response.headers.get('x-accel-buffering')).toBe('no'); expect(response.headers.get('mcp-session-id')).toBeDefined(); }); @@ -357,6 +358,7 @@ describe('Zod v4', () => { expect(response.status).toBe(200); expect(response.headers.get('content-type')).toBe('text/event-stream'); + expect(response.headers.get('x-accel-buffering')).toBe('no'); expect(response.headers.get('mcp-session-id')).toBe(sessionId); }); @@ -857,6 +859,7 @@ describe('Zod v4', () => { createRequest('GET', undefined, { sessionId, extraHeaders: { 'Last-Event-ID': primingId! } }) ); expect(reconnect.status).toBe(200); + expect(reconnect.headers.get('x-accel-buffering')).toBe('no'); release(); const replayed = await readSSEEvent(reconnect); expect(replayed).toContain('notifications/progress'); @@ -1472,22 +1475,26 @@ describe('WebStandardStreamableHTTPServerTransport SSE keep-alive', () => { await transport.close(); }); - it('should disable keep-alive for a non-finite keepAliveMs instead of arming a clamped interval', async () => { - const { transport, sessionId } = await createTransport({ keepAliveMs: Number.NaN }); + it.each([Number.NaN, Number.POSITIVE_INFINITY, 2_147_483_648])( + 'should disable keep-alive for invalid keepAliveMs %s instead of arming a clamped interval', + async keepAliveMs => { + const { transport, sessionId } = await createTransport({ keepAliveMs }); - const response = await transport.handleRequest(createRequest('GET', undefined, { sessionId })); - expect(response.status).toBe(200); + const response = await transport.handleRequest(createRequest('GET', undefined, { sessionId })); + expect(response.status).toBe(200); - // No timer may be armed: setInterval(fn, NaN) would be clamped by - // Node to ~1ms and flood the stream with keep-alive frames. - expect(vi.getTimerCount()).toBe(0); - const reader = response.body!.getReader(); - await vi.advanceTimersByTimeAsync(60000); - const raced = await Promise.race([reader.read(), Promise.resolve('pending')]); - expect(raced).toBe('pending'); + // No timer may be armed: setInterval with a NaN/out-of-range delay + // is clamped by Node to ~1ms and would flood the stream with + // keep-alive frames. + expect(vi.getTimerCount()).toBe(0); + 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(); - }); + await transport.close(); + } + ); it('should stop keep-alive frames after the stream is closed', async () => { const { transport, sessionId } = await createTransport(); @@ -1581,10 +1588,18 @@ describe('WebStandardStreamableHTTPServerTransport SSE keep-alive lifecycle', () // Close the transport mid-await, then let the replay continuation run await transport.close(); releaseReplay?.(); - await pendingGet; + const replayResponse = await pendingGet; // The deferred continuation must not have armed a timer close() can never sweep expect(vi.getTimerCount()).toBe(0); + + // The continuation must not re-register the stream on the closed + // transport: the client observes stream end instead of hanging on a + // dead session, and a later resume isn't 409-blocked by a stale entry. + const { done } = await replayResponse.body!.getReader().read(); + expect(done).toBe(true); + const internals = transport as unknown as { _streamMapping: Map }; + expect(internals._streamMapping.size).toBe(0); }); it('should not leak a keep-alive timer when the priming event write fails on a POST SSE stream', async () => { @@ -1616,7 +1631,7 @@ describe('WebStandardStreamableHTTPServerTransport SSE keep-alive lifecycle', () expect(vi.getTimerCount()).toBe(0); storeFails = true; - const response = await transport.handleRequest( + await transport.handleRequest( createRequest( 'POST', { jsonrpc: '2.0', method: 'tools/call', params: { name: 'noop', arguments: {} }, id: 'call-1' } as JSONRPCMessage, @@ -1625,11 +1640,20 @@ describe('WebStandardStreamableHTTPServerTransport SSE keep-alive lifecycle', () } ) ); - expect(response.status).toBe(400); // The discarded stream must not carry a permanently-firing timer expect(vi.getTimerCount()).toBe(0); + // The stream bookkeeping registered before the failed priming write + // must be reclaimed too: repeated failures during an event-store + // outage must not accrete orphaned stream entries or request mappings. + const internals = transport as unknown as { + _streamMapping: Map; + _requestToStreamMapping: Map; + }; + expect(internals._streamMapping.size).toBe(0); + expect(internals._requestToStreamMapping.size).toBe(0); + await transport.close(); }); }); From 49f886cdda1e3fa8ad6cb270171d2570c33c510c Mon Sep 17 00:00:00 2001 From: Matt Carey Date: Fri, 24 Jul 2026 15:37:37 +0100 Subject: [PATCH 6/7] refactor: centralize SSE keep-alive timer policy Share the default, safe timer range, and Node unref behavior across session, per-request, and listen streams. Reject sub-millisecond delays that JavaScript timers clamp to 1ms, and clarify when auto-mode exchanges can emit heartbeats. --- .changeset/streamable-http-sse-keepalive.md | 2 +- .../server/src/server/createMcpHandler.ts | 4 ++- packages/server/src/server/listenRouter.ts | 14 +++-------- .../server/src/server/perRequestTransport.ts | 23 +++++++---------- packages/server/src/server/sseKeepAlive.ts | 25 +++++++++++++++++++ packages/server/src/server/streamableHttp.ts | 24 ++++++++---------- .../server/createMcpHandlerListen.test.ts | 2 +- .../test/server/perRequestStreaming.test.ts | 2 +- .../server/test/server/streamableHttp.test.ts | 2 +- 9 files changed, 55 insertions(+), 43 deletions(-) create mode 100644 packages/server/src/server/sseKeepAlive.ts diff --git a/.changeset/streamable-http-sse-keepalive.md b/.changeset/streamable-http-sse-keepalive.md index 77347b39fb..b1bfaba114 100644 --- a/.changeset/streamable-http-sse-keepalive.md +++ b/.changeset/streamable-http-sse-keepalive.md @@ -2,4 +2,4 @@ '@modelcontextprotocol/server': patch --- -SSE streams served by the SDK now emit keep-alive comment frames (`: keepalive`) so idle connections (e.g. the standalone GET stream, or a stream during a long-running tool call) are not killed by intermediaries or server idle timeouts. `WebStandardStreamableHTTPServerTransport` and `PerRequestHTTPServerTransport` gain a `keepAliveMs` option (default 15000; set 0 to disable), and `createMcpHandler`'s existing `keepAliveMs` now covers modern per-request exchange streams and the legacy stateless fallback in addition to `subscriptions/listen` streams. +SSE streams served by the SDK now emit keep-alive comment frames (`: keepalive`) so idle connections are not killed by client body-idle timeouts or intermediaries. `WebStandardStreamableHTTPServerTransport` and `PerRequestHTTPServerTransport` gain a `keepAliveMs` option (default 15000; set 0 to disable), and `createMcpHandler`'s existing `keepAliveMs` now covers modern per-request exchange streams and the legacy stateless fallback in addition to `subscriptions/listen` streams. Streaming responses also disable nginx-style proxy buffering. diff --git a/packages/server/src/server/createMcpHandler.ts b/packages/server/src/server/createMcpHandler.ts index fb924c101e..360b4f066f 100644 --- a/packages/server/src/server/createMcpHandler.ts +++ b/packages/server/src/server/createMcpHandler.ts @@ -198,7 +198,9 @@ export interface CreateMcpHandlerOptions { * SSE comment-frame keepalive interval, in milliseconds, applied to every * SSE stream this handler serves: `subscriptions/listen` streams, modern * per-request exchange streams, and the legacy stateless fallback's - * per-request transport. Set to `0` to disable. + * per-request transport. In modern `auto` mode, the timer starts only after + * the exchange upgrades to SSE; use `responseMode: 'sse'` when a silent + * long-running handler needs heartbeat bytes. Set to `0` to disable. * @default 15000 */ keepAliveMs?: number; diff --git a/packages/server/src/server/listenRouter.ts b/packages/server/src/server/listenRouter.ts index c6e70c4bab..0a076e613b 100644 --- a/packages/server/src/server/listenRouter.ts +++ b/packages/server/src/server/listenRouter.ts @@ -34,9 +34,10 @@ import { codecForVersion, MODERN_WIRE_REVISION, SERVER_INFO_META_KEY, SUBSCRIPTI import type { ServerEventBus } from './serverEventBus'; import { honoredSubset, listenFilterAccepts, serverEventToNotification } from './serverEventBus'; +import { armSseKeepAlive, DEFAULT_SSE_KEEP_ALIVE_MS } from './sseKeepAlive'; /** Default SSE comment-frame keepalive interval for listen streams. */ -export const DEFAULT_LISTEN_KEEPALIVE_MS = 15_000; +export const DEFAULT_LISTEN_KEEPALIVE_MS = DEFAULT_SSE_KEEP_ALIVE_MS; /** Default capacity guard: refuse a new subscription when this many are already open. */ export const DEFAULT_MAX_SUBSCRIPTIONS = 1024; @@ -218,16 +219,7 @@ export function createListenRouter(options: ListenRouterOptions): ListenRouter { writeNotification(note.method, note.params); }); - // Invalid timer delays disable keep-alive rather than letting - // setInterval clamp them to ~1ms and flood the stream. - if (Number.isFinite(keepAliveMs) && keepAliveMs > 0 && keepAliveMs <= 2_147_483_647) { - keepAliveTimer = setInterval(() => writeFrame(': keepalive\n\n'), keepAliveMs); - // Do not hold the event loop open on idle subscriptions. Node's - // setInterval returns a Timeout with .unref(); browsers/Workers - // return a number — the cast is an environment shim, not a - // workaround for SDK typing. - (keepAliveTimer as { unref?: () => void }).unref?.(); - } + keepAliveTimer = armSseKeepAlive(keepAliveMs, () => writeFrame(': keepalive\n\n')); open.add(teardown); }, diff --git a/packages/server/src/server/perRequestTransport.ts b/packages/server/src/server/perRequestTransport.ts index cd88a8dc65..be797e02cf 100644 --- a/packages/server/src/server/perRequestTransport.ts +++ b/packages/server/src/server/perRequestTransport.ts @@ -58,6 +58,8 @@ import { SdkErrorCode } from '@modelcontextprotocol/core-internal'; +import { armSseKeepAlive, DEFAULT_SSE_KEEP_ALIVE_MS } from './sseKeepAlive'; + /** * How the transport shapes its HTTP response for a request: * @@ -81,17 +83,15 @@ export interface PerRequestHTTPServerTransportOptions { responseMode?: PerRequestResponseMode; /** * Interval in milliseconds between SSE keep-alive comment frames - * (`: keepalive`) written while the exchange's SSE stream is open, so a - * long-running handler with no mid-call output doesn't idle past - * intermediary and server idle timeouts. Set to `0` to disable. + * (`: keepalive`) written while the exchange's SSE stream is open. With + * `responseMode: 'sse'`, this also protects long-running handlers that emit + * no mid-call output. Set to `0` to disable; values below `1`, above + * `2147483647`, or non-finite values also disable the timer. * @default 15000 */ keepAliveMs?: number; } -/** Default interval between SSE keep-alive comment frames. */ -const DEFAULT_KEEP_ALIVE_MS = 15_000; - /** Per-exchange context handed to {@linkcode PerRequestHTTPServerTransport.handleMessage}. */ export interface PerRequestMessageExtra { /** @@ -157,7 +157,7 @@ export class PerRequestHTTPServerTransport implements Transport { constructor(options: PerRequestHTTPServerTransportOptions) { this._classification = options.classification; this._responseMode = options.responseMode ?? 'auto'; - this._keepAliveMs = options.keepAliveMs ?? DEFAULT_KEEP_ALIVE_MS; + this._keepAliveMs = options.keepAliveMs ?? DEFAULT_SSE_KEEP_ALIVE_MS; } async start(): Promise { @@ -420,17 +420,12 @@ export class PerRequestHTTPServerTransport implements Transport { * already drops frames once the exchange is closed or the stream is * finalized, so the interval body needs no extra guards; the timer itself * is cleared on stream finalization and transport close. - * Invalid timer delays disable keep-alive rather than letting setInterval - * clamp them to ~1ms and flood the stream. */ private startKeepAlive(): void { - if (!Number.isFinite(this._keepAliveMs) || this._keepAliveMs <= 0 || this._keepAliveMs > 2_147_483_647 || this._closed) { + if (this._closed) { return; } - const timer = setInterval(() => this.writeCommentFrame('keepalive'), this._keepAliveMs); - // Don't let the keep-alive timer hold the process open (Node.js only) - (timer as { unref?: () => void }).unref?.(); - this._keepAliveTimer = timer; + this._keepAliveTimer = armSseKeepAlive(this._keepAliveMs, () => this.writeCommentFrame('keepalive')); } private stopKeepAlive(): void { diff --git a/packages/server/src/server/sseKeepAlive.ts b/packages/server/src/server/sseKeepAlive.ts new file mode 100644 index 0000000000..fa43faf8ec --- /dev/null +++ b/packages/server/src/server/sseKeepAlive.ts @@ -0,0 +1,25 @@ +/** Default interval between SSE keep-alive comment frames. */ +export const DEFAULT_SSE_KEEP_ALIVE_MS = 15_000; + +/** Largest delay accepted by JavaScript timers without overflow clamping. */ +const MAX_TIMER_DELAY_MS = 2_147_483_647; + +/** + * Arms an SSE keep-alive timer when `intervalMs` is a valid timer delay. + * + * Invalid delays disable keep-alive. In Node.js, sub-millisecond, non-finite, + * and overflowing delays are clamped to roughly 1 ms, which would otherwise + * flood every open stream with comment frames. + * + * @returns The timer handle, or `undefined` when keep-alive is disabled. + */ +export function armSseKeepAlive(intervalMs: number, onTick: () => void): ReturnType | undefined { + if (!Number.isFinite(intervalMs) || intervalMs < 1 || intervalMs > MAX_TIMER_DELAY_MS) { + return undefined; + } + + const timer = setInterval(onTick, intervalMs); + // Node.js timers expose `unref`; browser and Workers timers are numbers. + (timer as { unref?: () => void }).unref?.(); + return timer; +} diff --git a/packages/server/src/server/streamableHttp.ts b/packages/server/src/server/streamableHttp.ts index 6ca26aa818..6f93f828c7 100644 --- a/packages/server/src/server/streamableHttp.ts +++ b/packages/server/src/server/streamableHttp.ts @@ -19,6 +19,8 @@ import { SUPPORTED_PROTOCOL_VERSIONS } from '@modelcontextprotocol/core-internal'; +import { armSseKeepAlive, DEFAULT_SSE_KEEP_ALIVE_MS } from './sseKeepAlive'; + export type StreamId = string; export type EventId = string; @@ -157,7 +159,8 @@ export interface WebStandardStreamableHTTPServerTransportOptions { * * 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. + * 15 seconds). Set to `0` to disable keep-alive frames; values below `1`, above + * `2147483647`, or non-finite values also disable the timer. */ keepAliveMs?: number; @@ -174,9 +177,6 @@ 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 */ @@ -282,7 +282,7 @@ 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; + this._keepAliveMs = options.keepAliveMs ?? DEFAULT_SSE_KEEP_ALIVE_MS; } /** @@ -300,22 +300,20 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { ): void { // A deferred arm (e.g. after an event-store await) must not outlive the // transport: close()'s timer sweep has already run and never runs again. - // Invalid timer delays disable keep-alive rather than letting - // setInterval clamp them to ~1ms and flood every stream. - if (!Number.isFinite(this._keepAliveMs) || this._keepAliveMs <= 0 || this._keepAliveMs > 2_147_483_647 || this._closed) { + if (this._closed) { return; } this.stopKeepAlive(streamId); - const timer = setInterval(() => { + const timer = armSseKeepAlive(this._keepAliveMs, () => { 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); + }); + if (timer !== undefined) { + this._keepAliveTimers.set(streamId, timer); + } } /** diff --git a/packages/server/test/server/createMcpHandlerListen.test.ts b/packages/server/test/server/createMcpHandlerListen.test.ts index 29c0069061..421ce36079 100644 --- a/packages/server/test/server/createMcpHandlerListen.test.ts +++ b/packages/server/test/server/createMcpHandlerListen.test.ts @@ -116,7 +116,7 @@ describe('createMcpHandler — subscriptions/listen', () => { await handler.close(); }); - it.each([Number.NaN, Number.POSITIVE_INFINITY, 2_147_483_648])( + it.each([0.5, Number.NaN, Number.POSITIVE_INFINITY, 2_147_483_648])( 'disables keep-alive for invalid keepAliveMs %s instead of arming a clamped interval', async keepAliveMs => { vi.useFakeTimers(); diff --git a/packages/server/test/server/perRequestStreaming.test.ts b/packages/server/test/server/perRequestStreaming.test.ts index 929b0efd39..20949b68b9 100644 --- a/packages/server/test/server/perRequestStreaming.test.ts +++ b/packages/server/test/server/perRequestStreaming.test.ts @@ -327,7 +327,7 @@ describe('keep-alive', () => { expect(frames.some(frame => frame.startsWith(': keepalive'))).toBe(false); }); - it.each([Number.NaN, Number.POSITIVE_INFINITY, 2_147_483_648])( + it.each([0.5, Number.NaN, Number.POSITIVE_INFINITY, 2_147_483_648])( 'disables keep-alive for invalid keepAliveMs %s instead of arming a clamped interval', async keepAliveMs => { let release!: () => void; diff --git a/packages/server/test/server/streamableHttp.test.ts b/packages/server/test/server/streamableHttp.test.ts index 7de000866b..0b33ce90d9 100644 --- a/packages/server/test/server/streamableHttp.test.ts +++ b/packages/server/test/server/streamableHttp.test.ts @@ -1475,7 +1475,7 @@ describe('WebStandardStreamableHTTPServerTransport SSE keep-alive', () => { await transport.close(); }); - it.each([Number.NaN, Number.POSITIVE_INFINITY, 2_147_483_648])( + it.each([0.5, Number.NaN, Number.POSITIVE_INFINITY, 2_147_483_648])( 'should disable keep-alive for invalid keepAliveMs %s instead of arming a clamped interval', async keepAliveMs => { const { transport, sessionId } = await createTransport({ keepAliveMs }); From 8a6a55a85f09bc5ea58cf13aa4f1a56dbff7b226 Mon Sep 17 00:00:00 2001 From: Matt Carey Date: Fri, 24 Jul 2026 15:58:08 +0100 Subject: [PATCH 7/7] fix: close remaining SSE lifecycle races Prevent post-await work from registering streams after close, preserve successor request mappings during failed priming cleanup, align anti-buffering headers, and classify the new public API as a minor change. --- .changeset/streamable-http-sse-keepalive.md | 4 +- packages/server/src/server/listenRouter.ts | 2 +- .../server/src/server/perRequestTransport.ts | 2 +- packages/server/src/server/streamableHttp.ts | 14 ++- .../server/createMcpHandlerListen.test.ts | 2 + .../test/server/perRequestStreaming.test.ts | 2 +- .../server/test/server/streamableHttp.test.ts | 86 +++++++++++++++++++ 7 files changed, 105 insertions(+), 7 deletions(-) diff --git a/.changeset/streamable-http-sse-keepalive.md b/.changeset/streamable-http-sse-keepalive.md index b1bfaba114..f9c107a17f 100644 --- a/.changeset/streamable-http-sse-keepalive.md +++ b/.changeset/streamable-http-sse-keepalive.md @@ -1,5 +1,5 @@ --- -'@modelcontextprotocol/server': patch +'@modelcontextprotocol/server': minor --- -SSE streams served by the SDK now emit keep-alive comment frames (`: keepalive`) so idle connections are not killed by client body-idle timeouts or intermediaries. `WebStandardStreamableHTTPServerTransport` and `PerRequestHTTPServerTransport` gain a `keepAliveMs` option (default 15000; set 0 to disable), and `createMcpHandler`'s existing `keepAliveMs` now covers modern per-request exchange streams and the legacy stateless fallback in addition to `subscriptions/listen` streams. Streaming responses also disable nginx-style proxy buffering. +SSE streams served by the SDK now emit keep-alive comment frames (`: keepalive`) so idle connections are not killed by client body-idle timeouts or intermediaries. `WebStandardStreamableHTTPServerTransport` and `PerRequestHTTPServerTransport` gain a `keepAliveMs` option (default 15000; set 0 to disable), and `createMcpHandler`'s existing `keepAliveMs` now covers modern per-request exchange streams and the legacy stateless fallback in addition to `subscriptions/listen` streams. Streaming responses disable proxy transformations and nginx-style buffering, and deferred work cannot register streams after transport close. diff --git a/packages/server/src/server/listenRouter.ts b/packages/server/src/server/listenRouter.ts index 0a076e613b..7ccce8dadd 100644 --- a/packages/server/src/server/listenRouter.ts +++ b/packages/server/src/server/listenRouter.ts @@ -245,7 +245,7 @@ export function createListenRouter(options: ListenRouterOptions): ListenRouter { status: 200, headers: { 'Content-Type': 'text/event-stream', - 'Cache-Control': 'no-cache', + 'Cache-Control': 'no-cache, no-transform', Connection: 'keep-alive', 'X-Accel-Buffering': 'no' } diff --git a/packages/server/src/server/perRequestTransport.ts b/packages/server/src/server/perRequestTransport.ts index be797e02cf..53400a05f5 100644 --- a/packages/server/src/server/perRequestTransport.ts +++ b/packages/server/src/server/perRequestTransport.ts @@ -404,7 +404,7 @@ export class PerRequestHTTPServerTransport implements Transport { status: 200, headers: { 'Content-Type': 'text/event-stream', - 'Cache-Control': 'no-cache', + 'Cache-Control': 'no-cache, no-transform', Connection: 'keep-alive', // Disable proxy buffering so streamed messages are // delivered as they are written. diff --git a/packages/server/src/server/streamableHttp.ts b/packages/server/src/server/streamableHttp.ts index 6f93f828c7..79a4c136be 100644 --- a/packages/server/src/server/streamableHttp.ts +++ b/packages/server/src/server/streamableHttp.ts @@ -413,6 +413,10 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { * Returns a `Response` object (Web Standard) */ async handleRequest(req: Request, options?: HandleRequestOptions): Promise { + if (this._closed) { + return this.createJsonErrorResponse(404, -32_001, 'Session not found'); + } + // Validate request headers for DNS rebinding protection const validationError = this.validateRequestHeaders(req); if (validationError) { @@ -866,6 +870,12 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } } + // Request parsing and session initialization may await user/runtime + // work. Do not register or dispatch after close() has swept state. + if (this._closed) { + return this.createJsonErrorResponse(404, -32_001, 'Session not found'); + } + // check if it contains requests const hasRequests = messages.some(element => isJSONRPCRequest(element)); @@ -934,7 +944,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { const headers: Record = { 'Content-Type': 'text/event-stream', - 'Cache-Control': 'no-cache', + 'Cache-Control': 'no-cache, no-transform', Connection: 'keep-alive', 'X-Accel-Buffering': 'no' }; @@ -968,7 +978,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { reclaimSseBookkeeping = () => { this._streamMapping.get(streamId)?.cleanup(); for (const message of messages) { - if (isJSONRPCRequest(message)) { + if (isJSONRPCRequest(message) && this._requestToStreamMapping.get(message.id) === streamId) { this._requestToStreamMapping.delete(message.id); } } diff --git a/packages/server/test/server/createMcpHandlerListen.test.ts b/packages/server/test/server/createMcpHandlerListen.test.ts index 421ce36079..7a524d0c9f 100644 --- a/packages/server/test/server/createMcpHandlerListen.test.ts +++ b/packages/server/test/server/createMcpHandlerListen.test.ts @@ -105,6 +105,8 @@ describe('createMcpHandler — subscriptions/listen', () => { ); const response = await handler.fetch(listenRequest(1, { toolsListChanged: true })); expect(response.status).toBe(200); + expect(response.headers.get('cache-control')).toBe('no-cache, no-transform'); + expect(response.headers.get('x-accel-buffering')).toBe('no'); const [ack] = await readMessages(response, 1); // The factory is consulted exactly once (capabilities probe only); the // instance is never connected and is closed immediately after the diff --git a/packages/server/test/server/perRequestStreaming.test.ts b/packages/server/test/server/perRequestStreaming.test.ts index 20949b68b9..9440cbdf4e 100644 --- a/packages/server/test/server/perRequestStreaming.test.ts +++ b/packages/server/test/server/perRequestStreaming.test.ts @@ -94,7 +94,7 @@ describe('lazy upgrade matrix', () => { const response = await transport.handleMessage(toolsCall()); expect(response.status).toBe(200); expect(response.headers.get('content-type')).toBe('text/event-stream'); - expect(response.headers.get('cache-control')).toBe('no-cache'); + expect(response.headers.get('cache-control')).toBe('no-cache, no-transform'); expect(response.headers.get('x-accel-buffering')).toBe('no'); const frames = await sseFrames(response); diff --git a/packages/server/test/server/streamableHttp.test.ts b/packages/server/test/server/streamableHttp.test.ts index 0b33ce90d9..f54708c8f0 100644 --- a/packages/server/test/server/streamableHttp.test.ts +++ b/packages/server/test/server/streamableHttp.test.ts @@ -165,6 +165,7 @@ describe('Zod v4', () => { expect(response.status).toBe(200); expect(response.headers.get('content-type')).toBe('text/event-stream'); + expect(response.headers.get('cache-control')).toBe('no-cache, no-transform'); expect(response.headers.get('x-accel-buffering')).toBe('no'); expect(response.headers.get('mcp-session-id')).toBeDefined(); }); @@ -358,6 +359,7 @@ describe('Zod v4', () => { expect(response.status).toBe(200); expect(response.headers.get('content-type')).toBe('text/event-stream'); + expect(response.headers.get('cache-control')).toBe('no-cache, no-transform'); expect(response.headers.get('x-accel-buffering')).toBe('no'); expect(response.headers.get('mcp-session-id')).toBe(sessionId); }); @@ -859,6 +861,7 @@ describe('Zod v4', () => { createRequest('GET', undefined, { sessionId, extraHeaders: { 'Last-Event-ID': primingId! } }) ); expect(reconnect.status).toBe(200); + expect(reconnect.headers.get('cache-control')).toBe('no-cache, no-transform'); expect(reconnect.headers.get('x-accel-buffering')).toBe('no'); release(); const replayed = await readSSEEvent(reconnect); @@ -1656,4 +1659,87 @@ describe('WebStandardStreamableHTTPServerTransport SSE keep-alive lifecycle', () await transport.close(); }); + + it('should not reclaim a successor mapping that reused the failed POST request id', async () => { + let failPriming = false; + let rejectPriming: (() => void) | undefined; + const eventStore: EventStore = { + async storeEvent(): Promise { + if (failPriming) { + return new Promise((_resolve, reject) => { + rejectPriming = () => reject(new Error('event store unavailable')); + }); + } + return `evt-${randomUUID()}`; + }, + async replayEventsAfter(): Promise { + return 'stream-1'; + } + }; + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), eventStore }); + const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }); + mcpServer.registerTool('noop', { description: 'noop' }, async (): Promise => ({ content: [] })); + await mcpServer.connect(transport); + + const initResponse = await transport.handleRequest(createRequest('POST', TEST_MESSAGES.initialize)); + const sessionId = initResponse.headers.get('mcp-session-id') as string; + await vi.advanceTimersByTimeAsync(0); + failPriming = true; + + const pending = transport.handleRequest( + createRequest( + 'POST', + { jsonrpc: '2.0', method: 'tools/call', params: { name: 'noop', arguments: {} }, id: 'same-id' } as JSONRPCMessage, + { sessionId } + ) + ); + await vi.advanceTimersByTimeAsync(0); + expect(rejectPriming).toBeDefined(); + + const internals = transport as unknown as { + _streamMapping: Map; + _requestToStreamMapping: Map; + }; + const failedStreamId = internals._requestToStreamMapping.get('same-id'); + expect(failedStreamId).toBeDefined(); + internals._requestToStreamMapping.set('same-id', 'successor-stream'); + + rejectPriming?.(); + await pending; + + expect(internals._streamMapping.has(failedStreamId!)).toBe(false); + expect(internals._requestToStreamMapping.get('same-id')).toBe('successor-stream'); + internals._requestToStreamMapping.delete('same-id'); + await transport.close(); + }); + + it('should not register a POST stream after close races session initialization', async () => { + let releaseInitialization: (() => void) | undefined; + const transport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID(), + onsessioninitialized: async () => { + await new Promise(resolve => { + releaseInitialization = resolve; + }); + } + }); + await new McpServer({ name: 'test-server', version: '1.0.0' }).connect(transport); + + const pendingInit = transport.handleRequest(createRequest('POST', TEST_MESSAGES.initialize)); + await vi.advanceTimersByTimeAsync(0); + expect(releaseInitialization).toBeDefined(); + + await transport.close(); + releaseInitialization?.(); + const response = await pendingInit; + + expect(response.status).toBe(404); + expect(vi.getTimerCount()).toBe(0); + const internals = transport as unknown as { + _streamMapping: Map; + _requestToStreamMapping: Map; + }; + expect(internals._streamMapping.size).toBe(0); + expect(internals._requestToStreamMapping.size).toBe(0); + }); });