modelcontextprotocol/typescript-sdk
Watch
1
0
Fork
You've already forked typescript-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

fix: send SSE keep-alive comment frames from WebStandardStreamableHTTPServerTransport (#2541)

Browse source
This commit is contained in:
Matt 2026-07-27 14:21:49 +01:00 • committed by GitHub
commit 470678d56b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
13 changed files with 480 additions and 36 deletions
Download patch file Download diff file Expand all files Collapse all files

5
.changeset/streamable-http-sse-keepalive.md Normal file
View file

@ -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.

4
docs/troubleshooting.md
View file

@ -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.

30
packages/server/src/server/createMcpHandler.ts
View file

@ -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<Response> {
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

5
packages/server/src/server/invoke.ts
View file

@ -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<Response> {
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, {

23
packages/server/src/server/listenRouter.ts
View file

@ -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'
}

16
packages/server/src/server/perRequestTransport.ts
View file

@ -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<Uint8Array>;
encoder: InstanceType<typeof TextEncoder>;
closed: boolean;
keepAliveTimer?: ReturnType<typeof setInterval>;
}
/**
@ -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<void> {
@ -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 {

15
packages/server/src/server/sseKeepAlive.ts Normal file
View file

@ -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<typeof setInterval> | 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;
}

88
packages/server/src/server/streamableHttp.ts
View file

@ -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<Uint8Array>,
encoder: InstanceType<typeof TextEncoder>
): ReturnType<typeof setInterval> | 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<Response> {
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<Uint8Array>;
// Captured by cancel/cleanup before it is assigned after stream setup.
// eslint-disable-next-line prefer-const
let keepAliveTimer: ReturnType<typeof setInterval> | undefined;
// Create a ReadableStream with a controller we can use to push SSE events
const readable = new ReadableStream<Uint8Array>({
@ -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<string, string> = {
'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<string, string> = {
'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<Uint8Array>;
let keepAliveTimer: ReturnType<typeof setInterval> | 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<Uint8Array>;
let keepAliveTimer: ReturnType<typeof setInterval> | undefined;
const readable = new ReadableStream<Uint8Array>({
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<string, string> = {
'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();
}
}
/**

54
packages/server/test/server/createMcpHandler.test.ts
View file

@ -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<void>(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();
}
});
});

42
packages/server/test/server/createMcpHandlerListen.test.ts
View file

@ -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 }));

65
packages/server/test/server/perRequestStreaming.test.ts
View file

@ -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<CallToolResult>,
responseMode?: PerRequestResponseMode
responseMode?: PerRequestResponseMode,
keepAliveMs?: number
): Promise<StreamingSetup> {
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<void>(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<void>(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);
});
});

29
packages/server/test/server/sseKeepAlive.test.ts Normal file
View file

@ -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);
});
});

140
packages/server/test/server/streamableHttp.test.ts
View file

@ -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<CallToolResult> => {
await new Promise<void>(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<Uint8Array>({
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<void>(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);
});
});