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 Streamable HTTP server transport (v1.x) (#2538)

Browse source
This commit is contained in:
Matt 2026-07-23 17:22:29 +01:00 • committed by GitHub
commit 1dad2634ce
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 252 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -0,0 +1,6 @@
---
'@modelcontextprotocol/sdk': patch
---
`StreamableHTTPServerTransport` and `WebStandardStreamableHTTPServerTransport` now write SSE keep-alive comment frames (`: keepalive`) to open SSE streams so idle connections (e.g. the standalone GET stream, or a POST stream during a long-running tool call) are not killed by
intermediaries or server idle timeouts. Configurable via the new `keepAliveMs` option (default 15000; set 0 to disable).

70
src/server/webStandardStreamableHttp.ts
View file

@ -146,8 +146,24 @@ export interface WebStandardStreamableHTTPServerTransportOptions {
* client reconnection timing for polling behavior.
*/
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;
}
/** Default interval between SSE keep-alive comment frames. */
const DEFAULT_KEEP_ALIVE_MS = 15_000;
/**
* Options for handling a request
*/
@ -225,6 +241,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
private _allowedOrigins?: string[];
private _enableDnsRebindingProtection: boolean;
private _retryInterval?: number;
private _keepAliveMs: number;
private _keepAliveTimers: Map<string, ReturnType<typeof setInterval>> = new Map();
sessionId?: string;
onclose?: () => void;
@ -241,6 +259,43 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
this._allowedOrigins = options.allowedOrigins;
this._enableDnsRebindingProtection = options.enableDnsRebindingProtection ?? false;
this._retryInterval = options.retryInterval;
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 stopKeepAlive when the stream is cleaned up, and
* clears itself if a write fails (stream already closed/cancelled).
*/
private startKeepAlive(streamId: string, controller: ReadableStreamDefaultController<Uint8Array>, encoder: TextEncoder): void {
if (this._keepAliveMs <= 0) {
return;
}
this.stopKeepAlive(streamId);
const timer = setInterval(() => {
try {
controller.enqueue(encoder.encode(': keepalive\n\n'));
} catch {
this.stopKeepAlive(streamId);
}
}, this._keepAliveMs);
// Don't let the keep-alive timer hold the process open (Node.js only)
(timer as { unref?: () => void }).unref?.();
this._keepAliveTimers.set(streamId, timer);
}
/**
* Clears the keep-alive interval for a stream, if one is armed.
*/
private stopKeepAlive(streamId: string): void {
const timer = this._keepAliveTimers.get(streamId);
if (timer !== undefined) {
clearInterval(timer);
this._keepAliveTimers.delete(streamId);
}
}
/**
@ -425,6 +480,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
},
cancel: () => {
// Stream was cancelled by client
this.stopKeepAlive(this._standaloneSseStreamId);
this._streamMapping.delete(this._standaloneSseStreamId);
}
});
@ -445,6 +501,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
controller: streamController!,
encoder,
cleanup: () => {
this.stopKeepAlive(this._standaloneSseStreamId);
this._streamMapping.delete(this._standaloneSseStreamId);
try {
streamController!.close();
@ -454,6 +511,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
}
});
this.startKeepAlive(this._standaloneSseStreamId, streamController!, encoder);
return new Response(readable, { headers });
}
@ -528,6 +587,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
controller: streamController!,
encoder,
cleanup: () => {
this.stopKeepAlive(replayedStreamId);
this._streamMapping.delete(replayedStreamId);
try {
streamController!.close();
@ -537,6 +597,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
}
});
this.startKeepAlive(replayedStreamId, streamController!, encoder);
return new Response(readable, { headers });
} catch (error) {
this.onerror?.(error as Error);
@ -743,6 +805,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
},
cancel: () => {
// Stream was cancelled by client
this.stopKeepAlive(streamId);
this._streamMapping.delete(streamId);
}
});
@ -766,6 +829,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
controller: streamController!,
encoder,
cleanup: () => {
this.stopKeepAlive(streamId);
this._streamMapping.delete(streamId);
try {
streamController!.close();
@ -778,6 +842,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
}
}
this.startKeepAlive(streamId, streamController!, encoder);
// Write priming event if event store is configured (after mapping is set up)
await this.writePrimingEvent(streamController!, encoder, streamId, clientProtocolVersion);
@ -901,6 +967,10 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
});
this._streamMapping.clear();
// Clear any keep-alive timers not already cleared by stream cleanup
this._keepAliveTimers.forEach(timer => clearInterval(timer));
this._keepAliveTimers.clear();
// Clear any pending responses
this._requestResponseMap.clear();
this.onclose?.();

176
test/server/streamableHttp.test.ts
View file

@ -3272,3 +3272,179 @@ describe('WebStandardStreamableHTTPServerTransport - onerror callback', () => {
await storeTransport.close();
});
});
describe('WebStandardStreamableHTTPServerTransport SSE keep-alive', () => {
/** Shorthand to build a Web Standard Request for direct transport testing. */
function req(method: string, opts?: { body?: unknown; headers?: Record<string, string> }): Request {
const headers: Record<string, string> = { ...opts?.headers };
if (method === 'POST') {
headers['Accept'] ??= 'application/json, text/event-stream';
headers['Content-Type'] ??= 'application/json';
} else if (method === 'GET') {
headers['Accept'] ??= 'text/event-stream';
}
return new Request('http://localhost/mcp', {
method,
headers,
body: opts?.body !== undefined ? JSON.stringify(opts.body) : undefined
});
}
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(req('POST', { body: 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(
req('GET', { headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' } })
);
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(
req('GET', { headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' } })
);
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(
req('GET', { headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' } })
);
const reader = response.body!.getReader();
await vi.advanceTimersByTimeAsync(60000);
const read = reader.read();
const raced = await Promise.race([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(
req('GET', { headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' } })
);
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.tool('slow', async () => {
await new Promise<void>(resolve => {
resolveTool = resolve;
});
return { content: [{ type: 'text', text: 'done' }] };
});
await mcpServer.connect(transport);
const initResponse = await transport.handleRequest(req('POST', { body: TEST_MESSAGES.initialize }));
const sessionId = initResponse.headers.get('mcp-session-id') as string;
const response = await transport.handleRequest(
req('POST', {
body: { jsonrpc: '2.0', method: 'tools/call', params: { name: 'slow', arguments: {} }, id: 'call-1' },
headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' }
})
);
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();
});
it('should supersede the previous keep-alive timer when a replayed stream re-registers under the same stream id', async () => {
// Event store WITHOUT the optional getStreamIdForEventId — the replay
// path then skips its 409 conflict check, so a reconnect re-registers
// the same stream id. The predecessor's timer must be replaced, not
// orphaned (an orphaned timer's failing write would clear the live
// stream's keep-alive via stopKeepAlive on the shared stream id).
const eventStore: EventStore = {
async storeEvent(): Promise<EventId> {
return 'evt-1';
},
async replayEventsAfter(): Promise<StreamId> {
return 'stream-1';
}
};
const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), eventStore });
await new McpServer({ name: 'test-server', version: '1.0.0' }).connect(transport);
const initResponse = await transport.handleRequest(req('POST', { body: TEST_MESSAGES.initialize }));
const sessionId = initResponse.headers.get('mcp-session-id') as string;
const replayHeaders = { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25', 'Last-Event-ID': 'evt-1' };
const first = await transport.handleRequest(req('GET', { headers: replayHeaders }));
expect(first.status).toBe(200);
// Reconnect with the same Last-Event-ID — re-registers 'stream-1'
const second = await transport.handleRequest(req('GET', { headers: replayHeaders }));
expect(second.status).toBe(200);
// Exactly one keep-alive timer must remain armed (plus none orphaned)
expect(vi.getTimerCount()).toBe(1);
// The live (second) stream still receives keep-alive frames
const reader = second.body!.getReader();
await vi.advanceTimersByTimeAsync(15000);
const { value } = await reader.read();
expect(new TextDecoder().decode(value)).toBe(': keepalive\n\n');
await transport.close();
});
});