Skip to content

Commit 5d1206e

Browse files
felixweinbergerfelixweinberger
committed
fix: skip priming events for clients with old protocol versions
Priming events (SEP-1699) have empty SSE data which older clients cannot handle - they try to JSON.parse("") and crash. Only send priming events to clients with protocol version >= 2025-11-25, which includes the fix for handling empty SSE data. For the initialize request, the protocol version is extracted from the request params. For subsequent requests, it's taken from the mcp-protocol-version header.
1 parent 423f5b1 commit 5d1206e

File tree

2 files changed

+146
-14
lines changed

2 files changed

+146
-14
lines changed

‎src/server/streamableHttp.test.ts‎

Lines changed: 122 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1671,7 +1671,7 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
16711671
'Content-Type': 'application/json',
16721672
Accept: 'text/event-stream, application/json',
16731673
'mcp-session-id': sessionId,
1674-
'mcp-protocol-version': '2025-03-26'
1674+
'mcp-protocol-version': '2025-11-25'
16751675
},
16761676
body: JSON.stringify(toolCallRequest)
16771677
});
@@ -1690,6 +1690,57 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
16901690
expect(text).toContain('data: ');
16911691
});
16921692

1693+
it('should NOT send priming event for old protocol versions (backwards compatibility)', async () => {
1694+
const result = await createTestServer({
1695+
sessionIdGenerator: () => randomUUID(),
1696+
eventStore: createEventStore(),
1697+
retryInterval: 5000
1698+
});
1699+
server = result.server;
1700+
transport = result.transport;
1701+
baseUrl = result.baseUrl;
1702+
mcpServer = result.mcpServer;
1703+
1704+
// Initialize to get session ID
1705+
const initResponse = await sendPostRequest(baseUrl, TEST_MESSAGES.initialize);
1706+
sessionId = initResponse.headers.get('mcp-session-id') as string;
1707+
expect(sessionId).toBeDefined();
1708+
1709+
// Send a tool call request with OLD protocol version
1710+
const toolCallRequest: JSONRPCMessage = {
1711+
jsonrpc: '2.0',
1712+
id: 100,
1713+
method: 'tools/call',
1714+
params: { name: 'greet', arguments: { name: 'Test' } }
1715+
};
1716+
1717+
const postResponse = await fetch(baseUrl, {
1718+
method: 'POST',
1719+
headers: {
1720+
'Content-Type': 'application/json',
1721+
Accept: 'text/event-stream, application/json',
1722+
'mcp-session-id': sessionId,
1723+
'mcp-protocol-version': '2025-03-26'
1724+
},
1725+
body: JSON.stringify(toolCallRequest)
1726+
});
1727+
1728+
expect(postResponse.status).toBe(200);
1729+
expect(postResponse.headers.get('content-type')).toBe('text/event-stream');
1730+
1731+
// Read the first chunk - should be the actual response, not a priming event
1732+
const reader = postResponse.body?.getReader();
1733+
const { value } = await reader!.read();
1734+
const text = new TextDecoder().decode(value);
1735+
1736+
// Should NOT contain a priming event (empty data line before the response)
1737+
// The first message should be the actual tool result
1738+
expect(text).toContain('event: message');
1739+
expect(text).toContain('"result"');
1740+
// Should NOT have a separate priming event line with empty data
1741+
expect(text).not.toMatch(/^id:.*\ndata:\s*\n\n/);
1742+
});
1743+
16931744
it('should send priming event without retry field when retryInterval is not configured', async () => {
16941745
const result = await createTestServer({
16951746
sessionIdGenerator: () => randomUUID(),
@@ -1720,7 +1771,7 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
17201771
'Content-Type': 'application/json',
17211772
Accept: 'text/event-stream, application/json',
17221773
'mcp-session-id': sessionId,
1723-
'mcp-protocol-version': '2025-03-26'
1774+
'mcp-protocol-version': '2025-11-25'
17241775
},
17251776
body: JSON.stringify(toolCallRequest)
17261777
});
@@ -1786,7 +1837,7 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
17861837
'Content-Type': 'application/json',
17871838
Accept: 'text/event-stream, application/json',
17881839
'mcp-session-id': sessionId,
1789-
'mcp-protocol-version': '2025-03-26'
1840+
'mcp-protocol-version': '2025-11-25'
17901841
},
17911842
body: JSON.stringify(toolCallRequest)
17921843
});
@@ -1849,7 +1900,7 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
18491900
'Content-Type': 'application/json',
18501901
Accept: 'text/event-stream, application/json',
18511902
'mcp-session-id': sessionId,
1852-
'mcp-protocol-version': '2025-03-26'
1903+
'mcp-protocol-version': '2025-11-25'
18531904
},
18541905
body: JSON.stringify(toolCallRequest)
18551906
});
@@ -1868,6 +1919,67 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
18681919
expect(typeof receivedCloseSSEStream).toBe('function');
18691920
});
18701921

1922+
it('should NOT provide closeSSEStream callback for old protocol versions (backwards compatibility)', async () => {
1923+
const result = await createTestServer({
1924+
sessionIdGenerator: () => randomUUID(),
1925+
eventStore: createEventStore(),
1926+
retryInterval: 1000
1927+
});
1928+
server = result.server;
1929+
transport = result.transport;
1930+
baseUrl = result.baseUrl;
1931+
mcpServer = result.mcpServer;
1932+
1933+
// Track whether closeSSEStream callback was provided
1934+
let receivedCloseSSEStream: (() => void) | undefined;
1935+
let receivedCloseStandaloneSSEStream: (() => void) | undefined;
1936+
1937+
// Register a tool that captures the extra.closeSSEStream callback
1938+
mcpServer.tool('test-old-version-tool', 'Test tool', {}, async (_args, extra) => {
1939+
receivedCloseSSEStream = extra.closeSSEStream;
1940+
receivedCloseStandaloneSSEStream = extra.closeStandaloneSSEStream;
1941+
return { content: [{ type: 'text', text: 'Done' }] };
1942+
});
1943+
1944+
// Initialize to get session ID
1945+
const initResponse = await sendPostRequest(baseUrl, TEST_MESSAGES.initialize);
1946+
sessionId = initResponse.headers.get('mcp-session-id') as string;
1947+
expect(sessionId).toBeDefined();
1948+
1949+
// Call the tool with OLD protocol version
1950+
const toolCallRequest: JSONRPCMessage = {
1951+
jsonrpc: '2.0',
1952+
id: 200,
1953+
method: 'tools/call',
1954+
params: { name: 'test-old-version-tool', arguments: {} }
1955+
};
1956+
1957+
const postResponse = await fetch(baseUrl, {
1958+
method: 'POST',
1959+
headers: {
1960+
'Content-Type': 'application/json',
1961+
Accept: 'text/event-stream, application/json',
1962+
'mcp-session-id': sessionId,
1963+
'mcp-protocol-version': '2025-03-26'
1964+
},
1965+
body: JSON.stringify(toolCallRequest)
1966+
});
1967+
1968+
expect(postResponse.status).toBe(200);
1969+
1970+
// Read all events to completion
1971+
const reader = postResponse.body?.getReader();
1972+
while (true) {
1973+
const { done } = await reader!.read();
1974+
if (done) break;
1975+
}
1976+
1977+
// Verify closeSSEStream callbacks were NOT provided for old protocol version
1978+
// even though eventStore is configured
1979+
expect(receivedCloseSSEStream).toBeUndefined();
1980+
expect(receivedCloseStandaloneSSEStream).toBeUndefined();
1981+
});
1982+
18711983
it('should NOT provide closeSSEStream callback when eventStore is NOT configured', async () => {
18721984
const result = await createTestServer({
18731985
sessionIdGenerator: () => randomUUID()
@@ -1963,7 +2075,7 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
19632075
'Content-Type': 'application/json',
19642076
Accept: 'text/event-stream, application/json',
19652077
'mcp-session-id': sessionId,
1966-
'mcp-protocol-version': '2025-03-26'
2078+
'mcp-protocol-version': '2025-11-25'
19672079
},
19682080
body: JSON.stringify(toolCallRequest)
19692081
});
@@ -2010,7 +2122,7 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
20102122
headers: {
20112123
Accept: 'text/event-stream',
20122124
'mcp-session-id': sessionId,
2013-
'mcp-protocol-version': '2025-03-26'
2125+
'mcp-protocol-version': '2025-11-25'
20142126
}
20152127
});
20162128
expect(sseResponse.status).toBe(200);
@@ -2040,7 +2152,7 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
20402152
'Content-Type': 'application/json',
20412153
Accept: 'text/event-stream, application/json',
20422154
'mcp-session-id': sessionId,
2043-
'mcp-protocol-version': '2025-03-26'
2155+
'mcp-protocol-version': '2025-11-25'
20442156
},
20452157
body: JSON.stringify(toolCallRequest)
20462158
});
@@ -2091,7 +2203,7 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
20912203
headers: {
20922204
Accept: 'text/event-stream',
20932205
'mcp-session-id': sessionId,
2094-
'mcp-protocol-version': '2025-03-26'
2206+
'mcp-protocol-version': '2025-11-25'
20952207
}
20962208
});
20972209
expect(sseResponse.status).toBe(200);
@@ -2122,7 +2234,7 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
21222234
'Content-Type': 'application/json',
21232235
Accept: 'text/event-stream, application/json',
21242236
'mcp-session-id': sessionId,
2125-
'mcp-protocol-version': '2025-03-26'
2237+
'mcp-protocol-version': '2025-11-25'
21262238
},
21272239
body: JSON.stringify(toolCallRequest)
21282240
});
@@ -2152,7 +2264,7 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
21522264
headers: {
21532265
Accept: 'text/event-stream',
21542266
'mcp-session-id': sessionId,
2155-
'mcp-protocol-version': '2025-03-26',
2267+
'mcp-protocol-version': '2025-11-25',
21562268
'last-event-id': lastEventId
21572269
}
21582270
});

‎src/server/streamableHttp.ts‎

Lines changed: 24 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -276,13 +276,21 @@ export class StreamableHTTPServerTransport implements Transport {
276276

277277
/**
278278
* Writes a priming event to establish resumption capability.
279-
* Only sends if eventStore is configured (opt-in for resumability).
279+
* Only sends if eventStore is configured (opt-in for resumability) and
280+
* the client's protocol version supports empty SSE data (>= 2025-11-25).
280281
*/
281-
private async _maybeWritePrimingEvent(res: ServerResponse, streamId: string): Promise<void> {
282+
private async _maybeWritePrimingEvent(res: ServerResponse, streamId: string, protocolVersion: string): Promise<void> {
282283
if (!this._eventStore) {
283284
return;
284285
}
285286

287+
// Priming events have empty data which older clients cannot handle.
288+
// Only send priming events to clients with protocol version >= 2025-11-25
289+
// which includes the fix for handling empty SSE data.
290+
if (protocolVersion < '2025-11-25') {
291+
return;
292+
}
293+
286294
const primingEventId = await this._eventStore.storeEvent(streamId, {} as JSONRPCMessage);
287295

288296
let primingEvent = `id: ${primingEventId}\ndata: \n\n`;
@@ -619,6 +627,15 @@ export class StreamableHTTPServerTransport implements Transport {
619627
// The default behavior is to use SSE streaming
620628
// but in some cases server will return JSON responses
621629
const streamId = randomUUID();
630+
631+
// Extract protocol version for priming event decision.
632+
// For initialize requests, get from request params.
633+
// For other requests, get from header (already validated).
634+
const initRequest = messages.find(m => isInitializeRequest(m));
635+
const clientProtocolVersion = initRequest
636+
? initRequest.params.protocolVersion
637+
: ((req.headers['mcp-protocol-version'] as string) ?? DEFAULT_NEGOTIATED_PROTOCOL_VERSION);
638+
622639
if (!this._enableJsonResponse) {
623640
const headers: Record<string, string> = {
624641
'Content-Type': 'text/event-stream',
@@ -633,7 +650,7 @@ export class StreamableHTTPServerTransport implements Transport {
633650

634651
res.writeHead(200, headers);
635652

636-
await this._maybeWritePrimingEvent(res, streamId);
653+
await this._maybeWritePrimingEvent(res, streamId, clientProtocolVersion);
637654
}
638655
// Store the response for this request to send messages back through this connection
639656
// We need to track by request ID to maintain the connection
@@ -656,9 +673,12 @@ export class StreamableHTTPServerTransport implements Transport {
656673
// handle each message
657674
for (const message of messages) {
658675
// Build closeSSEStream callback for requests when eventStore is configured
676+
// AND client supports resumability (protocol version >= 2025-11-25).
677+
// Old clients can't resume if the stream is closed early because they
678+
// didn't receive a priming event with an event ID.
659679
let closeSSEStream: (() => void) | undefined;
660680
let closeStandaloneSSEStream: (() => void) | undefined;
661-
if (isJSONRPCRequest(message) && this._eventStore) {
681+
if (isJSONRPCRequest(message) && this._eventStore && clientProtocolVersion >= '2025-11-25') {
662682
closeSSEStream = () => {
663683
this.closeSSEStream(message.id);
664684
};

0 commit comments

Comments
 (0)