From de8a6cc97f07e06abec17e2cc5e551930e6169c3 Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Mon, 29 Jun 2026 11:24:34 +0000 Subject: [PATCH 1/9] =?UTF-8?q?fix(server):=20restore=20v1=20transport=20l?= =?UTF-8?q?ifecycle=20parity=20=E2=80=94=20single-use=20stateless=20transp?= =?UTF-8?q?orts,=20fail-fast=20on=20double=20connect?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .changeset/stateless-transport-reuse-guard.md | 11 + docs/migration/upgrade-to-v2.md | 37 +++ docs/serving/sessions-state-scaling.md | 2 +- examples/README.md | 2 + packages/client/src/client/client.ts | 11 + packages/client/test/client/listen.test.ts | 31 ++- packages/core-internal/src/shared/protocol.ts | 58 ++++- .../test/shared/protocolConnectGuard.test.ts | 188 +++++++++++++++ .../shared/protocolTransportHandling.test.ts | 122 +++++----- packages/middleware/express/README.md | 8 +- packages/middleware/fastify/README.md | 13 +- packages/middleware/hono/README.md | 13 +- packages/middleware/node/README.md | 9 +- .../node/src/streamableHttp.examples.ts | 23 +- .../middleware/node/src/streamableHttp.ts | 20 +- .../node/test/streamableHttp.test.ts | 98 ++++++-- .../src/server/streamableHttp.examples.ts | 24 +- packages/server/src/server/streamableHttp.ts | 29 +++ .../server/classificationCarrierPin.test.ts | 14 +- .../test/server/statelessReuseGuard.test.ts | 225 ++++++++++++++++++ .../server/test/server/streamableHttp.test.ts | 33 ++- test/e2e/requirements.ts | 7 +- test/integration/test/client/client.test.ts | 7 + .../test/client/discoverRoundtrip.test.ts | 75 +++--- .../test/client/versionNegotiation.test.ts | 58 ++++- test/integration/test/server/bun.test.ts | 23 +- .../test/server/cloudflareWorkers.test.ts | 30 ++- test/integration/test/server/deno.test.ts | 49 ++-- .../stateManagementStreamableHttp.test.ts | 59 +++-- 29 files changed, 1050 insertions(+), 229 deletions(-) create mode 100644 .changeset/stateless-transport-reuse-guard.md create mode 100644 packages/core-internal/test/shared/protocolConnectGuard.test.ts create mode 100644 packages/server/test/server/statelessReuseGuard.test.ts diff --git a/.changeset/stateless-transport-reuse-guard.md b/.changeset/stateless-transport-reuse-guard.md new file mode 100644 index 0000000000..7b0bc4e1b6 --- /dev/null +++ b/.changeset/stateless-transport-reuse-guard.md @@ -0,0 +1,11 @@ +--- +'@modelcontextprotocol/server': patch +'@modelcontextprotocol/core-internal': patch +--- + +Restores two v1 transport lifecycle invariants dropped in the v2 rewrite: + +- A stateless `WebStandardStreamableHTTPServerTransport` (`sessionIdGenerator: undefined`) handles a single request exchange and throws on reuse (matching v1 behavior since 1.26.0): `"Stateless transport cannot be reused across requests. Create a new transport per request."` Construct a fresh transport (and server instance) per request, or use `createMcpHandler`, which already serves per-request pairs. Stateful transports (`sessionIdGenerator` set) are unaffected; `NodeStreamableHTTPServerTransport` inherits the behavior. +- `Protocol.connect()` (and `Client.connect()`) throw when the instance is already connected, instead of silently rebinding the transport: `"Already connected to a transport. Call close() before connecting to a new transport, or use a separate Protocol instance per connection."` Sequential `close()` then `connect()` keeps working. After `close()` aborts an in-flight request handler, `ctx.mcpReq.notify()` resolves as a no-op and `ctx.mcpReq.send()` rejects with `SdkError(ConnectionClosed)`, consistent with the above. + +Docs, middleware READMEs, and examples that showed a shared stateless transport now show the per-request pattern. diff --git a/docs/migration/upgrade-to-v2.md b/docs/migration/upgrade-to-v2.md index 89188aa893..dd18f69f03 100644 --- a/docs/migration/upgrade-to-v2.md +++ b/docs/migration/upgrade-to-v2.md @@ -1616,6 +1616,43 @@ rewrite required unless noted. unchanged from v1. This `-32001` is an SDK convention, not spec-assigned; client code should key off the HTTP `404` status, not `-32001`. +#### Transport / connection lifecycle (v1 parity) + +Two v1 transport lifecycle invariants (in effect since sdk@1.26.0) apply in v2: + +- **A stateless `WebStandardStreamableHTTPServerTransport` serves exactly one request.** + With `sessionIdGenerator: undefined`, the second `handleRequest()` call throws + `"Stateless transport cannot be reused across requests. Create a new transport per + request."` — matching v1 behavior since 1.26.0. Stateful transports + (`sessionIdGenerator` set) accept multiple requests as before; + `NodeStreamableHTTPServerTransport` inherits the behavior. Migrate shared-transport + hosting to the per-request pattern: + + ```typescript + app.all('/mcp', async c => { + // Fresh transport + server pair per request. + const server = new McpServer({ name: 'my-server', version: '1.0.0' }); + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); + await server.connect(transport); + return transport.handleRequest(c.req.raw); + }); + ``` + + Or use `createMcpHandler`, which constructs the per-request pair for you (its + `legacy: 'stateless'` fallback and the modern per-request transport are both + one-exchange by construction). + +- **`Protocol.connect()` throws when already connected** (`"Already connected to a + transport. Call close() before connecting to a new transport, or use a separate + Protocol instance per connection."`) instead of silently rebinding `this._transport`. + Sequential `close()` then `connect()` keeps working. `Client.connect()` performs the + same check up front, before any per-connection state is reset. + +- **Aborted request handlers cannot write to a replacement transport.** After the + handler's `ctx.mcpReq.signal` aborts (connection closed or request cancelled), + `ctx.mcpReq.notify()` resolves as a no-op and `ctx.mcpReq.send()` rejects with + `SdkError(ConnectionClosed)`. + #### Server (deprecated accessors and app-factory Origin validation) - `Server.getClientCapabilities()`, `getClientVersion()`, `getNegotiatedProtocolVersion()` diff --git a/docs/serving/sessions-state-scaling.md b/docs/serving/sessions-state-scaling.md index 0bbffe450a..0f51680a08 100644 --- a/docs/serving/sessions-state-scaling.md +++ b/docs/serving/sessions-state-scaling.md @@ -7,7 +7,7 @@ shape: how-to ## Pin a client to a session -A **session** pins a client to one long-lived transport instance; sessions belong to the hand-wired 2025-era transport — the 2026-07-28 revision is per-request and has no `Mcp-Session-Id` ([Protocol versions](../protocol-versions.md)). On `NodeStreamableHTTPServerTransport`, `sessionIdGenerator` turns sessions on; leaving it `undefined` is stateless mode. +A **session** pins a client to one long-lived transport instance; sessions belong to the hand-wired 2025-era transport — the 2026-07-28 revision is per-request and has no `Mcp-Session-Id` ([Protocol versions](../protocol-versions.md)). On `NodeStreamableHTTPServerTransport`, `sessionIdGenerator` turns sessions on; leaving it `undefined` is stateless mode. A stateless transport serves exactly one request exchange; construct a fresh transport + server pair per request — reuse throws. For stateless HTTP hosting prefer `createMcpHandler`, which builds the per-request pair for you. ```ts source="../../examples/guides/serving/sessions-state-scaling.examples.ts#sessions_stateful" import { NodeStreamableHTTPServerTransport } from '@modelcontextprotocol/node'; diff --git a/examples/README.md b/examples/README.md index fc57f2de7d..8e97fd4362 100644 --- a/examples/README.md +++ b/examples/README.md @@ -92,6 +92,8 @@ To enable stateless mode, configure the `NodeStreamableHTTPServerTransport` with sessionIdGenerator: undefined; ``` +A stateless transport serves exactly one request: construct a fresh transport + server pair per request (reuse throws). `createMcpHandler` does this for you — see [`stateless-legacy/`](./stateless-legacy/README.md). + ``` ┌─────────────────────────────────────────────┐ │ Client │ diff --git a/packages/client/src/client/client.ts b/packages/client/src/client/client.ts index 5028e937cd..44a12cdf4a 100644 --- a/packages/client/src/client/client.ts +++ b/packages/client/src/client/client.ts @@ -927,6 +927,17 @@ export class Client extends Protocol { * ``` */ override async connect(transport: Transport, options?: ConnectOptions): Promise { + // Fail fast while still connected: the negotiated paths below reset + // per-connection state and run the discover probe BEFORE the base + // Protocol.connect() reuse guard would fire — without this early + // check, a connect() on a connected instance would corrupt the live + // connection's state before throwing. Same guard + message as + // Protocol.connect(). + if (this.transport) { + throw new Error( + 'Already connected to a transport. Call close() before connecting to a new transport, or use a separate Protocol instance per connection.' + ); + } if (options?.prior !== undefined) { // Zero-round-trip reconnect from a previously-obtained // DiscoverResult: bypasses versionNegotiation resolution entirely. diff --git a/packages/client/test/client/listen.test.ts b/packages/client/test/client/listen.test.ts index 408fe1ea39..88d5a68f68 100644 --- a/packages/client/test/client/listen.test.ts +++ b/packages/client/test/client/listen.test.ts @@ -837,26 +837,41 @@ describe('Client.listen()', () => { await client.close(); }); - it('a fresh connect without an intervening close settles in-flight listen() from the prior connection', async () => { - // Edge: prior transport never fires onclose; consumer calls connect() - // again. The in-flight listen() promise from the old connection must - // reject with a clear "client reconnected/closed" error rather than - // hang on the (now-discarded) ack timer. + it('connect() without an intervening close throws; close() settles in-flight listen() from the prior connection', async () => { + // Edge: prior transport never spontaneously fires onclose; consumer + // calls connect() again. The transport-reuse guard rejects the second + // connect outright WITHOUT touching the live connection's state (the + // in-flight listen stays pending), and the close() now required in + // between settles the in-flight listen() promise with a clear + // "client reconnected/closed" error rather than leaving it hanging on + // the (now-discarded) ack timer. const { clientTx } = await scriptedModernNoAck(); const client = new Client({ name: 'c', version: '1' }, { versionNegotiation: { mode: 'auto' } }); await client.connect(clientTx); const pending = client.listen({ toolsListChanged: true }); await flush(); expect((client as unknown as { _listenState: Map })._listenState.size).toBe(1); - // Fresh connect on a new transport — _resetConnectionState runs. + // Fresh connect on a new transport without close() — rejected by the + // reuse guard, and the prior connection's listen state is untouched. const { clientTx: clientTx2 } = await scriptedModern(); - await client.connect(clientTx2); + await expect(client.connect(clientTx2)).rejects.toThrow( + 'Already connected to a transport. Call close() before connecting to a new transport, or use a separate Protocol instance per connection.' + ); + expect((client as unknown as { _listenState: Map })._listenState.size).toBe(1); + // close() settles the in-flight listen even though the prior server + // never acked or closed. + await client.close(); const error = await pending.catch(e => e as Error); expect(error).toBeInstanceOf(SdkError); expect((error as SdkError).code).toBe(SdkErrorCode.ConnectionClosed); - expect((error as SdkError).message).toContain('reconnected or closed'); + // Settled via the transport onclose path ('Connection closed') or the + // reset path ('reconnected or closed') — both are clean + // ConnectionClosed settlements, never a hang. + expect((error as SdkError).message).toMatch(/Connection closed|reconnected or closed/); // No leaked per-listen state from the old connection. expect((client as unknown as { _listenState: Map })._listenState.size).toBe(0); + // The instance is reusable after close(): a fresh connect succeeds. + await client.connect(clientTx2); await client.close(); }); diff --git a/packages/core-internal/src/shared/protocol.ts b/packages/core-internal/src/shared/protocol.ts index ffab6e8df8..5a1243cbce 100644 --- a/packages/core-internal/src/shared/protocol.ts +++ b/packages/core-internal/src/shared/protocol.ts @@ -780,6 +780,12 @@ export abstract class Protocol { * The caller assumes ownership of the {@linkcode Transport}, replacing any callbacks that have already been set, and expects that it is the only user of the {@linkcode Transport} instance going forward. */ async connect(transport: Transport): Promise { + if (this._transport) { + throw new Error( + 'Already connected to a transport. Call close() before connecting to a new transport, or use a separate Protocol instance per connection.' + ); + } + this._transport = transport; const _onclose = this.transport?.onclose; this._transport.onclose = () => { @@ -813,7 +819,16 @@ export abstract class Protocol { // Pass supported protocol versions to transport for header validation transport.setSupportedProtocolVersions?.(this._supportedProtocolVersions); - await this._transport.start(); + try { + await this._transport.start(); + } catch (error) { + // A transport that failed to start was never connected: unwind so + // the instance can retry connect() — the reuse guard above would + // otherwise lock the instance onto a transport that never carried + // traffic. + this._transport = undefined; + throw error; + } } /** @@ -1024,22 +1039,41 @@ export abstract class Protocol { return; } + const abortController = new AbortController(); + this._requestHandlerAbortControllers.set(request.id, abortController); + // Related sends resolve through the SAME instance era as every other // sender (the per-request/instance asymmetry is deliberately gone): // the codec is resolved at send time from the connection state. - const sendNotification = (notification: Notification, options?: NotificationOptions) => - this._notificationViaCodec(this._resolveOutboundCodec(notification.method), notification, { + // + // Once this request's handler has been aborted (connection closed or + // request cancelled), its related sends must not reach the transport: + // `this._transport` is read live at send time, so a handler still + // running after close() could otherwise write onto a transport + // connected later. Notifications no-op; requests reject with + // ConnectionClosed. + const sendNotification = (notification: Notification, options?: NotificationOptions): Promise => { + if (abortController.signal.aborted) { + return Promise.resolve(); + } + return this._notificationViaCodec(this._resolveOutboundCodec(notification.method), notification, { ...options, relatedRequestId: request.id }); - const sendRequest = (r: Request, resultSchema: U, options?: RequestOptions) => - this._requestWithSchemaViaCodec(this._resolveOutboundCodec(r.method), r, resultSchema, { + }; + const sendRequest = ( + r: Request, + resultSchema: U, + options?: RequestOptions + ): Promise> => { + if (abortController.signal.aborted) { + return Promise.reject(new SdkError(SdkErrorCode.ConnectionClosed, 'Request was cancelled')); + } + return this._requestWithSchemaViaCodec(this._resolveOutboundCodec(r.method), r, resultSchema, { ...options, relatedRequestId: request.id }); - - const abortController = new AbortController(); - this._requestHandlerAbortControllers.set(request.id, abortController); + }; // Multi-round-trip retry material: only BARE response objects are // surfaced to the handler; entries that look like a wrapped @@ -1071,6 +1105,14 @@ export abstract class Protocol { // that overloaded property type. The cast is sound: this impl dispatches both overload paths via the // isStandardSchema guard, and sendRequest validates the result against the resolved schema either way. send: ((r: Request, schemaOrOptions?: StandardSchemaV1 | RequestOptions, maybeOptions?: RequestOptions) => { + // The abort guard comes FIRST — before era/codec + // resolution — so an aborted handler always gets the + // documented ConnectionClosed rejection, never a + // synchronous era throw evaluated against whatever + // connection replaced this request's. + if (abortController.signal.aborted) { + return Promise.reject(new SdkError(SdkErrorCode.ConnectionClosed, 'Request was cancelled')); + } // Related requests resolve through the instance era at // send time, exactly like direct sends: era-gate first, // then method-keyed schema resolution. diff --git a/packages/core-internal/test/shared/protocolConnectGuard.test.ts b/packages/core-internal/test/shared/protocolConnectGuard.test.ts new file mode 100644 index 0000000000..a5752149b0 --- /dev/null +++ b/packages/core-internal/test/shared/protocolConnectGuard.test.ts @@ -0,0 +1,188 @@ +/** + * Protocol.connect() single-transport invariant + post-close handler + * teardown (restores v1 behavior, in effect since sdk@1.26.0). + * + * connect() on an already-connected Protocol instance throws instead of + * silently overwriting `this._transport` (the silent overwrite routed + * in-flight handler output to whichever transport connected last). After + * close() the instance can connect again — sequential reuse stays supported. + * + * The companion guards cover the window AFTER close() aborts an in-flight + * handler: its `ctx.mcpReq.notify` / `ctx.mcpReq.send` must not write to a + * replacement transport connected afterwards (notify no-ops; send rejects + * with ConnectionClosed). The abort itself is covered by + * 'should abort in-flight request handlers when the connection is closed' + * in protocol.test.ts. + */ +import { describe, expect, test } from 'vitest'; +import * as z from 'zod/v4'; + +import { SdkError, SdkErrorCode } from '../../src/errors/sdkErrors'; +import type { BaseContext } from '../../src/shared/protocol'; +import { Protocol } from '../../src/shared/protocol'; +import type { Transport } from '../../src/shared/transport'; +import type { JSONRPCMessage } from '../../src/types/index'; + +class MockTransport implements Transport { + id: string; + onclose?: () => void; + onerror?: (error: Error) => void; + onmessage?: (message: unknown) => void; + sentMessages: JSONRPCMessage[] = []; + + constructor(id: string) { + this.id = id; + } + + async start(): Promise {} + + async close(): Promise { + this.onclose?.(); + } + + async send(message: JSONRPCMessage): Promise { + this.sentMessages.push(message); + } +} + +function createProtocol(): Protocol { + return new (class extends Protocol { + protected assertCapabilityForMethod(): void {} + protected assertNotificationCapability(): void {} + protected assertRequestHandlerCapability(): void {} + protected buildContext(ctx: BaseContext): BaseContext { + return ctx; + } + })(); +} + +describe('Protocol.connect() reuse guard', () => { + test('connect() while already connected throws', async () => { + const protocol = createProtocol(); + const transportA = new MockTransport('A'); + const transportB = new MockTransport('B'); + + await protocol.connect(transportA); + + await expect(protocol.connect(transportB)).rejects.toThrow( + 'Already connected to a transport. Call close() before connecting to a new transport, or use a separate Protocol instance per connection.' + ); + + // The original connection is untouched by the rejected connect. + expect(protocol.transport).toBe(transportA); + }); + + test('close() then connect() succeeds (sequential reuse)', async () => { + const protocol = createProtocol(); + const transportA = new MockTransport('A'); + const transportB = new MockTransport('B'); + + protocol.setRequestHandler('ping', async () => ({})); + + await protocol.connect(transportA); + await protocol.close(); + expect(protocol.transport).toBeUndefined(); + + await protocol.connect(transportB); + expect(protocol.transport).toBe(transportB); + + // The new connection serves requests normally. + transportB.onmessage?.({ jsonrpc: '2.0', method: 'ping', id: 1 }); + await new Promise(resolve => setTimeout(resolve, 10)); + expect(transportB.sentMessages).toHaveLength(1); + expect(transportB.sentMessages[0]).toMatchObject({ jsonrpc: '2.0', id: 1, result: {} }); + }); + + test('transport-initiated close also frees the instance for a new connect()', async () => { + const protocol = createProtocol(); + const transportA = new MockTransport('A'); + const transportB = new MockTransport('B'); + + await protocol.connect(transportA); + await transportA.close(); + + await protocol.connect(transportB); + expect(protocol.transport).toBe(transportB); + }); + + test('a failed transport start() leaves the instance reconnectable', async () => { + // A transport whose start() rejects was never connected — connect() + // unwinds, so a retry does not hit the reuse guard. + const protocol = createProtocol(); + const failing: Transport = { + async start() { + throw new Error('spawn failed'); + }, + async close() {}, + async send() {} + }; + + await expect(protocol.connect(failing)).rejects.toThrow('spawn failed'); + expect(protocol.transport).toBeUndefined(); + + const transportB = new MockTransport('B'); + await protocol.connect(transportB); + expect(protocol.transport).toBe(transportB); + }); +}); + +describe('aborted-handler sends after close()', () => { + test('an aborted handler cannot write to a replacement transport: notify no-ops, send rejects with ConnectionClosed', async () => { + const protocol = createProtocol(); + const transportA = new MockTransport('A'); + const transportB = new MockTransport('B'); + + let ctxRef: BaseContext | undefined; + let handlerEnteredResolve!: () => void; + const handlerEntered = new Promise(resolve => { + handlerEnteredResolve = resolve; + }); + let releaseHandler!: () => void; + const handlerGate = new Promise(resolve => { + releaseHandler = resolve; + }); + + protocol.setRequestHandler('ping', async (_request, ctx) => { + ctxRef = ctx; + handlerEnteredResolve(); + await handlerGate; + return {}; + }); + + await protocol.connect(transportA); + transportA.onmessage?.({ jsonrpc: '2.0', method: 'ping', id: 1 }); + await handlerEntered; + + // close() aborts the in-flight handler and clears the transport... + await protocol.close(); + expect(ctxRef!.mcpReq.signal.aborted).toBe(true); + + // ...then the instance is reconnected to a REPLACEMENT transport. + await protocol.connect(transportB); + + // The aborted handler's related notification is a no-op (v1 shape): + // nothing reaches the newly connected transport. + await expect( + ctxRef!.mcpReq.notify({ + method: 'notifications/progress', + params: { progressToken: 1, progress: 1 } + }) + ).resolves.toBeUndefined(); + expect(transportB.sentMessages).toHaveLength(0); + + // The aborted handler's related request rejects with ConnectionClosed + // (v1 shape) instead of going out on the newly connected transport. + await expect(ctxRef!.mcpReq.send({ method: 'custom/probe' }, z.object({}))).rejects.toSatisfy( + error => error instanceof SdkError && error.code === SdkErrorCode.ConnectionClosed + ); + expect(transportB.sentMessages).toHaveLength(0); + + // Releasing the handler must not deliver its (stale) result anywhere: + // the result path is keyed to the captured transport and the aborted + // signal drops it. + releaseHandler(); + await new Promise(resolve => setTimeout(resolve, 10)); + expect(transportA.sentMessages).toHaveLength(0); + expect(transportB.sentMessages).toHaveLength(0); + }); +}); diff --git a/packages/core-internal/test/shared/protocolTransportHandling.test.ts b/packages/core-internal/test/shared/protocolTransportHandling.test.ts index 731914265b..5fe3613172 100644 --- a/packages/core-internal/test/shared/protocolTransportHandling.test.ts +++ b/packages/core-internal/test/shared/protocolTransportHandling.test.ts @@ -3,7 +3,7 @@ import { beforeEach, describe, expect, test } from 'vitest'; import type { BaseContext } from '../../src/shared/protocol'; import { Protocol } from '../../src/shared/protocol'; import type { Transport } from '../../src/shared/transport'; -import type { EmptyResult, JSONRPCMessage, Notification, Request, Result } from '../../src/types/index'; +import type { EmptyResult, JSONRPCMessage } from '../../src/types/index'; // Mock Transport class class MockTransport implements Transport { @@ -28,96 +28,112 @@ class MockTransport implements Transport { } } -describe('Protocol transport handling bug', () => { - let protocol: Protocol; +function createProtocol(): Protocol { + return new (class extends Protocol { + protected assertCapabilityForMethod(): void {} + protected assertNotificationCapability(): void {} + protected assertRequestHandlerCapability(): void {} + protected buildContext(ctx: BaseContext): BaseContext { + return ctx; + } + })(); +} + +describe('Protocol transport handling', () => { let transportA: MockTransport; let transportB: MockTransport; beforeEach(() => { - protocol = new (class extends Protocol { - protected assertCapabilityForMethod(): void {} - protected assertNotificationCapability(): void {} - protected assertRequestHandlerCapability(): void {} - protected buildContext(ctx: BaseContext): BaseContext { - return ctx; - } - })(); - transportA = new MockTransport('A'); transportB = new MockTransport('B'); }); - test('should send response to the correct transport when multiple clients are connected', async () => { - // Set up a request handler that simulates processing time - let resolveHandler: (value: EmptyResult) => void; - const handlerPromise = new Promise(resolve => { - resolveHandler = resolve; + test('should send response to the correct transport when using separate protocol instances', async () => { + const protocolA = createProtocol(); + const protocolB = createProtocol(); + + // Each protocol gets its own resolver so we can verify responses route correctly + let resolveA: (value: EmptyResult) => void; + let resolveB: (value: EmptyResult) => void; + let handlerAEnteredResolve: () => void; + let handlerBEnteredResolve: () => void; + const handlerAEntered = new Promise(resolve => { + handlerAEnteredResolve = resolve; + }); + const handlerBEntered = new Promise(resolve => { + handlerBEnteredResolve = resolve; + }); + + protocolA.setRequestHandler('ping', async () => { + return new Promise(resolve => { + resolveA = resolve; + handlerAEnteredResolve(); + }); }); - protocol.setRequestHandler('ping', async () => handlerPromise); + protocolB.setRequestHandler('ping', async () => { + return new Promise(resolve => { + resolveB = resolve; + handlerBEnteredResolve(); + }); + }); // Client A connects and sends a request - await protocol.connect(transportA); + await protocolA.connect(transportA); transportA.onmessage?.({ jsonrpc: '2.0', method: 'ping', id: 1 }); - // While A's request is being processed, client B connects - // This overwrites the transport reference in the protocol - await protocol.connect(transportB); + // Client B connects to a separate protocol instance + await protocolB.connect(transportB); transportB.onmessage?.({ jsonrpc: '2.0', method: 'ping', id: 2 }); - // Now complete A's request - resolveHandler!({}); + // Wait for both handlers to be invoked so resolvers are captured + await handlerAEntered; + await handlerBEntered; - // Wait for async operations to complete - await new Promise(resolve => setTimeout(resolve, 10)); + // Resolve each handler + resolveA!({}); + resolveB!({}); - // Check where the responses went - console.log('Transport A received:', transportA.sentMessages); - console.log('Transport B received:', transportB.sentMessages); + // Wait for response delivery (transport.send is async) + await new Promise(resolve => setTimeout(resolve, 10)); - // Transport A should receive response for request ID 1 + // Each transport receives its own response expect(transportA.sentMessages).toHaveLength(1); expect(transportA.sentMessages[0]).toMatchObject({ jsonrpc: '2.0', id: 1, result: {} }); - // Transport B should receive response for request ID 2 expect(transportB.sentMessages).toHaveLength(1); expect(transportB.sentMessages[0]).toMatchObject({ jsonrpc: '2.0', id: 2, result: {} }); }); - test('demonstrates the timing issue with multiple rapid connections', async () => { - const results: { transport: string; response: JSONRPCMessage[] }[] = []; - - // Set up handler with variable delay based on request id - protocol.setRequestHandler('ping', async (_request, ctx) => { - const delay = ctx.mcpReq.id === 1 ? 50 : 10; - await new Promise(resolve => setTimeout(resolve, delay)); - return {}; - }); + test('demonstrates isolation with separate protocol instances for rapid connections', async () => { + const protocolA = createProtocol(); + const protocolB = createProtocol(); + + // Set up handler with variable delay based on request id on each protocol + for (const protocol of [protocolA, protocolB]) { + protocol.setRequestHandler('ping', async (_request, ctx) => { + const delay = ctx.mcpReq.id === 1 ? 50 : 10; + await new Promise(resolve => setTimeout(resolve, delay)); + return {}; + }); + } - // Rapid succession of connections and requests - await protocol.connect(transportA); + // Connect and send requests + await protocolA.connect(transportA); transportA.onmessage?.({ jsonrpc: '2.0', method: 'ping', id: 1 }); // Connect B while A is processing setTimeout(async () => { - await protocol.connect(transportB); + await protocolB.connect(transportB); transportB.onmessage?.({ jsonrpc: '2.0', method: 'ping', id: 2 }); }, 10); // Wait for all processing await new Promise(resolve => setTimeout(resolve, 100)); - // Collect results - if (transportA.sentMessages.length > 0) { - results.push({ transport: 'A', response: transportA.sentMessages }); - } - if (transportB.sentMessages.length > 0) { - results.push({ transport: 'B', response: transportB.sentMessages }); - } - - console.log('Timing test results:', results); - expect(transportA.sentMessages).toHaveLength(1); + expect(transportA.sentMessages[0]).toMatchObject({ jsonrpc: '2.0', id: 1, result: {} }); expect(transportB.sentMessages).toHaveLength(1); + expect(transportB.sentMessages[0]).toMatchObject({ jsonrpc: '2.0', id: 2, result: {} }); }); }); diff --git a/packages/middleware/express/README.md b/packages/middleware/express/README.md index b837f5b623..82a5883e59 100644 --- a/packages/middleware/express/README.md +++ b/packages/middleware/express/README.md @@ -48,11 +48,13 @@ import { NodeStreamableHTTPServerTransport } from '@modelcontextprotocol/node'; import { McpServer } from '@modelcontextprotocol/server'; const app = createMcpExpressApp(); -const server = new McpServer({ name: 'my-server', version: '1.0.0' }); app.post('/mcp', async (req, res) => { - // Stateless example: create a transport per request. - // For stateful mode (sessions), keep a transport instance around and reuse it. + // Stateless example: create a fresh transport + server pair per request. + // A stateless transport serves exactly one request (reuse throws), and a + // connected server must be close()d before it can connect a new transport. + // For stateful mode (sessions), keep a transport + server pair around and reuse it. + const server = new McpServer({ name: 'my-server', version: '1.0.0' }); const transport = new NodeStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); await server.connect(transport); await transport.handleRequest(req, res, req.body); diff --git a/packages/middleware/fastify/README.md b/packages/middleware/fastify/README.md index 2355170002..350014b21f 100644 --- a/packages/middleware/fastify/README.md +++ b/packages/middleware/fastify/README.md @@ -42,24 +42,27 @@ import { NodeStreamableHTTPServerTransport } from '@modelcontextprotocol/node'; import { McpServer } from '@modelcontextprotocol/server'; const app = createMcpFastifyApp(); -const mcpServer = new McpServer({ name: 'my-server', version: '1.0.0' }); app.post('/mcp', async (request, reply) => { - // Stateless example: create a transport per request. - // For stateful mode (sessions), keep a transport instance around and reuse it. + // Stateless example: create a fresh transport + server pair per request. + // A stateless transport serves exactly one request (reuse throws), and a + // connected server must be close()d before it can connect a new transport. + // For stateful mode (sessions), keep a transport + server pair around and reuse it. + const mcpServer = new McpServer({ name: 'my-server', version: '1.0.0' }); const transport = new NodeStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); await mcpServer.connect(transport); // Clean up when the client closes the connection (e.g. during SSE streaming). reply.raw.on('close', () => { - transport.close(); + void mcpServer.close().catch(() => {}); + void transport.close().catch(() => {}); }); await transport.handleRequest(request.raw, reply.raw, request.body); }); ``` -If you create a new `McpServer` per request in stateless mode, also call `mcpServer.close()` in the `close` handler. To reject non-POST requests with 405 Method Not Allowed, add routes for GET and DELETE that send a JSON-RPC error response. +To reject non-POST requests with 405 Method Not Allowed, add routes for GET and DELETE that send a JSON-RPC error response. ### Host header validation (DNS rebinding protection) diff --git a/packages/middleware/hono/README.md b/packages/middleware/hono/README.md index f591888506..deac2943c5 100644 --- a/packages/middleware/hono/README.md +++ b/packages/middleware/hono/README.md @@ -30,12 +30,15 @@ npm install @modelcontextprotocol/server @modelcontextprotocol/hono hono import { McpServer, WebStandardStreamableHTTPServerTransport } from '@modelcontextprotocol/server'; import { createMcpHonoApp } from '@modelcontextprotocol/hono'; -const server = new McpServer({ name: 'my-server', version: '1.0.0' }); -const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); -await server.connect(transport); - const app = createMcpHonoApp(); -app.all('/mcp', c => transport.handleRequest(c.req.raw, { parsedBody: c.get('parsedBody') })); +app.all('/mcp', async c => { + // Stateless serving: a fresh transport + server pair per request. + // A stateless transport serves exactly one request (reuse throws). + const server = new McpServer({ name: 'my-server', version: '1.0.0' }); + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); + await server.connect(transport); + return transport.handleRequest(c.req.raw, { parsedBody: c.get('parsedBody') }); +}); ``` ### Host header validation (DNS rebinding protection) diff --git a/packages/middleware/node/README.md b/packages/middleware/node/README.md index bc279cae3a..9aa80a25e2 100644 --- a/packages/middleware/node/README.md +++ b/packages/middleware/node/README.md @@ -33,10 +33,13 @@ import { createMcpExpressApp } from '@modelcontextprotocol/express'; import { NodeStreamableHTTPServerTransport } from '@modelcontextprotocol/node'; import { McpServer } from '@modelcontextprotocol/server'; -const server = new McpServer({ name: 'my-server', version: '1.0.0' }); const app = createMcpExpressApp(); app.post('/mcp', async (req, res) => { + // Stateless serving: a fresh transport + server pair per request. + // A stateless transport serves exactly one request (reuse throws), and a + // connected server must be close()d before it can connect a new transport. + const server = new McpServer({ name: 'my-server', version: '1.0.0' }); const transport = new NodeStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); await server.connect(transport); @@ -59,13 +62,13 @@ import { createServer } from 'node:http'; import { localhostHostValidation, localhostOriginValidation, NodeStreamableHTTPServerTransport } from '@modelcontextprotocol/node'; import { McpServer } from '@modelcontextprotocol/server'; -const server = new McpServer({ name: 'my-server', version: '1.0.0' }); - const validateHost = localhostHostValidation(); const validateOrigin = localhostOriginValidation(); createServer(async (req, res) => { if (!validateHost(req, res) || !validateOrigin(req, res)) return; + // Stateless serving: a fresh transport + server pair per request. + const server = new McpServer({ name: 'my-server', version: '1.0.0' }); const transport = new NodeStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); await server.connect(transport); await transport.handleRequest(req, res); diff --git a/packages/middleware/node/src/streamableHttp.examples.ts b/packages/middleware/node/src/streamableHttp.examples.ts index 4ac8778c92..0329fdd957 100644 --- a/packages/middleware/node/src/streamableHttp.examples.ts +++ b/packages/middleware/node/src/streamableHttp.examples.ts @@ -29,16 +29,27 @@ async function NodeStreamableHTTPServerTransport_stateful() { //#endregion NodeStreamableHTTPServerTransport_stateful } +// Stubs for Node.js request handling examples +declare const incomingRequest: IncomingMessage; +declare const serverResponse: ServerResponse; + /** * Example: Stateless Streamable HTTP transport (Node.js). */ async function NodeStreamableHTTPServerTransport_stateless() { //#region NodeStreamableHTTPServerTransport_stateless + // A stateless transport serves exactly one request: construct a fresh + // transport + server pair per request — reusing a stateless transport + // across requests throws. + const server = new McpServer({ name: 'my-server', version: '1.0.0' }); + const transport = new NodeStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); + + await server.connect(transport); + await transport.handleRequest(incomingRequest, serverResponse); //#endregion NodeStreamableHTTPServerTransport_stateless - return transport; } // Stubs for Express-style app @@ -47,10 +58,14 @@ declare const app: { post(path: string, handler: (req: IncomingMessage & { body? /** * Example: Using with a pre-parsed request body (e.g. Express). */ -function NodeStreamableHTTPServerTransport_express(transport: NodeStreamableHTTPServerTransport) { +function NodeStreamableHTTPServerTransport_express() { //#region NodeStreamableHTTPServerTransport_express - app.post('/mcp', (req, res) => { - transport.handleRequest(req, res, req.body); + app.post('/mcp', async (req, res) => { + // Stateless serving: a fresh transport + server pair per request. + const server = new McpServer({ name: 'my-server', version: '1.0.0' }); + const transport = new NodeStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); + await server.connect(transport); + await transport.handleRequest(req, res, req.body); }); //#endregion NodeStreamableHTTPServerTransport_express } diff --git a/packages/middleware/node/src/streamableHttp.ts b/packages/middleware/node/src/streamableHttp.ts index a6f1c43a6b..602c55d4a2 100644 --- a/packages/middleware/node/src/streamableHttp.ts +++ b/packages/middleware/node/src/streamableHttp.ts @@ -44,6 +44,10 @@ export type StreamableHTTPServerTransportOptions = WebStandardStreamableHTTPServ * In stateless mode: * - No Session ID is included in any responses * - No session validation is performed + * - Each transport instance serves exactly ONE request: construct a fresh + * transport (and server instance) per request — reusing a stateless + * transport across requests throws (the guard lives in the wrapped + * {@linkcode WebStandardStreamableHTTPServerTransport}) * * @example Stateful setup * ```ts source="./streamableHttp.examples.ts#NodeStreamableHTTPServerTransport_stateful" @@ -58,15 +62,27 @@ export type StreamableHTTPServerTransportOptions = WebStandardStreamableHTTPServ * * @example Stateless setup * ```ts source="./streamableHttp.examples.ts#NodeStreamableHTTPServerTransport_stateless" + * // A stateless transport serves exactly one request: construct a fresh + * // transport + server pair per request — reusing a stateless transport + * // across requests throws. + * const server = new McpServer({ name: 'my-server', version: '1.0.0' }); + * * const transport = new NodeStreamableHTTPServerTransport({ * sessionIdGenerator: undefined * }); + * + * await server.connect(transport); + * await transport.handleRequest(incomingRequest, serverResponse); * ``` * * @example Using with a pre-parsed request body (e.g. Express) * ```ts source="./streamableHttp.examples.ts#NodeStreamableHTTPServerTransport_express" - * app.post('/mcp', (req, res) => { - * transport.handleRequest(req, res, req.body); + * app.post('/mcp', async (req, res) => { + * // Stateless serving: a fresh transport + server pair per request. + * const server = new McpServer({ name: 'my-server', version: '1.0.0' }); + * const transport = new NodeStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); + * await server.connect(transport); + * await transport.handleRequest(req, res, req.body); * }); * ``` */ diff --git a/packages/middleware/node/test/streamableHttp.test.ts b/packages/middleware/node/test/streamableHttp.test.ts index 140717c6bb..79f5e72ea2 100644 --- a/packages/middleware/node/test/streamableHttp.test.ts +++ b/packages/middleware/node/test/streamableHttp.test.ts @@ -1599,20 +1599,64 @@ describe('Zod v4', () => { // Test stateless mode describe('NodeStreamableHTTPServerTransport in stateless mode', () => { let server: Server; - let transport: NodeStreamableHTTPServerTransport; let baseUrl: URL; + const perRequestServers: McpServer[] = []; + // In stateless mode, each request must use a fresh transport + server pair. + // The HTTP server creates these per-request and delegates accordingly. beforeEach(async () => { - const result = await createTestServer({ sessionIdGenerator: undefined }); - server = result.server; - transport = result.transport; - baseUrl = result.baseUrl; + server = createServer(async (req, res) => { + try { + const { transport, mcpServer } = await createStatelessHandler(); + perRequestServers.push(mcpServer); + await transport.handleRequest(req, res); + } catch (error) { + console.error('Error handling request:', error); + if (!res.headersSent) res.writeHead(500).end(); + } + }); + baseUrl = await listenOnRandomPort(server); }); afterEach(async () => { - await stopTestServer({ server, transport }); + while (perRequestServers.length > 0) { + await perRequestServers + .pop()! + .close() + .catch(() => {}); + } + server.close(); }); + /** + * Creates a fresh transport + mcpServer pair for a single stateless request. + */ + async function createStatelessHandler(): Promise<{ + transport: NodeStreamableHTTPServerTransport; + mcpServer: McpServer; + }> { + const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }, { capabilities: { logging: {} } }); + + mcpServer.registerTool( + 'greet', + { + description: 'A simple greeting tool', + inputSchema: z.object({ name: z.string().describe('Name to greet') }) + }, + async ({ name }): Promise => { + return { content: [{ type: 'text', text: `Hello, ${name}!` }] }; + } + ); + + const transport = new NodeStreamableHTTPServerTransport({ + sessionIdGenerator: undefined + }); + + await mcpServer.connect(transport); + + return { transport, mcpServer }; + } + it('should operate without session ID validation', async () => { // Initialize the server first const initResponse = await sendPostRequest(baseUrl, TEST_MESSAGES.initialize); @@ -1622,6 +1666,7 @@ describe('Zod v4', () => { expect(initResponse.headers.get('mcp-session-id')).toBeNull(); // Try request without session ID - should work in stateless mode + // (a fresh transport is created per request) const toolsResponse = await sendPostRequest(baseUrl, TEST_MESSAGES.toolsList); expect(toolsResponse.status).toBe(200); @@ -1655,14 +1700,14 @@ describe('Zod v4', () => { expect(response2.status).toBe(200); }); - it('should reject second SSE stream even in stateless mode', async () => { - // Despite no session ID requirement, the transport still only allows - // one standalone SSE stream at a time + it('should allow multiple SSE streams in stateless mode with per-request transports', async () => { + // Each request gets its own transport, so multiple SSE streams can + // coexist since they are handled by separate transport instances // Initialize the server first await sendPostRequest(baseUrl, TEST_MESSAGES.initialize); - // Open first SSE stream + // Open first SSE stream - this uses its own per-request transport const stream1 = await fetch(baseUrl, { method: 'GET', headers: { @@ -1672,7 +1717,8 @@ describe('Zod v4', () => { }); expect(stream1.status).toBe(200); - // Open second SSE stream - should still be rejected, stateless mode still only allows one + // Open second SSE stream - also gets its own per-request transport, + // so it should also succeed (each transport only handles one request) const stream2 = await fetch(baseUrl, { method: 'GET', headers: { @@ -1680,7 +1726,7 @@ describe('Zod v4', () => { 'mcp-protocol-version': '2025-11-25' } }); - expect(stream2.status).toBe(409); // Conflict - only one stream allowed + expect(stream2.status).toBe(200); }); }); @@ -2941,17 +2987,20 @@ describe('Zod v4', () => { describe('Combined validations', () => { it('should validate both host and origin when both are configured', async () => { - const result = await createTestServerWithDnsProtection({ + // In stateless mode, each request needs a fresh transport, so we + // test invalid and valid origins with separate server instances. + + // Test with invalid origin (host will be automatically correct via fetch) + const result1 = await createTestServerWithDnsProtection({ sessionIdGenerator: undefined, allowedHosts: ['localhost'], allowedOrigins: ['http://localhost:3001'], enableDnsRebindingProtection: true }); - server = result.server; - transport = result.transport; - baseUrl = result.baseUrl; + server = result1.server; + transport = result1.transport; + baseUrl = result1.baseUrl; - // Test with invalid origin (host will be automatically correct via fetch) const response1 = await fetch(baseUrl, { method: 'POST', headers: { @@ -2966,7 +3015,20 @@ describe('Zod v4', () => { const body1 = (await response1.json()) as JSONRPCErrorResponse; expect(body1.error.message).toBe('Invalid Origin header: http://evil.com'); - // Test with valid origin + // Clean up first server + await stopTestServer({ server, transport }); + + // Test with valid origin using a fresh server+transport + const result2 = await createTestServerWithDnsProtection({ + sessionIdGenerator: undefined, + allowedHosts: ['localhost'], + allowedOrigins: ['http://localhost:3001'], + enableDnsRebindingProtection: true + }); + server = result2.server; + transport = result2.transport; + baseUrl = result2.baseUrl; + const response2 = await fetch(baseUrl, { method: 'POST', headers: { diff --git a/packages/server/src/server/streamableHttp.examples.ts b/packages/server/src/server/streamableHttp.examples.ts index 678c43c928..9d3ebdd740 100644 --- a/packages/server/src/server/streamableHttp.examples.ts +++ b/packages/server/src/server/streamableHttp.examples.ts @@ -28,13 +28,21 @@ async function WebStandardStreamableHTTPServerTransport_stateful() { /** * Example: Stateless Streamable HTTP transport (Web Standard). */ -async function WebStandardStreamableHTTPServerTransport_stateless() { +async function WebStandardStreamableHTTPServerTransport_stateless(request: Request) { //#region WebStandardStreamableHTTPServerTransport_stateless + // A stateless transport serves exactly one request: construct a fresh + // transport + server pair per request — reusing a stateless transport + // across requests throws. + const server = new McpServer({ name: 'my-server', version: '1.0.0' }); + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); + + await server.connect(transport); + const response = await transport.handleRequest(request); //#endregion WebStandardStreamableHTTPServerTransport_stateless - return transport; + return response; } // Stubs for framework-specific examples @@ -43,9 +51,13 @@ declare const app: { all(path: string, handler: (c: { req: { raw: Request } }) = /** * Example: Using with Hono.js. */ -function WebStandardStreamableHTTPServerTransport_hono(transport: WebStandardStreamableHTTPServerTransport) { +function WebStandardStreamableHTTPServerTransport_hono() { //#region WebStandardStreamableHTTPServerTransport_hono app.all('/mcp', async c => { + // Stateless serving: a fresh transport + server pair per request. + const server = new McpServer({ name: 'my-server', version: '1.0.0' }); + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); + await server.connect(transport); return transport.handleRequest(c.req.raw); }); //#endregion WebStandardStreamableHTTPServerTransport_hono @@ -54,10 +66,14 @@ function WebStandardStreamableHTTPServerTransport_hono(transport: WebStandardStr /** * Example: Using with Cloudflare Workers. */ -function WebStandardStreamableHTTPServerTransport_workers(transport: WebStandardStreamableHTTPServerTransport) { +function WebStandardStreamableHTTPServerTransport_workers() { //#region WebStandardStreamableHTTPServerTransport_workers const worker = { async fetch(request: Request): Promise { + // Stateless serving: a fresh transport + server pair per request. + const server = new McpServer({ name: 'my-server', version: '1.0.0' }); + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); + await server.connect(transport); return transport.handleRequest(request); } }; diff --git a/packages/server/src/server/streamableHttp.ts b/packages/server/src/server/streamableHttp.ts index 7da5fb853c..aabce85fcd 100644 --- a/packages/server/src/server/streamableHttp.ts +++ b/packages/server/src/server/streamableHttp.ts @@ -193,6 +193,11 @@ export interface HandleRequestOptions { * In stateless mode: * - No Session ID is included in any responses * - No session validation is performed + * - Each transport instance serves exactly ONE request: construct a fresh + * transport (and server instance) per request — reusing a stateless + * transport across requests throws. Reuse would share the request-id → + * stream mapping between unrelated clients, so colliding JSON-RPC ids + * could deliver one client's response to another client. * * @example Stateful setup * ```ts source="./streamableHttp.examples.ts#WebStandardStreamableHTTPServerTransport_stateful" @@ -207,14 +212,26 @@ export interface HandleRequestOptions { * * @example Stateless setup * ```ts source="./streamableHttp.examples.ts#WebStandardStreamableHTTPServerTransport_stateless" + * // A stateless transport serves exactly one request: construct a fresh + * // transport + server pair per request — reusing a stateless transport + * // across requests throws. + * const server = new McpServer({ name: 'my-server', version: '1.0.0' }); + * * const transport = new WebStandardStreamableHTTPServerTransport({ * sessionIdGenerator: undefined * }); + * + * await server.connect(transport); + * const response = await transport.handleRequest(request); * ``` * * @example Hono.js * ```ts source="./streamableHttp.examples.ts#WebStandardStreamableHTTPServerTransport_hono" * app.all('/mcp', async c => { + * // Stateless serving: a fresh transport + server pair per request. + * const server = new McpServer({ name: 'my-server', version: '1.0.0' }); + * const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); + * await server.connect(transport); * return transport.handleRequest(c.req.raw); * }); * ``` @@ -223,6 +240,10 @@ export interface HandleRequestOptions { * ```ts source="./streamableHttp.examples.ts#WebStandardStreamableHTTPServerTransport_workers" * const worker = { * async fetch(request: Request): Promise { + * // Stateless serving: a fresh transport + server pair per request. + * const server = new McpServer({ name: 'my-server', version: '1.0.0' }); + * const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); + * await server.connect(transport); * return transport.handleRequest(request); * } * }; @@ -232,6 +253,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { // when sessionId is not set (undefined), it means the transport is in stateless mode private sessionIdGenerator: (() => string) | undefined; private _started: boolean = false; + private _hasHandledRequest: boolean = false; private _closed: boolean = false; private _streamMapping: Map = new Map(); private _requestToStreamMapping: Map = new Map(); @@ -352,6 +374,13 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { * Returns a `Response` object (Web Standard) */ async handleRequest(req: Request, options?: HandleRequestOptions): Promise { + // In stateless mode (no sessionIdGenerator), each request must use a fresh transport. + // Reusing a stateless transport causes message ID collisions between clients. + if (!this.sessionIdGenerator && this._hasHandledRequest) { + throw new Error('Stateless transport cannot be reused across requests. Create a new transport per request.'); + } + this._hasHandledRequest = true; + // Validate request headers for DNS rebinding protection const validationError = this.validateRequestHeaders(req); if (validationError) { diff --git a/packages/server/test/server/classificationCarrierPin.test.ts b/packages/server/test/server/classificationCarrierPin.test.ts index 4694f65abb..b53fd08137 100644 --- a/packages/server/test/server/classificationCarrierPin.test.ts +++ b/packages/server/test/server/classificationCarrierPin.test.ts @@ -73,18 +73,22 @@ const modernEnvelope = { describe('B-2: hand-wired legacy-transport traffic is never Protocol-classified', () => { it('extra.request is set and extra.classification stays unset for every delivered message', async () => { - const { server, seen, post } = await setupHandWired(); + // One hand-wired pair per request: a stateless transport cannot be + // reused across requests (per-request idiom). + const plainSetup = await setupHandWired(); + await plainSetup.post(toolsCall()); + await plainSetup.server.close(); - await post(toolsCall()); - await post(toolsCall(modernEnvelope)); + const stampedSetup = await setupHandWired(); + await stampedSetup.post(toolsCall(modernEnvelope)); + await stampedSetup.server.close(); + const seen = [...plainSetup.seen, ...stampedSetup.seen]; expect(seen.length).toBeGreaterThanOrEqual(2); for (const { extra } of seen) { expect(extra?.request).toBeInstanceOf(Request); expect(extra?.classification).toBeUndefined(); } - - await server.close(); }); it('a modern-stamped body through the legacy transport gets today’s exact legacy semantics, byte-identical', async () => { diff --git a/packages/server/test/server/statelessReuseGuard.test.ts b/packages/server/test/server/statelessReuseGuard.test.ts new file mode 100644 index 0000000000..fb12cc5791 --- /dev/null +++ b/packages/server/test/server/statelessReuseGuard.test.ts @@ -0,0 +1,225 @@ +/** + * Stateless transport single-use invariant (restores v1 behavior, in effect + * since sdk@1.26.0). + * + * A stateless `WebStandardStreamableHTTPServerTransport` (`sessionIdGenerator: + * undefined`) serves exactly one HTTP request exchange; any further + * `handleRequest()` call throws. Stateful transports (sessionIdGenerator set) + * accept multiple requests on the same session. + */ +import { randomUUID } from 'node:crypto'; + +import type { CallToolResult, JSONRPCMessage } from '@modelcontextprotocol/core-internal'; +import { afterEach, describe, expect, it } from 'vitest'; +import * as z from 'zod/v4'; + +import { McpServer } from '../../src/server/mcp'; +import { WebStandardStreamableHTTPServerTransport } from '../../src/server/streamableHttp'; + +const REUSE_ERROR = /Stateless transport cannot be reused across requests/; + +function postRequest(body: JSONRPCMessage, extraHeaders: Record = {}): Request { + return new Request('http://localhost/mcp', { + method: 'POST', + headers: { + accept: 'application/json, text/event-stream', + 'content-type': 'application/json', + 'mcp-protocol-version': '2025-11-25', + ...extraHeaders + }, + body: JSON.stringify(body) + }); +} + +function getRequest(): Request { + return new Request('http://localhost/mcp', { + method: 'GET', + headers: { accept: 'text/event-stream', 'mcp-protocol-version': '2025-11-25' } + }); +} + +function deleteRequest(): Request { + return new Request('http://localhost/mcp', { + method: 'DELETE', + headers: { 'mcp-protocol-version': '2025-11-25' } + }); +} + +const initializeMessage: JSONRPCMessage = { + jsonrpc: '2.0', + id: 0, + method: 'initialize', + params: { + protocolVersion: '2025-11-25', + capabilities: {}, + clientInfo: { name: 'reuse-guard-test', version: '1.0.0' } + } +}; + +const toolsListMessage: JSONRPCMessage = { jsonrpc: '2.0', id: 1, method: 'tools/list', params: {} }; + +describe('stateless transport reuse guard', () => { + const cleanups: Array<() => Promise> = []; + afterEach(async () => { + while (cleanups.length > 0) await cleanups.pop()!(); + }); + + async function setupStateless(options: { enableJsonResponse?: boolean; configure?: (mcpServer: McpServer) => void } = {}): Promise<{ + transport: WebStandardStreamableHTTPServerTransport; + mcpServer: McpServer; + }> { + const mcpServer = new McpServer({ name: 'reuse-guard-server', version: '1.0.0' }, { capabilities: { logging: {} } }); + options.configure?.(mcpServer); + const transport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: undefined, + ...(options.enableJsonResponse !== undefined && { enableJsonResponse: options.enableJsonResponse }) + }); + await mcpServer.connect(transport); + cleanups.push(async () => { + await mcpServer.close().catch(() => {}); + await transport.close().catch(() => {}); + }); + return { transport, mcpServer }; + } + + it('throws on a second POST', async () => { + const { transport } = await setupStateless(); + + const first = await transport.handleRequest(postRequest(initializeMessage)); + expect(first.status).toBe(200); + + await expect(transport.handleRequest(postRequest(toolsListMessage))).rejects.toThrow(REUSE_ERROR); + }); + + it('throws on a GET after a handled request', async () => { + const { transport } = await setupStateless(); + + const first = await transport.handleRequest(postRequest(initializeMessage)); + expect(first.status).toBe(200); + + await expect(transport.handleRequest(getRequest())).rejects.toThrow(REUSE_ERROR); + }); + + it('throws on a DELETE after a handled request', async () => { + const { transport } = await setupStateless(); + + const first = await transport.handleRequest(postRequest(initializeMessage)); + expect(first.status).toBe(200); + + await expect(transport.handleRequest(deleteRequest())).rejects.toThrow(REUSE_ERROR); + }); + + it('stateful transport (sessionIdGenerator set) still accepts multiple requests on the same session', async () => { + const mcpServer = new McpServer({ name: 'stateful-server', version: '1.0.0' }, { capabilities: { logging: {} } }); + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID() }); + await mcpServer.connect(transport); + cleanups.push(async () => { + await mcpServer.close().catch(() => {}); + await transport.close().catch(() => {}); + }); + + const initResponse = await transport.handleRequest(postRequest(initializeMessage)); + expect(initResponse.status).toBe(200); + const sessionId = initResponse.headers.get('mcp-session-id'); + expect(sessionId).toBeTruthy(); + + const second = await transport.handleRequest(postRequest(toolsListMessage, { 'mcp-session-id': sessionId! })); + expect(second.status).toBe(200); + + const third = await transport.handleRequest( + postRequest({ jsonrpc: '2.0', id: 2, method: 'tools/list', params: {} }, { 'mcp-session-id': sessionId! }) + ); + expect(third.status).toBe(200); + }); + + /** Registers a gated `echo` tool so a request can be held in flight. */ + function registerGatedEchoTool(mcpServer: McpServer): { echoStarted: Promise; releaseEcho: () => void } { + let releaseEcho!: () => void; + const echoGate = new Promise(resolve => { + releaseEcho = resolve; + }); + let echoEntered!: () => void; + const echoStarted = new Promise(resolve => { + echoEntered = resolve; + }); + mcpServer.registerTool( + 'echo', + { description: 'returns the caller-supplied marker once released', inputSchema: z.object({ marker: z.string() }) }, + async ({ marker }): Promise => { + echoEntered(); + await echoGate; + return { content: [{ type: 'text', text: `echo:${marker}` }] }; + } + ); + return { echoStarted, releaseEcho }; + } + + const callBody = (marker: string): JSONRPCMessage => ({ + jsonrpc: '2.0', + id: 7, // both exchanges use the same JSON-RPC id (client counters start identically) + method: 'tools/call', + params: { name: 'echo', arguments: { marker } } + }); + + it('a second POST during an in-flight stateless exchange throws; the first exchange completes with its own response (JSON shape)', async () => { + // The second handleRequest() call (same JSON-RPC id - client counters + // start identically) is rejected synchronously, before any of the + // first exchange's routing state is touched; the first exchange then + // completes normally. + let gates!: { echoStarted: Promise; releaseEcho: () => void }; + const { transport } = await setupStateless({ + enableJsonResponse: true, + configure: mcpServer => { + gates = registerGatedEchoTool(mcpServer); + } + }); + const { echoStarted, releaseEcho } = gates; + + // First exchange: in-flight request on the transport. + const aPromise = transport.handleRequest(postRequest(callBody('marker-a'))); + await echoStarted; + + // Second exchange with the same request id while the first is in + // flight - must throw. + await expect(transport.handleRequest(postRequest(callBody('marker-b')))).rejects.toThrow(REUSE_ERROR); + + // The first exchange completes with its own response. + releaseEcho(); + const aResponse = await aPromise; + expect(aResponse.status).toBe(200); + const aBody = await aResponse.text(); + expect(aBody).toContain('marker-a'); + expect(aBody).not.toContain('marker-b'); + }); + + it('a second POST during an in-flight stateless SSE exchange throws; the first stream delivers its own response (SSE shape)', async () => { + // Default SSE response mode: the second handleRequest() call is + // rejected while the first exchange's stream is still open; the first + // stream then delivers its own response. + let gates!: { echoStarted: Promise; releaseEcho: () => void }; + const { transport } = await setupStateless({ + configure: mcpServer => { + gates = registerGatedEchoTool(mcpServer); + } + }); + const { echoStarted, releaseEcho } = gates; + + // First exchange: SSE response in flight (response headers arrive + // before the tool resolves). + const aResponse = await transport.handleRequest(postRequest(callBody('marker-a'))); + expect(aResponse.status).toBe(200); + expect(aResponse.headers.get('content-type')).toContain('text/event-stream'); + await echoStarted; + + // Second exchange with the same request id while the first stream is + // open - must throw. + await expect(transport.handleRequest(postRequest(callBody('marker-b')))).rejects.toThrow(REUSE_ERROR); + + // The first stream delivers its own response (the stream closes after + // the JSON-RPC response event, so text() terminates). + releaseEcho(); + const aBody = await aResponse.text(); + expect(aBody).toContain('marker-a'); + expect(aBody).not.toContain('marker-b'); + }); +}); diff --git a/packages/server/test/server/streamableHttp.test.ts b/packages/server/test/server/streamableHttp.test.ts index beca451113..c88a2c32f8 100644 --- a/packages/server/test/server/streamableHttp.test.ts +++ b/packages/server/test/server/streamableHttp.test.ts @@ -428,11 +428,16 @@ describe('Zod v4', () => { }); describe('HTTPServerTransport - Stateless Mode', () => { - let transport: WebStandardStreamableHTTPServerTransport; - let mcpServer: McpServer; + // In stateless mode, each request must use a fresh transport + server + // pair (a stateless transport throws when reused across requests). + const statelessCleanups: Array<() => Promise> = []; - beforeEach(async () => { - mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }, { capabilities: { logging: {} } }); + afterEach(async () => { + while (statelessCleanups.length > 0) await statelessCleanups.pop()!(); + }); + + async function handleStatelessRequest(request: Request): Promise { + const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }, { capabilities: { logging: {} } }); mcpServer.registerTool( 'echo', @@ -442,33 +447,35 @@ describe('Zod v4', () => { } ); - transport = new WebStandardStreamableHTTPServerTransport({ + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); await mcpServer.connect(transport); - }); + statelessCleanups.push(async () => { + await mcpServer.close().catch(() => {}); + await transport.close().catch(() => {}); + }); - afterEach(async () => { - await transport.close(); - }); + return transport.handleRequest(request); + } it('should work without session management', async () => { const request = createRequest('POST', TEST_MESSAGES.initialize); - const response = await transport.handleRequest(request); + const response = await handleStatelessRequest(request); expect(response.status).toBe(200); expect(response.headers.get('mcp-session-id')).toBeNull(); }); it('should not require session ID on subsequent requests', async () => { - // Initialize + // Initialize (served by its own per-request transport) const initRequest = createRequest('POST', TEST_MESSAGES.initialize); - await transport.handleRequest(initRequest); + await handleStatelessRequest(initRequest); // Subsequent request without session ID should work const request = createRequest('POST', TEST_MESSAGES.toolsList); - const response = await transport.handleRequest(request); + const response = await handleStatelessRequest(request); expect(response.status).toBe(200); }); diff --git a/test/e2e/requirements.ts b/test/e2e/requirements.ts index 4fda9e661f..349124727e 100644 --- a/test/e2e/requirements.ts +++ b/test/e2e/requirements.ts @@ -1507,12 +1507,7 @@ export const REQUIREMENTS: Record = { source: 'sdk', behavior: 'A stateless per-request transport cannot be reused for a second request.', transports: ['streamableHttp'], - note: 'This exercises the HTTP hosting layer and stateless mode; the matrix transport arg is ignored, so it runs as a single streamableHttp-labelled cell to avoid duplicate runs.', - knownFailures: [ - { - note: 'changed in v2: the stateless reuse guard was removed, so a second request on the same per-request transport is processed instead of rejected.' - } - ] + note: 'This exercises the HTTP hosting layer and stateless mode; the matrix transport arg is ignored, so it runs as a single streamableHttp-labelled cell to avoid duplicate runs.' }, 'hosting:stateless:no-session-id': { source: 'sdk', diff --git a/test/integration/test/client/client.test.ts b/test/integration/test/client/client.test.ts index 5cfa15abf2..ada5d56898 100644 --- a/test/integration/test/client/client.test.ts +++ b/test/integration/test/client/client.test.ts @@ -152,6 +152,13 @@ test('should restore negotiated protocol version on transport when reconnecting expect(setProtocolVersion).toHaveBeenCalledWith(LATEST_PROTOCOL_VERSION); expect(client.getNegotiatedProtocolVersion()).toBe(LATEST_PROTOCOL_VERSION); + // The original connection drops (transport fires onclose). The negotiated + // protocol version survives a transport-initiated close -- only close()/a + // fresh handshake reset it -- which is exactly what session resumption + // relies on. (connect() on a still-connected client now throws; resumption + // happens after the old connection is gone.) + initialTransport.onclose?.(); + // Now simulate reconnection: new transport with a pre-existing sessionId. // connect() will early-return without re-initializing, but MUST restore the protocol version // so HTTP transports can keep sending the required mcp-protocol-version header. diff --git a/test/integration/test/client/discoverRoundtrip.test.ts b/test/integration/test/client/discoverRoundtrip.test.ts index 13bddc46fb..8801e42a40 100644 --- a/test/integration/test/client/discoverRoundtrip.test.ts +++ b/test/integration/test/client/discoverRoundtrip.test.ts @@ -41,29 +41,39 @@ describe('server/discover round-trip against a modern server', () => { async function startServer(options: { modernEraInstance: boolean }) { const httpServer: HttpServer = createServer(); - const mcpServer = new McpServer( - { name: 'dual-era-server', version: '2.0.0' }, - { - capabilities: { tools: { listChanged: true } }, - supportedProtocolVersions: DUAL_ERA_VERSIONS, - instructions: 'dual era' - } - ); - mcpServer.registerTool('echo', { inputSchema: z.object({ text: z.string() }) }, ({ text }) => ({ - content: [{ type: 'text', text }] - })); - const serverTransport = new NodeStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); - await mcpServer.connect(serverTransport); - if (options.modernEraInstance) { - // Stand-in for the server-side entry (instance binding): mark the - // instance as serving the modern era so it can answer the probe. - setNegotiatedProtocolVersion(mcpServer.server, MODERN); - } - httpServer.on('request', (req, res) => void serverTransport.handleRequest(req, res)); + // Stateless serving is per-request: each inbound request gets a fresh + // McpServer + stateless transport pair (a stateless transport throws + // when reused across requests). Era binding is instance state, so the + // stand-in entry binding is re-applied to every fresh instance. + httpServer.on('request', (req, res) => { + void (async () => { + const mcpServer = new McpServer( + { name: 'dual-era-server', version: '2.0.0' }, + { + capabilities: { tools: { listChanged: true } }, + supportedProtocolVersions: DUAL_ERA_VERSIONS, + instructions: 'dual era' + } + ); + mcpServer.registerTool('echo', { inputSchema: z.object({ text: z.string() }) }, ({ text }) => ({ + content: [{ type: 'text', text }] + })); + const serverTransport = new NodeStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); + await mcpServer.connect(serverTransport); + if (options.modernEraInstance) { + // Stand-in for the server-side entry (instance binding): mark the + // instance as serving the modern era so it can answer the probe. + setNegotiatedProtocolVersion(mcpServer.server, MODERN); + } + cleanups.push(async () => { + await mcpServer.close().catch(() => {}); + await serverTransport.close().catch(() => {}); + }); + await serverTransport.handleRequest(req, res); + })(); + }); const baseUrl = await listenOnRandomPort(httpServer); - cleanups.push(async () => { - await mcpServer.close().catch(() => {}); - await serverTransport.close().catch(() => {}); + cleanups.push(() => { httpServer.close(); }); return baseUrl; @@ -149,14 +159,21 @@ describe('server/discover round-trip against a modern server', () => { // round-trip over HTTP completes once every modern request carries the // per-request _meta envelope.) const httpServer: HttpServer = createServer(); - const mcpServer = new McpServer({ name: 'legacy-only', version: '1.0.0' }, { capabilities: { tools: {} } }); - const serverTransport = new NodeStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); - await mcpServer.connect(serverTransport); - httpServer.on('request', (req, res) => void serverTransport.handleRequest(req, res)); + // Per-request stateless pair (a stateless transport throws on reuse). + httpServer.on('request', (req, res) => { + void (async () => { + const mcpServer = new McpServer({ name: 'legacy-only', version: '1.0.0' }, { capabilities: { tools: {} } }); + const serverTransport = new NodeStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); + await mcpServer.connect(serverTransport); + cleanups.push(async () => { + await mcpServer.close().catch(() => {}); + await serverTransport.close().catch(() => {}); + }); + await serverTransport.handleRequest(req, res); + })(); + }); const baseUrl = await listenOnRandomPort(httpServer); - cleanups.push(async () => { - await mcpServer.close().catch(() => {}); - await serverTransport.close().catch(() => {}); + cleanups.push(() => { httpServer.close(); }); diff --git a/test/integration/test/client/versionNegotiation.test.ts b/test/integration/test/client/versionNegotiation.test.ts index 2bd93eba93..a78d358731 100644 --- a/test/integration/test/client/versionNegotiation.test.ts +++ b/test/integration/test/client/versionNegotiation.test.ts @@ -58,19 +58,57 @@ function recordingFetch() { const NEGOTIATION_HEADERS = ['mcp-protocol-version', 'mcp-method', 'mcp-name'] as const; -async function setupLegacyServer(stateful: boolean) { - const httpServer: Server = createServer(); +function buildLegacyMcpServer(): McpServer { const mcpServer = new McpServer({ name: 'deployed-2025-server', version: '1.0.0' }, { capabilities: { tools: {} } }); mcpServer.registerTool('echo', { inputSchema: z.object({ text: z.string() }) }, ({ text }) => ({ content: [{ type: 'text', text }] })); - const serverTransport = new NodeStreamableHTTPServerTransport({ - sessionIdGenerator: stateful ? () => randomUUID() : undefined + return mcpServer; +} + +async function setupLegacyServer(stateful: boolean): Promise<{ httpServer: Server; baseUrl: URL; close: () => Promise }> { + const httpServer: Server = createServer(); + if (stateful) { + const mcpServer = buildLegacyMcpServer(); + const serverTransport = new NodeStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID() }); + await mcpServer.connect(serverTransport); + httpServer.on('request', (req, res) => void serverTransport.handleRequest(req, res)); + const baseUrl = await listenOnRandomPort(httpServer); + return { + httpServer, + baseUrl, + close: async () => { + await mcpServer.close().catch(() => {}); + await serverTransport.close().catch(() => {}); + httpServer.close(); + } + }; + } + // Stateless deployment serves per request: a fresh McpServer + stateless + // transport pair per inbound request (a stateless transport throws when + // reused across requests). + const pairCleanups: Array<() => Promise> = []; + httpServer.on('request', (req, res) => { + void (async () => { + const mcpServer = buildLegacyMcpServer(); + const serverTransport = new NodeStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); + await mcpServer.connect(serverTransport); + pairCleanups.push(async () => { + await mcpServer.close().catch(() => {}); + await serverTransport.close().catch(() => {}); + }); + await serverTransport.handleRequest(req, res); + })(); }); - await mcpServer.connect(serverTransport); - httpServer.on('request', (req, res) => void serverTransport.handleRequest(req, res)); const baseUrl = await listenOnRandomPort(httpServer); - return { httpServer, mcpServer, serverTransport, baseUrl }; + return { + httpServer, + baseUrl, + close: async () => { + while (pairCleanups.length > 0) await pairCleanups.pop()!(); + httpServer.close(); + } + }; } describe('version negotiation against real legacy servers (wire-real first-contact shapes)', () => { @@ -81,11 +119,7 @@ describe('version negotiation against real legacy servers (wire-real first-conta async function startLegacy(stateful: boolean) { const setup = await setupLegacyServer(stateful); - cleanups.push(async () => { - await setup.mcpServer.close().catch(() => {}); - await setup.serverTransport.close().catch(() => {}); - setup.httpServer.close(); - }); + cleanups.push(setup.close); return setup; } diff --git a/test/integration/test/server/bun.test.ts b/test/integration/test/server/bun.test.ts index 229145aa86..ee97631942 100644 --- a/test/integration/test/server/bun.test.ts +++ b/test/integration/test/server/bun.test.ts @@ -13,9 +13,11 @@ import * as z from 'zod/v4'; describe('MCP on Bun', () => { let httpServer: ReturnType; - let transport: WebStandardStreamableHTTPServerTransport; + const perRequestServers: McpServer[] = []; - beforeAll(async () => { + // Stateless serving is per-request: a fresh transport + server pair per + // fetch (a stateless transport throws when reused across requests). + function buildServer(): McpServer { const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }); mcpServer.registerTool( @@ -29,17 +31,26 @@ describe('MCP on Bun', () => { }) ); - transport = new WebStandardStreamableHTTPServerTransport(); - await mcpServer.connect(transport); + return mcpServer; + } + beforeAll(async () => { httpServer = Bun.serve({ port: 0, - fetch: req => transport.handleRequest(req) + fetch: async req => { + const mcpServer = buildServer(); + const transport = new WebStandardStreamableHTTPServerTransport(); + await mcpServer.connect(transport); + perRequestServers.push(mcpServer); + return transport.handleRequest(req); + } }); }); afterAll(async () => { - await transport?.close(); + for (const mcpServer of perRequestServers) { + await mcpServer.close().catch(() => {}); + } httpServer?.stop(); }); diff --git a/test/integration/test/server/cloudflareWorkers.test.ts b/test/integration/test/server/cloudflareWorkers.test.ts index 44af77575a..997318d9c5 100644 --- a/test/integration/test/server/cloudflareWorkers.test.ts +++ b/test/integration/test/server/cloudflareWorkers.test.ts @@ -292,25 +292,33 @@ describe('Cloudflare Workers compatibility (no nodejs_compat)', () => { }; fs.writeFileSync(path.join(tempDir, 'wrangler.jsonc'), JSON.stringify(wranglerConfig, null, 2)); - // Write server source + // Write server source. Stateless serving is per-request: a fresh + // transport + server pair per fetch (a stateless transport throws + // when reused across requests). const serverSource = ` import { McpServer, WebStandardStreamableHTTPServerTransport } from '@modelcontextprotocol/server'; import { z } from 'zod'; -const server = new McpServer({ name: "test-server", version: "${SERVER_VERSION_NONCE}" }); +function buildServer() { + const server = new McpServer({ name: "test-server", version: "${SERVER_VERSION_NONCE}" }); -server.registerTool("greet", { - description: "Greet someone", - inputSchema: z.object({ name: z.string() }) -}, async ({ name }) => ({ - content: [{ type: "text", text: "Hello, " + name + "!" }] -})); + server.registerTool("greet", { + description: "Greet someone", + inputSchema: z.object({ name: z.string() }) + }, async ({ name }) => ({ + content: [{ type: "text", text: "Hello, " + name + "!" }] + })); -const transport = new WebStandardStreamableHTTPServerTransport(); -await server.connect(transport); + return server; +} export default { - fetch: (request) => transport.handleRequest(request) + fetch: async (request) => { + const server = buildServer(); + const transport = new WebStandardStreamableHTTPServerTransport(); + await server.connect(transport); + return transport.handleRequest(request); + } }; `; fs.writeFileSync(path.join(tempDir, 'server.ts'), serverSource); diff --git a/test/integration/test/server/deno.test.ts b/test/integration/test/server/deno.test.ts index 7d45f815cf..cf1434fbdf 100644 --- a/test/integration/test/server/deno.test.ts +++ b/test/integration/test/server/deno.test.ts @@ -16,23 +16,34 @@ Deno.test({ sanitizeOps: false, sanitizeResources: false, async fn() { - const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }); - - mcpServer.registerTool( - 'greet', - { - description: 'Greet someone', - inputSchema: z.object({ name: z.string() }) - }, - async ({ name }) => ({ - content: [{ type: 'text' as const, text: `Hello, ${name}!` }] - }) - ); - - const transport = new WebStandardStreamableHTTPServerTransport(); - await mcpServer.connect(transport); - - const httpServer = Deno.serve({ port: 0 }, req => transport.handleRequest(req)); + // Stateless serving is per-request: a fresh transport + server pair + // per request (a stateless transport throws when reused across + // requests). + const buildServer = () => { + const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }); + + mcpServer.registerTool( + 'greet', + { + description: 'Greet someone', + inputSchema: z.object({ name: z.string() }) + }, + async ({ name }) => ({ + content: [{ type: 'text' as const, text: `Hello, ${name}!` }] + }) + ); + + return mcpServer; + }; + + const perRequestServers: McpServer[] = []; + const httpServer = Deno.serve({ port: 0 }, async req => { + const mcpServer = buildServer(); + const transport = new WebStandardStreamableHTTPServerTransport(); + await mcpServer.connect(transport); + perRequestServers.push(mcpServer); + return transport.handleRequest(req); + }); const port = httpServer.addr.port; try { @@ -46,7 +57,9 @@ Deno.test({ await client.close(); } finally { - await transport.close(); + for (const mcpServer of perRequestServers) { + await mcpServer.close().catch(() => {}); + } await httpServer.shutdown(); } } diff --git a/test/integration/test/stateManagementStreamableHttp.test.ts b/test/integration/test/stateManagementStreamableHttp.test.ts index 3f32c64a34..3737e40601 100644 --- a/test/integration/test/stateManagementStreamableHttp.test.ts +++ b/test/integration/test/stateManagementStreamableHttp.test.ts @@ -8,8 +8,7 @@ import { LATEST_PROTOCOL_VERSION, McpServer } from '@modelcontextprotocol/server import { listenOnRandomPort } from '@modelcontextprotocol/test-helpers'; import * as z from 'zod/v4'; -async function setupServer(withSessionManagement: boolean) { - const server: Server = createServer(); +function buildMcpServer(): McpServer { const mcpServer = new McpServer( { name: 'test-server', version: '1.0.0' }, { @@ -59,11 +58,20 @@ async function setupServer(withSessionManagement: boolean) { } ); - // Create transport with or without session management + return mcpServer; +} + +async function setupServer(): Promise<{ + server: Server; + mcpServer: McpServer; + serverTransport: NodeStreamableHTTPServerTransport; + baseUrl: URL; +}> { + const server: Server = createServer(); + const mcpServer = buildMcpServer(); + const serverTransport = new NodeStreamableHTTPServerTransport({ - sessionIdGenerator: withSessionManagement - ? () => randomUUID() // With session management, generate UUID - : undefined // Without session management, return undefined + sessionIdGenerator: () => randomUUID() }); await mcpServer.connect(serverTransport); @@ -78,27 +86,48 @@ async function setupServer(withSessionManagement: boolean) { return { server, mcpServer, serverTransport, baseUrl }; } +/** + * Stateless hosting: each request gets a fresh transport + mcpServer pair (a + * stateless transport throws when reused across requests). The per-request + * mcpServer is closed after handling to release resources. + */ +async function setupStatelessServer(): Promise<{ server: Server; baseUrl: URL }> { + const server: Server = createServer(async (req, res) => { + try { + const mcpServer = buildMcpServer(); + const serverTransport = new NodeStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); + await mcpServer.connect(serverTransport); + res.on('close', () => { + void mcpServer.close().catch(() => {}); + }); + await serverTransport.handleRequest(req, res); + } catch (error) { + console.error('Error handling request:', error); + if (!res.headersSent) res.writeHead(500).end(); + } + }); + + const baseUrl = await listenOnRandomPort(server); + + return { server, baseUrl }; +} + describe('Zod v4', () => { describe('Streamable HTTP Transport Session Management', () => { // Function to set up the server with optional session management describe('Stateless Mode', () => { let server: Server; - let mcpServer: McpServer; - let serverTransport: NodeStreamableHTTPServerTransport; let baseUrl: URL; beforeEach(async () => { - const setup = await setupServer(false); + const setup = await setupStatelessServer(); server = setup.server; - mcpServer = setup.mcpServer; - serverTransport = setup.serverTransport; baseUrl = setup.baseUrl; }); afterEach(async () => { - // Clean up resources - await mcpServer.close().catch(() => {}); - await serverTransport.close().catch(() => {}); + // Clean up resources (per-request pairs are closed as their + // responses complete). server.close(); }); @@ -234,7 +263,7 @@ describe('Zod v4', () => { let baseUrl: URL; beforeEach(async () => { - const setup = await setupServer(true); + const setup = await setupServer(); server = setup.server; mcpServer = setup.mcpServer; serverTransport = setup.serverTransport; From acccb8e7cb62551439908cdc17d0a9a3e369e1c6 Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Thu, 2 Jul 2026 15:38:52 +0000 Subject: [PATCH 2/9] address review: shared connect guard, unwind callback restore, Node adapter single-use propagation, client changeset entry --- .changeset/stateless-transport-reuse-guard.md | 1 + docs/migration/upgrade-to-v2.md | 2 +- packages/client/src/client/client.ts | 13 +++----- packages/core-internal/src/shared/protocol.ts | 27 ++++++++++++--- .../test/shared/protocolConnectGuard.test.ts | 30 +++++++++++++++++ .../middleware/node/src/streamableHttp.ts | 24 +++++++++++--- .../node/test/streamableHttp.test.ts | 33 +++++++++++++++++++ 7 files changed, 111 insertions(+), 19 deletions(-) diff --git a/.changeset/stateless-transport-reuse-guard.md b/.changeset/stateless-transport-reuse-guard.md index 7b0bc4e1b6..e13b46ab29 100644 --- a/.changeset/stateless-transport-reuse-guard.md +++ b/.changeset/stateless-transport-reuse-guard.md @@ -1,6 +1,7 @@ --- '@modelcontextprotocol/server': patch '@modelcontextprotocol/core-internal': patch +'@modelcontextprotocol/client': patch --- Restores two v1 transport lifecycle invariants dropped in the v2 rewrite: diff --git a/docs/migration/upgrade-to-v2.md b/docs/migration/upgrade-to-v2.md index dd18f69f03..086d96d244 100644 --- a/docs/migration/upgrade-to-v2.md +++ b/docs/migration/upgrade-to-v2.md @@ -1618,7 +1618,7 @@ rewrite required unless noted. #### Transport / connection lifecycle (v1 parity) -Two v1 transport lifecycle invariants (in effect since sdk@1.26.0) apply in v2: +The following v1 transport lifecycle invariants (in effect since sdk@1.26.0) apply in v2: - **A stateless `WebStandardStreamableHTTPServerTransport` serves exactly one request.** With `sessionIdGenerator: undefined`, the second `handleRequest()` call throws diff --git a/packages/client/src/client/client.ts b/packages/client/src/client/client.ts index 44a12cdf4a..8193a76078 100644 --- a/packages/client/src/client/client.ts +++ b/packages/client/src/client/client.ts @@ -929,15 +929,10 @@ export class Client extends Protocol { override async connect(transport: Transport, options?: ConnectOptions): Promise { // Fail fast while still connected: the negotiated paths below reset // per-connection state and run the discover probe BEFORE the base - // Protocol.connect() reuse guard would fire — without this early - // check, a connect() on a connected instance would corrupt the live - // connection's state before throwing. Same guard + message as - // Protocol.connect(). - if (this.transport) { - throw new Error( - 'Already connected to a transport. Call close() before connecting to a new transport, or use a separate Protocol instance per connection.' - ); - } + // Protocol.connect() guard would fire — without this early check, a + // connect() on a connected instance would corrupt the live + // connection's state before throwing. + this.assertNotConnected(); if (options?.prior !== undefined) { // Zero-round-trip reconnect from a previously-obtained // DiscoverResult: bypasses versionNegotiation resolution entirely. diff --git a/packages/core-internal/src/shared/protocol.ts b/packages/core-internal/src/shared/protocol.ts index 5a1243cbce..261fc1e242 100644 --- a/packages/core-internal/src/shared/protocol.ts +++ b/packages/core-internal/src/shared/protocol.ts @@ -775,16 +775,26 @@ export abstract class Protocol { } /** - * Attaches to the given transport, starts it, and starts listening for messages. - * - * The caller assumes ownership of the {@linkcode Transport}, replacing any callbacks that have already been set, and expects that it is the only user of the {@linkcode Transport} instance going forward. + * Throws if this instance is already bound to a transport. Called by + * {@linkcode Protocol.connect} and, earlier, by subclass connect + * overrides whose pre-connect work mutates per-connection state (see + * `Client.connect()`): the check must run before any of that work. */ - async connect(transport: Transport): Promise { + protected assertNotConnected(): void { if (this._transport) { throw new Error( 'Already connected to a transport. Call close() before connecting to a new transport, or use a separate Protocol instance per connection.' ); } + } + + /** + * Attaches to the given transport, starts it, and starts listening for messages. + * + * The caller assumes ownership of the {@linkcode Transport}, replacing any callbacks that have already been set, and expects that it is the only user of the {@linkcode Transport} instance going forward. + */ + async connect(transport: Transport): Promise { + this.assertNotConnected(); this._transport = transport; const _onclose = this.transport?.onclose; @@ -825,7 +835,14 @@ export abstract class Protocol { // A transport that failed to start was never connected: unwind so // the instance can retry connect() — the reuse guard above would // otherwise lock the instance onto a transport that never carried - // traffic. + // traffic. Restore the transport's own callbacks too: leaving the + // wrapped closures installed would let a late event from the + // failed transport (e.g. a child process 'close' after a failed + // spawn) tear down whatever connection a successful retry + // established. + transport.onclose = _onclose; + transport.onerror = _onerror; + transport.onmessage = _onmessage; this._transport = undefined; throw error; } diff --git a/packages/core-internal/test/shared/protocolConnectGuard.test.ts b/packages/core-internal/test/shared/protocolConnectGuard.test.ts index a5752149b0..4e72b7e94e 100644 --- a/packages/core-internal/test/shared/protocolConnectGuard.test.ts +++ b/packages/core-internal/test/shared/protocolConnectGuard.test.ts @@ -57,6 +57,36 @@ function createProtocol(): Protocol { } describe('Protocol.connect() reuse guard', () => { + test('a transport that failed to start cannot disturb a later connection', async () => { + const protocol = createProtocol(); + const failing = new MockTransport('failing'); + const userOnClose = (): void => {}; + failing.onclose = userOnClose; + failing.start = async (): Promise => { + throw new Error('boot failure'); + }; + + await expect(protocol.connect(failing)).rejects.toThrow('boot failure'); + expect(protocol.transport).toBeUndefined(); + // The unwind restored the transport's own callbacks. + expect(failing.onclose).toBe(userOnClose); + + // A later connection is undisturbed by events from the failed + // transport (e.g. a child process 'close' arriving after a failed + // spawn already rejected start()). + const transportB = new MockTransport('B'); + await protocol.connect(transportB); + let sawClose = false; + protocol.onclose = () => { + sawClose = true; + }; + + failing.onclose?.(); + + expect(protocol.transport).toBe(transportB); + expect(sawClose).toBe(false); + }); + test('connect() while already connected throws', async () => { const protocol = createProtocol(); const transportA = new MockTransport('A'); diff --git a/packages/middleware/node/src/streamableHttp.ts b/packages/middleware/node/src/streamableHttp.ts index 602c55d4a2..9dcb2e6e70 100644 --- a/packages/middleware/node/src/streamableHttp.ts +++ b/packages/middleware/node/src/streamableHttp.ts @@ -203,12 +203,18 @@ export class NodeStreamableHTTPServerTransport implements Transport { // Create a custom handler that includes our context // overrideGlobalObjects: false prevents Hono from overwriting global Response, which would // break frameworks like Next.js whose response classes extend the native Response + let dispatchError: unknown; const handler = getRequestListener( async (webRequest: Request) => { - return this._webStandardTransport.handleRequest(webRequest, { - authInfo, - parsedBody - }); + try { + return await this._webStandardTransport.handleRequest(webRequest, { + authInfo, + parsedBody + }); + } catch (error) { + dispatchError = error; + throw error; + } }, { overrideGlobalObjects: false } ); @@ -216,6 +222,16 @@ export class NodeStreamableHTTPServerTransport implements Transport { // Delegate to the request listener which handles all the Node.js <-> Web Standard conversion // including proper SSE streaming support await handler(req, res); + + // getRequestListener absorbs a rejected dispatch into a generic 500 + // response, which would swallow the wrapped transport's single-use + // throw. Re-raise it so the documented behavior (reusing a stateless + // transport across requests throws at the call site) holds for the + // Node adapter too. Other dispatch errors keep the existing + // 500-response behavior. + if (dispatchError instanceof Error && dispatchError.message.includes('Stateless transport cannot be reused across requests')) { + throw dispatchError; + } } /** diff --git a/packages/middleware/node/test/streamableHttp.test.ts b/packages/middleware/node/test/streamableHttp.test.ts index 79f5e72ea2..f63e2eee00 100644 --- a/packages/middleware/node/test/streamableHttp.test.ts +++ b/packages/middleware/node/test/streamableHttp.test.ts @@ -1657,6 +1657,39 @@ describe('Zod v4', () => { return { transport, mcpServer }; } + it('reusing one stateless transport rejects at the handleRequest call site (single-use invariant)', async () => { + // Deliberately share ONE stateless transport across two requests: + // the second handleRequest call must reject with the single-use + // message at the Node call site (not be absorbed into a bare 500). + const mcpServer = new McpServer({ name: 'single-use-test', version: '1.0.0' }, { capabilities: { logging: {} } }); + const transport = new NodeStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); + await mcpServer.connect(transport); + perRequestServers.push(mcpServer); + + const callSiteErrors: Error[] = []; + const sharedServer = createServer(async (req, res) => { + try { + await transport.handleRequest(req, res); + } catch (error) { + callSiteErrors.push(error as Error); + if (!res.headersSent) res.writeHead(500); + res.end(); + } + }); + const sharedUrl = await listenOnRandomPort(sharedServer); + try { + const first = await sendPostRequest(sharedUrl, TEST_MESSAGES.initialize); + expect(first.status).toBe(200); + expect(callSiteErrors).toHaveLength(0); + + await sendPostRequest(sharedUrl, TEST_MESSAGES.toolsList); + expect(callSiteErrors).toHaveLength(1); + expect(callSiteErrors[0]!.message).toContain('Stateless transport cannot be reused across requests'); + } finally { + sharedServer.close(); + } + }); + it('should operate without session ID validation', async () => { // Initialize the server first const initResponse = await sendPostRequest(baseUrl, TEST_MESSAGES.initialize); From 6e83ce27c050dd7b4ca4d29633f93c1c33feb11e Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Thu, 2 Jul 2026 16:51:53 +0000 Subject: [PATCH 3/9] review: type the lifecycle guard errors (SdkErrorCode.AlreadyConnected, new SdkErrorCode.StatelessTransportReuse) instead of raw Error --- .changeset/stateless-transport-reuse-guard.md | 4 +- docs/migration/upgrade-to-v2.md | 38 ++++++++++--------- .../core-internal/src/errors/sdkErrors.ts | 2 + packages/core-internal/src/shared/protocol.ts | 3 +- .../test/shared/protocolConnectGuard.test.ts | 8 +++- .../test/types/errorSurfacePins.test.ts | 1 + .../middleware/node/src/streamableHttp.ts | 8 ++-- .../node/test/streamableHttp.test.ts | 1 + packages/server/src/server/streamableHttp.ts | 7 +++- .../test/server/statelessReuseGuard.test.ts | 9 ++++- 10 files changed, 54 insertions(+), 27 deletions(-) diff --git a/.changeset/stateless-transport-reuse-guard.md b/.changeset/stateless-transport-reuse-guard.md index e13b46ab29..a58c4bbcf9 100644 --- a/.changeset/stateless-transport-reuse-guard.md +++ b/.changeset/stateless-transport-reuse-guard.md @@ -1,10 +1,10 @@ --- '@modelcontextprotocol/server': patch -'@modelcontextprotocol/core-internal': patch +'@modelcontextprotocol/core-internal': minor '@modelcontextprotocol/client': patch --- -Restores two v1 transport lifecycle invariants dropped in the v2 rewrite: +Restores two v1 transport lifecycle invariants dropped in the v2 rewrite. Both throw typed errors — `SdkError` with `SdkErrorCode.StatelessTransportReuse` (new) and `SdkErrorCode.AlreadyConnected` — while keeping the message strings byte-identical to v1.x, so structured handling can use the code and existing message matching keeps working: - A stateless `WebStandardStreamableHTTPServerTransport` (`sessionIdGenerator: undefined`) handles a single request exchange and throws on reuse (matching v1 behavior since 1.26.0): `"Stateless transport cannot be reused across requests. Create a new transport per request."` Construct a fresh transport (and server instance) per request, or use `createMcpHandler`, which already serves per-request pairs. Stateful transports (`sessionIdGenerator` set) are unaffected; `NodeStreamableHTTPServerTransport` inherits the behavior. - `Protocol.connect()` (and `Client.connect()`) throw when the instance is already connected, instead of silently rebinding the transport: `"Already connected to a transport. Call close() before connecting to a new transport, or use a separate Protocol instance per connection."` Sequential `close()` then `connect()` keeps working. After `close()` aborts an in-flight request handler, `ctx.mcpReq.notify()` resolves as a no-op and `ctx.mcpReq.send()` rejects with `SdkError(ConnectionClosed)`, consistent with the above. diff --git a/docs/migration/upgrade-to-v2.md b/docs/migration/upgrade-to-v2.md index 086d96d244..f9a4d69fd1 100644 --- a/docs/migration/upgrade-to-v2.md +++ b/docs/migration/upgrade-to-v2.md @@ -1623,28 +1623,30 @@ The following v1 transport lifecycle invariants (in effect since sdk@1.26.0) app - **A stateless `WebStandardStreamableHTTPServerTransport` serves exactly one request.** With `sessionIdGenerator: undefined`, the second `handleRequest()` call throws `"Stateless transport cannot be reused across requests. Create a new transport per - request."` — matching v1 behavior since 1.26.0. Stateful transports +request."` (an `SdkError` with code `StatelessTransportReuse`) — matching v1 + message behavior since 1.26.0. Stateful transports (`sessionIdGenerator` set) accept multiple requests as before; `NodeStreamableHTTPServerTransport` inherits the behavior. Migrate shared-transport hosting to the per-request pattern: - ```typescript - app.all('/mcp', async c => { - // Fresh transport + server pair per request. - const server = new McpServer({ name: 'my-server', version: '1.0.0' }); - const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); - await server.connect(transport); - return transport.handleRequest(c.req.raw); - }); - ``` - - Or use `createMcpHandler`, which constructs the per-request pair for you (its - `legacy: 'stateless'` fallback and the modern per-request transport are both - one-exchange by construction). - -- **`Protocol.connect()` throws when already connected** (`"Already connected to a - transport. Call close() before connecting to a new transport, or use a separate - Protocol instance per connection."`) instead of silently rebinding `this._transport`. + ```typescript + app.all('/mcp', async c => { + // Fresh transport + server pair per request. + const server = new McpServer({ name: 'my-server', version: '1.0.0' }); + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: undefined }); + await server.connect(transport); + return transport.handleRequest(c.req.raw); + }); + ``` + + Or use `createMcpHandler`, which constructs the per-request pair for you (its + `legacy: 'stateless'` fallback and the modern per-request transport are both + one-exchange by construction). + +- **`Protocol.connect()` throws when already connected** (an `SdkError` with code + `AlreadyConnected`: `"Already connected to a transport. Call close() before connecting +to a new transport, or use a separate Protocol instance per connection."`) instead of + silently rebinding `this._transport`. Sequential `close()` then `connect()` keeps working. `Client.connect()` performs the same check up front, before any per-connection state is reset. diff --git a/packages/core-internal/src/errors/sdkErrors.ts b/packages/core-internal/src/errors/sdkErrors.ts index 21f42f6ef8..96fcfa6e95 100644 --- a/packages/core-internal/src/errors/sdkErrors.ts +++ b/packages/core-internal/src/errors/sdkErrors.ts @@ -14,6 +14,8 @@ export enum SdkErrorCode { NotConnected = 'NOT_CONNECTED', /** Transport is already connected */ AlreadyConnected = 'ALREADY_CONNECTED', + /** A stateless (sessionless) transport was asked to handle a second request exchange */ + StatelessTransportReuse = 'STATELESS_TRANSPORT_REUSE', /** Protocol is not initialized */ NotInitialized = 'NOT_INITIALIZED', diff --git a/packages/core-internal/src/shared/protocol.ts b/packages/core-internal/src/shared/protocol.ts index 261fc1e242..0d631ae2f7 100644 --- a/packages/core-internal/src/shared/protocol.ts +++ b/packages/core-internal/src/shared/protocol.ts @@ -782,7 +782,8 @@ export abstract class Protocol { */ protected assertNotConnected(): void { if (this._transport) { - throw new Error( + throw new SdkError( + SdkErrorCode.AlreadyConnected, 'Already connected to a transport. Call close() before connecting to a new transport, or use a separate Protocol instance per connection.' ); } diff --git a/packages/core-internal/test/shared/protocolConnectGuard.test.ts b/packages/core-internal/test/shared/protocolConnectGuard.test.ts index 4e72b7e94e..08c2871098 100644 --- a/packages/core-internal/test/shared/protocolConnectGuard.test.ts +++ b/packages/core-internal/test/shared/protocolConnectGuard.test.ts @@ -94,7 +94,13 @@ describe('Protocol.connect() reuse guard', () => { await protocol.connect(transportA); - await expect(protocol.connect(transportB)).rejects.toThrow( + const rejection = await protocol.connect(transportB).then( + () => undefined, + (error: unknown) => error + ); + expect(rejection).toBeInstanceOf(SdkError); + expect((rejection as SdkError).code).toBe(SdkErrorCode.AlreadyConnected); + expect((rejection as SdkError).message).toBe( 'Already connected to a transport. Call close() before connecting to a new transport, or use a separate Protocol instance per connection.' ); diff --git a/packages/core-internal/test/types/errorSurfacePins.test.ts b/packages/core-internal/test/types/errorSurfacePins.test.ts index cc01cf4c57..b3dac31337 100644 --- a/packages/core-internal/test/types/errorSurfacePins.test.ts +++ b/packages/core-internal/test/types/errorSurfacePins.test.ts @@ -70,6 +70,7 @@ describe('SdkErrorCode', () => { expect({ ...SdkErrorCode }).toEqual({ NotConnected: 'NOT_CONNECTED', AlreadyConnected: 'ALREADY_CONNECTED', + StatelessTransportReuse: 'STATELESS_TRANSPORT_REUSE', NotInitialized: 'NOT_INITIALIZED', CapabilityNotSupported: 'CAPABILITY_NOT_SUPPORTED', RequestTimeout: 'REQUEST_TIMEOUT', diff --git a/packages/middleware/node/src/streamableHttp.ts b/packages/middleware/node/src/streamableHttp.ts index 9dcb2e6e70..fb3177bd9c 100644 --- a/packages/middleware/node/src/streamableHttp.ts +++ b/packages/middleware/node/src/streamableHttp.ts @@ -18,7 +18,7 @@ import type { Transport, WebStandardStreamableHTTPServerTransportOptions } from '@modelcontextprotocol/server'; -import { WebStandardStreamableHTTPServerTransport } from '@modelcontextprotocol/server'; +import { SdkErrorCode, WebStandardStreamableHTTPServerTransport } from '@modelcontextprotocol/server'; /** * Configuration options for {@linkcode NodeStreamableHTTPServerTransport} @@ -228,8 +228,10 @@ export class NodeStreamableHTTPServerTransport implements Transport { // throw. Re-raise it so the documented behavior (reusing a stateless // transport across requests throws at the call site) holds for the // Node adapter too. Other dispatch errors keep the existing - // 500-response behavior. - if (dispatchError instanceof Error && dispatchError.message.includes('Stateless transport cannot be reused across requests')) { + // 500-response behavior. Matched by code, not `instanceof`, so the + // check also holds if bundling ever yields two copies of the error + // class. + if (dispatchError instanceof Error && (dispatchError as { code?: unknown }).code === SdkErrorCode.StatelessTransportReuse) { throw dispatchError; } } diff --git a/packages/middleware/node/test/streamableHttp.test.ts b/packages/middleware/node/test/streamableHttp.test.ts index f63e2eee00..6603086f85 100644 --- a/packages/middleware/node/test/streamableHttp.test.ts +++ b/packages/middleware/node/test/streamableHttp.test.ts @@ -1685,6 +1685,7 @@ describe('Zod v4', () => { await sendPostRequest(sharedUrl, TEST_MESSAGES.toolsList); expect(callSiteErrors).toHaveLength(1); expect(callSiteErrors[0]!.message).toContain('Stateless transport cannot be reused across requests'); + expect((callSiteErrors[0] as { code?: unknown }).code).toBe('STATELESS_TRANSPORT_REUSE'); } finally { sharedServer.close(); } diff --git a/packages/server/src/server/streamableHttp.ts b/packages/server/src/server/streamableHttp.ts index aabce85fcd..90bb953bef 100644 --- a/packages/server/src/server/streamableHttp.ts +++ b/packages/server/src/server/streamableHttp.ts @@ -16,6 +16,8 @@ import { isJSONRPCRequest, isJSONRPCResultResponse, JSONRPCMessageSchema, + SdkError, + SdkErrorCode, SUPPORTED_PROTOCOL_VERSIONS } from '@modelcontextprotocol/core-internal'; @@ -377,7 +379,10 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { // In stateless mode (no sessionIdGenerator), each request must use a fresh transport. // Reusing a stateless transport causes message ID collisions between clients. if (!this.sessionIdGenerator && this._hasHandledRequest) { - throw new Error('Stateless transport cannot be reused across requests. Create a new transport per request.'); + throw new SdkError( + SdkErrorCode.StatelessTransportReuse, + 'Stateless transport cannot be reused across requests. Create a new transport per request.' + ); } this._hasHandledRequest = true; diff --git a/packages/server/test/server/statelessReuseGuard.test.ts b/packages/server/test/server/statelessReuseGuard.test.ts index fb12cc5791..0752f99a82 100644 --- a/packages/server/test/server/statelessReuseGuard.test.ts +++ b/packages/server/test/server/statelessReuseGuard.test.ts @@ -10,6 +10,7 @@ import { randomUUID } from 'node:crypto'; import type { CallToolResult, JSONRPCMessage } from '@modelcontextprotocol/core-internal'; +import { SdkError, SdkErrorCode } from '@modelcontextprotocol/core-internal'; import { afterEach, describe, expect, it } from 'vitest'; import * as z from 'zod/v4'; @@ -88,7 +89,13 @@ describe('stateless transport reuse guard', () => { const first = await transport.handleRequest(postRequest(initializeMessage)); expect(first.status).toBe(200); - await expect(transport.handleRequest(postRequest(toolsListMessage))).rejects.toThrow(REUSE_ERROR); + const rejection = await transport.handleRequest(postRequest(toolsListMessage)).then( + () => undefined, + (error: unknown) => error + ); + expect(rejection).toBeInstanceOf(SdkError); + expect((rejection as SdkError).code).toBe(SdkErrorCode.StatelessTransportReuse); + expect((rejection as SdkError).message).toMatch(REUSE_ERROR); }); it('throws on a GET after a handled request', async () => { From a95a303ce3905360b5b7673b7872bfd85189aa33 Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Thu, 2 Jul 2026 17:01:03 +0000 Subject: [PATCH 4/9] review: reject stateless reuse before delegating to getRequestListener so the caller keeps control of the response --- .../middleware/node/src/streamableHttp.ts | 24 ++++++++++++------- .../node/test/streamableHttp.test.ts | 6 +++++ packages/server/src/server/streamableHttp.ts | 14 ++++++++++- 3 files changed, 35 insertions(+), 9 deletions(-) diff --git a/packages/middleware/node/src/streamableHttp.ts b/packages/middleware/node/src/streamableHttp.ts index fb3177bd9c..605334b31b 100644 --- a/packages/middleware/node/src/streamableHttp.ts +++ b/packages/middleware/node/src/streamableHttp.ts @@ -196,6 +196,13 @@ export class NodeStreamableHTTPServerTransport implements Transport { * @param parsedBody - Optional pre-parsed body from body-parser middleware */ async handleRequest(req: IncomingMessage & { auth?: AuthInfo }, res: ServerResponse, parsedBody?: unknown): Promise { + // Fail fast before delegating: once getRequestListener is invoked, a + // rejected dispatch is committed to `res` as an empty 500 before the + // rejection can surface, leaving the caller unable to shape its own + // error response. Checking up front rejects with nothing written to + // `res`. The check lives on the wrapped transport (single source). + this._webStandardTransport.assertNotReused(); + // Store context for this request to pass through auth and parsedBody // We need to intercept the request creation to attach this context const authInfo = req.auth; @@ -223,14 +230,15 @@ export class NodeStreamableHTTPServerTransport implements Transport { // including proper SSE streaming support await handler(req, res); - // getRequestListener absorbs a rejected dispatch into a generic 500 - // response, which would swallow the wrapped transport's single-use - // throw. Re-raise it so the documented behavior (reusing a stateless - // transport across requests throws at the call site) holds for the - // Node adapter too. Other dispatch errors keep the existing - // 500-response behavior. Matched by code, not `instanceof`, so the - // check also holds if bundling ever yields two copies of the error - // class. + // Backstop for concurrent calls racing the up-front check (the + // single-use flag is set inside the wrapped handleRequest, after the + // request conversion): getRequestListener absorbs a rejected dispatch + // into a generic 500, which would swallow the single-use throw. + // Re-raise it so the documented behavior (reusing a stateless + // transport across requests throws at the call site) holds on this + // path too. Other dispatch errors keep the existing 500-response + // behavior. Matched by code, not `instanceof`, so the check also + // holds if bundling ever yields two copies of the error class. if (dispatchError instanceof Error && (dispatchError as { code?: unknown }).code === SdkErrorCode.StatelessTransportReuse) { throw dispatchError; } diff --git a/packages/middleware/node/test/streamableHttp.test.ts b/packages/middleware/node/test/streamableHttp.test.ts index 6603086f85..524e309fad 100644 --- a/packages/middleware/node/test/streamableHttp.test.ts +++ b/packages/middleware/node/test/streamableHttp.test.ts @@ -1667,11 +1667,13 @@ describe('Zod v4', () => { perRequestServers.push(mcpServer); const callSiteErrors: Error[] = []; + const headersSentAtRejection: boolean[] = []; const sharedServer = createServer(async (req, res) => { try { await transport.handleRequest(req, res); } catch (error) { callSiteErrors.push(error as Error); + headersSentAtRejection.push(res.headersSent); if (!res.headersSent) res.writeHead(500); res.end(); } @@ -1686,6 +1688,10 @@ describe('Zod v4', () => { expect(callSiteErrors).toHaveLength(1); expect(callSiteErrors[0]!.message).toContain('Stateless transport cannot be reused across requests'); expect((callSiteErrors[0] as { code?: unknown }).code).toBe('STATELESS_TRANSPORT_REUSE'); + // The rejection reached the caller before anything was + // committed to `res` — the caller keeps control of the + // response it sends for the rejected request. + expect(headersSentAtRejection).toEqual([false]); } finally { sharedServer.close(); } diff --git a/packages/server/src/server/streamableHttp.ts b/packages/server/src/server/streamableHttp.ts index 90bb953bef..9bf4ad042d 100644 --- a/packages/server/src/server/streamableHttp.ts +++ b/packages/server/src/server/streamableHttp.ts @@ -375,7 +375,15 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { * Handles an incoming HTTP request, whether `GET`, `POST`, or `DELETE` * Returns a `Response` object (Web Standard) */ - async handleRequest(req: Request, options?: HandleRequestOptions): Promise { + /** + * Throws when a stateless transport has already handled its single + * request exchange. This is the same check `handleRequest()` performs on + * entry; exposed so wrapping adapters (e.g. the Node middleware) can fail + * fast before committing anything to their own response stream. + * + * @internal + */ + assertNotReused(): void { // In stateless mode (no sessionIdGenerator), each request must use a fresh transport. // Reusing a stateless transport causes message ID collisions between clients. if (!this.sessionIdGenerator && this._hasHandledRequest) { @@ -384,6 +392,10 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { 'Stateless transport cannot be reused across requests. Create a new transport per request.' ); } + } + + async handleRequest(req: Request, options?: HandleRequestOptions): Promise { + this.assertNotReused(); this._hasHandledRequest = true; // Validate request headers for DNS rebinding protection From c9505ae59e2077f70ab9c2b635fd1edd40869081 Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Thu, 2 Jul 2026 17:28:22 +0000 Subject: [PATCH 5/9] review: apply the handler abort gate to ctx.mcpReq.elicitInput and requestSampling --- .changeset/stateless-transport-reuse-guard.md | 2 +- docs/migration/upgrade-to-v2.md | 5 +- packages/server/src/server/server.ts | 19 ++++- .../test/server/abortedContextSenders.test.ts | 82 +++++++++++++++++++ 4 files changed, 103 insertions(+), 5 deletions(-) create mode 100644 packages/server/test/server/abortedContextSenders.test.ts diff --git a/.changeset/stateless-transport-reuse-guard.md b/.changeset/stateless-transport-reuse-guard.md index a58c4bbcf9..7335ac572b 100644 --- a/.changeset/stateless-transport-reuse-guard.md +++ b/.changeset/stateless-transport-reuse-guard.md @@ -7,6 +7,6 @@ Restores two v1 transport lifecycle invariants dropped in the v2 rewrite. Both throw typed errors — `SdkError` with `SdkErrorCode.StatelessTransportReuse` (new) and `SdkErrorCode.AlreadyConnected` — while keeping the message strings byte-identical to v1.x, so structured handling can use the code and existing message matching keeps working: - A stateless `WebStandardStreamableHTTPServerTransport` (`sessionIdGenerator: undefined`) handles a single request exchange and throws on reuse (matching v1 behavior since 1.26.0): `"Stateless transport cannot be reused across requests. Create a new transport per request."` Construct a fresh transport (and server instance) per request, or use `createMcpHandler`, which already serves per-request pairs. Stateful transports (`sessionIdGenerator` set) are unaffected; `NodeStreamableHTTPServerTransport` inherits the behavior. -- `Protocol.connect()` (and `Client.connect()`) throw when the instance is already connected, instead of silently rebinding the transport: `"Already connected to a transport. Call close() before connecting to a new transport, or use a separate Protocol instance per connection."` Sequential `close()` then `connect()` keeps working. After `close()` aborts an in-flight request handler, `ctx.mcpReq.notify()` resolves as a no-op and `ctx.mcpReq.send()` rejects with `SdkError(ConnectionClosed)`, consistent with the above. +- `Protocol.connect()` (and `Client.connect()`) throw when the instance is already connected, instead of silently rebinding the transport: `"Already connected to a transport. Call close() before connecting to a new transport, or use a separate Protocol instance per connection."` Sequential `close()` then `connect()` keeps working. After `close()` aborts an in-flight request handler, `ctx.mcpReq.notify()` resolves as a no-op and `ctx.mcpReq.send()`, `ctx.mcpReq.elicitInput()`, and `ctx.mcpReq.requestSampling()` reject with `SdkError(ConnectionClosed)`, consistent with the above. Docs, middleware READMEs, and examples that showed a shared stateless transport now show the per-request pattern. diff --git a/docs/migration/upgrade-to-v2.md b/docs/migration/upgrade-to-v2.md index f9a4d69fd1..1340ca1300 100644 --- a/docs/migration/upgrade-to-v2.md +++ b/docs/migration/upgrade-to-v2.md @@ -1652,8 +1652,9 @@ to a new transport, or use a separate Protocol instance per connection."`) inste - **Aborted request handlers cannot write to a replacement transport.** After the handler's `ctx.mcpReq.signal` aborts (connection closed or request cancelled), - `ctx.mcpReq.notify()` resolves as a no-op and `ctx.mcpReq.send()` rejects with - `SdkError(ConnectionClosed)`. + `ctx.mcpReq.notify()` (and `ctx.mcpReq.log()`) resolves as a no-op, and + `ctx.mcpReq.send()`, `ctx.mcpReq.elicitInput()`, and `ctx.mcpReq.requestSampling()` + reject with `SdkError(ConnectionClosed)`. #### Server (deprecated accessors and app-factory Origin validation) diff --git a/packages/server/src/server/server.ts b/packages/server/src/server/server.ts index 4151849494..bc96717fa5 100644 --- a/packages/server/src/server/server.ts +++ b/packages/server/src/server/server.ts @@ -400,8 +400,23 @@ export class Server extends Protocol { // session-wide stream to deliver it on. return ctx.mcpReq.notify({ method: 'notifications/message', params: { level, data, logger } }); }, - elicitInput: (params, options) => this.elicitInput(params, options), - requestSampling: (params, options) => this.createMessage(params, options) + // Same abort gate as `ctx.mcpReq.send`: once this request's + // handler has been aborted (connection closed or request + // cancelled), its per-request senders must not reach the + // transport, which is read live at send time and may belong + // to a later connection. + elicitInput: (params, options) => { + if (ctx.mcpReq.signal.aborted) { + return Promise.reject(new SdkError(SdkErrorCode.ConnectionClosed, 'Request was cancelled')); + } + return this.elicitInput(params, options); + }, + requestSampling: (params, options) => { + if (ctx.mcpReq.signal.aborted) { + return Promise.reject(new SdkError(SdkErrorCode.ConnectionClosed, 'Request was cancelled')); + } + return this.createMessage(params, options); + } }, http: hasHttpInfo ? { diff --git a/packages/server/test/server/abortedContextSenders.test.ts b/packages/server/test/server/abortedContextSenders.test.ts new file mode 100644 index 0000000000..21c1f83e0f --- /dev/null +++ b/packages/server/test/server/abortedContextSenders.test.ts @@ -0,0 +1,82 @@ +/** + * After close() aborts an in-flight request handler, the ServerContext + * per-request senders (`ctx.mcpReq.elicitInput`, `ctx.mcpReq.requestSampling`) + * reject with `ConnectionClosed` instead of reaching a transport — the same + * gate `ctx.mcpReq.notify` / `ctx.mcpReq.send` apply. The transport reference + * is read live at send time, so without the gate an aborted handler could + * write to a transport connected after the close. + */ +import type { JSONRPCMessage } from '@modelcontextprotocol/core-internal'; +import { InMemoryTransport, SdkError, SdkErrorCode } from '@modelcontextprotocol/core-internal'; +import { describe, expect, it } from 'vitest'; + +import { Server } from '../../src/server/server'; + +describe('aborted-handler context senders', () => { + it('elicitInput and requestSampling reject with ConnectionClosed after close(), and never reach a later connection', async () => { + const server = new Server({ name: 'abort-gate-test', version: '1.0.0' }, { capabilities: { tools: {} } }); + const [peerTx, serverTx] = InMemoryTransport.createLinkedPair(); + + let elicitCall!: () => Promise; + let samplingCall!: () => Promise; + let handlerEntered!: () => void; + const entered = new Promise(resolve => { + handlerEntered = resolve; + }); + server.setRequestHandler('tools/call', async (_request, ctx) => { + elicitCall = () => + ctx.mcpReq.elicitInput({ + message: 'pick one', + requestedSchema: { type: 'object', properties: {} } + }); + samplingCall = () => + ctx.mcpReq.requestSampling({ + messages: [], + maxTokens: 1 + }); + handlerEntered(); + // Held open until close() aborts it. + await new Promise(() => {}); + return { content: [] }; + }); + + await server.connect(serverTx); + await peerTx.start(); + await peerTx.send({ + jsonrpc: '2.0', + id: 0, + method: 'initialize', + params: { + protocolVersion: '2025-11-25', + capabilities: { elicitation: {}, sampling: {} }, + clientInfo: { name: 'peer', version: '0' } + } + }); + await peerTx.send({ jsonrpc: '2.0', method: 'notifications/initialized' }); + await peerTx.send({ jsonrpc: '2.0', id: 1, method: 'tools/call', params: { name: 'hold', arguments: {} } }); + await entered; + + await server.close(); + + // A later connection on the same instance must not receive the + // aborted handler's sends. + const [latePeerTx, lateServerTx] = InMemoryTransport.createLinkedPair(); + const latePeerMessages: JSONRPCMessage[] = []; + latePeerTx.onmessage = message => { + latePeerMessages.push(message as JSONRPCMessage); + }; + await server.connect(lateServerTx); + + for (const call of [elicitCall, samplingCall]) { + const rejection = await call().then( + () => undefined, + (error: unknown) => error + ); + expect(rejection).toBeInstanceOf(SdkError); + expect((rejection as SdkError).code).toBe(SdkErrorCode.ConnectionClosed); + } + expect(latePeerMessages).toHaveLength(0); + + await server.close(); + }); +}); From 652694c902751b669b638485f7c84505c8ecc52c Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Thu, 9 Jul 2026 11:47:47 +0000 Subject: [PATCH 6/9] review: adopt the transport-carried protocol version when resuming a session close() wipes the client's negotiated state, so the documented close-then-connect sequence silently broke session resumption: the resume branches skipped setProtocolVersion and subsequent requests went out without the MCP-Protocol-Version header. Both resume branches now fall back to the version the transport itself carries (new optional Transport.protocolVersion, which StreamableHTTPClientTransport already exposes) and re-adopt it as connection state. --- packages/client/src/client/client.ts | 45 ++++++++----------- .../test/client/versionNegotiation.test.ts | 18 ++++++++ .../core-internal/src/shared/transport.ts | 3 ++ test/integration/test/client/client.test.ts | 26 +++++++++++ 4 files changed, 65 insertions(+), 27 deletions(-) diff --git a/packages/client/src/client/client.ts b/packages/client/src/client/client.ts index 8193a76078..6746f63820 100644 --- a/packages/client/src/client/client.ts +++ b/packages/client/src/client/client.ts @@ -927,11 +927,8 @@ export class Client extends Protocol { * ``` */ override async connect(transport: Transport, options?: ConnectOptions): Promise { - // Fail fast while still connected: the negotiated paths below reset - // per-connection state and run the discover probe BEFORE the base - // Protocol.connect() guard would fire — without this early check, a - // connect() on a connected instance would corrupt the live - // connection's state before throwing. + // Guard before the paths below mutate per-connection state — the base + // Protocol.connect() check would fire only after that corruption. this.assertNotConnected(); if (options?.prior !== undefined) { // Zero-round-trip reconnect from a previously-obtained @@ -944,27 +941,20 @@ export class Client extends Protocol { } // Plain legacy connect — the pinned 2025 sequence, byte-untouched. await super.connect(transport); - // 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. + // Session resume (transport carries a sessionId): skip re-init, restore + // the originally negotiated version — from this instance, or from the + // transport itself after close() wiped the instance's copy. if (transport.sessionId !== undefined) { - const negotiatedProtocolVersion = this._negotiatedProtocolVersion; - if (negotiatedProtocolVersion !== undefined) { - // Resuming keeps the original negotiation: the instance still - // holds the negotiated version (and with it the wire era) — - // only the new transport needs the header pushed again. - transport.setProtocolVersion?.(negotiatedProtocolVersion); + const version = this._negotiatedProtocolVersion ?? transport.protocolVersion; + if (version !== undefined) { + this._negotiatedProtocolVersion = version; + transport.setProtocolVersion?.(version); } return; } - // Fresh connect: per-connection state left over from a previous - // connection must not survive into a new handshake. Clearing it puts - // the instance back in the pre-negotiation phase, so the initialize - // exchange below rides the bootstrap method pins (legacy era) instead - // of a dead session's era. Without this, an instance that once - // negotiated a modern era could never re-run a fresh handshake: - // `initialize` is physically absent from the modern registry. (The - // resume branch above keeps it instead.) + // Fresh connect: clear leftover connection state so `initialize` rides + // the legacy bootstrap pins — a stale modern era would make it + // unsendable (absent from the modern registry). this._resetConnectionState(); await this._legacyHandshake(transport, options); } @@ -1052,13 +1042,14 @@ export class Client extends Protocol { negotiation: Extract, options?: RequestOptions ): Promise { - // Session-resuming reconnect: restore the previously negotiated version, - // never re-probe mid-session. + // Session resume: never re-probe; restore the negotiated version from + // this instance or, after close() wiped it, from the transport itself. if (transport.sessionId !== undefined) { await super.connect(transport); - const negotiatedProtocolVersion = this._negotiatedProtocolVersion; - if (negotiatedProtocolVersion !== undefined && transport.setProtocolVersion) { - transport.setProtocolVersion(negotiatedProtocolVersion); + const version = this._negotiatedProtocolVersion ?? transport.protocolVersion; + if (version !== undefined) { + this._negotiatedProtocolVersion = version; + transport.setProtocolVersion?.(version); } return; } diff --git a/packages/client/test/client/versionNegotiation.test.ts b/packages/client/test/client/versionNegotiation.test.ts index 698bccd6ee..75fbdbaeb0 100644 --- a/packages/client/test/client/versionNegotiation.test.ts +++ b/packages/client/test/client/versionNegotiation.test.ts @@ -82,6 +82,7 @@ class ScriptedTransport implements Transport { onerror?: (error: Error) => void; onmessage?: (message: JSONRPCMessage) => void; sessionId?: string; + protocolVersion?: string; startCalls = 0; sent: JSONRPCMessage[] = []; @@ -698,6 +699,23 @@ describe('era scope discipline', () => { await client.close(); }); + test('session resume on a fresh auto client adopts the transport-carried version: no probe, no initialize', async () => { + // After close() (or on a brand-new client) the instance holds no + // negotiated version — a resuming transport constructed with + // { sessionId, protocolVersion } supplies it instead. + const transport = new ScriptedTransport(() => {}); + transport.sessionId = 'sess-1'; + transport.protocolVersion = MODERN; + const client = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'auto' } }); + + await client.connect(transport); + + expect(requests(transport.sent)).toHaveLength(0); + expect(transport.setProtocolVersionCalls).toEqual([MODERN]); + expect(client.getNegotiatedProtocolVersion()).toBe(MODERN); + await client.close(); + }); + test('no era state exists before the first connect, and none is persisted anywhere', () => { const client = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'auto' } }); expect(client.getNegotiatedProtocolVersion()).toBeUndefined(); diff --git a/packages/core-internal/src/shared/transport.ts b/packages/core-internal/src/shared/transport.ts index 226f6ab0bd..f077a39024 100644 --- a/packages/core-internal/src/shared/transport.ts +++ b/packages/core-internal/src/shared/transport.ts @@ -171,6 +171,9 @@ export interface Transport { */ setProtocolVersion?: ((version: string) => void) | undefined; + /** The negotiated protocol version, when the transport already carries one (e.g. a session-resuming transport constructed with a saved version). */ + readonly protocolVersion?: string | undefined; + /** * Sets the supported protocol versions for header validation (called during connect). * This allows the server to pass its supported versions to the transport. diff --git a/test/integration/test/client/client.test.ts b/test/integration/test/client/client.test.ts index ada5d56898..bf2c3f10bb 100644 --- a/test/integration/test/client/client.test.ts +++ b/test/integration/test/client/client.test.ts @@ -179,6 +179,32 @@ test('should restore negotiated protocol version on transport when reconnecting expect(reconnectSetProtocolVersion).toHaveBeenCalledWith(LATEST_PROTOCOL_VERSION); }); +/*** + * Test: Session resumption on a FRESH client adopts the transport-carried protocol version. + * After close() (or on a brand-new Client) the instance holds no negotiated version; a + * resuming transport constructed with `{ sessionId, protocolVersion }` supplies it instead. + */ +test('resuming with a fresh client adopts the protocol version carried by the transport', async () => { + const setProtocolVersion = vi.fn(); + const resumeTransport: Transport = { + start: vi.fn().mockResolvedValue(undefined), + close: vi.fn().mockResolvedValue(undefined), + setProtocolVersion, + send: vi.fn().mockResolvedValue(undefined), + sessionId: 'existing-session-id', + protocolVersion: LATEST_PROTOCOL_VERSION + }; + + const client = new Client({ name: 'test client', version: '1.0' }); + await client.connect(resumeTransport); + + // Resumption: no initialize, and the carried version lands on the header + // hook and on the client's own negotiated state. + expect(resumeTransport.send).not.toHaveBeenCalledWith(expect.objectContaining({ method: 'initialize' }), expect.anything()); + expect(setProtocolVersion).toHaveBeenCalledWith(LATEST_PROTOCOL_VERSION); + expect(client.getNegotiatedProtocolVersion()).toBe(LATEST_PROTOCOL_VERSION); +}); + /*** * Test: The negotiated protocol version (and with it the wire era) is connection state — it must * not survive into a fresh connect. A client whose previous connection negotiated the modern From b38fa44463ad13fe7f9807fc3448a117b8acd01d Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Thu, 9 Jul 2026 11:47:50 +0000 Subject: [PATCH 7/9] review: run close() teardown without relying on the transport onclose round-trip Protocol.close() only awaited transport.close(); a transport whose close() resolves without invoking onclose left the instance permanently wedged on the AlreadyConnected guard. close() now runs the teardown itself when the callback has not, and the wrapped onclose gained an identity guard so a deferred onclose from the old transport cannot tear down a replacement connection established afterwards. --- packages/core-internal/src/shared/protocol.ts | 50 +++++++++---------- .../test/shared/protocolConnectGuard.test.ts | 27 ++++++++++ 2 files changed, 52 insertions(+), 25 deletions(-) diff --git a/packages/core-internal/src/shared/protocol.ts b/packages/core-internal/src/shared/protocol.ts index 0d631ae2f7..66900a8313 100644 --- a/packages/core-internal/src/shared/protocol.ts +++ b/packages/core-internal/src/shared/protocol.ts @@ -775,10 +775,8 @@ export abstract class Protocol { } /** - * Throws if this instance is already bound to a transport. Called by - * {@linkcode Protocol.connect} and, earlier, by subclass connect - * overrides whose pre-connect work mutates per-connection state (see - * `Client.connect()`): the check must run before any of that work. + * Throws if already bound to a transport. Subclass connect overrides that + * mutate per-connection state before `super.connect()` must call this first. */ protected assertNotConnected(): void { if (this._transport) { @@ -803,7 +801,12 @@ export abstract class Protocol { try { _onclose?.(); } finally { - this._onclose(); + // Identity guard: close() may have already torn this + // connection down (and a replacement may be live) by the time + // a transport fires a deferred onclose. + if (this._transport === transport) { + this._onclose(); + } } }; @@ -833,14 +836,9 @@ export abstract class Protocol { try { await this._transport.start(); } catch (error) { - // A transport that failed to start was never connected: unwind so - // the instance can retry connect() — the reuse guard above would - // otherwise lock the instance onto a transport that never carried - // traffic. Restore the transport's own callbacks too: leaving the - // wrapped closures installed would let a late event from the - // failed transport (e.g. a child process 'close' after a failed - // spawn) tear down whatever connection a successful retry - // established. + // Unwind a failed start so connect() can be retried: restore the + // transport's own callbacks — a late event from the failed + // transport must not tear down a successful retry's connection. transport.onclose = _onclose; transport.onerror = _onerror; transport.onmessage = _onmessage; @@ -1064,12 +1062,9 @@ export abstract class Protocol { // sender (the per-request/instance asymmetry is deliberately gone): // the codec is resolved at send time from the connection state. // - // Once this request's handler has been aborted (connection closed or - // request cancelled), its related sends must not reach the transport: - // `this._transport` is read live at send time, so a handler still - // running after close() could otherwise write onto a transport - // connected later. Notifications no-op; requests reject with - // ConnectionClosed. + // After the handler aborts, related sends must not reach the transport + // (read live at send time, it may belong to a later connection): + // notifications no-op, requests reject with ConnectionClosed. const sendNotification = (notification: Notification, options?: NotificationOptions): Promise => { if (abortController.signal.aborted) { return Promise.resolve(); @@ -1123,11 +1118,9 @@ export abstract class Protocol { // that overloaded property type. The cast is sound: this impl dispatches both overload paths via the // isStandardSchema guard, and sendRequest validates the result against the resolved schema either way. send: ((r: Request, schemaOrOptions?: StandardSchemaV1 | RequestOptions, maybeOptions?: RequestOptions) => { - // The abort guard comes FIRST — before era/codec - // resolution — so an aborted handler always gets the - // documented ConnectionClosed rejection, never a - // synchronous era throw evaluated against whatever - // connection replaced this request's. + // Abort guard FIRST, before era/codec resolution: an + // aborted handler gets ConnectionClosed, never an era + // throw evaluated against a replacement connection. if (abortController.signal.aborted) { return Promise.reject(new SdkError(SdkErrorCode.ConnectionClosed, 'Request was cancelled')); } @@ -1282,7 +1275,14 @@ export abstract class Protocol { * Closes the connection. */ async close(): Promise { - await this._transport?.close(); + const transport = this._transport; + await transport?.close(); + // A transport whose close() resolves without firing onclose would + // otherwise leave the instance wedged (still "connected" per the + // connect() guard). Run the teardown ourselves if it hasn't happened. + if (transport !== undefined && this._transport === transport) { + this._onclose(); + } } /** diff --git a/packages/core-internal/test/shared/protocolConnectGuard.test.ts b/packages/core-internal/test/shared/protocolConnectGuard.test.ts index 08c2871098..9dd299dcba 100644 --- a/packages/core-internal/test/shared/protocolConnectGuard.test.ts +++ b/packages/core-internal/test/shared/protocolConnectGuard.test.ts @@ -129,6 +129,33 @@ describe('Protocol.connect() reuse guard', () => { expect(transportB.sentMessages[0]).toMatchObject({ jsonrpc: '2.0', id: 1, result: {} }); }); + test('close() frees the instance even when the transport never fires onclose', async () => { + const protocol = createProtocol(); + const silent = new MockTransport('silent'); + // Some transports resolve close() without invoking onclose; + // Protocol.close() must not depend on that round-trip. + silent.close = async (): Promise => {}; + + await protocol.connect(silent); + await protocol.close(); + expect(protocol.transport).toBeUndefined(); + + const transportB = new MockTransport('B'); + await protocol.connect(transportB); + expect(protocol.transport).toBe(transportB); + + // A deferred onclose from the old transport (e.g. a stdio child's + // 'close' event arriving after close() resolved) must not tear down + // the replacement connection or re-fire the user-facing onclose. + let sawClose = false; + protocol.onclose = () => { + sawClose = true; + }; + silent.onclose?.(); + expect(protocol.transport).toBe(transportB); + expect(sawClose).toBe(false); + }); + test('transport-initiated close also frees the instance for a new connect()', async () => { const protocol = createProtocol(); const transportA = new MockTransport('A'); From 9aaea878e79b482659a8e48ef4d189d60f012834 Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Thu, 9 Jul 2026 11:47:54 +0000 Subject: [PATCH 8/9] review: version the node package, complete the error-code table, tighten comments MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - changeset: add @modelcontextprotocol/node — the call-site re-raise changed its runtime behavior, and server is only a peer/dev dependency so the adapter would not release otherwise - migration guide: StatelessTransportReuse row in the SdkErrorCode table; close-then-resume example carrying { sessionId, protocolVersion } - reattach the handleRequest JSDoc that the inserted assertNotReused() had orphaned - trim mechanism-inventory comments down to their load-bearing lines --- .changeset/stateless-transport-reuse-guard.md | 1 + docs/migration/upgrade-to-v2.md | 11 +++++++++ .../node/src/streamableHttp.examples.ts | 5 ++-- .../middleware/node/src/streamableHttp.ts | 24 ++++++------------- packages/server/src/server/server.ts | 7 ++---- .../src/server/streamableHttp.examples.ts | 5 ++-- packages/server/src/server/streamableHttp.ts | 22 +++++++---------- 7 files changed, 34 insertions(+), 41 deletions(-) diff --git a/.changeset/stateless-transport-reuse-guard.md b/.changeset/stateless-transport-reuse-guard.md index 7335ac572b..18b99b1157 100644 --- a/.changeset/stateless-transport-reuse-guard.md +++ b/.changeset/stateless-transport-reuse-guard.md @@ -2,6 +2,7 @@ '@modelcontextprotocol/server': patch '@modelcontextprotocol/core-internal': minor '@modelcontextprotocol/client': patch +'@modelcontextprotocol/node': patch --- Restores two v1 transport lifecycle invariants dropped in the v2 rewrite. Both throw typed errors — `SdkError` with `SdkErrorCode.StatelessTransportReuse` (new) and `SdkErrorCode.AlreadyConnected` — while keeping the message strings byte-identical to v1.x, so structured handling can use the code and existing message matching keeps working: diff --git a/docs/migration/upgrade-to-v2.md b/docs/migration/upgrade-to-v2.md index 1340ca1300..d9a71f694d 100644 --- a/docs/migration/upgrade-to-v2.md +++ b/docs/migration/upgrade-to-v2.md @@ -986,6 +986,7 @@ the third argument — `new SdkHttpError(SdkErrorCode.ClientHttpNotImplemented, | ------------------------------------- | -------------------------------------------------------------------------- | | `NotConnected` | Transport is not connected | | `AlreadyConnected` | Transport is already connected | +| `StatelessTransportReuse` | Stateless transport reused across requests | | `NotInitialized` | Protocol is not initialized | | `CapabilityNotSupported` | Required capability is not supported | | `RequestTimeout` | Request timed out waiting for response | @@ -1649,6 +1650,16 @@ to a new transport, or use a separate Protocol instance per connection."`) inste silently rebinding `this._transport`. Sequential `close()` then `connect()` keeps working. `Client.connect()` performs the same check up front, before any per-connection state is reset. + To resume a Streamable HTTP session across that sequence, carry both the session ID + and the negotiated protocol version onto the new transport — `close()` wipes the + client's copy, and the resuming connect reads it back from the transport: + + ```typescript + const sessionId = transport.sessionId; + const protocolVersion = client.getNegotiatedProtocolVersion(); + await client.close(); + await client.connect(new StreamableHTTPClientTransport(url, { sessionId, protocolVersion })); + ``` - **Aborted request handlers cannot write to a replacement transport.** After the handler's `ctx.mcpReq.signal` aborts (connection closed or request cancelled), diff --git a/packages/middleware/node/src/streamableHttp.examples.ts b/packages/middleware/node/src/streamableHttp.examples.ts index 0329fdd957..cd50bedad0 100644 --- a/packages/middleware/node/src/streamableHttp.examples.ts +++ b/packages/middleware/node/src/streamableHttp.examples.ts @@ -38,9 +38,8 @@ declare const serverResponse: ServerResponse; */ async function NodeStreamableHTTPServerTransport_stateless() { //#region NodeStreamableHTTPServerTransport_stateless - // A stateless transport serves exactly one request: construct a fresh - // transport + server pair per request — reusing a stateless transport - // across requests throws. + // A stateless transport serves exactly one request — reuse throws. + // Construct a fresh transport + server pair per request. const server = new McpServer({ name: 'my-server', version: '1.0.0' }); const transport = new NodeStreamableHTTPServerTransport({ diff --git a/packages/middleware/node/src/streamableHttp.ts b/packages/middleware/node/src/streamableHttp.ts index 605334b31b..c560b4e75a 100644 --- a/packages/middleware/node/src/streamableHttp.ts +++ b/packages/middleware/node/src/streamableHttp.ts @@ -62,9 +62,8 @@ export type StreamableHTTPServerTransportOptions = WebStandardStreamableHTTPServ * * @example Stateless setup * ```ts source="./streamableHttp.examples.ts#NodeStreamableHTTPServerTransport_stateless" - * // A stateless transport serves exactly one request: construct a fresh - * // transport + server pair per request — reusing a stateless transport - * // across requests throws. + * // A stateless transport serves exactly one request — reuse throws. + * // Construct a fresh transport + server pair per request. * const server = new McpServer({ name: 'my-server', version: '1.0.0' }); * * const transport = new NodeStreamableHTTPServerTransport({ @@ -196,11 +195,8 @@ export class NodeStreamableHTTPServerTransport implements Transport { * @param parsedBody - Optional pre-parsed body from body-parser middleware */ async handleRequest(req: IncomingMessage & { auth?: AuthInfo }, res: ServerResponse, parsedBody?: unknown): Promise { - // Fail fast before delegating: once getRequestListener is invoked, a - // rejected dispatch is committed to `res` as an empty 500 before the - // rejection can surface, leaving the caller unable to shape its own - // error response. Checking up front rejects with nothing written to - // `res`. The check lives on the wrapped transport (single source). + // Check before delegating: getRequestListener may swallow a + // mid-dispatch rejection after bytes are written to `res`. this._webStandardTransport.assertNotReused(); // Store context for this request to pass through auth and parsedBody @@ -230,15 +226,9 @@ export class NodeStreamableHTTPServerTransport implements Transport { // including proper SSE streaming support await handler(req, res); - // Backstop for concurrent calls racing the up-front check (the - // single-use flag is set inside the wrapped handleRequest, after the - // request conversion): getRequestListener absorbs a rejected dispatch - // into a generic 500, which would swallow the single-use throw. - // Re-raise it so the documented behavior (reusing a stateless - // transport across requests throws at the call site) holds on this - // path too. Other dispatch errors keep the existing 500-response - // behavior. Matched by code, not `instanceof`, so the check also - // holds if bundling ever yields two copies of the error class. + // Backstop for concurrent calls racing the up-front check. + // getRequestListener absorbs rejections into a 500; re-raise the + // single-use throw (matched by code, robust to dual bundling). if (dispatchError instanceof Error && (dispatchError as { code?: unknown }).code === SdkErrorCode.StatelessTransportReuse) { throw dispatchError; } diff --git a/packages/server/src/server/server.ts b/packages/server/src/server/server.ts index bc96717fa5..661963f4ea 100644 --- a/packages/server/src/server/server.ts +++ b/packages/server/src/server/server.ts @@ -400,11 +400,8 @@ export class Server extends Protocol { // session-wide stream to deliver it on. return ctx.mcpReq.notify({ method: 'notifications/message', params: { level, data, logger } }); }, - // Same abort gate as `ctx.mcpReq.send`: once this request's - // handler has been aborted (connection closed or request - // cancelled), its per-request senders must not reach the - // transport, which is read live at send time and may belong - // to a later connection. + // Same abort gate as `ctx.mcpReq.send`: an aborted handler's + // senders must not reach a possibly-replaced transport. elicitInput: (params, options) => { if (ctx.mcpReq.signal.aborted) { return Promise.reject(new SdkError(SdkErrorCode.ConnectionClosed, 'Request was cancelled')); diff --git a/packages/server/src/server/streamableHttp.examples.ts b/packages/server/src/server/streamableHttp.examples.ts index 9d3ebdd740..845dffdefb 100644 --- a/packages/server/src/server/streamableHttp.examples.ts +++ b/packages/server/src/server/streamableHttp.examples.ts @@ -30,9 +30,8 @@ async function WebStandardStreamableHTTPServerTransport_stateful() { */ async function WebStandardStreamableHTTPServerTransport_stateless(request: Request) { //#region WebStandardStreamableHTTPServerTransport_stateless - // A stateless transport serves exactly one request: construct a fresh - // transport + server pair per request — reusing a stateless transport - // across requests throws. + // A stateless transport serves exactly one request — reuse throws. + // Construct a fresh transport + server pair per request. const server = new McpServer({ name: 'my-server', version: '1.0.0' }); const transport = new WebStandardStreamableHTTPServerTransport({ diff --git a/packages/server/src/server/streamableHttp.ts b/packages/server/src/server/streamableHttp.ts index 9bf4ad042d..3b00b262e2 100644 --- a/packages/server/src/server/streamableHttp.ts +++ b/packages/server/src/server/streamableHttp.ts @@ -214,9 +214,8 @@ export interface HandleRequestOptions { * * @example Stateless setup * ```ts source="./streamableHttp.examples.ts#WebStandardStreamableHTTPServerTransport_stateless" - * // A stateless transport serves exactly one request: construct a fresh - * // transport + server pair per request — reusing a stateless transport - * // across requests throws. + * // A stateless transport serves exactly one request — reuse throws. + * // Construct a fresh transport + server pair per request. * const server = new McpServer({ name: 'my-server', version: '1.0.0' }); * * const transport = new WebStandardStreamableHTTPServerTransport({ @@ -372,20 +371,13 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } /** - * Handles an incoming HTTP request, whether `GET`, `POST`, or `DELETE` - * Returns a `Response` object (Web Standard) - */ - /** - * Throws when a stateless transport has already handled its single - * request exchange. This is the same check `handleRequest()` performs on - * entry; exposed so wrapping adapters (e.g. the Node middleware) can fail - * fast before committing anything to their own response stream. + * The single-use check `handleRequest()` performs on entry, exposed so + * wrapping adapters can fail fast before writing to their own response. * * @internal */ assertNotReused(): void { - // In stateless mode (no sessionIdGenerator), each request must use a fresh transport. - // Reusing a stateless transport causes message ID collisions between clients. + // Stateless reuse would collide JSON-RPC message ids across clients. if (!this.sessionIdGenerator && this._hasHandledRequest) { throw new SdkError( SdkErrorCode.StatelessTransportReuse, @@ -394,6 +386,10 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } } + /** + * Handles an incoming HTTP request, whether `GET`, `POST`, or `DELETE` + * Returns a `Response` object (Web Standard) + */ async handleRequest(req: Request, options?: HandleRequestOptions): Promise { this.assertNotReused(); this._hasHandledRequest = true; From 18747af0479503af0da599a9e26574788d3f7f1b Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Thu, 9 Jul 2026 12:24:40 +0000 Subject: [PATCH 9/9] test: pin ctx sender gating for requests cancelled on a live connection --- .../test/server/abortedContextSenders.test.ts | 127 +++++++++++++++--- 1 file changed, 112 insertions(+), 15 deletions(-) diff --git a/packages/server/test/server/abortedContextSenders.test.ts b/packages/server/test/server/abortedContextSenders.test.ts index 21c1f83e0f..f7de115aab 100644 --- a/packages/server/test/server/abortedContextSenders.test.ts +++ b/packages/server/test/server/abortedContextSenders.test.ts @@ -1,10 +1,14 @@ /** - * After close() aborts an in-flight request handler, the ServerContext + * After an in-flight request handler is aborted, the ServerContext * per-request senders (`ctx.mcpReq.elicitInput`, `ctx.mcpReq.requestSampling`) * reject with `ConnectionClosed` instead of reaching a transport — the same * gate `ctx.mcpReq.notify` / `ctx.mcpReq.send` apply. The transport reference * is read live at send time, so without the gate an aborted handler could * write to a transport connected after the close. + * + * The abort has two triggers, each pinned below: close() tearing the + * connection down, and the peer cancelling one request via + * `notifications/cancelled` while the connection stays open. */ import type { JSONRPCMessage } from '@modelcontextprotocol/core-internal'; import { InMemoryTransport, SdkError, SdkErrorCode } from '@modelcontextprotocol/core-internal'; @@ -12,6 +16,24 @@ import { describe, expect, it } from 'vitest'; import { Server } from '../../src/server/server'; +/** Connects, runs the initialize handshake, and issues the held-open `tools/call` (id 1, name 'hold'). */ +async function connectAndHoldToolCall(server: Server, serverTx: InMemoryTransport, peerTx: InMemoryTransport): Promise { + await server.connect(serverTx); + await peerTx.start(); + await peerTx.send({ + jsonrpc: '2.0', + id: 0, + method: 'initialize', + params: { + protocolVersion: '2025-11-25', + capabilities: { elicitation: {}, sampling: {} }, + clientInfo: { name: 'peer', version: '0' } + } + }); + await peerTx.send({ jsonrpc: '2.0', method: 'notifications/initialized' }); + await peerTx.send({ jsonrpc: '2.0', id: 1, method: 'tools/call', params: { name: 'hold', arguments: {} } }); +} + describe('aborted-handler context senders', () => { it('elicitInput and requestSampling reject with ConnectionClosed after close(), and never reach a later connection', async () => { const server = new Server({ name: 'abort-gate-test', version: '1.0.0' }, { capabilities: { tools: {} } }); @@ -40,20 +62,7 @@ describe('aborted-handler context senders', () => { return { content: [] }; }); - await server.connect(serverTx); - await peerTx.start(); - await peerTx.send({ - jsonrpc: '2.0', - id: 0, - method: 'initialize', - params: { - protocolVersion: '2025-11-25', - capabilities: { elicitation: {}, sampling: {} }, - clientInfo: { name: 'peer', version: '0' } - } - }); - await peerTx.send({ jsonrpc: '2.0', method: 'notifications/initialized' }); - await peerTx.send({ jsonrpc: '2.0', id: 1, method: 'tools/call', params: { name: 'hold', arguments: {} } }); + await connectAndHoldToolCall(server, serverTx, peerTx); await entered; await server.close(); @@ -79,4 +88,92 @@ describe('aborted-handler context senders', () => { await server.close(); }); + + it('gates all ctx senders after notifications/cancelled while the connection stays open, and still serves fresh requests', async () => { + const server = new Server({ name: 'abort-gate-test', version: '1.0.0' }, { capabilities: { tools: {}, logging: {} } }); + const [peerTx, serverTx] = InMemoryTransport.createLinkedPair(); + + const peerMessages: JSONRPCMessage[] = []; + peerTx.onmessage = message => { + peerMessages.push(message as JSONRPCMessage); + }; + + let notifyCall!: () => Promise; + let sendCall!: () => Promise; + let elicitCall!: () => Promise; + let samplingCall!: () => Promise; + let aborted!: Promise; + let handlerEntered!: () => void; + const entered = new Promise(resolve => { + handlerEntered = resolve; + }); + server.setRequestHandler('tools/call', async (request, ctx) => { + if (request.params?.name !== 'hold') { + return { content: [] }; + } + notifyCall = () => ctx.mcpReq.notify({ method: 'notifications/message', params: { level: 'info', data: 'late' } }); + sendCall = () => ctx.mcpReq.send({ method: 'ping' }); + elicitCall = () => + ctx.mcpReq.elicitInput({ + message: 'pick one', + requestedSchema: { type: 'object', properties: {} } + }); + samplingCall = () => + ctx.mcpReq.requestSampling({ + messages: [], + maxTokens: 1 + }); + aborted = new Promise(resolve => { + ctx.mcpReq.signal.addEventListener('abort', () => resolve(), { once: true }); + }); + handlerEntered(); + // Held open across the cancellation; never resolves. + await new Promise(() => {}); + return { content: [] }; + }); + + await connectAndHoldToolCall(server, serverTx, peerTx); + await entered; + + // The peer cancels just that request; the connection stays open. + await peerTx.send({ + jsonrpc: '2.0', + method: 'notifications/cancelled', + params: { requestId: 1, reason: 'peer changed its mind' } + }); + await aborted; + + const wireCountAfterCancel = peerMessages.length; + + // Notifications no-op without reaching the wire. + await expect(notifyCall()).resolves.toBeUndefined(); + + // Requests reject with the gate's exact error, without reaching the wire. + for (const call of [sendCall, elicitCall, samplingCall]) { + const rejection = await call().then( + () => undefined, + (error: unknown) => error + ); + expect(rejection).toBeInstanceOf(SdkError); + expect((rejection as SdkError).code).toBe(SdkErrorCode.ConnectionClosed); + expect((rejection as SdkError).message).toBe('Request was cancelled'); + } + expect(peerMessages).toHaveLength(wireCountAfterCancel); + + // The live connection still serves a fresh request end to end. + // (peerMessages has no assertions past this point, so the collector + // can be replaced outright.) + const freshResponse = new Promise(resolve => { + peerTx.onmessage = message => { + const received = message as JSONRPCMessage; + if ('id' in received && received.id === 2) { + resolve(received); + } + }; + }); + await peerTx.send({ jsonrpc: '2.0', id: 2, method: 'tools/call', params: { name: 'quick', arguments: {} } }); + expect(await freshResponse).toMatchObject({ id: 2, result: { content: [] } }); + + await server.close(); + }); });