diff --git a/.changeset/export-classify-entry-request.md b/.changeset/export-classify-entry-request.md new file mode 100644 index 0000000000..a72828094a --- /dev/null +++ b/.changeset/export-classify-entry-request.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/server': minor +--- + +Export `classifyEntryRequest` (with its `EntryClassification` outcome type) from `@modelcontextprotocol/server`: the Request-shaped sibling of `classifyInboundRequest`, and the single classification step already behind `createMcpHandler` and `isLegacyRequest`. It takes the web-standard `Request`, extracts the HTTP method and the `MCP-Protocol-Version` / `Mcp-Method` / `Mcp-Name` headers, performs the entry's single body read (distinguishing an unreadable stream from a non-JSON body), and returns the full routing outcome plus a body-preserving `forwardRequest` — so hybrid deployments that need the routing reason (for example: route the legacy `initialize` handshake to a sessionful host, everything else to the stateless handler) no longer hand-assemble the classifier's fields or lose the reason to the boolean `isLegacyRequest`. diff --git a/.changeset/refused-handshake-fires-close.md b/.changeset/refused-handshake-fires-close.md new file mode 100644 index 0000000000..eaecbf3809 --- /dev/null +++ b/.changeset/refused-handshake-fires-close.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/server': minor +--- + +Add a `closeOnRefusedHandshake` option (default `false`) to `WebStandardStreamableHTTPServerTransport` for transports created per handshake attempt — the session-map recipe, where a fresh transport and connected server pair serves one expected `initialize`. When enabled, a first completed request that fails to establish the session — a refused or throwing handshake, a pre-session `GET`/`DELETE`, an unsupported method — schedules the transport's close chain behind its response, so `onclose` fires and the paired server tears down instead of leaking. The close defers while another request is still in flight and re-checks before running, so a refusal settling next to an in-flight `initialize` never tears it down. The default stays off: long-lived sessionful endpoints must answer pre-session refusals (for example the SDK client's version-negotiation probe) without ending the transport, and stateless transports never close on refusals. diff --git a/docs/serving/legacy-clients.md b/docs/serving/legacy-clients.md index 62980e01cc..7da0d784b1 100644 --- a/docs/serving/legacy-clients.md +++ b/docs/serving/legacy-clients.md @@ -70,7 +70,7 @@ async function serve(request: Request): Promise { } ``` -`legacyStatelessFallback(factory)` is the entry's default legacy serving as a standalone handler — it holds the legacy leg's place here. Put your existing wiring there instead and it keeps its sessions, its event store, and its clients: [`legacy-routing/server.ts`](https://github.com/modelcontextprotocol/typescript-sdk/blob/main/examples/legacy-routing/server.ts) runs a sessionful `StreamableHTTPServerTransport` deployment behind this exact branch. Route every `false` to the strict handler — the modern path owns the error answers for malformed modern requests. +`legacyStatelessFallback(factory)` is the entry's default legacy serving as a standalone handler — it holds the legacy leg's place here. Put your existing wiring there instead and it keeps its sessions, its event store, and its clients: [`legacy-routing/server.ts`](https://github.com/modelcontextprotocol/typescript-sdk/blob/main/examples/legacy-routing/server.ts) runs a sessionful `StreamableHTTPServerTransport` deployment behind this exact branch. Route every `false` to the strict handler — the modern path owns the error answers for malformed modern requests. When the branch needs the routing reason rather than a boolean — say, sending only the legacy `initialize` handshake to a session host and everything else to the stateless handler — `classifyEntryRequest` is the same classification step returning the full outcome (`classified.outcome.reason === 'initialize'`) plus a body-preserving `forwardRequest` to hand to whichever handler wins. The `initialize` the strict handler rejected above now completes the 2025 handshake on the legacy leg: diff --git a/docs/serving/sessions-state-scaling.md b/docs/serving/sessions-state-scaling.md index 0bbffe450a..119ddb4d80 100644 --- a/docs/serving/sessions-state-scaling.md +++ b/docs/serving/sessions-state-scaling.md @@ -34,6 +34,10 @@ const route = async (req: Request, res: Response) => { if (!sessionId && isInitializeRequest(req.body)) { const transport = new NodeStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), + // This pair exists for one expected handshake: if the + // initialize is refused, close the transport (onclose below + // runs) so the just-connected server never leaks. + closeOnRefusedHandshake: true, onsessioninitialized: id => { sessions.set(id, transport); } @@ -59,7 +63,7 @@ app.get('/mcp', route); app.delete('/mcp', route); ``` -The map cleans itself up: `transport.onclose` fires when the session ends, whether the client sent `DELETE` or you called `transport.close()`. A request with an unknown `Mcp-Session-Id` gets the `404` above, which tells the client to start a new session; a request with no session header at all gets the `400`, which tells it to re-send the id it already has instead of re-initializing. +The map cleans itself up: `transport.onclose` fires when the session ends, whether the client sent `DELETE` or you called `transport.close()`. `closeOnRefusedHandshake` extends that to the handshake itself — each transport here exists for one expected `initialize`, so if that first request is refused instead (a malformed body, a missing `Accept` header), the transport closes and `onclose` runs, and the server connected just above never leaks. Leave the option off for a sessionful transport that serves an endpoint long-term, where pre-session refusals (a client's version-negotiation probe, say) must not end it. A request with an unknown `Mcp-Session-Id` gets the `404` above, which tells the client to start a new session; a request with no session header at all gets the `400`, which tells it to re-send the id it already has instead of re-initializing. ::: tip On shutdown, close every stored transport — `for (const [, transport] of sessions) await transport.close()` — before exiting; `close()` ends the session's SSE streams and rejects its pending requests. diff --git a/examples/elicitation/README.md b/examples/elicitation/README.md index de3443bd68..6a2e4d8e1c 100644 --- a/examples/elicitation/README.md +++ b/examples/elicitation/README.md @@ -11,7 +11,7 @@ Server requests user input. One factory, both protocol eras: elicitation works o `plan_trip` chains **two** form elicitations inside one tool call (destination → dates for that destination): two sequential `ctx.mcpReq.elicitInput` pushes on 2025, two `inputRequired` rounds with `requestState` carry-over on 2026. The `register_user` form schema includes an `enumNames` field (display labels for the `plan` enum). For the secure `requestState` round-trip pattern see [`../mrtr/`](../mrtr/README.md). -Runs all four transport/era legs: `server.ts` inlines a sessionful `NodeStreamableHTTPServerTransport` arm for 2025 traffic (the same `isLegacyRequest` composition `../legacy-routing/` shows by hand), so push server→client requests reach the client over either transport. +Runs all four transport/era legs: `server.ts` inlines a sessionful `NodeStreamableHTTPServerTransport` arm for 2025 traffic (the `classifyEntryRequest` flavor of the routing composition `../legacy-routing/` shows with `isLegacyRequest` — one classification step also yields the parsed body both arms need), so push server→client requests reach the client over either transport. ```bash pnpm --filter @mcp-examples/elicitation client # 2026-07-28 (inputRequired) diff --git a/examples/elicitation/server.ts b/examples/elicitation/server.ts index 37808e9d21..0df82db031 100644 --- a/examples/elicitation/server.ts +++ b/examples/elicitation/server.ts @@ -33,10 +33,10 @@ import type { } from '@modelcontextprotocol/server'; import { acceptedContent, + classifyEntryRequest, createMcpHandler, inputRequired, isInitializeRequest, - isLegacyRequest, McpServer, UrlElicitationRequiredError } from '@modelcontextprotocol/server'; @@ -254,6 +254,9 @@ if (transport === 'stdio') { } else if (!sid && isInitializeRequest(body)) { const t = new NodeStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), + // This pair serves one expected handshake: a refused initialize + // closes the transport (onclose below runs) instead of leaking it. + closeOnRefusedHandshake: true, onsessioninitialized: id => { sessions.set(id, t); } @@ -276,14 +279,18 @@ if (transport === 'stdio') { createServer((req, res) => { void (async () => { - // `toWebRequest` reads the Node body into a web-standard `Request`, - // so the body now lives in `request`, not `req`. Ask the predicate - // first — it classifies an internal clone, leaving `request` - // readable for the `.json()` both arms need (reading `.json()` - // first would make the predicate's internal clone throw). + // `toWebRequest` reads the Node body into a web-standard `Request`. + // One classification step reads it exactly once and returns the + // routing outcome together with the parsed body both arms need — + // no second `.json()` read, no clone-ordering pitfall. const request = await toWebRequest(req); - const legacy = await isLegacyRequest(request); - const body: unknown = req.method === 'POST' ? await request.json().catch(() => {}) : undefined; + const classified = await classifyEntryRequest(request); + if (classified.step === 'unreadable-body') { + res.writeHead(400).end(); + return; + } + const legacy = classified.step === 'no-json-body' || classified.outcome.kind === 'legacy'; + const body: unknown = classified.step === 'classified' ? classified.parsedBody : undefined; await (legacy ? handleLegacy(req, res, body) : modern(req, res, body)); })().catch(error => { console.error('[server] request error:', error instanceof Error ? error.message : error); diff --git a/examples/guides/serving/sessions-state-scaling.examples.ts b/examples/guides/serving/sessions-state-scaling.examples.ts index 9d9ef5ba41..e4f2adb869 100644 --- a/examples/guides/serving/sessions-state-scaling.examples.ts +++ b/examples/guides/serving/sessions-state-scaling.examples.ts @@ -46,6 +46,10 @@ function sessions_routing(app: Express, buildServer: () => McpServer) { if (!sessionId && isInitializeRequest(req.body)) { const transport = new NodeStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), + // This pair exists for one expected handshake: if the + // initialize is refused, close the transport (onclose below + // runs) so the just-connected server never leaks. + closeOnRefusedHandshake: true, onsessioninitialized: id => { sessions.set(id, transport); } diff --git a/examples/legacy-routing/server.ts b/examples/legacy-routing/server.ts index 33547b2384..40dffddd3b 100644 --- a/examples/legacy-routing/server.ts +++ b/examples/legacy-routing/server.ts @@ -39,6 +39,9 @@ const handleLegacy = async (req: Request, res: Response) => { } else if (!sid && isInitializeRequest(req.body)) { const transport = new NodeStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), + // This pair serves one expected handshake: a refused initialize + // closes the transport (onclose below runs) instead of leaking it. + closeOnRefusedHandshake: true, onsessioninitialized: id => { sessions.set(id, transport); } diff --git a/examples/repl/server.ts b/examples/repl/server.ts index 33f70bf631..d9d6abd7eb 100644 --- a/examples/repl/server.ts +++ b/examples/repl/server.ts @@ -267,6 +267,9 @@ app.all('/mcp', async (req: Request, res: Response) => { const transport = new NodeStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), eventStore, // resumability — events are persisted for replay on GET reconnect + // This pair serves one expected handshake: a refused initialize + // closes the transport (onclose below runs) instead of leaking it. + closeOnRefusedHandshake: true, onsessioninitialized: id => { sessions.set(id, transport); } diff --git a/examples/sse-polling/server.ts b/examples/sse-polling/server.ts index dbc011a84b..3cc2479159 100644 --- a/examples/sse-polling/server.ts +++ b/examples/sse-polling/server.ts @@ -107,6 +107,9 @@ app.all('/mcp', async (req: Request, res: Response) => { sessionIdGenerator: () => randomUUID(), eventStore, retryInterval: 300, // Default retry interval for priming events + // This pair serves one expected handshake: a refused initialize + // closes the transport (onclose below runs) instead of leaking it. + closeOnRefusedHandshake: true, onsessioninitialized: id => { console.error(`[${id}] Session initialized`); transports.set(id, transport); diff --git a/examples/standalone-get/server.ts b/examples/standalone-get/server.ts index 5852cb40e8..738786e389 100644 --- a/examples/standalone-get/server.ts +++ b/examples/standalone-get/server.ts @@ -65,6 +65,9 @@ app.post('/mcp', async (req: Request, res: Response) => { } else if (!sid && isInitializeRequest(req.body)) { const transport = new NodeStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), + // This pair serves one expected handshake: a refused initialize + // closes the transport (onclose below runs) instead of leaking it. + closeOnRefusedHandshake: true, onsessioninitialized: id => { sessions.set(id, transport); } diff --git a/packages/server/src/index.ts b/packages/server/src/index.ts index 1c48319bc0..3deb32d0bc 100644 --- a/packages/server/src/index.ts +++ b/packages/server/src/index.ts @@ -10,13 +10,14 @@ export type { CompletableSchema, CompleteCallback } from './server/completable'; export { completable, isCompletable } from './server/completable'; export type { CreateMcpHandlerOptions, + EntryClassification, LegacyHttpHandler, McpHandlerRequestOptions, McpHttpHandler, McpRequestContext, McpServerFactory } from './server/createMcpHandler'; -export { createMcpHandler, isLegacyRequest, legacyStatelessFallback } from './server/createMcpHandler'; +export { classifyEntryRequest, createMcpHandler, isLegacyRequest, legacyStatelessFallback } from './server/createMcpHandler'; export type { AnyToolHandler, BaseToolCallback, @@ -69,6 +70,7 @@ export { fromJsonSchema } from './fromJsonSchema'; // Inbound HTTP request classification (dual-era serving): the body-primary era // predicate used by createMcpHandler, exported for hand-wired compositions. +// Its Request-shaped sibling, classifyEntryRequest, is exported above. export type { InboundClassificationOutcome, InboundHttpRequest, diff --git a/packages/server/src/server/createMcpHandler.ts b/packages/server/src/server/createMcpHandler.ts index 3f7632d570..4e37fdd46e 100644 --- a/packages/server/src/server/createMcpHandler.ts +++ b/packages/server/src/server/createMcpHandler.ts @@ -389,11 +389,12 @@ export function legacyStatelessFallback(factory: McpServerFactory, onerror?: (er } /* ------------------------------------------------------------------------ * - * The entry's classification step (shared with isLegacyRequest) + * The entry's classification step (shared with isLegacyRequest, exported for + * hand-wired hybrid routing) * ------------------------------------------------------------------------ */ -/** The outcome of the entry's classification step for one inbound HTTP request. */ -type EntryClassification = +/** The outcome of the entry's classification step ({@linkcode classifyEntryRequest}) for one inbound HTTP request. */ +export type EntryClassification = /** The body bytes could not be read at all (a failing stream, not malformed JSON). */ | { step: 'unreadable-body' } /** A POST with an empty or non-JSON body: nothing to classify, so there is no envelope claim. */ @@ -402,17 +403,55 @@ type EntryClassification = | { step: 'classified'; outcome: InboundClassificationOutcome; body: unknown; parsedBody: unknown; forwardRequest: Request }; /** - * The entry's classification step: read the request body exactly once (unless - * a pre-parsed body is supplied) and classify the request with - * {@linkcode classifyInboundRequest}. This is the single code path behind both - * {@linkcode createMcpHandler}'s routing and the exported - * {@linkcode isLegacyRequest} predicate, so the two can never disagree. + * The entry's classification step, taking the web-standard `Request` itself — + * the Request-shaped sibling of {@linkcode classifyInboundRequest}. It + * extracts the HTTP method and the `MCP-Protocol-Version` / `Mcp-Method` / + * `Mcp-Name` headers, reads the request body exactly once (unless a + * pre-parsed body is supplied) with the unreadable-stream vs. no-JSON-body + * distinction, and classifies with {@linkcode classifyInboundRequest} — so a + * hand-wired composition never hand-assembles those fields. This is the + * single code path behind both {@linkcode createMcpHandler}'s routing and the + * exported {@linkcode isLegacyRequest} predicate, so the three can never + * disagree. * - * Pass `needsForward: false` when the caller never reads `forwardRequest` — - * the body-preserving clone is then skipped and `forwardRequest` is the - * (consumed) input request. + * Where {@linkcode isLegacyRequest} collapses the decision to a boolean, this + * returns the full routing outcome — for hybrid deployments that need the + * routing reason. The canonical pattern routes the legacy `initialize` + * handshake to a sessionful host and everything else (modern traffic, other + * legacy traffic, rejections) to a stateless or strict handler: + * + * ```ts + * import { classifyEntryRequest, createMcpHandler } from '@modelcontextprotocol/server'; + * + * const stateless = createMcpHandler(factory); + * + * async function serve(request: Request): Promise { + * const classified = await classifyEntryRequest(request); + * if (classified.step === 'unreadable-body') { + * return new Response(null, { status: 400 }); + * } + * if (classified.step === 'classified' && classified.outcome.kind === 'legacy' && classified.outcome.reason === 'initialize') { + * // The 2025 handshake: open a session on the sessionful legacy host. + * return sessionHost(classified.forwardRequest); + * } + * // Hand the already-parsed body along so the entry classifies from the + * // value instead of reading the forwarded clone a second time. + * return stateless.fetch(classified.forwardRequest, classified.step === 'classified' ? { parsedBody: classified.parsedBody } : undefined); + * } + * ``` + * + * For a `POST` the body is read from the request you pass; `forwardRequest` + * is a body-preserving clone (or the input itself for body-less methods and + * pre-parsed bodies) that stays fully readable for whichever handler the + * outcome routes to. Pass `needsForward: false` when the caller never reads + * `forwardRequest` — the body-preserving clone is then skipped and + * `forwardRequest` is the (consumed) input request. */ -async function classifyEntryRequest(request: Request, providedParsedBody?: unknown, needsForward = true): Promise { +export async function classifyEntryRequest( + request: Request, + providedParsedBody?: unknown, + needsForward = true +): Promise { const httpMethod = request.method.toUpperCase(); let body: unknown; diff --git a/packages/server/src/server/streamableHttp.ts b/packages/server/src/server/streamableHttp.ts index 1f2cf94b00..520743ef60 100644 --- a/packages/server/src/server/streamableHttp.ts +++ b/packages/server/src/server/streamableHttp.ts @@ -106,6 +106,29 @@ export interface WebStandardStreamableHTTPServerTransportOptions { */ onsessionclosed?: ((sessionId: string) => void | Promise) | undefined; + /** + * Close the transport when its first completed request fails to establish + * a session. + * + * Enable this on transports created per handshake attempt — the + * session-map recipe, where a fresh transport and connected server pair + * serves one expected `initialize`. When the pair's first completed + * request does not establish the session (a refused or throwing + * handshake, a pre-session `GET`/`DELETE`, an unsupported method), the + * transport closes and fires {@linkcode WebStandardStreamableHTTPServerTransport.onclose | onclose}, + * so the paired server tears down instead of leaking. A request still in + * flight defers the close, and an established session is never affected. + * + * Leave it off (the default) for a transport that serves an endpoint + * long-term: pre-session refusals there — for example the SDK client's + * version-negotiation probe, answered `400` before `initialize` — must + * not end the transport. Only meaningful with `sessionIdGenerator` set; + * stateless transports never close on refusals. + * + * @default false + */ + closeOnRefusedHandshake?: boolean; + /** * If `true`, the server will return JSON responses instead of starting an SSE stream. * This can be useful for simple request/response scenarios without streaming. @@ -236,6 +259,9 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { private _requestToStreamMapping: Map = new Map(); private _requestResponseMap: Map = new Map(); private _initialized: boolean = false; + private _closeOnRefusedHandshake: boolean = false; + /** Requests currently inside {@linkcode handleRequest} — the refused-handshake close is deferred while any are in flight. */ + private _activeRequests: number = 0; private _enableJsonResponse: boolean = false; private _standaloneSseStreamId: string = '_GET_stream'; private _eventStore?: EventStore; @@ -254,6 +280,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { constructor(options: WebStandardStreamableHTTPServerTransportOptions = {}) { this.sessionIdGenerator = options.sessionIdGenerator; + this._closeOnRefusedHandshake = options.closeOnRefusedHandshake ?? false; this._enableJsonResponse = options.enableJsonResponse ?? false; this._eventStore = options.eventStore; this._onsessioninitialized = options.onsessioninitialized; @@ -349,26 +376,66 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { /** * Handles an incoming HTTP request, whether `GET`, `POST`, or `DELETE` * Returns a `Response` object (Web Standard) + * + * With {@linkcode WebStandardStreamableHTTPServerTransportOptions.closeOnRefusedHandshake | closeOnRefusedHandshake} + * enabled, enforces one invariant behind every response: a sessionful + * transport's first completed request either establishes the session or + * ends the transport. On a per-handshake transport, a request that leaves + * it without a session — a refused or throwing handshake, a pre-session + * `GET`/`DELETE`, an unsupported method — can never be followed by one + * that reaches this instance, so the close chain is scheduled behind the + * response; without it, anything the consumer paired with the transport + * before the handshake (a connected server, a session-map entry keyed off + * {@linkcode onclose}) would never hear about the teardown and leak. + * The close runs on a microtask (the response is never delayed), is + * deferred while any other request is still in flight (a refusal settling + * next to an `initialize` still parsing its body must not tear it down), + * and re-checks the predicate before closing; a successful `initialize` + * assigns `sessionId` and its initialized state synchronously before its + * response is produced, so it never matches. Established sessions and + * stateless transports never match the predicate — refusals leave them + * untouched. {@linkcode close} is idempotent, so caller-side closes still + * work. */ async handleRequest(req: Request, options?: HandleRequestOptions): Promise { - // Validate request headers for DNS rebinding protection - const validationError = this.validateRequestHeaders(req); - if (validationError) { - return validationError; - } - - switch (req.method) { - case 'POST': { - return this.handlePostRequest(req, options); - } - case 'GET': { - return this.handleGetRequest(req); + this._activeRequests += 1; + try { + // Validate request headers for DNS rebinding protection + const validationError = this.validateRequestHeaders(req); + if (validationError) { + return validationError; } - case 'DELETE': { - return this.handleDeleteRequest(req); + + switch (req.method) { + case 'POST': { + return await this.handlePostRequest(req, options); + } + case 'GET': { + return await this.handleGetRequest(req); + } + case 'DELETE': { + return await this.handleDeleteRequest(req); + } + default: { + return this.handleUnsupportedRequest(); + } } - default: { - return this.handleUnsupportedRequest(); + } finally { + // The awaits above mean this runs once the response (or throw) has + // settled — after a successful initialize has assigned its state. + this._activeRequests -= 1; + if ( + this._closeOnRefusedHandshake && + this._activeRequests === 0 && + this.sessionIdGenerator !== undefined && + !this._initialized && + this.sessionId === undefined + ) { + queueMicrotask(() => { + if (this._activeRequests === 0 && !this._initialized && this.sessionId === undefined) { + void this.close().catch(() => {}); + } + }); } } } diff --git a/packages/server/test/server/createMcpHandler.test.ts b/packages/server/test/server/createMcpHandler.test.ts index f3237221e0..154e98f7aa 100644 --- a/packages/server/test/server/createMcpHandler.test.ts +++ b/packages/server/test/server/createMcpHandler.test.ts @@ -7,12 +7,17 @@ * faces, the per-request era write + client-identity backfill, notification * routing, the response-mode knob, and close() teardown of the modern leg. */ -import { CLIENT_CAPABILITIES_META_KEY, CLIENT_INFO_META_KEY, PROTOCOL_VERSION_META_KEY } from '@modelcontextprotocol/core-internal'; +import { + CLIENT_CAPABILITIES_META_KEY, + CLIENT_INFO_META_KEY, + classifyInboundRequest, + PROTOCOL_VERSION_META_KEY +} from '@modelcontextprotocol/core-internal'; import { describe, expect, it, vi } from 'vitest'; import * as z from 'zod/v4'; import type { McpRequestContext } from '../../src/server/createMcpHandler'; -import { createMcpHandler, isLegacyRequest } from '../../src/server/createMcpHandler'; +import { classifyEntryRequest, createMcpHandler, isLegacyRequest } from '../../src/server/createMcpHandler'; import { McpServer } from '../../src/server/mcp'; import { PerRequestHTTPServerTransport } from '../../src/server/perRequestTransport'; @@ -736,6 +741,130 @@ describe('createMcpHandler — user-land routing with isLegacyRequest (replaces }); }); +describe('classifyEntryRequest — the Request-shaped sibling of classifyInboundRequest', () => { + const legacyInitialize = { + jsonrpc: '2.0', + id: 'init-1', + method: 'initialize', + params: { protocolVersion: '2025-11-25', clientInfo: { name: 'legacy', version: '1.0' }, capabilities: {} } + }; + + /** The fields a hand-wired composition would extract itself to call {@linkcode classifyInboundRequest} directly. */ + function handExtractedFields(request: Request, body?: unknown): Parameters[0] { + return { + httpMethod: request.method.toUpperCase(), + protocolVersionHeader: request.headers.get('mcp-protocol-version') ?? undefined, + mcpMethodHeader: request.headers.get('mcp-method') ?? undefined, + mcpNameHeader: request.headers.get('mcp-name') ?? undefined, + ...(body !== undefined && { body }) + }; + } + + it('classifies representative requests exactly as classifyInboundRequest classifies the hand-extracted fields', async () => { + const cases: Array<{ name: string; body?: unknown; request: () => Request }> = [ + { + name: 'modern envelope POST', + body: modernToolsCall('echo', { text: 'x' }), + request: () => postRequest(modernToolsCall('echo', { text: 'x' })) + }, + { name: 'legacy initialize POST', body: legacyInitialize, request: () => postRequest(legacyInitialize) }, + { name: 'GET session operation', request: () => new Request('http://localhost/mcp', { method: 'GET' }) }, + { + name: 'header-only modern (modern header, claim-less body)', + body: { jsonrpc: '2.0', id: 11, method: 'tools/list', params: {} }, + request: () => + postRequest( + { jsonrpc: '2.0', id: 11, method: 'tools/list', params: {} }, + { 'mcp-protocol-version': MODERN_REVISION, 'mcp-method': 'tools/list' } + ) + } + ]; + + for (const { name, body, request } of cases) { + const classified = await classifyEntryRequest(request()); + expect(classified.step, name).toBe('classified'); + if (classified.step !== 'classified') continue; + expect(classified.outcome, name).toEqual(classifyInboundRequest(handExtractedFields(request(), body))); + expect(classified.body, name).toEqual(body); + } + }); + + it('reports a POST with a non-JSON body as no-json-body (nothing to hand-extract), with the bytes preserved for routing', async () => { + const classified = await classifyEntryRequest(postRequest('{not json')); + expect(classified.step).toBe('no-json-body'); + if (classified.step !== 'no-json-body') return; + expect(await classified.forwardRequest.text()).toBe('{not json'); + }); + + it('reads the input body exactly once and hands routing a still-readable forwardRequest (the entry contract)', async () => { + const original = { jsonrpc: '2.0', id: 3, method: 'tools/list', params: {} }; + const request = postRequest(original); + const classified = await classifyEntryRequest(request); + + expect(classified.step).toBe('classified'); + if (classified.step !== 'classified') return; + // The single read consumed the input request... + expect(request.bodyUsed).toBe(true); + // ...and forwardRequest carries the original bytes, unread — exactly + // what createMcpHandler hands its legacy leg. + expect(classified.forwardRequest.bodyUsed).toBe(false); + expect(await classified.forwardRequest.text()).toBe(JSON.stringify(original)); + expect(classified.parsedBody).toEqual(original); + + // With a pre-parsed body the stream is never touched and no clone is made. + const preParsed = postRequest(original); + const classifiedPreParsed = await classifyEntryRequest(preParsed, original); + expect(classifiedPreParsed.step).toBe('classified'); + if (classifiedPreParsed.step !== 'classified') return; + expect(preParsed.bodyUsed).toBe(false); + expect(classifiedPreParsed.forwardRequest).toBe(preParsed); + + // needsForward: false skips the body-preserving clone — forwardRequest + // is then the (consumed) input request, the isLegacyRequest fast path. + const noForward = postRequest(original); + const classifiedNoForward = await classifyEntryRequest(noForward, undefined, false); + expect(classifiedNoForward.step).toBe('classified'); + if (classifiedNoForward.step !== 'classified') return; + expect(classifiedNoForward.forwardRequest).toBe(noForward); + expect(noForward.bodyUsed).toBe(true); + }); + + it('routes on the outcome reason — legacy initialize to a session host, everything else to the entry (the hybrid pattern)', async () => { + const { factory } = testFactory(); + const handler = createMcpHandler(factory, { legacy: 'reject' }); + const sessionHost = vi.fn(async (request: Request) => new Response(await request.text(), { status: 299 })); + + const route = async (request: Request): Promise => { + const classified = await classifyEntryRequest(request); + if (classified.step === 'unreadable-body') { + return new Response(null, { status: 400 }); + } + if (classified.step === 'classified' && classified.outcome.kind === 'legacy' && classified.outcome.reason === 'initialize') { + return sessionHost(classified.forwardRequest); + } + // Hand the already-parsed body along so the entry classifies from + // the value instead of reading the forwarded clone a second time. + return handler.fetch( + classified.forwardRequest, + classified.step === 'classified' ? { parsedBody: classified.parsedBody } : undefined + ); + }; + + // The 2025 handshake reaches the session host with its bytes intact. + const initResponse = await route(postRequest(legacyInitialize)); + expect(initResponse.status).toBe(299); + expect(await initResponse.text()).toBe(JSON.stringify(legacyInitialize)); + + // Modern traffic is served by the entry from the forwarded clone — + // the entry's own single body read still works downstream. + const modernResponse = await route(postRequest(modernToolsCall('echo', { text: 'routed' }))); + expect(modernResponse.status).toBe(200); + const body = (await modernResponse.json()) as { result: { content: Array<{ text: string }> } }; + expect(body.result.content[0]?.text).toBe('routed'); + expect(sessionHost).toHaveBeenCalledTimes(1); + }); +}); + describe('createMcpHandler — responseMode', () => { it('defaults to the lazy upgrade: a handler emitting a related notification streams the exchange over SSE', async () => { const { factory } = testFactory(); diff --git a/packages/server/test/server/streamableHttp.test.ts b/packages/server/test/server/streamableHttp.test.ts index beca451113..bf7f279de6 100644 --- a/packages/server/test/server/streamableHttp.test.ts +++ b/packages/server/test/server/streamableHttp.test.ts @@ -1370,6 +1370,265 @@ describe('Zod v4', () => { }); }); + describe('HTTPServerTransport - refused handshake close chain (closeOnRefusedHandshake: true)', () => { + let transport: WebStandardStreamableHTTPServerTransport; + let mcpServer: McpServer; + let oncloseSpy: ReturnType void>>; + + beforeEach(async () => { + mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }, { capabilities: {} }); + transport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID(), + closeOnRefusedHandshake: true + }); + await mcpServer.connect(transport); + + // connect() installed the Protocol's own onclose handler — chain it + // so the spy observes the close without detaching server teardown. + const protocolOnClose = transport.onclose; + oncloseSpy = vi.fn<() => void>(); + transport.onclose = () => { + oncloseSpy(); + protocolOnClose?.(); + }; + }); + + afterEach(async () => { + await transport.close(); + }); + + /** The refusal-path close is scheduled on a microtask — let it run. */ + const flushCloseChain = () => new Promise(resolve => setTimeout(resolve, 0)); + + it('fires onclose when a schema-invalid initialize is refused on a never-established transport', async () => { + // `initialize` with params failing the schema-validated guard: it + // is not recognized as an initialization request, so the handshake + // is refused with 400 before any session exists. + const invalidInit = { jsonrpc: '2.0', method: 'initialize', params: {}, id: 'init-bad' } as JSONRPCMessage; + const response = await transport.handleRequest(createRequest('POST', invalidInit)); + + expect(response.status).toBe(400); + await flushCloseChain(); + expect(oncloseSpy).toHaveBeenCalledTimes(1); + }); + + it('fires onclose when the initialize Accept header is refused on a never-established transport', async () => { + const response = await transport.handleRequest(createRequest('POST', TEST_MESSAGES.initialize, { accept: 'application/json' })); + + expect(response.status).toBe(406); + await flushCloseChain(); + expect(oncloseSpy).toHaveBeenCalledTimes(1); + }); + + it('runs the connected server close chain when a batch initialize is refused', async () => { + const serverClosed = vi.fn(); + mcpServer.server.onclose = serverClosed; + + const response = await transport.handleRequest(createRequest('POST', [TEST_MESSAGES.initialize, TEST_MESSAGES.initialize])); + + expect(response.status).toBe(400); + await flushCloseChain(); + expect(oncloseSpy).toHaveBeenCalledTimes(1); + expect(serverClosed).toHaveBeenCalledTimes(1); + }); + + it('fires onclose when the handshake attempt itself throws on a never-established transport', async () => { + // A throwing sessionIdGenerator lands the handshake in + // handlePostRequest's outer catch before any session state was + // written — the 400 there is a refusal like the explicit paths. + const throwingTransport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: () => { + throw new Error('generator boom'); + }, + closeOnRefusedHandshake: true + }); + const throwingServer = new McpServer({ name: 'test-server', version: '1.0.0' }, { capabilities: {} }); + await throwingServer.connect(throwingTransport); + const protocolOnClose = throwingTransport.onclose; + const closeSpy = vi.fn<() => void>(); + throwingTransport.onclose = () => { + closeSpy(); + protocolOnClose?.(); + }; + + const response = await throwingTransport.handleRequest(createRequest('POST', TEST_MESSAGES.initialize)); + + expect(response.status).toBe(400); + await flushCloseChain(); + expect(closeSpy).toHaveBeenCalledTimes(1); + }); + + it('fires onclose when a pre-session GET is refused on a never-established transport', async () => { + const response = await transport.handleRequest(createRequest('GET', undefined, { accept: 'application/json' })); + + expect(response.status).toBe(406); + await flushCloseChain(); + expect(oncloseSpy).toHaveBeenCalledTimes(1); + }); + + it('fires onclose when a pre-session request is refused on a path no refusal branch instruments (405 unsupported method)', async () => { + // The invariant lives at the handleRequest entry, not at the + // individual refusal branches — an unsupported method's 405 is a + // pre-session refusal like any other. + const response = await transport.handleRequest(new Request('http://localhost/mcp', { method: 'PATCH' })); + + expect(response.status).toBe(405); + await flushCloseChain(); + expect(oncloseSpy).toHaveBeenCalledTimes(1); + }); + + it('does not fire onclose when an established session receives refused GET and DELETE requests', async () => { + const initResponse = await transport.handleRequest(createRequest('POST', TEST_MESSAGES.initialize)); + expect(initResponse.status).toBe(200); + const sessionId = initResponse.headers.get('mcp-session-id') as string; + + const badGet = await transport.handleRequest(createRequest('GET', undefined, { sessionId, accept: 'application/json' })); + expect(badGet.status).toBe(406); + + const badDelete = await transport.handleRequest(createRequest('DELETE', undefined, { sessionId: 'invalid-session' })); + expect(badDelete.status).toBe(404); + + await flushCloseChain(); + expect(oncloseSpy).not.toHaveBeenCalled(); + + // The session stays serviceable after both refusals. + const ping: JSONRPCMessage = { jsonrpc: '2.0', method: 'ping', id: 'ping-2' }; + const pingResponse = await transport.handleRequest(createRequest('POST', ping, { sessionId })); + expect(pingResponse.status).toBe(200); + }); + + it('does not close a stateless transport on a refused request even with the option on (shared stateless endpoints stay up)', async () => { + // The Hono/Workers wirings share ONE stateless transport across + // all clients; a malformed request from one client must not take + // the endpoint down for everyone — the option only applies to + // sessionful transports. + const statelessServer = new McpServer({ name: 'test-server', version: '1.0.0' }, { capabilities: {} }); + const statelessTransport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: undefined, + closeOnRefusedHandshake: true + }); + await statelessServer.connect(statelessTransport); + const protocolOnClose = statelessTransport.onclose; + const statelessOnclose = vi.fn<() => void>(); + statelessTransport.onclose = () => { + statelessOnclose(); + protocolOnClose?.(); + }; + + const refused = await statelessTransport.handleRequest( + createRequest('POST', TEST_MESSAGES.initialize, { accept: 'application/json' }) + ); + expect(refused.status).toBe(406); + await flushCloseChain(); + expect(statelessOnclose).not.toHaveBeenCalled(); + + // The shared endpoint still serves the next client. + const initResponse = await statelessTransport.handleRequest(createRequest('POST', TEST_MESSAGES.initialize)); + expect(initResponse.status).toBe(200); + await statelessTransport.close(); + }); + + it('does not fire onclose when an established session receives a malformed request', async () => { + const initResponse = await transport.handleRequest(createRequest('POST', TEST_MESSAGES.initialize)); + expect(initResponse.status).toBe(200); + const sessionId = initResponse.headers.get('mcp-session-id') as string; + + const malformed = new Request('http://localhost/mcp', { + method: 'POST', + headers: { + Accept: 'application/json, text/event-stream', + 'Content-Type': 'application/json', + 'mcp-session-id': sessionId, + 'mcp-protocol-version': '2025-11-25' + }, + body: 'not valid json' + }); + const response = await transport.handleRequest(malformed); + + expect(response.status).toBe(400); + await flushCloseChain(); + expect(oncloseSpy).not.toHaveBeenCalled(); + + // The session stays serviceable after the refusal. + const ping: JSONRPCMessage = { jsonrpc: '2.0', method: 'ping', id: 'ping-1' }; + const pingResponse = await transport.handleRequest(createRequest('POST', ping, { sessionId })); + expect(pingResponse.status).toBe(200); + }); + + it('leaves a long-lived sessionful transport open on pre-session refusals by default (version-negotiation probes)', async () => { + // The SDK client's own version negotiation sends a pre-session + // non-initialize probe, expects the 400, then initializes on the + // SAME transport — the default posture must not end it. + const endpointServer = new McpServer({ name: 'test-server', version: '1.0.0' }, { capabilities: {} }); + const endpointTransport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID() + }); + await endpointServer.connect(endpointTransport); + const protocolOnClose = endpointTransport.onclose; + const endpointOnclose = vi.fn<() => void>(); + endpointTransport.onclose = () => { + endpointOnclose(); + protocolOnClose?.(); + }; + + const probe = await endpointTransport.handleRequest(createRequest('POST', TEST_MESSAGES.toolsList)); + expect(probe.status).toBe(400); + await flushCloseChain(); + expect(endpointOnclose).not.toHaveBeenCalled(); + + const initResponse = await endpointTransport.handleRequest(createRequest('POST', TEST_MESSAGES.initialize)); + expect(initResponse.status).toBe(200); + expect(initResponse.headers.get('mcp-session-id')).toBeDefined(); + await endpointTransport.close(); + }); + + it('defers the close while an initialize is in flight when a concurrent refusal settles (active-request guard)', async () => { + // The initialize's session state is only assigned after its body is + // read; a refusal settling inside that window must not tear the + // handshake down. A gated streaming body holds the initialize + // in flight deterministically. (The option's sessionIdGenerator is + // sync-typed, so the async injection point is the body read.) + const encoder = new TextEncoder(); + let releaseBody!: () => void; + const gate = new Promise(resolve => { + releaseBody = resolve; + }); + const body = new ReadableStream({ + start: async controller => { + await gate; + controller.enqueue(encoder.encode(JSON.stringify(TEST_MESSAGES.initialize))); + controller.close(); + } + }); + const initRequest = new Request('http://localhost/mcp', { + method: 'POST', + headers: { + Accept: 'application/json, text/event-stream', + 'Content-Type': 'application/json' + }, + body, + // Streaming request bodies require half-duplex in undici. + duplex: 'half' + } as RequestInit); + + const initPromise = transport.handleRequest(initRequest); + // Let the initialize enter handleRequest and start reading. + await new Promise(resolve => setTimeout(resolve, 0)); + + const refused = await transport.handleRequest(new Request('http://localhost/mcp', { method: 'PATCH' })); + expect(refused.status).toBe(405); + await flushCloseChain(); + expect(oncloseSpy).not.toHaveBeenCalled(); + + releaseBody(); + const initResponse = await initPromise; + expect(initResponse.status).toBe(200); + expect(initResponse.headers.get('mcp-session-id')).toBeDefined(); + await flushCloseChain(); + expect(oncloseSpy).not.toHaveBeenCalled(); + }); + }); + describe('close() re-entrancy guard', () => { it('should not recurse when onclose triggers a second close()', async () => { const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: randomUUID });