From 91b29cb59ebc8c03499308b907685165ae1440bd Mon Sep 17 00:00:00 2001 From: jxom <7336481+jxom@users.noreply.github.com> Date: Sun, 19 Jul 2026 11:59:55 +1000 Subject: [PATCH] fix: isolate stateless MCP requests --- .changeset/fresh-mcp-transports.md | 5 + src/Cli.test.ts | 145 ++++++++++++++++++++++++++--- src/Cli.ts | 99 +++++++++++++++----- test/fixtures/mcp-memory.ts | 41 +++++++- 4 files changed, 248 insertions(+), 42 deletions(-) create mode 100644 .changeset/fresh-mcp-transports.md diff --git a/.changeset/fresh-mcp-transports.md b/.changeset/fresh-mcp-transports.md new file mode 100644 index 0000000..5037455 --- /dev/null +++ b/.changeset/fresh-mcp-transports.md @@ -0,0 +1,5 @@ +--- +'incur': patch +--- + +Isolated stateless MCP requests, preventing cross-client response collisions and retained aborted requests. diff --git a/src/Cli.test.ts b/src/Cli.test.ts index 35ff72d..59c0517 100644 --- a/src/Cli.test.ts +++ b/src/Cli.test.ts @@ -58,18 +58,20 @@ function mockMcpServeResponses(responses: unknown[]) { }) } -function countRetainedMcpResponses() { - return new Promise((resolve, reject) => { - execFile( - process.execPath, - ['--expose-gc', '--import', 'tsx', 'test/fixtures/mcp-memory.ts'], - { cwd: join(import.meta.dirname, '..'), timeout: 30_000 }, - (error, stdout, stderr) => { - if (error) reject(new Error(stderr.trim() || stdout.trim() || error.message)) - else resolve(Number(stdout.trim())) - }, - ) - }) +function countRetainedMcpExchanges() { + return new Promise<{ aborted: number; requests: number; responses: number }>( + (resolve, reject) => { + execFile( + process.execPath, + ['--expose-gc', '--import', 'tsx', 'test/fixtures/mcp-memory.ts'], + { cwd: join(import.meta.dirname, '..'), timeout: 30_000 }, + (error, stdout, stderr) => { + if (error) reject(new Error(stderr.trim() || stdout.trim() || error.message)) + else resolve(JSON.parse(stdout.trim())) + }, + ) + }, + ) } function createConfigCli(flag?: string) { @@ -5708,6 +5710,7 @@ describe('fetch', () => { body: unknown, sessionId?: string, extraHeaders: Record = {}, + signal?: AbortSignal, ) { const headers: Record = { 'content-type': 'application/json', @@ -5720,6 +5723,7 @@ describe('fetch', () => { method: 'POST', headers, body: JSON.stringify(body), + ...(signal ? { signal } : {}), }), ) } @@ -5803,8 +5807,76 @@ describe('fetch', () => { `) }) - test('completed JSON responses are released', async () => { - expect(await countRetainedMcpResponses()).toBe(0) + test('completed and aborted MCP exchanges are released', async () => { + expect(await countRetainedMcpExchanges()).toMatchInlineSnapshot(` + { + "aborted": 50, + "requests": 0, + "responses": 0, + } + `) + }) + + test('concurrent clients may reuse JSON-RPC ids', async () => { + let startFirst!: () => void + let startSecond!: () => void + let releaseFirst!: () => void + let releaseSecond!: () => void + const firstStarted = new Promise((resolve) => (startFirst = resolve)) + const secondStarted = new Promise((resolve) => (startSecond = resolve)) + const firstReleased = new Promise((resolve) => (releaseFirst = resolve)) + const secondReleased = new Promise((resolve) => (releaseSecond = resolve)) + const cli = Cli.create('test', { + version: '1.0.0', + mcp: { tools: { discovery: 'direct' } }, + }).command('identify', { + run: async (c) => { + const authorization = c.request?.headers.get('authorization') + if (authorization === 'Bearer first') { + startFirst() + await firstReleased + } else { + startSecond() + await secondReleased + } + return { authorization } + }, + }) + + const call = async (authorization: string) => { + const response = await mcpRequest( + cli, + { + jsonrpc: '2.0', + id: 1, + method: 'tools/call', + params: { name: 'identify', arguments: {} }, + }, + undefined, + { authorization }, + ) + const body = await response.json() + return JSON.parse(body.result.content[0].text) + } + + const first = call('Bearer first') + await firstStarted + const second = call('Bearer second') + await secondStarted + releaseFirst() + await new Promise(setImmediate) + releaseSecond() + + expect(await Promise.all([first, second])).toMatchInlineSnapshot(` + [ + { + "authorization": "Bearer first", + }, + { + "authorization": "Bearer second", + }, + ] + `) }) test('POST /mcp with tools/list → returns registered tools', async () => { @@ -5916,6 +5988,51 @@ describe('fetch', () => { expect(await res.text()).toBe('') }) + test('POST /mcp rejects pre-aborted requests', async () => { + const controller = new AbortController() + controller.abort() + await expect( + mcpRequest( + mcpCli(), + { jsonrpc: '2.0', id: 1, method: 'tools/list', params: {} }, + undefined, + {}, + controller.signal, + ), + ).rejects.toThrowErrorMatchingInlineSnapshot(`[AbortError: This operation was aborted]`) + }) + + test('POST /mcp rejects requests aborted during server startup', async () => { + let runs = 0 + const cli = Cli.create('test', { + version: '1.0.0', + mcp: { tools: { discovery: 'direct' } }, + }).command('run', { + run: () => { + runs++ + return { ok: true } + }, + }) + const controller = new AbortController() + const response = mcpRequest( + cli, + { + jsonrpc: '2.0', + id: 1, + method: 'tools/call', + params: { name: 'run', arguments: {} }, + }, + undefined, + {}, + controller.signal, + ) + controller.abort() + await expect(response).rejects.toThrowErrorMatchingInlineSnapshot( + `[AbortError: This operation was aborted]`, + ) + expect(runs).toMatchInlineSnapshot(`0`) + }) + test('mcp.stateless false keeps stateful session handling', async () => { const cli = Cli.create('test', { version: '1.0.0', mcp: { stateless: false } }) cli.command('ping', { diff --git a/src/Cli.ts b/src/Cli.ts index dca5c59..78b8a83 100644 --- a/src/Cli.ts +++ b/src/Cli.ts @@ -1824,7 +1824,46 @@ function createMcpHttpHandler( version: string, options: createMcpHttpHandler.Options = {}, ) { - let transport: any + let session: ReturnType | undefined + + async function createServer( + commands: Map, + mcpOptions: + | { + middlewares?: MiddlewareHandler[] | undefined + env?: z.ZodObject | undefined + vars?: z.ZodObject | undefined + } + | undefined, + stateless: boolean, + ) { + const { fromJsonSchema, McpServer, WebStandardStreamableHTTPServerTransport } = + await import('@modelcontextprotocol/server') + + const server = new McpServer({ name, version }) + Mcp.registerTools(server, commands, { + env: mcpOptions?.env, + fromJsonSchema, + middlewares: mcpOptions?.middlewares, + name, + request: (extra) => extra?.http?.req, + sendNotification: (notification) => server.server.notification(notification), + tools: options.tools, + vars: mcpOptions?.vars, + version, + }) + + const transport = new WebStandardStreamableHTTPServerTransport( + stateless + ? { enableJsonResponse: true } + : { + sessionIdGenerator: () => crypto.randomUUID(), + enableJsonResponse: true, + }, + ) + await server.connect(transport) + return { server, transport } + } return async ( req: Request, @@ -1839,33 +1878,43 @@ function createMcpHttpHandler( if (stateless && req.method !== 'POST') return new Response(null, { status: 405, headers: { Allow: 'POST' } }) - if (!transport) { - const { fromJsonSchema, McpServer, WebStandardStreamableHTTPServerTransport } = - await import('@modelcontextprotocol/server') - - const server = new McpServer({ name, version }) - Mcp.registerTools(server, commands, { - env: mcpOptions?.env, - fromJsonSchema, - middlewares: mcpOptions?.middlewares, - name, - request: (extra) => extra?.http?.req, - sendNotification: (notification) => server.server.notification(notification), - tools: options.tools, - vars: mcpOptions?.vars, - version, + if (!stateless) { + session ??= createServer(commands, mcpOptions, false).catch((error) => { + session = undefined + throw error }) + return (await session).transport.handleRequest(req) + } - const transportOptions = stateless - ? { enableJsonResponse: true } - : { - sessionIdGenerator: () => crypto.randomUUID(), - enableJsonResponse: true, - } - transport = new WebStandardStreamableHTTPServerTransport(transportOptions) - await server.connect(transport) + const abortReason = () => + req.signal.reason ?? new DOMException('This operation was aborted', 'AbortError') + if (req.signal.aborted) throw abortReason() + + const { server, transport } = await createServer(commands, mcpOptions, true) + let closing: Promise | undefined + const close = () => (closing ??= server.close()) + // Transport closure does not settle `handleRequest`; reject the public fetch separately. + let rejectAbort!: (reason?: unknown) => void + const aborted = new Promise((_resolve, reject) => { + rejectAbort = reject + }) + let didAbort = false + const abort = () => { + if (didAbort) return + didAbort = true + rejectAbort(abortReason()) + void close() + } + req.signal.addEventListener('abort', abort, { once: true }) + // Catch aborts that happened during asynchronous server creation. + if (req.signal.aborted) abort() + try { + if (req.signal.aborted) return await aborted + return await Promise.race([transport.handleRequest(req), aborted]) + } finally { + req.signal.removeEventListener('abort', abort) + await close() } - return transport.handleRequest(req) } } diff --git a/test/fixtures/mcp-memory.ts b/test/fixtures/mcp-memory.ts index 30dd769..f120de6 100644 --- a/test/fixtures/mcp-memory.ts +++ b/test/fixtures/mcp-memory.ts @@ -1,13 +1,23 @@ import { Cli } from '../../src/index.js' +let started = 0 +const requests: WeakRef[] = [] const cli = Cli.create('memory-test', { mcp: { tools: { discovery: 'direct' } }, version: '1.0.0', -}).command('ping', { run: () => ({ pong: true }) }) +}) + .command('ping', { run: () => ({ pong: true }) }) + .command('stall', { + run: (c) => { + if (c.request) requests.push(new WeakRef(c.request)) + started++ + return new Promise(() => {}) + }, + }) let id = 0 -function request(method: string, params: Record = {}) { +function request(method: string, params: Record = {}, signal?: AbortSignal) { return cli.fetch( new Request('http://localhost/mcp', { body: JSON.stringify({ id: ++id, jsonrpc: '2.0', method, params }), @@ -16,6 +26,7 @@ function request(method: string, params: Record = {}) { 'content-type': 'application/json', }, method: 'POST', + ...(signal ? { signal } : {}), }), ) } @@ -40,6 +51,24 @@ await consume('initialize', { const responses: WeakRef[] = [] for (let index = 0; index < 100; index++) responses.push(await weakResponse()) +let calls: Promise[] | undefined = [] +let controllers: AbortController[] | undefined = [] +for (let index = 0; index < 50; index++) { + const controller = new AbortController() + controllers.push(controller) + calls.push( + request('tools/call', { name: 'stall', arguments: {} }, controller.signal).then( + () => false, + () => true, + ), + ) +} +while (started < controllers.length) await new Promise(setImmediate) +for (const controller of controllers) controller.abort() +controllers = undefined +const aborted = (await Promise.all(calls)).filter(Boolean).length +calls = undefined + const gc = globalThis.gc if (!gc) throw new Error('garbage collection is unavailable') for (let index = 0; index < 5; index++) { @@ -47,4 +76,10 @@ for (let index = 0; index < 5; index++) { gc() } -console.log(responses.filter((response) => response.deref()).length) +console.log( + JSON.stringify({ + aborted, + requests: requests.filter((request) => request.deref()).length, + responses: responses.filter((response) => response.deref()).length, + }), +)