diff --git a/.changeset/shttp-session-capture-on-ok.md b/.changeset/shttp-session-capture-on-ok.md new file mode 100644 index 0000000000..5c52e696a9 --- /dev/null +++ b/.changeset/shttp-session-capture-on-ok.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/client': patch +--- + +`StreamableHTTPClientTransport` session-id handling now follows the spec's assignment model: the `Mcp-Session-Id` response header is captured only from a successful response to a POST that carried an `InitializeRequest` — the spec assigns the id "at initialization time ... on the HTTP response containing the `InitializeResult`", and is silent on the header elsewhere. Session headers on any other response are now ignored entirely (previously any 2xx reply could install one), so an error reply, a late reply from an older session, or a stray header on a normal request contributes nothing to session state. Previously a 404 probe reply carrying a session id made the legacy fallback `initialize` present an id the server never issued, which strict stateful servers reject. A 404 to a POST that carried the current id now clears it (the session is gone per spec), and `terminateSession()` treats 404 as successful termination. On a legacy-era connection the client then starts a new session automatically — a fresh session-less `InitializeRequest`, per the spec's session-management requirement — and retries the failed request or notification once (pagination continuations re-establish the session but surface the 404 — the dead session's cursor is meaningless to the new one); server-side per-session state (e.g. resource subscriptions) does not carry over. diff --git a/docs/migration/upgrade-to-v2.md b/docs/migration/upgrade-to-v2.md index 89188aa893..e5d3623dda 100644 --- a/docs/migration/upgrade-to-v2.md +++ b/docs/migration/upgrade-to-v2.md @@ -1560,6 +1560,23 @@ rewrite required unless noted. open is benign (the client proceeds without the stream), and a `405` answering the session DELETE resolves `terminateSession()` normally — stateless-topology servers that decline both verbs keep working without changes, as in v1. +- **Changed: session-id capture is handshake-only.** The `Mcp-Session-Id` response + header is stored only from a successful response to a POST that carried an + `InitializeRequest` — the spec assigns the id "at initialization time ... on the + HTTP response containing the `InitializeResult`". v1 captured the header from any + response (including error replies), so a server that (re)assigned session ids on + ordinary responses could silently rotate the client's session. The spec defines + that single assignment point and is silent on the header elsewhere, so the client + now ignores it on every other response — a server that wants a session must + assign the id on the `initialize` response, as SDK servers always have. +- **Changed: session `404` handling.** A `404` answering the session DELETE now + resolves `terminateSession()` (v1 threw) — the session is gone either way, and the + stored id is cleared. A `404` to a POST that carried the current session id clears + `transport.sessionId`, and on a pre-2026 connection the client re-initializes + automatically and retries the failed request once (pagination continuations are not + retried, and server-side per-session state such as subscriptions or log level does + not carry over). Host code keyed off that `404` now only sees it when the automatic + recovery has also failed, at which point the client is closed. #### stdio transport @@ -1614,7 +1631,9 @@ rewrite required unless noted. request body no longer enable it. - Session-ID mismatch still responds `404` with JSON-RPC `-32001` (`Session not found`), unchanged from v1. This `-32001` is an SDK convention, not spec-assigned; client code - should key off the HTTP `404` status, not `-32001`. + should key off the HTTP `404` status, not `-32001`. The v2 client handles this `404` + itself on pre-2026 connections (see the session `404` bullet under the client + Streamable HTTP section above) — hosts only observe it when that recovery fails. #### Server (deprecated accessors and app-factory Origin validation) diff --git a/docs/serving/sessions-state-scaling.md b/docs/serving/sessions-state-scaling.md index 0bbffe450a..400e88015b 100644 --- a/docs/serving/sessions-state-scaling.md +++ b/docs/serving/sessions-state-scaling.md @@ -59,7 +59,7 @@ app.get('/mcp', route); app.delete('/mcp', route); ``` -The map cleans itself up: `transport.onclose` fires when the session ends, whether the client sent `DELETE` or you called `transport.close()`. A request with an unknown `Mcp-Session-Id` gets the `404` above, which tells the client to start a new session; a request with no session header at all gets the `400`, which tells it to re-send the id it already has instead of re-initializing. +The map cleans itself up: `transport.onclose` fires when the session ends, whether the client sent `DELETE` or you called `transport.close()`. A request with an unknown `Mcp-Session-Id` gets the `404` above, which tells the client to start a new session (the v2 client does this automatically on pre-2026 connections, re-initializing and retrying the failed request once); a request with no session header at all gets the `400`, which tells it to re-send the id it already has instead of re-initializing. ::: tip On shutdown, close every stored transport — `for (const [, transport] of sessions) await transport.close()` — before exiting; `close()` ends the session's SSE streams and rejects its pending requests. diff --git a/examples/repl/client.ts b/examples/repl/client.ts index ccb1ca2c0c..bb5fb50b78 100644 --- a/examples/repl/client.ts +++ b/examples/repl/client.ts @@ -531,23 +531,17 @@ async function terminateSession(): Promise { try { console.log('Terminating session with ID:', transport.sessionId); + // terminateSession() also resolves on 405 (server declines DELETE) + // and 404 (session already gone), clearing the session id in every + // resolving case — the outcome is not distinguishable here. await transport.terminateSession(); - console.log('Session terminated successfully'); + console.log('Session terminated; session ID cleared'); + sessionId = undefined; - // Check if sessionId was cleared after termination - if (transport.sessionId) { - console.log('Server responded with 405 Method Not Allowed (session termination not supported)'); - console.log('Session ID is still active:', transport.sessionId); - } else { - console.log('Session ID has been cleared'); - sessionId = undefined; - - // Also close the transport and clear client objects - await transport.close(); - console.log('Transport closed after session termination'); - client = null; - transport = null; - } + await transport.close(); + console.log('Transport closed after session termination'); + client = null; + transport = null; } catch (error) { console.error('Error terminating session:', error); } diff --git a/packages/client/src/client/client.ts b/packages/client/src/client/client.ts index 5028e937cd..ecd9c58049 100644 --- a/packages/client/src/client/client.ts +++ b/packages/client/src/client/client.ts @@ -34,7 +34,9 @@ import type { LoggingLevel, MessageExtraInfo, NonCompleteResultFlow, + Notification, NotificationMethod, + NotificationOptions, ProtocolEra, ProtocolOptions, ReadResourceRequest, @@ -43,6 +45,7 @@ import type { RequestOptions, ResolvedInputRequiredDriverConfig, Result, + ResultTypeMap, ServerCapabilities, StandardSchemaV1, SubscribeRequest, @@ -74,6 +77,7 @@ import { scanXMcpHeaderDeclarations, SdkError, SdkErrorCode, + SdkHttpError, SUBSCRIPTION_ID_META_KEY, SUPPORTED_MODERN_PROTOCOL_VERSIONS } from '@modelcontextprotocol/core-internal'; @@ -897,6 +901,92 @@ export class Client extends Protocol { } } + private _sessionReinit?: Promise; + + override request( + request: { method: M; params?: Record }, + options?: RequestOptions + ): Promise; + override request( + request: { method: string; params?: Record }, + resultSchema: T, + options?: RequestOptions + ): Promise>; + override async request( + request: { method: string; params?: Record }, + schemaOrOptions?: StandardSchemaV1 | RequestOptions, + maybeOptions?: RequestOptions + ): Promise { + try { + return await super.request(request as never, schemaOrOptions as never, maybeOptions); + } catch (error) { + // Spec (Session Management): a 404 to a request that carried the + // session id means the session is gone — start a new one with a + // fresh InitializeRequest, then retry the failed request once. + if (request.method === 'initialize' || !this._isSessionTerminated(error)) { + throw error; + } + await this._recoverSession(error); + // A cursor minted by the dead session is meaningless to the new + // one: the session is recovered, but the 404 surfaces so the + // caller restarts pagination. + if (request.params !== undefined && 'cursor' in request.params) { + throw error; + } + return await super.request(request as never, schemaOrOptions as never, maybeOptions); + } + } + + override async notification(notification: Notification, options?: NotificationOptions): Promise { + try { + return await super.notification(notification, options); + } catch (error) { + // `notifications/initialized` is part of the recovery handshake + // itself — recovering on it would await the in-flight recovery. + if (notification.method === 'notifications/initialized' || !this._isSessionTerminated(error)) { + throw error; + } + await this._recoverSession(error); + return await super.notification(notification, options); + } + } + + private async _recoverSession(cause: unknown): Promise { + this._sessionReinit ??= this._reinitializeSession().finally(() => { + this._sessionReinit = undefined; + }); + try { + await this._sessionReinit; + } catch (reinitError) { + // Re-establishment failed (the client is closed at this + // point): surface the request's own 404; the handshake + // failure goes to onerror. + this.onerror?.(reinitError instanceof Error ? reinitError : new Error(String(reinitError))); + throw cause; + } + } + + private _isSessionTerminated(error: unknown): boolean { + return ( + error instanceof SdkHttpError && + error.data.status === 404 && + error.data['sessionTerminated'] === true && + // Legacy era only: the 2026-07-28 handshake is discover-based. + this._negotiatedProtocolVersion !== undefined && + legacyProtocolVersions([this._negotiatedProtocolVersion]).length === 1 + ); + } + + private async _reinitializeSession(): Promise { + const transport = this.transport; + if (transport === undefined) { + throw new SdkError(SdkErrorCode.ConnectionClosed, 'Cannot reinitialize: not connected'); + } + // No caller options: the handshake is shared connection state, so no + // single request's signal/timeout/onprogress may govern it. + await this._legacyHandshake(transport); + } + /** * Connects to a server via the given transport and performs the MCP initialization handshake. * diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index 0d8eff917b..68f16a48db 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -5,6 +5,7 @@ import { createFetchWithInit, encodeMcpParamValue, isInitializedNotification, + isInitializeRequest, isJSONRPCErrorResponse, isJSONRPCRequest, isJSONRPCResultResponse, @@ -315,6 +316,7 @@ export class StreamableHTTPClientTransport implements Transport { private _fetch?: FetchLike; private _fetchWithInit: FetchLike; private _sessionId?: string; + private _terminatedSessionId?: string; private _reconnectionOptions: StreamableHTTPReconnectionOptions; private _protocolVersion?: string; private _onInsufficientScope: 'reauthorize' | 'throw'; @@ -322,6 +324,7 @@ export class StreamableHTTPClientTransport implements Transport { private _serverRetryMs?: number; // Server-provided retry delay from SSE retry field private readonly _reconnectionScheduler?: ReconnectionScheduler; private _cancelReconnection?: () => void; + private _pendingReconnectOptions?: StartSSEOptions; onclose?: () => void; onerror?: (error: Error) => void; @@ -652,6 +655,7 @@ export class StreamableHTTPClientTransport implements Transport { const reconnect = (): void => { this._cancelReconnection = undefined; + this._pendingReconnectOptions = undefined; // Honour BOTH the transport-wide abort and the per-request abort // (a listen subscription closed during the backoff delay): do not // resurrect a stream the caller already tore down. @@ -674,6 +678,7 @@ export class StreamableHTTPClientTransport implements Transport { const handle = setTimeout(reconnect, delay); this._cancelReconnection = () => clearTimeout(handle); } + this._pendingReconnectOptions = options; } private _handleSseStream(stream: ReadableStream | null, options: StartSSEOptions, isReconnectable: boolean): void { @@ -971,15 +976,46 @@ export class StreamableHTTPClientTransport implements Transport { signal }; + // 404s are judged against the id this request carried — late + // replies from an older session race `_sessionId`. + const sentSessionId = headers.get('mcp-session-id'); + const response = await (this._fetch ?? fetch)(this._url, init); - // Handle session ID received during initialization - const sessionId = response.headers.get('mcp-session-id'); - if (sessionId) { - this._sessionId = sessionId; + // Spec: the session id is assigned "at initialization time ... on the HTTP + // response containing the InitializeResult" — non-handshake echoes are ignored. + if (response.ok && (Array.isArray(message) ? message.some(m => isInitializeRequest(m)) : isInitializeRequest(message))) { + const sessionId = response.headers.get('mcp-session-id'); + if (sessionId) { + this._sessionId = sessionId; + this._terminatedSessionId = undefined; + } } if (!response.ok) { + // 404 to the current session id: session is gone (spec) — + // drop it. POSTs only; SSE reconnect GETs 404 transiently. + // `_terminatedSessionId` marks raced in-flight requests of the + // same dead session until a new one is assigned; the stale + // version header is dropped so the recovery initialize goes + // out like a fresh connect. + let sessionTerminated = false; + if (response.status === 404 && sentSessionId !== null && sentSessionId === this._sessionId) { + this._terminatedSessionId = sentSessionId; + this._sessionId = undefined; + this._protocolVersion = undefined; + // A reconnect pending for the dead session's SSE stream + // would fire into the new session with a stale + // last-event-id — drop it, and settle any per-request + // caller waiting on that stream (it is definitively gone). + this._cancelReconnection?.(); + this._cancelReconnection = undefined; + this._pendingReconnectOptions?.onRequestStreamEnd?.(); + this._pendingReconnectOptions = undefined; + sessionTerminated = true; + } else if (response.status === 404 && sentSessionId !== null && sentSessionId === this._terminatedSessionId) { + sessionTerminated = true; + } if (response.status === 401 && this._authProvider) { // Store WWW-Authenticate params for interactive finishAuth() path if (response.headers.has('www-authenticate')) { @@ -1060,7 +1096,8 @@ export class StreamableHTTPClientTransport implements Transport { throw new SdkHttpError(SdkErrorCode.ClientHttpNotImplemented, `Error POSTing to endpoint: ${text}`, { status: response.status, statusText: response.statusText, - text + text, + ...(sessionTerminated ? { sessionTerminated: true } : {}) }); } @@ -1145,8 +1182,9 @@ export class StreamableHTTPClientTransport implements Transport { * HTTP `DELETE` to the MCP endpoint with the `Mcp-Session-Id` header to explicitly * terminate the session. * - * The server MAY respond with HTTP `405 Method Not Allowed`, indicating that - * the server does not allow clients to terminate sessions. + * A `405 Method Not Allowed` (server does not allow client termination) + * or `404 Not Found` (session already expired) response also resolves; + * the stored session id is cleared in every resolving case. */ async terminateSession(): Promise { if (!this._sessionId) { @@ -1166,9 +1204,9 @@ export class StreamableHTTPClientTransport implements Transport { const response = await (this._fetch ?? fetch)(this._url, init); await response.text?.().catch(() => {}); - // We specifically handle 405 as a valid response according to the spec, - // meaning the server does not support explicit session termination - if (!response.ok && response.status !== 405) { + // 405: server does not support explicit termination (spec). + // 404: session already gone — the goal state; still drop the id. + if (!response.ok && response.status !== 405 && response.status !== 404) { throw new SdkHttpError( SdkErrorCode.ClientHttpFailedToTerminateSession, `Failed to terminate session: ${response.statusText}`, diff --git a/packages/client/test/client/sessionReinitialize.test.ts b/packages/client/test/client/sessionReinitialize.test.ts new file mode 100644 index 0000000000..fe99aedfc9 --- /dev/null +++ b/packages/client/test/client/sessionReinitialize.test.ts @@ -0,0 +1,178 @@ +import type { JSONRPCMessage } from '@modelcontextprotocol/core-internal'; +import { SdkErrorCode, SdkHttpError } from '@modelcontextprotocol/core-internal'; +import { describe, expect, it } from 'vitest'; + +import { Client } from '../../src/client/client'; +import type { Transport } from '../../src/index'; + +const INIT_RESULT = { + protocolVersion: '2025-11-25', + capabilities: { tools: {} }, + serverInfo: { name: 's', version: '1' } +}; + +function terminated404(): SdkHttpError { + return new SdkHttpError(SdkErrorCode.ClientHttpNotImplemented, 'Error POSTing to endpoint: session not found', { + status: 404, + statusText: 'Not Found', + sessionTerminated: true + }); +} + +class ScriptedTransport implements Transport { + onmessage?: (message: JSONRPCMessage) => void; + onerror?: (error: Error) => void; + onclose?: () => void; + sent: JSONRPCMessage[] = []; + constructor(private readonly script: (message: JSONRPCMessage, t: ScriptedTransport) => void) {} + async start(): Promise {} + async close(): Promise { + this.onclose?.(); + } + async send(message: JSONRPCMessage): Promise { + this.sent.push(message); + this.script(message, this); + } + reply(message: JSONRPCMessage): void { + queueMicrotask(() => this.onmessage?.(message)); + } +} + +function isRequest(m: JSONRPCMessage): m is JSONRPCMessage & { id: number | string; method: string } { + return 'method' in m && 'id' in m; +} + +function echoScript(opts: { failNthToolCall?: number[]; failReinit?: boolean; failFirstRootsNotification?: boolean }) { + let toolCalls = 0; + let initCount = 0; + let rootsNotifications = 0; + return (message: JSONRPCMessage, t: ScriptedTransport) => { + if (!isRequest(message)) { + if ( + 'method' in message && + message.method === 'notifications/roots/list_changed' && + opts.failFirstRootsNotification && + ++rootsNotifications === 1 + ) { + throw terminated404(); + } + return; + } + if (message.method === 'initialize') { + initCount++; + if (opts.failReinit && initCount > 1) { + throw new SdkHttpError(SdkErrorCode.ClientHttpNotImplemented, 'unavailable', { status: 500 }); + } + t.reply({ jsonrpc: '2.0', id: message.id, result: INIT_RESULT }); + return; + } + if (message.method === 'tools/call') { + toolCalls++; + if (opts.failNthToolCall?.includes(toolCalls)) throw terminated404(); + t.reply({ jsonrpc: '2.0', id: message.id, result: { content: [{ type: 'text', text: 'ok' }] } }); + return; + } + if (message.method === 'tools/list') { + toolCalls++; + if (opts.failNthToolCall?.includes(toolCalls)) throw terminated404(); + t.reply({ jsonrpc: '2.0', id: message.id, result: { tools: [] } }); + } + }; +} + +describe('session re-initialization after a terminated-session 404', () => { + it('re-initializes and retries the failed request once', async () => { + const transport = new ScriptedTransport(echoScript({ failNthToolCall: [1] })); + const client = new Client({ name: 'c', version: '0' }, {}); + await client.connect(transport); + + const result = await client.callTool({ name: 'echo', arguments: {} }); + expect(result.content).toEqual([{ type: 'text', text: 'ok' }]); + + const methods = transport.sent.filter(isRequest).map(m => m.method); + expect(methods).toEqual(['initialize', 'tools/call', 'initialize', 'tools/call']); + await client.close(); + }); + + it('deduplicates concurrent recoveries into one handshake', async () => { + const transport = new ScriptedTransport(echoScript({ failNthToolCall: [1, 2] })); + const client = new Client({ name: 'c', version: '0' }, {}); + await client.connect(transport); + + const [a, b] = await Promise.all([ + client.callTool({ name: 'echo', arguments: {} }), + client.callTool({ name: 'echo', arguments: {} }) + ]); + expect(a.content).toEqual([{ type: 'text', text: 'ok' }]); + expect(b.content).toEqual([{ type: 'text', text: 'ok' }]); + + const inits = transport.sent.filter(isRequest).filter(m => m.method === 'initialize'); + expect(inits).toHaveLength(2); + await client.close(); + }); + + it('surfaces the original 404 when re-establishment fails, reporting the handshake error via onerror', async () => { + const transport = new ScriptedTransport(echoScript({ failNthToolCall: [1], failReinit: true })); + const client = new Client({ name: 'c', version: '0' }, {}); + const reported: Error[] = []; + client.onerror = e => void reported.push(e); + await client.connect(transport); + + const rejection = await client.callTool({ name: 'echo', arguments: {} }).catch((e: unknown) => e); + expect(rejection).toBeInstanceOf(SdkHttpError); + expect((rejection as SdkHttpError).data.status).toBe(404); + expect(reported.some(e => e instanceof SdkHttpError && e.data.status === 500)).toBe(true); + }); + + it('recovers the session on a pagination continuation but does not retry it', async () => { + const transport = new ScriptedTransport(echoScript({ failNthToolCall: [1] })); + const client = new Client({ name: 'c', version: '0' }, {}); + await client.connect(transport); + + const rejection = await client.request({ method: 'tools/list', params: { cursor: 'stale' } }).catch((e: unknown) => e); + expect(rejection).toBeInstanceOf(SdkHttpError); + expect((rejection as SdkHttpError).data.status).toBe(404); + // The session is re-established (so later requests work), but the + // dead session's cursor is not replayed into the new one. + const methods = transport.sent.filter(isRequest).map(m => m.method); + expect(methods).toEqual(['initialize', 'tools/list', 'initialize']); + + const result = await client.request({ method: 'tools/list' }); + expect(result.tools).toEqual([]); + await client.close(); + }); + + it('recovers and resends a notification that hit a terminated session', async () => { + const transport = new ScriptedTransport(echoScript({ failFirstRootsNotification: true })); + const client = new Client({ name: 'c', version: '0' }, { capabilities: { roots: { listChanged: true } } }); + await client.connect(transport); + + await client.sendRootsListChanged(); + + const methods = transport.sent.map(m => ('method' in m ? m.method : '')).filter(m => m !== ''); + expect(methods).toEqual([ + 'initialize', + 'notifications/initialized', + 'notifications/roots/list_changed', + 'initialize', + 'notifications/initialized', + 'notifications/roots/list_changed' + ]); + await client.close(); + }); + + it('runs the recovery handshake without the failing request options', async () => { + const transport = new ScriptedTransport(echoScript({ failNthToolCall: [1] })); + const client = new Client({ name: 'c', version: '0' }, {}); + await client.connect(transport); + + await client.callTool({ name: 'echo', arguments: {} }, { onprogress: () => {} }); + + const inits = transport.sent.filter(isRequest).filter(m => m.method === 'initialize'); + expect(inits).toHaveLength(2); + // The caller's onprogress must not decorate the shared recovery + // initialize with a progress token. + expect((inits[1] as unknown as { params: { _meta?: unknown } }).params._meta).toBeUndefined(); + await client.close(); + }); +}); diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index a20fb92252..41e29200f0 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -93,7 +93,8 @@ describe('StreamableHTTPClientTransport', () => { method: 'initialize', params: { clientInfo: { name: 'test-client', version: '1.0' }, - protocolVersion: '2025-03-26' + protocolVersion: '2025-03-26', + capabilities: {} }, id: 'init-id' }; @@ -122,6 +123,324 @@ describe('StreamableHTTPClientTransport', () => { expect(lastCall[1].headers.get('mcp-session-id')).toBe('test-session-id'); }); + it('should NOT store a session ID from an error response', async () => { + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: false, + status: 404, + statusText: 'Not Found', + headers: new Headers({ 'mcp-session-id': 'from-error-reply-123' }), + text: () => Promise.resolve('not found') + }); + + await expect(transport.send({ jsonrpc: '2.0', method: 'server/discover', id: 'probe-id' } as JSONRPCMessage)).rejects.toThrow(); + + expect(transport.sessionId).toBeUndefined(); + + // The next request must go out session-less. + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: true, + status: 202, + headers: new Headers() + }); + await transport.send({ jsonrpc: '2.0', method: 'test', params: {} } as JSONRPCMessage); + + const lastSend = (globalThis.fetch as Mock).mock.calls.at(-1)!; + expect(lastSend[1].headers.get('mcp-session-id')).toBeNull(); + }); + + it('should still store the session ID assigned by a successful response after an earlier error response', async () => { + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: false, + status: 404, + statusText: 'Not Found', + headers: new Headers({ 'mcp-session-id': 'from-error-reply-123' }), + text: () => Promise.resolve('not found') + }); + await expect(transport.send({ jsonrpc: '2.0', method: 'server/discover', id: 'probe-id' } as JSONRPCMessage)).rejects.toThrow(); + + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream', 'mcp-session-id': 'real-session' }) + }); + await transport.send({ + jsonrpc: '2.0', + method: 'initialize', + params: { clientInfo: { name: 'test-client', version: '1.0' }, protocolVersion: '2025-03-26', capabilities: {} }, + id: 'init-id' + } as JSONRPCMessage); + + expect(transport.sessionId).toBe('real-session'); + }); + + it('stores the session ID when the initialize request is part of a batch', async () => { + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream', 'mcp-session-id': 'batch-session' }) + }); + await transport.send([ + { + jsonrpc: '2.0', + method: 'initialize', + params: { clientInfo: { name: 'test-client', version: '1.0' }, protocolVersion: '2025-03-26', capabilities: {} }, + id: 'init-id' + }, + { jsonrpc: '2.0', method: 'other', id: 'other-id' } + ] as JSONRPCMessage[]); + expect(transport.sessionId).toBe('batch-session'); + }); + + it('ignores a session id on a successful non-initialize response', async () => { + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream', 'mcp-session-id': 'handshake-session' }) + }); + await transport.send({ + jsonrpc: '2.0', + method: 'initialize', + params: { clientInfo: { name: 'test-client', version: '1.0' }, protocolVersion: '2025-03-26', capabilities: {} }, + id: 'init-id' + } as JSONRPCMessage); + expect(transport.sessionId).toBe('handshake-session'); + + // Spec: the session id is assigned "at initialization time ... on the + // HTTP response containing the InitializeResult" — the spec is silent on + // the header elsewhere, so the client ignores it on any other response. + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: true, + status: 202, + headers: new Headers({ 'mcp-session-id': 'unrelated-session' }) + }); + await transport.send({ jsonrpc: '2.0', method: 'notifications/roots/list_changed' } as JSONRPCMessage); + expect(transport.sessionId).toBe('handshake-session'); + + // A same-id echo on a request's response is equally inert. + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: true, + status: 202, + headers: new Headers({ 'mcp-session-id': 'handshake-session' }) + }); + await transport.send({ jsonrpc: '2.0', method: 'test', id: 'r1' } as JSONRPCMessage); + expect(transport.sessionId).toBe('handshake-session'); + }); + + it('marks raced 404s of the same dead session until a new one is assigned', async () => { + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream', 'mcp-session-id': 'doomed' }) + }); + await transport.send({ + jsonrpc: '2.0', + method: 'initialize', + params: { clientInfo: { name: 'test-client', version: '1.0' }, protocolVersion: '2025-03-26', capabilities: {} }, + id: 'init-id' + } as JSONRPCMessage); + + const mock404 = () => + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: false, + status: 404, + statusText: 'Not Found', + headers: new Headers(), + text: () => Promise.resolve('gone') + }); + + // First 404: terminates the session — marked. + mock404(); + const first = await transport.send({ jsonrpc: '2.0', method: 'a', id: 1 } as JSONRPCMessage).catch((e: unknown) => e); + expect((first as SdkHttpError).data['sessionTerminated']).toBe(true); + expect(transport.sessionId).toBeUndefined(); + + // A raced sibling that also went out under 'doomed' — still marked. + mock404(); + // Simulate: the sibling's request carried the dead id. + (transport as unknown as { _sessionId?: string })._sessionId = 'doomed'; + const sibling = await transport.send({ jsonrpc: '2.0', method: 'b', id: 2 } as JSONRPCMessage).catch((e: unknown) => e); + expect((sibling as SdkHttpError).data['sessionTerminated']).toBe(true); + + // A request that went out under 'doomed', answered 404 only AFTER a + // fresh session was assigned: no longer marked. + (transport as unknown as { _sessionId?: string })._sessionId = 'doomed'; + let resolveLate: (r: unknown) => void; + (globalThis.fetch as Mock).mockImplementationOnce(() => new Promise(resolve => (resolveLate = resolve))); + const latePromise = transport.send({ jsonrpc: '2.0', method: 'c', id: 3 } as JSONRPCMessage).catch((e: unknown) => e); + // Rotation completes while the late request is in flight. + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream', 'mcp-session-id': 'fresh' }) + }); + (transport as unknown as { _sessionId?: string })._sessionId = undefined; + await transport.send({ + jsonrpc: '2.0', + method: 'initialize', + params: { clientInfo: { name: 'test-client', version: '1.0' }, protocolVersion: '2025-03-26', capabilities: {} }, + id: 'init-2' + } as JSONRPCMessage); + resolveLate!({ + ok: false, + status: 404, + statusText: 'Not Found', + headers: new Headers(), + text: () => Promise.resolve('gone') + }); + const late = await latePromise; + expect((late as SdkHttpError).data['sessionTerminated']).toBeUndefined(); + expect(transport.sessionId).toBe('fresh'); + }); + + it('should clear the session ID on a 404 to a request that carried it', async () => { + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream', 'mcp-session-id': 'expired-later' }) + }); + await transport.send({ + jsonrpc: '2.0', + method: 'initialize', + params: { clientInfo: { name: 'test-client', version: '1.0' }, protocolVersion: '2025-03-26', capabilities: {} }, + id: 'init-id' + } as JSONRPCMessage); + expect(transport.sessionId).toBe('expired-later'); + + // Server expired the session: 404 to a request that presented it. + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: false, + status: 404, + statusText: 'Not Found', + headers: new Headers(), + text: () => Promise.resolve('session not found') + }); + await expect(transport.send({ jsonrpc: '2.0', method: 'test', id: 'r1' } as JSONRPCMessage)).rejects.toThrow(); + expect(transport.sessionId).toBeUndefined(); + + // Next request goes out session-less (fresh handshake possible). + (globalThis.fetch as Mock).mockResolvedValueOnce({ ok: true, status: 202, headers: new Headers() }); + await transport.send({ jsonrpc: '2.0', method: 'notifications/roots/list_changed' } as JSONRPCMessage); + expect((globalThis.fetch as Mock).mock.calls.at(-1)![1].headers.get('mcp-session-id')).toBeNull(); + }); + + it('does not let a late 404 from an old session wipe a newer session id', async () => { + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream', 'mcp-session-id': 'first-session' }) + }); + await transport.send({ + jsonrpc: '2.0', + method: 'initialize', + params: { clientInfo: { name: 'test-client', version: '1.0' }, protocolVersion: '2025-03-26', capabilities: {} }, + id: 'init-id' + } as JSONRPCMessage); + expect(transport.sessionId).toBe('first-session'); + + // Request A goes out under first-session but its response is delayed. + let resolveLate!: (r: unknown) => void; + (globalThis.fetch as Mock).mockImplementationOnce(() => new Promise(r => (resolveLate = r))); + const late = transport.send({ jsonrpc: '2.0', method: 'test', id: 'late' } as JSONRPCMessage); + + // Meanwhile request B hits a 404 — first-session is dead — and a + // fresh handshake establishes second-session. + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: false, + status: 404, + statusText: 'Not Found', + headers: new Headers(), + text: () => Promise.resolve('session not found') + }); + await expect(transport.send({ jsonrpc: '2.0', method: 'test', id: 'b' } as JSONRPCMessage)).rejects.toThrow(); + expect(transport.sessionId).toBeUndefined(); + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream', 'mcp-session-id': 'second-session' }) + }); + await transport.send({ + jsonrpc: '2.0', + method: 'initialize', + params: { clientInfo: { name: 'test-client', version: '1.0' }, protocolVersion: '2025-03-26', capabilities: {} }, + id: 'init-2' + } as JSONRPCMessage); + expect(transport.sessionId).toBe('second-session'); + + // The late 404 was for first-session; second-session is still live. + resolveLate({ + ok: false, + status: 404, + statusText: 'Not Found', + headers: new Headers(), + text: () => Promise.resolve('session not found') + }); + await expect(late).rejects.toThrow(); + expect(transport.sessionId).toBe('second-session'); + }); + + it('cancels a pending SSE reconnection when a 404 terminates the session', async () => { + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream', 'mcp-session-id': 'doomed' }) + }); + await transport.send({ + jsonrpc: '2.0', + method: 'initialize', + params: { clientInfo: { name: 'test-client', version: '1.0' }, protocolVersion: '2025-03-26', capabilities: {} }, + id: 'init-id' + } as JSONRPCMessage); + + // A reconnect scheduled for the dead session's stream must not fire + // into the recovered session with a stale last-event-id. + const cancel = vi.fn(); + const onRequestStreamEnd = vi.fn(); + const internals = transport as unknown as { + _cancelReconnection?: () => void; + _pendingReconnectOptions?: { onRequestStreamEnd?: () => void }; + }; + internals._cancelReconnection = cancel; + internals._pendingReconnectOptions = { onRequestStreamEnd }; + + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: false, + status: 404, + statusText: 'Not Found', + headers: new Headers(), + text: () => Promise.resolve('session not found') + }); + await expect(transport.send({ jsonrpc: '2.0', method: 'test', id: 'r1' } as JSONRPCMessage)).rejects.toThrow(); + expect(cancel).toHaveBeenCalledTimes(1); + // A per-request caller waiting on that stream's resume settles. + expect(onRequestStreamEnd).toHaveBeenCalledTimes(1); + }); + + it('treats a 404 to the termination DELETE as successful termination and drops the id', async () => { + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream', 'mcp-session-id': 'gone-session' }) + }); + await transport.send({ + jsonrpc: '2.0', + method: 'initialize', + params: { clientInfo: { name: 'test-client', version: '1.0' }, protocolVersion: '2025-03-26', capabilities: {} }, + id: 'init-id' + } as JSONRPCMessage); + expect(transport.sessionId).toBe('gone-session'); + + // Session already gone server-side — termination goal state. + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: false, + status: 404, + statusText: 'Not Found', + headers: new Headers(), + text: () => Promise.resolve('session not found') + }); + await expect(transport.terminateSession()).resolves.toBeUndefined(); + expect(transport.sessionId).toBeUndefined(); + }); + it('should accept protocolVersion constructor option and include it in request headers', async () => { // When reconnecting with a preserved sessionId, users need to also preserve the // negotiated protocol version so the required mcp-protocol-version header is sent. @@ -156,7 +475,8 @@ describe('StreamableHTTPClientTransport', () => { method: 'initialize', params: { clientInfo: { name: 'test-client', version: '1.0' }, - protocolVersion: '2025-03-26' + protocolVersion: '2025-03-26', + capabilities: {} }, id: 'init-id' }; @@ -196,7 +516,8 @@ describe('StreamableHTTPClientTransport', () => { method: 'initialize', params: { clientInfo: { name: 'test-client', version: '1.0' }, - protocolVersion: '2025-03-26' + protocolVersion: '2025-03-26', + capabilities: {} }, id: 'init-id' }; diff --git a/test/e2e/requirements.ts b/test/e2e/requirements.ts index 4fda9e661f..1b42402a6a 100644 --- a/test/e2e/requirements.ts +++ b/test/e2e/requirements.ts @@ -1884,21 +1884,17 @@ export const REQUIREMENTS: Record = { 'client-transport:http:404-surfaces': { source: 'sdk', - behavior: 'A 404 (session expired) on a request surfaces as an error to the caller.', + behavior: + 'When a 404 (session expired) is followed by a failed session re-establishment, the failure surfaces as an error to the caller.', transports: ['streamableHttp'], - note: 'Session-id continuity testing requires the per-session host (404 is session-not-found).' + note: 'Session-id continuity testing requires the per-session host (404 is session-not-found). The recovered path is client-transport:http:session-404-reinitialize.' }, 'client-transport:http:session-404-reinitialize': { source: 'https://modelcontextprotocol.io/specification/2025-11-25/basic/transports#session-management', behavior: 'A 404 in response to a request carrying a session ID makes the client start a new session with a fresh InitializeRequest and no session ID attached.', transports: ['streamableHttp'], - note: 'This exercises the StreamableHTTP client transport directly; the matrix transport arg is ignored, so it runs as a single streamableHttp-labelled cell to avoid duplicate runs.', - knownFailures: [ - { - note: 'On a 404 for an existing session the transport throws StreamableHTTPError (streamableHttp.ts:551) and never re-initializes — no session recovery is attempted.' - } - ] + note: 'This exercises the StreamableHTTP client transport directly; the matrix transport arg is ignored, so it runs as a single streamableHttp-labelled cell to avoid duplicate runs.' }, 'client-transport:http:accept-header-get': { source: 'https://modelcontextprotocol.io/specification/2025-11-25/basic/transports#listening-for-messages-from-the-server', diff --git a/test/e2e/scenarios/transport-http.test.ts b/test/e2e/scenarios/transport-http.test.ts index 7cda0f28aa..8e9381b18a 100644 --- a/test/e2e/scenarios/transport-http.test.ts +++ b/test/e2e/scenarios/transport-http.test.ts @@ -363,6 +363,11 @@ verifies('client-transport:http:404-surfaces', async (_args: TestArgs) => { if (sid && sid === sessionIdToBreak) { return Response.json({ error: 'Session not found' }, { status: 404 }); } + // Once the session is broken, refuse re-establishment too: + // the automatic session-less InitializeRequest gets a 500. + if (sessionIdToBreak !== undefined && !sid && req.method === 'POST') { + return new Response('unavailable', { status: 500 }); + } return handle.handleRequest(req); } });