fix(server): bound resumability version gates to supported versions, pin the unsupported-version rejection format (#2280)
This commit is contained in:
Felix Weinberger
GitHub
parent
278d725309
commit
e8c7180109
3 changed files with 356 additions and 7 deletions
|
|
@ -367,10 +367,27 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns true if the client's protocol version supports empty SSE data in
|
||||
* priming events (the fix shipped with protocol version `2025-11-25`).
|
||||
*
|
||||
* The version is checked for membership in this transport instance's
|
||||
* supported protocol versions rather than with an open-ended
|
||||
* `>= '2025-11-25'` comparison: the value may come from an `initialize`
|
||||
* request body, which (unlike the `MCP-Protocol-Version` header) is not
|
||||
* validated against `supportedProtocolVersions` before reaching this
|
||||
* check. An unknown future version string must not silently enable
|
||||
* behavior reserved for versions this transport actually supports.
|
||||
*/
|
||||
private supportsEmptySSEData(protocolVersion: string): boolean {
|
||||
return this._supportedProtocolVersions.includes(protocolVersion) && protocolVersion >= '2025-11-25';
|
||||
}
|
||||
|
||||
/**
|
||||
* Writes a priming event to establish resumption capability.
|
||||
* Only sends if `eventStore` is configured (opt-in for resumability) and
|
||||
* the client's protocol version supports empty SSE data (>= `2025-11-25`).
|
||||
* the client's protocol version supports empty SSE data (a supported
|
||||
* version that is >= `2025-11-25`).
|
||||
*/
|
||||
private async writePrimingEvent(
|
||||
controller: ReadableStreamDefaultController<Uint8Array>,
|
||||
|
|
@ -383,9 +400,9 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
|
|||
}
|
||||
|
||||
// Priming events have empty data which older clients cannot handle.
|
||||
// Only send priming events to clients with protocol version >= 2025-11-25
|
||||
// which includes the fix for handling empty SSE data.
|
||||
if (protocolVersion < '2025-11-25') {
|
||||
// Only send priming events to clients whose protocol version includes
|
||||
// the fix for handling empty SSE data.
|
||||
if (!this.supportsEmptySSEData(protocolVersion)) {
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
@ -795,12 +812,12 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
|
|||
// handle each message
|
||||
for (const message of messages) {
|
||||
// Build closeSSEStream callback for requests when eventStore is configured
|
||||
// AND client supports resumability (protocol version >= 2025-11-25).
|
||||
// AND client supports resumability (a supported protocol version >= 2025-11-25).
|
||||
// Old clients can't resume if the stream is closed early because they
|
||||
// didn't receive a priming event with an event ID.
|
||||
let closeSSEStream: (() => void) | undefined;
|
||||
let closeStandaloneSSEStream: (() => void) | undefined;
|
||||
if (isJSONRPCRequest(message) && this._eventStore && clientProtocolVersion >= '2025-11-25') {
|
||||
if (isJSONRPCRequest(message) && this._eventStore && this.supportsEmptySSEData(clientProtocolVersion)) {
|
||||
closeSSEStream = () => {
|
||||
this.closeSSEStream(message.id);
|
||||
};
|
||||
|
|
|
|||
|
|
@ -0,0 +1,166 @@
|
|||
import { randomUUID } from 'node:crypto';
|
||||
|
||||
import type { JSONRPCMessage, MessageExtraInfo } from '@modelcontextprotocol/core';
|
||||
|
||||
import { McpServer } from '../../src/server/mcp.js';
|
||||
import type { EventId, EventStore, StreamId } from '../../src/server/streamableHttp.js';
|
||||
import { WebStandardStreamableHTTPServerTransport } from '../../src/server/streamableHttp.js';
|
||||
|
||||
/**
|
||||
* Gate-closure tests for the two protocol-version checks that guard
|
||||
* resumability behavior (priming events and `closeSSEStream` callbacks).
|
||||
*
|
||||
* The protocol version in an `initialize` request body is NOT validated
|
||||
* against `supportedProtocolVersions` (unlike the `MCP-Protocol-Version`
|
||||
* header), so these gates must be bounded: only versions the transport
|
||||
* instance actually supports may enable the resumability behavior introduced
|
||||
* with protocol version 2025-11-25. An unknown future version string must
|
||||
* behave like a client that does not support it.
|
||||
*/
|
||||
|
||||
function initializeRequest(protocolVersion: string): Request {
|
||||
const body = {
|
||||
jsonrpc: '2.0',
|
||||
method: 'initialize',
|
||||
params: {
|
||||
clientInfo: { name: 'test-client', version: '1.0' },
|
||||
protocolVersion,
|
||||
capabilities: {}
|
||||
},
|
||||
id: 'init-1'
|
||||
} as JSONRPCMessage;
|
||||
|
||||
return new Request('http://localhost/mcp', {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
Accept: 'application/json, text/event-stream'
|
||||
},
|
||||
body: JSON.stringify(body)
|
||||
});
|
||||
}
|
||||
|
||||
async function readFirstSSEChunk(response: Response): Promise<string> {
|
||||
const reader = response.body?.getReader();
|
||||
const { value } = await reader!.read();
|
||||
return new TextDecoder().decode(value);
|
||||
}
|
||||
|
||||
/**
|
||||
* A priming event is an SSE event with an event ID and empty data,
|
||||
* e.g. `id: <eventId>\ndata: \n\n`.
|
||||
*/
|
||||
function isPrimingEvent(sseChunk: string): boolean {
|
||||
const lines = sseChunk.split('\n');
|
||||
const dataLine = lines.find(line => line.startsWith('data:'));
|
||||
return lines.some(line => line.startsWith('id:')) && dataLine !== undefined && dataLine.slice(5).trim() === '';
|
||||
}
|
||||
|
||||
describe('WebStandardStreamableHTTPServerTransport - future-version gate closure', () => {
|
||||
let transport: WebStandardStreamableHTTPServerTransport;
|
||||
let mcpServer: McpServer;
|
||||
let storedEvents: Map<EventId, { streamId: StreamId; message: JSONRPCMessage }>;
|
||||
let capturedExtras: Array<MessageExtraInfo | undefined>;
|
||||
|
||||
beforeEach(async () => {
|
||||
storedEvents = new Map();
|
||||
capturedExtras = [];
|
||||
|
||||
const eventStore: EventStore = {
|
||||
async storeEvent(streamId: StreamId, message: JSONRPCMessage): Promise<EventId> {
|
||||
const eventId = `${streamId}_${storedEvents.size}`;
|
||||
storedEvents.set(eventId, { streamId, message });
|
||||
return eventId;
|
||||
},
|
||||
async getStreamIdForEventId(eventId: EventId): Promise<StreamId | undefined> {
|
||||
return storedEvents.get(eventId)?.streamId;
|
||||
},
|
||||
async replayEventsAfter(): Promise<StreamId> {
|
||||
throw new Error('not used in these tests');
|
||||
}
|
||||
};
|
||||
|
||||
mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }, { capabilities: {} });
|
||||
transport = new WebStandardStreamableHTTPServerTransport({
|
||||
sessionIdGenerator: () => randomUUID(),
|
||||
eventStore
|
||||
});
|
||||
await mcpServer.connect(transport);
|
||||
|
||||
// Capture the per-message extras (closeSSEStream availability) while
|
||||
// preserving normal server dispatch.
|
||||
const serverOnMessage = transport.onmessage;
|
||||
transport.onmessage = (message, extra) => {
|
||||
capturedExtras.push(extra);
|
||||
serverOnMessage?.(message, extra);
|
||||
};
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
await transport.close();
|
||||
});
|
||||
|
||||
function expectPrimingEventStored(expected: boolean): void {
|
||||
// The priming event is stored as an empty message; real messages always
|
||||
// have at least a `jsonrpc` member.
|
||||
const primingStored = [...storedEvents.values()].some(event => Object.keys(event.message).length === 0);
|
||||
expect(primingStored).toBe(expected);
|
||||
}
|
||||
|
||||
describe('unknown future protocol versions in the initialize body', () => {
|
||||
// Far-future sentinels, deliberately not the next planned revision
|
||||
// (2026-07-28): these cases must stay "unknown" when real future
|
||||
// versions gain support, rather than silently inverting.
|
||||
it.each(['2099-01-01', '2099-12-31'])('does not send a priming event for protocol version %s', async futureVersion => {
|
||||
const response = await transport.handleRequest(initializeRequest(futureVersion));
|
||||
expect(response.status).toBe(200);
|
||||
|
||||
const firstChunk = await readFirstSSEChunk(response);
|
||||
// The first SSE event must be the initialize response, not a priming event.
|
||||
expect(isPrimingEvent(firstChunk)).toBe(false);
|
||||
expect(firstChunk).toContain('"result"');
|
||||
|
||||
expectPrimingEventStored(false);
|
||||
});
|
||||
|
||||
it.each(['2099-01-01', '2099-12-31'])('does not provide closeSSEStream callbacks for protocol version %s', async futureVersion => {
|
||||
const response = await transport.handleRequest(initializeRequest(futureVersion));
|
||||
expect(response.status).toBe(200);
|
||||
|
||||
expect(capturedExtras).toHaveLength(1);
|
||||
expect(capturedExtras[0]?.closeSSEStream).toBeUndefined();
|
||||
expect(capturedExtras[0]?.closeStandaloneSSEStream).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe('existing protocol versions keep their behavior', () => {
|
||||
// Only 2025-11-25 (the version that introduced the empty-SSE-data fix)
|
||||
// takes the resumability paths - exactly as before the gates were bounded.
|
||||
const expectations: Array<[string, boolean]> = [
|
||||
['2024-10-07', false],
|
||||
['2024-11-05', false],
|
||||
['2025-03-26', false],
|
||||
['2025-06-18', false],
|
||||
['2025-11-25', true]
|
||||
];
|
||||
|
||||
it.each(expectations)('priming event for protocol version %s: %s', async (version, expectPriming) => {
|
||||
const response = await transport.handleRequest(initializeRequest(version));
|
||||
expect(response.status).toBe(200);
|
||||
|
||||
const firstChunk = await readFirstSSEChunk(response);
|
||||
expect(isPrimingEvent(firstChunk)).toBe(expectPriming);
|
||||
|
||||
expectPrimingEventStored(expectPriming);
|
||||
});
|
||||
|
||||
it.each(expectations)('closeSSEStream callbacks for protocol version %s: %s', async (version, expectAvailable) => {
|
||||
const response = await transport.handleRequest(initializeRequest(version));
|
||||
expect(response.status).toBe(200);
|
||||
|
||||
expect(capturedExtras).toHaveLength(1);
|
||||
expect(capturedExtras[0]?.closeSSEStream !== undefined).toBe(expectAvailable);
|
||||
expect(capturedExtras[0]?.closeStandaloneSSEStream !== undefined).toBe(expectAvailable);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
@ -0,0 +1,139 @@
|
|||
import { randomUUID } from 'node:crypto';
|
||||
|
||||
import type { JSONRPCMessage } from '@modelcontextprotocol/core';
|
||||
import { SUPPORTED_PROTOCOL_VERSIONS } from '@modelcontextprotocol/core';
|
||||
|
||||
import { McpServer } from '../../src/server/mcp.js';
|
||||
import { WebStandardStreamableHTTPServerTransport } from '../../src/server/streamableHttp.js';
|
||||
|
||||
/**
|
||||
* Wire-level continuity tests for the "Unsupported protocol version" rejection.
|
||||
*
|
||||
* The load-bearing surface is the HTTP 400 status plus the literal substring
|
||||
* `Unsupported protocol version` in the response body: the go-sdk client
|
||||
* substring-matches exactly that phrase on non-2xx bodies to drive its
|
||||
* protocol-version fallback (`streamableClientConn.checkResponse` in
|
||||
* go-sdk's `mcp/streamable.go`). Its structured JSON-RPC parse path rejects
|
||||
* this server's `id: null` error body, so the substring is the operative
|
||||
* interop signal — it must keep appearing verbatim in the wire bytes across
|
||||
* refactors of the transport internals.
|
||||
*
|
||||
* The rest of the message (prefix, echoed version, supported-versions list)
|
||||
* is asserted against a string derived from `SUPPORTED_PROTOCOL_VERSIONS`
|
||||
* rather than a frozen byte literal, so these tests survive additions to the
|
||||
* supported-versions list while still catching any rewording.
|
||||
*/
|
||||
|
||||
const INITIALIZE_MESSAGE = {
|
||||
jsonrpc: '2.0',
|
||||
method: 'initialize',
|
||||
params: {
|
||||
clientInfo: { name: 'test-client', version: '1.0' },
|
||||
protocolVersion: '2025-11-25',
|
||||
capabilities: {}
|
||||
},
|
||||
id: 'init-1'
|
||||
} as JSONRPCMessage;
|
||||
|
||||
const TOOLS_LIST_MESSAGE = {
|
||||
jsonrpc: '2.0',
|
||||
method: 'tools/list',
|
||||
params: {},
|
||||
id: 'tools-1'
|
||||
} as JSONRPCMessage;
|
||||
|
||||
interface JSONRPCErrorBody {
|
||||
jsonrpc: string;
|
||||
id: unknown;
|
||||
error: { code: number; message: string };
|
||||
}
|
||||
|
||||
function postRequest(body: JSONRPCMessage, headers: Record<string, string> = {}): Request {
|
||||
return new Request('http://localhost/mcp', {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
Accept: 'application/json, text/event-stream',
|
||||
...headers
|
||||
},
|
||||
body: JSON.stringify(body)
|
||||
});
|
||||
}
|
||||
|
||||
async function initializeServer(transport: WebStandardStreamableHTTPServerTransport): Promise<string> {
|
||||
const response = await transport.handleRequest(postRequest(INITIALIZE_MESSAGE));
|
||||
expect(response.status).toBe(200);
|
||||
return response.headers.get('mcp-session-id') as string;
|
||||
}
|
||||
|
||||
async function connectedTransport(supportedProtocolVersions?: string[]): Promise<WebStandardStreamableHTTPServerTransport> {
|
||||
// `connect()` passes the server's supported protocol versions down to the
|
||||
// transport, so a custom list is configured on the server options.
|
||||
const mcpServer = new McpServer(
|
||||
{ name: 'test-server', version: '1.0.0' },
|
||||
{ capabilities: {}, ...(supportedProtocolVersions ? { supportedProtocolVersions } : {}) }
|
||||
);
|
||||
const transport = new WebStandardStreamableHTTPServerTransport({
|
||||
sessionIdGenerator: () => randomUUID()
|
||||
});
|
||||
await mcpServer.connect(transport);
|
||||
return transport;
|
||||
}
|
||||
|
||||
describe('Unsupported protocol version - wire literal continuity', () => {
|
||||
it('rejects an unsupported MCP-Protocol-Version header with HTTP 400, code -32000, and the sniffed literal substring', async () => {
|
||||
const transport = await connectedTransport();
|
||||
try {
|
||||
const sessionId = await initializeServer(transport);
|
||||
|
||||
const response = await transport.handleRequest(
|
||||
postRequest(TOOLS_LIST_MESSAGE, {
|
||||
'mcp-session-id': sessionId,
|
||||
'mcp-protocol-version': '2099-01-01'
|
||||
})
|
||||
);
|
||||
|
||||
expect(response.status).toBe(400);
|
||||
expect(response.headers.get('content-type')).toContain('application/json');
|
||||
|
||||
const rawBody = await response.text();
|
||||
// The substring deployed clients (go-sdk) sniff must appear
|
||||
// verbatim in the wire bytes.
|
||||
expect(rawBody).toContain('Unsupported protocol version');
|
||||
|
||||
const body = JSON.parse(rawBody) as JSONRPCErrorBody;
|
||||
expect(body.jsonrpc).toBe('2.0');
|
||||
expect(body.id).toBeNull();
|
||||
expect(body.error.code).toBe(-32_000);
|
||||
expect(body.error.message).toBe(
|
||||
`Bad Request: Unsupported protocol version: 2099-01-01 (supported versions: ${SUPPORTED_PROTOCOL_VERSIONS.join(', ')})`
|
||||
);
|
||||
} finally {
|
||||
await transport.close();
|
||||
}
|
||||
});
|
||||
|
||||
it('derives the supported-versions suffix from the per-instance supportedProtocolVersions', async () => {
|
||||
const transport = await connectedTransport(['2025-11-25', '2025-06-18']);
|
||||
try {
|
||||
const sessionId = await initializeServer(transport);
|
||||
|
||||
const response = await transport.handleRequest(
|
||||
postRequest(TOOLS_LIST_MESSAGE, {
|
||||
'mcp-session-id': sessionId,
|
||||
'mcp-protocol-version': '1999-01-01'
|
||||
})
|
||||
);
|
||||
|
||||
expect(response.status).toBe(400);
|
||||
|
||||
const body = (await response.json()) as JSONRPCErrorBody;
|
||||
expect(body.error.code).toBe(-32_000);
|
||||
expect(body.error.message).toBe(
|
||||
'Bad Request: Unsupported protocol version: 1999-01-01 (supported versions: 2025-11-25, 2025-06-18)'
|
||||
);
|
||||
} finally {
|
||||
await transport.close();
|
||||
}
|
||||
});
|
||||
});
|
||||
Loading…
Reference in a new issue