diff --git a/.changeset/steady-rivers-reinit.md b/.changeset/steady-rivers-reinit.md new file mode 100644 index 0000000000..76b3f5fc63 --- /dev/null +++ b/.changeset/steady-rivers-reinit.md @@ -0,0 +1,5 @@ +--- +"@modelcontextprotocol/client": patch +--- + +fix: reinitialize expired streamable sessions diff --git a/packages/client/src/client/client.ts b/packages/client/src/client/client.ts index 0b386a63e8..3079020210 100644 --- a/packages/client/src/client/client.ts +++ b/packages/client/src/client/client.ts @@ -347,6 +347,10 @@ export type ClientOptions = ProtocolOptions & { defaultCacheTtlMs?: number; }; +type SessionExpiringTransport = Transport & { + onsessionexpired?: () => void | Promise; +}; + /** * Options for {@linkcode Client.connect}. Extends {@linkcode RequestOptions} * (the timeout/signal apply to the connect-time handshake or probe) with the @@ -1000,6 +1004,13 @@ export class Client extends Protocol { */ private async _connectPlainLegacy(transport: Transport, options?: RequestOptions): Promise { await super.connect(transport); + (transport as SessionExpiringTransport).onsessionexpired = async () => { + this._serverCapabilities = undefined; + this._serverVersion = undefined; + this._negotiatedProtocolVersion = undefined; + await this._initialize(transport, options); + }; + // When transport sessionId is already set this means we are trying to reconnect. // Restore the protocol version negotiated during the original initialize handshake // so HTTP transports include the required mcp-protocol-version header, but skip re-init. @@ -1038,64 +1049,69 @@ export class Client extends Protocol { // revision is negotiated exclusively via server/discover. const legacyVersions = legacyProtocolVersions(this._supportedProtocolVersions); try { - const offeredVersion = legacyVersions[0]; - if (offeredVersion === undefined) { - throw new SdkError( - SdkErrorCode.EraNegotiationFailed, - 'Cannot run the initialize handshake: supportedProtocolVersions contains no pre-2026-07-28 protocol version' - ); - } - const result = await this.request( - { - method: 'initialize', - params: { - protocolVersion: offeredVersion, - capabilities: this._capabilities, - clientInfo: this._clientInfo - } - }, - options + await this._initialize(transport, options); + } catch (error) { + // Disconnect if initialization fails. + void this.close(); + throw error; + } + } + + private async _initialize(transport: Transport, options?: RequestOptions): Promise { + const legacyVersions = legacyProtocolVersions(this._supportedProtocolVersions); + const offeredVersion = legacyVersions[0]; + if (offeredVersion === undefined) { + throw new SdkError( + SdkErrorCode.EraNegotiationFailed, + 'Cannot run the initialize handshake: supportedProtocolVersions contains no pre-2026-07-28 protocol version' ); + } + const result = await this.request( + { + method: 'initialize', + params: { + protocolVersion: offeredVersion, + capabilities: this._capabilities, + clientInfo: this._clientInfo + } + }, + options + ); - if (result === undefined) { - throw new Error(`Server sent invalid initialize result: ${result}`); - } + if (result === undefined) { + throw new Error(`Server sent invalid initialize result: ${result}`); + } - if (!legacyVersions.includes(result.protocolVersion)) { - throw new Error(`Server's protocol version is not supported: ${result.protocolVersion}`); - } + if (!legacyVersions.includes(result.protocolVersion)) { + throw new Error(`Server's protocol version is not supported: ${result.protocolVersion}`); + } - this._serverCapabilities = result.capabilities; - this._serverVersion = result.serverInfo; - this._cache.setServerIdentity(this._deriveServerIdentity(transport)); - // HTTP transports must set the protocol version in each header after initialization. - if (transport.setProtocolVersion) { - transport.setProtocolVersion(result.protocolVersion); - } + this._serverCapabilities = result.capabilities; + this._serverVersion = result.serverInfo; + this._cache.setServerIdentity(this._deriveServerIdentity(transport)); + // HTTP transports must set the protocol version in each header after initialization. + if (transport.setProtocolVersion) { + transport.setProtocolVersion(result.protocolVersion); + } - this._instructions = result.instructions; + this._instructions = result.instructions; - await this.notification({ - method: 'notifications/initialized' - }); + await this.notification({ + method: 'notifications/initialized' + }); - // Handshake completion: the negotiated version becomes the - // instance's connection state, and with it the wire era for - // everything this connection sends/receives from here on (the - // negotiated version cashes out as the negotiated wire ERA — - // Q1-SD1). Set AFTER the initialized notification: the initialize - // EXCHANGE is the legacy handshake by definition and completes on - // that era. - this._negotiatedProtocolVersion = result.protocolVersion; - - // Set up list changed handlers now that we know server capabilities - if (this._listChangedConfig) { - this._setupListChangedHandlers(this._listChangedConfig); - } - } catch (error) { - // Disconnect if initialization fails. - void this.close(); - throw error; + // Handshake completion: the negotiated version becomes the + // instance's connection state, and with it the wire era for + // everything this connection sends/receives from here on (the + // negotiated version cashes out as the negotiated wire ERA — + // Q1-SD1). Set AFTER the initialized notification: the initialize + // EXCHANGE is the legacy handshake by definition and completes on + // that era. + this._negotiatedProtocolVersion = result.protocolVersion; + + // Set up list changed handlers now that we know server capabilities + if (this._listChangedConfig) { + this._setupListChangedHandlers(this._listChangedConfig); } } diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index 9067fd1ec4..73c3939a92 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -327,6 +327,7 @@ export class StreamableHTTPClientTransport implements Transport { onclose?: () => void; onerror?: (error: Error) => void; onmessage?: (message: JSONRPCMessage) => void; + onsessionexpired?: () => void | Promise; /** * Streamable HTTP opens one POST (and SSE response stream) per outbound @@ -901,7 +902,7 @@ export class StreamableHTTPClientTransport implements Transport { headers?: Readonly>; } ): Promise { - return this._send(message, options, false); + return this._send(message, options, false, false); } private async _send( @@ -916,6 +917,7 @@ export class StreamableHTTPClientTransport implements Transport { } | undefined, isAuthRetry: boolean, + isSessionRetry = false, stepUpRetries = 0 ): Promise { try { @@ -1004,7 +1006,7 @@ export class StreamableHTTPClientTransport implements Transport { }); await response.text?.().catch(() => {}); // Purposely _not_ awaited, so we don't call onerror twice - return this._send(message, options, true, stepUpRetries); + return this._send(message, options, true, isSessionRetry, stepUpRetries); } await response.text?.().catch(() => {}); if (isAuthRetry) { @@ -1018,6 +1020,15 @@ export class StreamableHTTPClientTransport implements Transport { const text = await response.text?.().catch(() => null); + if (response.status === 404 && this._sessionId && !isSessionRetry) { + this._sessionId = undefined; + await this.onsessionexpired?.(); + + if (this._sessionId) { + return this._send(message, options, isAuthRetry, true, stepUpRetries); + } + } + if (response.status === 403) { const { resourceMetadataUrl, scope, error, errorDescription } = extractWWWAuthenticateParams(response); @@ -1029,7 +1040,7 @@ export class StreamableHTTPClientTransport implements Transport { if (result !== 'AUTHORIZED') { throw new UnauthorizedError(); } - return this._send(message, options, isAuthRetry, stepUpRetries + 1); + return this._send(message, options, isAuthRetry, isSessionRetry, stepUpRetries + 1); } } diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index a36bbc0ad3..d6c2ffc1fd 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -1,9 +1,10 @@ import type { JSONRPCMessage, JSONRPCRequest, OAuthTokens } from '@modelcontextprotocol/core-internal'; -import { OAuthError, OAuthErrorCode, SdkErrorCode, SdkHttpError } from '@modelcontextprotocol/core-internal'; +import { LATEST_PROTOCOL_VERSION, OAuthError, OAuthErrorCode, SdkErrorCode, SdkHttpError } from '@modelcontextprotocol/core-internal'; import type { Mock, Mocked } from 'vitest'; import type { OAuthClientProvider } from '../../src/client/auth'; import { UnauthorizedError } from '../../src/client/auth'; +import { Client } from '../../src/client/client'; import type { ReconnectionScheduler, StartSSEOptions, StreamableHTTPReconnectionOptions } from '../../src/client/streamableHttp'; import { StreamableHTTPClientTransport } from '../../src/client/streamableHttp'; @@ -369,6 +370,96 @@ describe('StreamableHTTPClientTransport', () => { expect(errorSpy).toHaveBeenCalled(); }); + it('reinitializes and retries once when a persisted session expires', async () => { + const client = new Client({ name: 'test-client', version: '1.0.0' }); + const httpTransport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp')); + let initializeCount = 0; + let firstPing = true; + + (globalThis.fetch as Mock).mockImplementation(async (_url, init) => { + if (init.method === 'GET') { + return { + ok: false, + status: 405, + statusText: 'Method Not Allowed', + headers: new Headers(), + text: async () => '' + }; + } + + const body = JSON.parse(init.body as string) as JSONRPCRequest; + + if (body.method === 'initialize') { + const sessionId = initializeCount++ === 0 ? 'old-session-id' : 'new-session-id'; + + return { + ok: true, + status: 200, + headers: new Headers({ + 'content-type': 'application/json', + 'mcp-session-id': sessionId + }), + json: async () => ({ + jsonrpc: '2.0', + id: body.id, + result: { + protocolVersion: LATEST_PROTOCOL_VERSION, + capabilities: {}, + serverInfo: { name: 'test-server', version: '1.0.0' } + } + }) + }; + } + + if (body.method === 'notifications/initialized') { + return { + ok: true, + status: 202, + headers: new Headers(), + text: async () => '' + }; + } + + if (body.method === 'ping' && firstPing) { + firstPing = false; + + return { + ok: false, + status: 404, + statusText: 'Not Found', + headers: new Headers(), + text: async () => 'Session not found' + }; + } + + return { + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'application/json' }), + json: async () => ({ + jsonrpc: '2.0', + id: body.id, + result: {} + }) + }; + }); + + try { + await client.connect(httpTransport); + await expect(client.ping()).resolves.toEqual({}); + + const calls = (globalThis.fetch as Mock).mock.calls; + const postCalls = calls.filter(([, init]) => init.method === 'POST'); + expect(postCalls).toHaveLength(6); + expect(postCalls[2]![1].headers.get('mcp-session-id')).toBe('old-session-id'); + expect(postCalls[3]![1].headers.get('mcp-session-id')).toBeNull(); + expect(postCalls[5]![1].headers.get('mcp-session-id')).toBe('new-session-id'); + expect(httpTransport.sessionId).toBe('new-session-id'); + } finally { + await client.close().catch(() => {}); + } + }); + it('should handle non-streaming JSON response', async () => { const message: JSONRPCMessage = { jsonrpc: '2.0', diff --git a/test/e2e/requirements.ts b/test/e2e/requirements.ts index c81783d509..26a2bde53c 100644 --- a/test/e2e/requirements.ts +++ b/test/e2e/requirements.ts @@ -1901,7 +1901,8 @@ 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: + 'A 404 (session expired) on a request surfaces as an error to the caller when the transport has no session recovery hook.', transports: ['streamableHttp'], note: 'Session-id continuity testing requires the per-session host (404 is session-not-found).' }, @@ -1910,12 +1911,7 @@ export const REQUIREMENTS: Record = { 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..c2c9e39bc1 100644 --- a/test/e2e/scenarios/transport-http.test.ts +++ b/test/e2e/scenarios/transport-http.test.ts @@ -370,6 +370,7 @@ verifies('client-transport:http:404-surfaces', async (_args: TestArgs) => { await client.connect(transport); sessionIdToBreak = transport.sessionId; + transport.onsessionexpired = undefined; const call = client.ping(); await expect(call).rejects.toThrow();