From 9505f9339d32b398cd4bd2fdea7854d4c416c729 Mon Sep 17 00:00:00 2001 From: Matt Carey Date: Thu, 23 Jul 2026 16:58:44 +0100 Subject: [PATCH 1/4] 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/4] 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/4] 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/4] 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();