diff --git a/.changeset/streamable-http-sse-keepalive.md b/.changeset/streamable-http-sse-keepalive.md new file mode 100644 index 0000000000..77347b39fb --- /dev/null +++ b/.changeset/streamable-http-sse-keepalive.md @@ -0,0 +1,5 @@ +--- +'@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. diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 1fc325b7e4..543828747c 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. + +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`). + ## 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..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,8 +195,10 @@ 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 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; @@ -295,18 +298,26 @@ 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, 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 +328,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 +646,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; @@ -778,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/src/server/streamableHttp.ts b/packages/server/src/server/streamableHttp.ts index 7da5fb853c..c547507101 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,52 @@ 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 { + // 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) { + 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 +537,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 +559,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { controller: streamController!, encoder, cleanup: () => { + this.stopKeepAlive(this._standaloneSseStreamId); this._streamMapping.delete(this._standaloneSseStreamId); try { streamController!.close(); @@ -503,6 +569,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } }); + this.startKeepAlive(this._standaloneSseStreamId, streamController!, encoder); + return new Response(readable, { headers }); } @@ -564,6 +632,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 +659,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { encoder, replayedEventIds, cleanup: () => { + this.stopKeepAlive(replayedStreamId!); this._streamMapping.delete(replayedStreamId!); try { streamController!.close(); @@ -618,6 +688,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 +906,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 +931,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { controller: streamController!, encoder, cleanup: () => { + this.stopKeepAlive(streamId); this._streamMapping.delete(streamId); try { streamController!.close(); @@ -891,6 +969,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 @@ -986,6 +1072,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/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 beca451113..16a5ae02c6 100644 --- a/packages/server/test/server/streamableHttp.test.ts +++ b/packages/server/test/server/streamableHttp.test.ts @@ -1407,3 +1407,229 @@ 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 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(); + + 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(); + }); +}); + +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; + }); + // 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 }); + 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(); + }); +});