diff --git a/.changeset/streamable-http-sse-keepalive.md b/.changeset/streamable-http-sse-keepalive.md new file mode 100644 index 0000000000..705fa5f8f4 --- /dev/null +++ b/.changeset/streamable-http-sse-keepalive.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/server': minor +--- + +Add configurable SSE keep-alive comment frames to Streamable HTTP transports and apply `createMcpHandler`'s existing `keepAliveMs` option to every HTTP SSE stream it serves. diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 1fc325b7e4..bf96270c36 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -154,6 +154,10 @@ 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` + +HTTP SSE streams emit a `: keepalive` comment every 15 seconds by default so client body-idle timeouts and intermediaries do not terminate an otherwise idle connection. Configure the interval with `keepAliveMs` on the transport or `createMcpHandler`; set it to `0` to disable heartbeats. + ## 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..a484869944 100644 --- a/packages/server/src/server/createMcpHandler.ts +++ b/packages/server/src/server/createMcpHandler.ts @@ -59,13 +59,14 @@ import { } from '@modelcontextprotocol/core-internal'; import { invoke } from './invoke'; -import { createListenRouter, DEFAULT_LISTEN_KEEPALIVE_MS, DEFAULT_MAX_SUBSCRIPTIONS } from './listenRouter'; +import { createListenRouter, DEFAULT_MAX_SUBSCRIPTIONS } from './listenRouter'; import { McpServer } from './mcp'; import type { PerRequestResponseMode } from './perRequestTransport'; import type { Server } from './server'; import { installModernOnlyHandlers, seedClientIdentityFromEnvelope, serverIdentityOf } from './server'; import type { ServerEventBus, ServerNotifier } from './serverEventBus'; import { createServerNotifier, InMemoryServerEventBus } from './serverEventBus'; +import { DEFAULT_SSE_KEEP_ALIVE_MS } from './sseKeepAlive'; import { WebStandardStreamableHTTPServerTransport } from './streamableHttp'; /* ------------------------------------------------------------------------ * @@ -194,8 +195,8 @@ 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 for every SSE stream this handler + * serves. In modern `auto` mode it starts after SSE upgrade. 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 { +function createLegacyStatelessFallback( + factory: McpServerFactory, + onerror?: (error: Error) => void, + 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, + ...(keepAliveMs !== undefined && { keepAliveMs }) + }); await product.connect(transport); const teardown = () => { @@ -390,6 +398,10 @@ export function legacyStatelessFallback(factory: McpServerFactory, onerror?: (er }; } +export function legacyStatelessFallback(factory: McpServerFactory, onerror?: (error: Error) => void): LegacyHttpHandler { + return createLegacyStatelessFallback(factory, onerror); +} + /* ------------------------------------------------------------------------ * * The entry's classification step (shared with isLegacyRequest) * ------------------------------------------------------------------------ */ @@ -619,7 +631,7 @@ export function createMcpHandler(factory: McpServerFactory, options: CreateMcpHa const listenRouter = createListenRouter({ bus, maxSubscriptions: options.maxSubscriptions ?? DEFAULT_MAX_SUBSCRIPTIONS, - keepAliveMs: options.keepAliveMs ?? DEFAULT_LISTEN_KEEPALIVE_MS, + keepAliveMs: options.keepAliveMs ?? DEFAULT_SSE_KEEP_ALIVE_MS, onerror: reportError }); if (responseMode === 'json') { @@ -632,7 +644,8 @@ 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 : createLegacyStatelessFallback(factory, reportError, options.keepAliveMs); async function serveModern(route: InboundModernRoute, request: Request, authInfo: AuthInfo | undefined): Promise { const claimedRevision = route.classification.revision; @@ -778,7 +791,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..1a6984926c 100644 --- a/packages/server/src/server/invoke.ts +++ b/packages/server/src/server/invoke.ts @@ -35,6 +35,8 @@ export interface InvokeContext { authInfo?: AuthInfo; /** Response shaping for the exchange; defaults to `auto` (lazy SSE upgrade). */ responseMode?: PerRequestResponseMode; + /** SSE keep-alive interval for the exchange. */ + keepAliveMs?: number; } /** @@ -58,7 +60,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/listenRouter.ts b/packages/server/src/server/listenRouter.ts index 96dcb16beb..40c4a38cf2 100644 --- a/packages/server/src/server/listenRouter.ts +++ b/packages/server/src/server/listenRouter.ts @@ -34,9 +34,7 @@ import { codecForVersion, MODERN_WIRE_REVISION, SERVER_INFO_META_KEY, SUBSCRIPTI import type { ServerEventBus } from './serverEventBus'; import { honoredSubset, listenFilterAccepts, serverEventToNotification } from './serverEventBus'; - -/** Default SSE comment-frame keepalive interval for listen streams. */ -export const DEFAULT_LISTEN_KEEPALIVE_MS = 15_000; +import { armSseKeepAlive, DEFAULT_SSE_KEEP_ALIVE_MS } from './sseKeepAlive'; /** Default capacity guard: refuse a new subscription when this many are already open. */ export const DEFAULT_MAX_SUBSCRIPTIONS = 1024; @@ -124,7 +122,7 @@ export interface ListenRouter { export function createListenRouter(options: ListenRouterOptions): ListenRouter { const { bus, onerror } = options; const maxSubscriptions = options.maxSubscriptions ?? DEFAULT_MAX_SUBSCRIPTIONS; - const keepAliveMs = options.keepAliveMs ?? DEFAULT_LISTEN_KEEPALIVE_MS; + const keepAliveMs = options.keepAliveMs ?? DEFAULT_SSE_KEEP_ALIVE_MS; const open = new Set<(graceful: boolean) => void>(); @@ -188,7 +186,11 @@ export function createListenRouter(options: ListenRouterOptions): ListenRouter { ); } closed = true; - unsubscribe?.(); + try { + unsubscribe?.(); + } catch (error) { + onerror?.(error instanceof Error ? error : new Error(String(error))); + } if (keepAliveTimer !== undefined) clearInterval(keepAliveTimer); abortCleanup?.(); open.delete(teardown); @@ -218,14 +220,7 @@ export function createListenRouter(options: ListenRouterOptions): ListenRouter { writeNotification(note.method, note.params); }); - if (keepAliveMs > 0) { - 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); }, @@ -251,7 +246,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 5003946404..5c17f9c455 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: * @@ -79,6 +81,8 @@ export interface PerRequestHTTPServerTransportOptions { classification: MessageClassification; /** Response shaping for the exchange; defaults to `auto`. */ responseMode?: PerRequestResponseMode; + /** SSE keep-alive interval in milliseconds; defaults to `15000`, `0` disables. */ + keepAliveMs?: number; } /** Per-exchange context handed to {@linkcode PerRequestHTTPServerTransport.handleMessage}. */ @@ -107,6 +111,7 @@ interface SseSink { controller: ReadableStreamDefaultController; encoder: InstanceType; closed: boolean; + keepAliveTimer?: ReturnType; } /** @@ -140,10 +145,12 @@ export class PerRequestHTTPServerTransport implements Transport { private _deferredResponse?: DeferredResponse; private _sse?: SseSink; private _abortCleanup?: () => void; + private readonly _keepAliveMs: number; constructor(options: PerRequestHTTPServerTransportOptions) { this._classification = options.classification; this._responseMode = options.responseMode ?? 'auto'; + this._keepAliveMs = options.keepAliveMs ?? DEFAULT_SSE_KEEP_ALIVE_MS; } async start(): Promise { @@ -343,6 +350,9 @@ export class PerRequestHTTPServerTransport implements Transport { this._abortCleanup?.(); this._abortCleanup = undefined; + if (this._sse?.keepAliveTimer !== undefined) { + clearInterval(this._sse.keepAliveTimer); + } if (this._sse !== undefined && !this._sse.closed) { this._sse.closed = true; try { @@ -382,13 +392,14 @@ export class PerRequestHTTPServerTransport implements Transport { } }); this._sse = { controller, encoder: new TextEncoder(), closed: false }; + this._sse.keepAliveTimer = armSseKeepAlive(this._keepAliveMs, () => this.writeCommentFrame('keepalive')); this.settleResponse( new Response(readable, { 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. @@ -399,6 +410,9 @@ export class PerRequestHTTPServerTransport implements Transport { } private finalizeStream(): void { + if (this._sse?.keepAliveTimer !== undefined) { + clearInterval(this._sse.keepAliveTimer); + } if (this._sse !== undefined && !this._sse.closed) { this._sse.closed = true; try { diff --git a/packages/server/src/server/sseKeepAlive.ts b/packages/server/src/server/sseKeepAlive.ts new file mode 100644 index 0000000000..25df11871c --- /dev/null +++ b/packages/server/src/server/sseKeepAlive.ts @@ -0,0 +1,15 @@ +/** Default interval between SSE keep-alive comment frames. */ +export const DEFAULT_SSE_KEEP_ALIVE_MS = 15_000; + +const MAX_TIMER_DELAY_MS = 2 ** 31 - 1; + +/** Arms an unref'd timer, or disables keep-alive for invalid delays. */ +export function armSseKeepAlive(intervalMs: number, onTick: () => void): ReturnType | undefined { + if (!Number.isFinite(intervalMs) || intervalMs < 1) { + return undefined; + } + + const timer = setInterval(onTick, Math.min(intervalMs, MAX_TIMER_DELAY_MS)); + (timer as { unref?: () => void }).unref?.(); + return timer; +} diff --git a/packages/server/src/server/streamableHttp.ts b/packages/server/src/server/streamableHttp.ts index 7da5fb853c..c0f48560a2 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; @@ -148,6 +150,12 @@ export interface WebStandardStreamableHTTPServerTransportOptions { */ retryInterval?: number; + /** + * Interval in milliseconds between SSE keep-alive comment frames. + * Defaults to `15000`; set to `0` to disable. + */ + keepAliveMs?: number; + /** * List of protocol versions that this transport will accept. * Used to validate the `mcp-protocol-version` header in incoming requests. @@ -247,6 +255,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { private _enableDnsRebindingProtection: boolean; private _retryInterval?: number; private _supportedProtocolVersions: string[]; + private _keepAliveMs: number; sessionId?: string; onclose?: () => void; @@ -264,6 +273,23 @@ 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_SSE_KEEP_ALIVE_MS; + } + + private startKeepAlive( + controller: ReadableStreamDefaultController, + encoder: InstanceType + ): ReturnType | undefined { + if (this._closed) return undefined; + + const timer = armSseKeepAlive(this._keepAliveMs, () => { + try { + controller.enqueue(encoder.encode(': keepalive\n\n')); + } catch { + if (timer !== undefined) clearInterval(timer); + } + }); + return timer; } /** @@ -352,6 +378,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) { @@ -462,6 +492,9 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { const encoder = new TextEncoder(); let streamController: ReadableStreamDefaultController; + // Captured by cancel/cleanup before it is assigned after stream setup. + // eslint-disable-next-line prefer-const + let keepAliveTimer: ReturnType | undefined; // Create a ReadableStream with a controller we can use to push SSE events const readable = new ReadableStream({ @@ -469,6 +502,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { streamController = controller; }, cancel: () => { + if (keepAliveTimer !== undefined) clearInterval(keepAliveTimer); // Stream was cancelled by client. Only drop the mapping when // it still points at THIS controller — a stale cancel must not // delete a successor stream registered by a later GET/resume. @@ -481,7 +515,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 @@ -494,6 +529,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { controller: streamController!, encoder, cleanup: () => { + if (keepAliveTimer !== undefined) clearInterval(keepAliveTimer); this._streamMapping.delete(this._standaloneSseStreamId); try { streamController!.close(); @@ -503,6 +539,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } }); + keepAliveTimer = this.startKeepAlive(streamController!, encoder); return new Response(readable, { headers }); } @@ -537,7 +574,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) { @@ -547,6 +585,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { // Create a ReadableStream with controller for SSE const encoder = new TextEncoder(); let streamController: ReadableStreamDefaultController; + let keepAliveTimer: ReturnType | undefined; + let cancelled = false; // Captured by the cancel closure below before it's assigned (after // replayEventsAfter resolves) — must be `let`. // eslint-disable-next-line prefer-const @@ -557,6 +597,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { streamController = controller; }, cancel: () => { + cancelled = true; + if (keepAliveTimer !== undefined) clearInterval(keepAliveTimer); // Stream was cancelled by client — drop the mapping so a // subsequent reconnect with the same Last-Event-ID is not // refused with 409 by the conflict check above. Only delete @@ -585,11 +627,22 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } }); + if (this._closed || cancelled) { + try { + streamController!.close(); + } catch { + // Controller already closed/cancelled. + } + return this.createJsonErrorResponse(404, -32_001, 'Session not found'); + } + + this._streamMapping.get(replayedStreamId)?.cleanup(); this._streamMapping.set(replayedStreamId, { controller: streamController!, encoder, replayedEventIds, cleanup: () => { + if (keepAliveTimer !== undefined) clearInterval(keepAliveTimer); this._streamMapping.delete(replayedStreamId!); try { streamController!.close(); @@ -618,6 +671,9 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } } + if (this._streamMapping.get(replayedStreamId)?.controller === streamController!) { + keepAliveTimer = this.startKeepAlive(streamController!, encoder); + } return new Response(readable, { headers }); } catch (error) { this.onerror?.(error as Error); @@ -728,6 +784,10 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { return this.createJsonErrorResponse(400, -32_700, 'Parse error: Invalid JSON-RPC message'); } + if (this._closed) { + return this.createJsonErrorResponse(404, -32_001, 'Session not found'); + } + // Check if this is an initialization request // https://spec.modelcontextprotocol.io/specification/2025-03-26/basic/lifecycle/ // The schema-validated guard (types/guards.ts → types/schemas.ts — @@ -770,6 +830,10 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } } + if (this._closed) { + return this.createJsonErrorResponse(404, -32_001, 'Session not found'); + } + // check if it contains requests const hasRequests = messages.some(element => isJSONRPCRequest(element)); @@ -818,12 +882,14 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { // SSE streaming mode - use ReadableStream with controller for more reliable data pushing const encoder = new TextEncoder(); let streamController: ReadableStreamDefaultController; + let keepAliveTimer: ReturnType | undefined; const readable = new ReadableStream({ start: controller => { streamController = controller; }, cancel: () => { + if (keepAliveTimer !== undefined) clearInterval(keepAliveTimer); // Stream was cancelled by client. Only drop the mapping // when it still points at THIS controller — a stale cancel // (firing after a Last-Event-ID reconnect registered a @@ -837,8 +903,9 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { const headers: Record = { 'Content-Type': 'text/event-stream', - 'Cache-Control': 'no-cache', - Connection: 'keep-alive' + 'Cache-Control': 'no-cache, no-transform', + Connection: 'keep-alive', + 'X-Accel-Buffering': 'no' }; // After initialization, always include the session ID if we have one @@ -854,6 +921,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { controller: streamController!, encoder, cleanup: () => { + if (keepAliveTimer !== undefined) clearInterval(keepAliveTimer); this._streamMapping.delete(streamId); try { streamController!.close(); @@ -891,6 +959,9 @@ 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 + if (this._streamMapping.get(streamId)?.controller === streamController!) { + keepAliveTimer = this.startKeepAlive(streamController!, encoder); + } return new Response(readable, { status: 200, headers }); } catch (error) { // return JSON-RPC formatted error @@ -912,9 +983,12 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { return protocolError; } - await Promise.resolve(this._onsessionclosed?.(this.sessionId!)); - await this.close(); - return new Response(null, { status: 200 }); + try { + await Promise.resolve(this._onsessionclosed?.(this.sessionId!)); + return new Response(null, { status: 200 }); + } finally { + await this.close(); + } } /** 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..7a0d2c8675 100644 --- a/packages/server/test/server/createMcpHandlerListen.test.ts +++ b/packages/server/test/server/createMcpHandlerListen.test.ts @@ -13,10 +13,11 @@ 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'; +import type { ServerEventBus } from '../../src/server/serverEventBus'; const ENVELOPE = { [PROTOCOL_VERSION_META_KEY]: '2026-07-28', @@ -105,6 +106,7 @@ 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'); 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 @@ -116,6 +118,44 @@ describe('createMcpHandler — subscriptions/listen', () => { await handler.close(); }); + it.each([0.5, Number.NaN, Number.POSITIVE_INFINITY])('disables invalid keepAliveMs %s', 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(); + expect(vi.getTimerCount()).toBe(0); + await reader.cancel(); + await handler.close(); + } finally { + vi.useRealTimers(); + } + }); + + it('cleans up when a custom bus unsubscribe throws', async () => { + vi.useFakeTimers(); + try { + const onerror = vi.fn(); + const bus: ServerEventBus = { + publish() {}, + subscribe: () => () => { + throw new Error('unsubscribe failed'); + } + }; + const handler = createMcpHandler(trivialFactory(), { bus, keepAliveMs: 1_000, onerror }); + const response = await handler.fetch(listenRequest(1, { toolsListChanged: true })); + const reader = response.body!.getReader(); + await reader.read(); + await reader.cancel(); + expect(onerror).toHaveBeenCalledWith(expect.objectContaining({ message: 'unsubscribe failed' })); + expect(vi.getTimerCount()).toBe(0); + 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 6d350ed2ef..b9ac07e09a 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 { 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 }; @@ -92,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); @@ -249,3 +251,58 @@ 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('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); + }); +}); diff --git a/packages/server/test/server/sseKeepAlive.test.ts b/packages/server/test/server/sseKeepAlive.test.ts new file mode 100644 index 0000000000..0310849429 --- /dev/null +++ b/packages/server/test/server/sseKeepAlive.test.ts @@ -0,0 +1,29 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { armSseKeepAlive } from '../../src/server/sseKeepAlive'; + +describe('armSseKeepAlive', () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it.each([0, -1, 0.5, Number.NaN, Number.POSITIVE_INFINITY])('disables invalid delay %s', delay => { + expect(armSseKeepAlive(delay, () => {})).toBeUndefined(); + expect(vi.getTimerCount()).toBe(0); + }); + + it('ticks at the configured interval', async () => { + const tick = vi.fn(); + const timer = armSseKeepAlive(1_000, tick)!; + await vi.advanceTimersByTimeAsync(3_000); + expect(tick).toHaveBeenCalledTimes(3); + clearInterval(timer); + }); + + it('clamps overflowing delays instead of creating a 1ms timer', async () => { + const tick = vi.fn(); + const timer = armSseKeepAlive(2 ** 31, tick)!; + await vi.advanceTimersByTimeAsync(60_000); + expect(tick).not.toHaveBeenCalled(); + clearInterval(timer); + }); +}); diff --git a/packages/server/test/server/streamableHttp.test.ts b/packages/server/test/server/streamableHttp.test.ts index beca451113..9ec6baf46c 100644 --- a/packages/server/test/server/streamableHttp.test.ts +++ b/packages/server/test/server/streamableHttp.test.ts @@ -1407,3 +1407,143 @@ 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); + expect(response.headers.get('cache-control')).toBe('no-cache, no-transform'); + expect(response.headers.get('x-accel-buffering')).toBe('no'); + + 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(); + expect(vi.getTimerCount()).toBe(0); + }); + + 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 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); + expect(response.headers.get('cache-control')).toBe('no-cache, no-transform'); + expect(response.headers.get('x-accel-buffering')).toBe('no'); + 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(); + }); + + it('should not initialize after close races request body parsing', async () => { + const onsessioninitialized = vi.fn(); + const transport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID(), + onsessioninitialized + }); + await new McpServer({ name: 'test-server', version: '1.0.0' }).connect(transport); + + let releaseBody!: () => void; + const body = new ReadableStream({ + start(controller) { + releaseBody = () => { + controller.enqueue(new TextEncoder().encode(JSON.stringify(TEST_MESSAGES.initialize))); + controller.close(); + }; + } + }); + const pending = transport.handleRequest( + new Request('http://localhost/mcp', { + method: 'POST', + headers: { Accept: 'application/json, text/event-stream', 'Content-Type': 'application/json' }, + body, + duplex: 'half' + }) + ); + + await transport.close(); + releaseBody(); + expect((await pending).status).toBe(404); + expect(onsessioninitialized).not.toHaveBeenCalled(); + }); + + it('should not register a stream after close races session initialization', async () => { + let releaseInitialization!: () => void; + const transport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID(), + onsessioninitialized: () => + new Promise(resolve => { + releaseInitialization = resolve; + }) + }); + await new McpServer({ name: 'test-server', version: '1.0.0' }).connect(transport); + + const pending = transport.handleRequest(createRequest('POST', TEST_MESSAGES.initialize)); + await vi.advanceTimersByTimeAsync(0); + await transport.close(); + releaseInitialization(); + + expect((await pending).status).toBe(404); + expect(vi.getTimerCount()).toBe(0); + }); +});