diff --git a/.changeset/stateless-transport-reuse-guard.md b/.changeset/stateless-transport-reuse-guard.md new file mode 100644 index 0000000000..18b99b1157 --- /dev/null +++ b/.changeset/stateless-transport-reuse-guard.md @@ -0,0 +1,13 @@ +--- +'@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: + +- 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()`, `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 89188aa893..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 | @@ -1616,6 +1617,56 @@ 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) + +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 + `"Stateless transport cannot be reused across requests. Create a new transport per +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** (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. + 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), + `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) - `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..6746f63820 100644 --- a/packages/client/src/client/client.ts +++ b/packages/client/src/client/client.ts @@ -927,6 +927,9 @@ export class Client extends Protocol { * ``` */ override async connect(transport: Transport, options?: ConnectOptions): Promise { + // 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 // DiscoverResult: bypasses versionNegotiation resolution entirely. @@ -938,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); } @@ -1046,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/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/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/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 ffab6e8df8..66900a8313 100644 --- a/packages/core-internal/src/shared/protocol.ts +++ b/packages/core-internal/src/shared/protocol.ts @@ -774,19 +774,39 @@ export abstract class Protocol { } } + /** + * 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) { + 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.' + ); + } + } + /** * 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; this._transport.onclose = () => { 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(); + } } }; @@ -813,7 +833,18 @@ 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) { + // 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; + this._transport = undefined; + throw error; + } } /** @@ -1024,22 +1055,38 @@ 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, { + // + // 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(); + } + 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 +1118,12 @@ 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) => { + // 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')); + } // Related requests resolve through the instance era at // send time, exactly like direct sends: era-gate first, // then method-keyed schema resolution. @@ -1222,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/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/packages/core-internal/test/shared/protocolConnectGuard.test.ts b/packages/core-internal/test/shared/protocolConnectGuard.test.ts new file mode 100644 index 0000000000..9dd299dcba --- /dev/null +++ b/packages/core-internal/test/shared/protocolConnectGuard.test.ts @@ -0,0 +1,251 @@ +/** + * 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('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'); + const transportB = new MockTransport('B'); + + await protocol.connect(transportA); + + 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.' + ); + + // 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('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'); + 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/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/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..cd50bedad0 100644 --- a/packages/middleware/node/src/streamableHttp.examples.ts +++ b/packages/middleware/node/src/streamableHttp.examples.ts @@ -29,16 +29,26 @@ 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 — 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({ sessionIdGenerator: undefined }); + + await server.connect(transport); + await transport.handleRequest(incomingRequest, serverResponse); //#endregion NodeStreamableHTTPServerTransport_stateless - return transport; } // Stubs for Express-style app @@ -47,10 +57,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..c560b4e75a 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} @@ -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,26 @@ export type StreamableHTTPServerTransportOptions = WebStandardStreamableHTTPServ * * @example Stateless setup * ```ts source="./streamableHttp.examples.ts#NodeStreamableHTTPServerTransport_stateless" + * // 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({ * 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); * }); * ``` */ @@ -180,6 +195,10 @@ 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 { + // 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 // We need to intercept the request creation to attach this context const authInfo = req.auth; @@ -187,12 +206,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 } ); @@ -200,6 +225,13 @@ 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); + + // 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/middleware/node/test/streamableHttp.test.ts b/packages/middleware/node/test/streamableHttp.test.ts index 140717c6bb..524e309fad 100644 --- a/packages/middleware/node/test/streamableHttp.test.ts +++ b/packages/middleware/node/test/streamableHttp.test.ts @@ -1599,18 +1599,102 @@ 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('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 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(); + } + }); + 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'); + 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(); + } }); it('should operate without session ID validation', async () => { @@ -1622,6 +1706,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 +1740,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 +1757,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 +1766,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 +3027,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 +3055,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/server.ts b/packages/server/src/server/server.ts index 4151849494..661963f4ea 100644 --- a/packages/server/src/server/server.ts +++ b/packages/server/src/server/server.ts @@ -400,8 +400,20 @@ 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`: 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')); + } + 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/src/server/streamableHttp.examples.ts b/packages/server/src/server/streamableHttp.examples.ts index 678c43c928..845dffdefb 100644 --- a/packages/server/src/server/streamableHttp.examples.ts +++ b/packages/server/src/server/streamableHttp.examples.ts @@ -28,13 +28,20 @@ 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 — 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({ 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 +50,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 +65,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..3b00b262e2 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'; @@ -193,6 +195,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 +214,25 @@ export interface HandleRequestOptions { * * @example Stateless setup * ```ts source="./streamableHttp.examples.ts#WebStandardStreamableHTTPServerTransport_stateless" + * // 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({ * 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 +241,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 +254,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(); @@ -347,11 +370,30 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { return undefined; } + /** + * The single-use check `handleRequest()` performs on entry, exposed so + * wrapping adapters can fail fast before writing to their own response. + * + * @internal + */ + assertNotReused(): void { + // Stateless reuse would collide JSON-RPC message ids across clients. + if (!this.sessionIdGenerator && this._hasHandledRequest) { + throw new SdkError( + SdkErrorCode.StatelessTransportReuse, + 'Stateless transport cannot be reused across requests. Create a new transport per request.' + ); + } + } + /** * 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; + // Validate request headers for DNS rebinding protection const validationError = this.validateRequestHeaders(req); if (validationError) { diff --git a/packages/server/test/server/abortedContextSenders.test.ts b/packages/server/test/server/abortedContextSenders.test.ts new file mode 100644 index 0000000000..f7de115aab --- /dev/null +++ b/packages/server/test/server/abortedContextSenders.test.ts @@ -0,0 +1,179 @@ +/** + * 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'; +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: {} } }); + 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 connectAndHoldToolCall(server, serverTx, peerTx); + 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(); + }); + + 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(); + }); +}); 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..0752f99a82 --- /dev/null +++ b/packages/server/test/server/statelessReuseGuard.test.ts @@ -0,0 +1,232 @@ +/** + * 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 { SdkError, SdkErrorCode } 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); + + 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 () => { + 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..bf2c3f10bb 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. @@ -172,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 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;