From 56aec154bdee548555978b1522841b5b80440733 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sat, 26 Sep 2026 19:15:39 -0700 Subject: [PATCH] fix: bound discovery time and response size Discovery requests now time out after 30 seconds, covering the body read, and every client call takes an optional `timeoutMs` and `signal`. mcpTools and mcpServers bound initialize and tools/call at 60 seconds by default (their `timeoutMs` option) and honour the tool call's abort signal. A response body or SSE frame over 4 MiB is refused, and the reader is cancelled on overflow, on timeout and once the matching frame is read. --- README.md | 4 +- src/client.test.ts | 131 +++++++++++++++++++++++++++ src/client.ts | 178 ++++++++++++++++++++++++++----------- src/hub/discover.test.ts | 26 +++++- src/hub/discover.ts | 12 ++- src/sidecar-bundle.test.ts | 33 +++++++ src/sidecar-bundle.ts | 14 ++- src/test-server.ts | 4 +- src/tool.test.ts | 43 +++++++++ src/tool.ts | 17 +++- 10 files changed, 399 insertions(+), 63 deletions(-) diff --git a/README.md b/README.md index cc27f55..23107b4 100644 --- a/README.md +++ b/README.md @@ -38,7 +38,7 @@ for (const tool of await mcpListTools(url)) { } ``` -Every client call takes `{ fetch }` as its last argument, for example a fetch that adds an `Authorization` header. +Every client call takes `{ fetch, timeoutMs, signal }` as its last argument, for example a fetch that adds an `Authorization` header. `timeoutMs` covers the request and reading its body, and `signal` cancels both. `mcpTools` and `mcpServers` take a `timeoutMs` option, 60 seconds by default, and pass each tool call's abort signal through, so a stalling server fails that call instead of hanging it. A response body or event-stream frame over 4 MiB is refused and the stream cancelled. ## Where it fits @@ -114,7 +114,7 @@ Mount `mcpRoutes` on the hub app under `/api/tenants/:tenantId`, behind the hub' `POST /api/tenants/:tenantId/mcp/discover` with `{ url, credentialId? }` returns `{ data: { serverInfo, tools } }`. `url` must be https (http only on loopback). `credentialId` names a tenant credential whose secret is sent as a bearer; a credential holding `MCP_NO_TOKEN_SENTINEL` from `@corbits/credential-http` sends no `authorization` header. Errors: 400 for a bad body or URL, 404 for an unknown credential, 422 when the server fails discovery or the request would leave the credential's origin. `requireGrant` is the host's own grant middleware for this route. -When a secret is sent, the fetch is pinned to the origin of the credential's provider `apiBaseUrl`; with no credential or a keyless one, to the URL's origin. Redirects are always refused, so the secret never leaves that origin. +When a secret is sent, the fetch is pinned to the origin of the credential's provider `apiBaseUrl`; with no credential or a keyless one, to the URL's origin. Redirects are always refused, so the secret never leaves that origin. Each discovery request times out after 30 seconds. #### Extra origins diff --git a/src/client.test.ts b/src/client.test.ts index a416850..c0bd7f9 100644 --- a/src/client.test.ts +++ b/src/client.test.ts @@ -88,3 +88,134 @@ describe("SSE response framing", () => { }, ); }); + +describe("bounded reads", () => { + const enc = new TextEncoder(); + function sseServer(body: () => ReadableStream) { + return Bun.serve({ + port: 0, + fetch: () => + new Response(body(), { + headers: { "content-type": "text/event-stream" }, + }), + }); + } + + test("an endless stream is cut off by the timeout and cancelled", async () => { + let cancelled = false; + const server = sseServer( + () => + new ReadableStream({ + async pull(controller) { + await Bun.sleep(20); + controller.enqueue(enc.encode(": keepalive\n\n")); + }, + cancel() { + cancelled = true; + }, + }), + ); + try { + await expect( + mcpListTools(server.url.toString(), { timeoutMs: 200 }), + ).rejects.toThrow(/within 200ms/); + await Bun.sleep(50); + expect(cancelled).toBe(true); + } finally { + void server.stop(true); + } + }); + + test("a stalled server is cut off by the timeout", async () => { + const server = Bun.serve({ + port: 0, + fetch: async () => { + await Bun.sleep(5_000); + return new Response("{}"); + }, + }); + try { + await expect( + mcpListTools(server.url.toString(), { timeoutMs: 200 }), + ).rejects.toThrow(/within 200ms/); + } finally { + void server.stop(true); + } + }); + + test("an oversized frame is refused and the stream cancelled", async () => { + let sent = 0; + let cancelled = false; + const server = sseServer( + () => + new ReadableStream({ + pull(controller) { + sent += 1; + if (sent > 64) return controller.close(); + controller.enqueue(enc.encode(`data: ${"x".repeat(1 << 20)}`)); + }, + cancel() { + cancelled = true; + }, + }), + ); + try { + await expect(mcpListTools(server.url.toString())).rejects.toThrow( + /larger than/, + ); + await Bun.sleep(50); + expect(cancelled).toBe(true); + expect(sent).toBeLessThan(64); + } finally { + void server.stop(true); + } + }); + + test("an oversized JSON body is refused", async () => { + const server = Bun.serve({ + port: 0, + fetch: () => + new Response(`"${"x".repeat(5 * 1024 * 1024)}"`, { + headers: { "content-type": "application/json" }, + }), + }); + try { + await expect(mcpListTools(server.url.toString())).rejects.toThrow( + /larger than/, + ); + } finally { + void server.stop(true); + } + }); + + test("the stream is cancelled once the matching frame is read", async () => { + let cancelled = false; + const server = Bun.serve({ + port: 0, + async fetch(req) { + const parsed = RequestIdOnly(await req.json()); + if (parsed instanceof type.errors) + return new Response("", { status: 400 }); + const frame = `data: ${JSON.stringify({ jsonrpc: "2.0", id: parsed.id, result: { tools: [] } })}\n\n`; + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(enc.encode(frame)); + }, + cancel() { + cancelled = true; + }, + }), + { headers: { "content-type": "text/event-stream" } }, + ); + }, + }); + try { + expect(await mcpListTools(server.url.toString())).toEqual([]); + await Bun.sleep(50); + expect(cancelled).toBe(true); + } finally { + void server.stop(true); + } + }); +}); diff --git a/src/client.ts b/src/client.ts index 7fd9a6c..49e7c4b 100644 --- a/src/client.ts +++ b/src/client.ts @@ -26,8 +26,18 @@ export interface McpToolResult { export interface McpClientOptions { /** Injectable `fetch` for tests and mediated-credential handles. */ fetch?: FetchLike; + /** Abort the request, including reading its body, after this long. */ + timeoutMs?: number; + /** Abort the request when this fires, e.g. a cancelled tool call. */ + signal?: AbortSignal; } +/** The run-time `tools/call` and `initialize` bound when a caller sets none. */ +export const DEFAULT_TIMEOUT_MS = 60_000; + +/** Largest response body or SSE frame read before the stream is cancelled. */ +const MAX_MESSAGE_BYTES = 4 * 1024 * 1024; + const JsonRpcResponse = type({ jsonrpc: "'2.0'", id: "string | number", @@ -58,30 +68,51 @@ async function sendRequest( ): Promise { const fetchImpl = opts.fetch ?? fetch; const id = nextId++; - const response = await fetchImpl(url, { - method: "POST", - headers: { - "content-type": "application/json", - accept: "application/json, text/event-stream", - }, - body: JSON.stringify({ - jsonrpc: "2.0", - id, - method, - ...(params !== undefined ? { params } : {}), - }), - }); - - if (!response.ok) { - throw new McpError( - `MCP server ${url} responded ${response.status} to ${method}`, - ); - } + const timeout = + opts.timeoutMs === undefined + ? undefined + : AbortSignal.timeout(opts.timeoutMs); + const signals = [timeout, opts.signal].filter((s) => s !== undefined); + const signal = signals.length > 0 ? AbortSignal.any(signals) : undefined; + let message: unknown; + try { + const response = await fetchImpl(url, { + method: "POST", + headers: { + "content-type": "application/json", + accept: "application/json, text/event-stream", + }, + body: JSON.stringify({ + jsonrpc: "2.0", + id, + method, + ...(params !== undefined ? { params } : {}), + }), + ...(signal !== undefined ? { signal } : {}), + }); + + if (!response.ok) { + await response.body?.cancel(); + throw new McpError( + `MCP server ${url} responded ${response.status} to ${method}`, + ); + } - const contentType = response.headers.get("content-type") ?? ""; - const message = contentType.includes("text/event-stream") - ? await readSseJsonRpc(response, id) - : await response.json(); + const contentType = response.headers.get("content-type") ?? ""; + message = contentType.includes("text/event-stream") + ? await readSseJsonRpc(response, id, signal) + : parseJson(await readCapped(response, signal)); + } catch (cause) { + if (opts.signal?.aborted === true) { + throw new McpError(`${method} to MCP server ${url} was cancelled`); + } + if (timeout?.aborted === true) { + throw new McpError( + `MCP server ${url} did not answer ${method} within ${String(opts.timeoutMs)}ms`, + ); + } + throw cause; + } const parsed = JsonRpcResponse(message); if (parsed instanceof type.errors) { @@ -97,44 +128,89 @@ async function sendRequest( return parsed.result; } -/** Read an SSE body and return the first JSON-RPC message whose id matches. */ -async function readSseJsonRpc( +function parseJson(text: string): unknown { + try { + return JSON.parse(text); + } catch { + throw new McpError("MCP server sent a body that is not JSON"); + } +} + +function overflow(): McpError { + return new McpError( + `MCP server sent a message larger than ${String(MAX_MESSAGE_BYTES)} bytes`, + ); +} + +/** Read a body's chunks, cancelling the stream when the signal aborts. */ +async function* chunks( response: Response, - id: number, -): Promise { + signal: AbortSignal | undefined, +): AsyncGenerator { const body = response.body; if (body === null) { - throw new McpError("MCP server sent an event-stream response with no body"); + throw new McpError("MCP server sent a response with no body"); } const reader = body.getReader(); - const decoder = new TextDecoder(); - let buffer = ""; + const abort = () => void reader.cancel().catch(() => undefined); + signal?.addEventListener("abort", abort, { once: true }); try { for (;;) { const { done, value } = await reader.read(); - if (done) break; - buffer += decoder.decode(value, { stream: true }); - const frames = buffer.split(/\r\n\r\n|\n\n|\r\r/); - buffer = frames.pop() ?? ""; - for (const frame of frames) { - const dataLines = frame - .split(/\r\n|\r|\n/) - .filter((line) => line.startsWith("data:")) - .map((line) => line.slice(5).trim()); - if (dataLines.length === 0) continue; - const candidate: unknown = JSON.parse(dataLines.join("\n")); - if ( - typeof candidate === "object" && - candidate !== null && - "id" in candidate && - (candidate as { id: unknown }).id === id - ) { - return candidate; - } - } + if (done) return; + yield value; } } finally { - reader.releaseLock(); + signal?.removeEventListener("abort", abort); + await reader.cancel().catch(() => undefined); + } +} + +async function readCapped( + response: Response, + signal: AbortSignal | undefined, +): Promise { + const decoder = new TextDecoder(); + let text = ""; + let bytes = 0; + for await (const chunk of chunks(response, signal)) { + bytes += chunk.byteLength; + if (bytes > MAX_MESSAGE_BYTES) throw overflow(); + text += decoder.decode(chunk, { stream: true }); + } + return text + decoder.decode(); +} + +/** Read an SSE body and return the first JSON-RPC message whose id matches. */ +async function readSseJsonRpc( + response: Response, + id: number, + signal: AbortSignal | undefined, +): Promise { + const decoder = new TextDecoder(); + let buffer = ""; + for await (const chunk of chunks(response, signal)) { + buffer += decoder.decode(chunk, { stream: true }); + const frames = buffer.split(/\r\n\r\n|\n\n|\r\r/); + buffer = frames.pop() ?? ""; + if (buffer.length > MAX_MESSAGE_BYTES) throw overflow(); + for (const frame of frames) { + if (frame.length > MAX_MESSAGE_BYTES) throw overflow(); + const dataLines = frame + .split(/\r\n|\r|\n/) + .filter((line) => line.startsWith("data:")) + .map((line) => line.slice(5).trim()); + if (dataLines.length === 0) continue; + const candidate: unknown = JSON.parse(dataLines.join("\n")); + if ( + typeof candidate === "object" && + candidate !== null && + "id" in candidate && + (candidate as { id: unknown }).id === id + ) { + return candidate; + } + } } throw new McpError( "MCP server closed the event stream without a matching response", diff --git a/src/hub/discover.test.ts b/src/hub/discover.test.ts index 3010605..f256b92 100644 --- a/src/hub/discover.test.ts +++ b/src/hub/discover.test.ts @@ -3,7 +3,11 @@ import { MCP_NO_TOKEN_SENTINEL } from "@corbits/credential-http"; import type { TenantEnv } from "@intx/hub-api"; import { Hono } from "hono"; -import { mountMcpDiscovery, type MountMcpDiscoveryOpts } from "./discover.js"; +import { + discoverMcpServer, + mountMcpDiscovery, + type MountMcpDiscoveryOpts, +} from "./discover.js"; import { startTestMcpServer, type TestServerHandle } from "../test-server.js"; let handle: TestServerHandle | undefined; @@ -249,4 +253,24 @@ describe("POST /mcp/discover", () => { await redirector.stop(true); } }); + + test("discovery gives up on a server that never answers", async () => { + const stalled = Bun.serve({ + port: 0, + fetch: async () => { + await Bun.sleep(5_000); + return new Response("{}"); + }, + }); + try { + await expect( + discoverMcpServer({ + url: new URL("/mcp", stalled.url).href, + timeoutMs: 200, + }), + ).rejects.toThrow(/within 200ms/); + } finally { + await stalled.stop(true); + } + }); }); diff --git a/src/hub/discover.ts b/src/hub/discover.ts index 3d9b72b..1fb9548 100644 --- a/src/hub/discover.ts +++ b/src/hub/discover.ts @@ -101,6 +101,9 @@ function refusingRedirects(inner: FetchLike): FetchLike { }; } +/** How long one discovery request, including its body, may take. */ +const DISCOVERY_TIMEOUT_MS = 30_000; + export type McpDiscovery = { readonly serverInfo: McpServerInfo; readonly tools: readonly McpTool[]; @@ -115,6 +118,7 @@ export async function discoverMcpServer(args: { readonly credential?: McpCredential; readonly extraOrigins?: Readonly>; readonly fetch?: FetchLike; + readonly timeoutMs?: number; }): Promise { const target = parseMcpEndpoint(args.url); const secret = args.credential?.secret; @@ -138,8 +142,12 @@ export async function discoverMcpServer(args: { fetch: args.fetch ?? globalThis.fetch, }), ); - const serverInfo = await mcpInitialize(args.url, { fetch: pinned }); - const tools = await mcpListTools(args.url, { fetch: pinned }); + const client = { + fetch: pinned, + timeoutMs: args.timeoutMs ?? DISCOVERY_TIMEOUT_MS, + }; + const serverInfo = await mcpInitialize(args.url, client); + const tools = await mcpListTools(args.url, client); return { serverInfo, tools }; } diff --git a/src/sidecar-bundle.test.ts b/src/sidecar-bundle.test.ts index 463117b..8084fcf 100644 --- a/src/sidecar-bundle.test.ts +++ b/src/sidecar-bundle.test.ts @@ -279,4 +279,37 @@ describe("mcpServers round trip through a mediated handle", () => { await bundle.dispose?.(); expect(disposed).toBe(1); }); + + test("a stalling tools/call times out as that tool's error", async () => { + handle = startTestMcpServer({ stallToolsCall: true }); + const { impl } = pinnedFetch(new URL(handle.url).origin); + const { env } = envWith({ srv: { fetch: impl } }); + const bundle = mcpServers({ + servers: [{ handle: "srv", url: handle.url, tools: CATALOG }], + timeoutMs: 100, + })(env); + const result = await bundle.run( + { id: "1", name: "srv.echo", arguments: { text: "hi" } }, + new AbortController().signal, + ); + expect(result.isError).toBe(true); + expect(result.content).toContain("within 100ms"); + }); + + test("aborting the tool call cancels a stalling tools/call", async () => { + handle = startTestMcpServer({ stallToolsCall: true }); + const { impl } = pinnedFetch(new URL(handle.url).origin); + const { env } = envWith({ srv: { fetch: impl } }); + const bundle = mcpServers({ + servers: [{ handle: "srv", url: handle.url, tools: CATALOG }], + })(env); + const controller = new AbortController(); + setTimeout(() => controller.abort(), 50); + const result = await bundle.run( + { id: "1", name: "srv.echo", arguments: { text: "hi" } }, + controller.signal, + ); + expect(result.isError).toBe(true); + expect(result.content).toContain("was cancelled"); + }); }); diff --git a/src/sidecar-bundle.ts b/src/sidecar-bundle.ts index 01beee1..51eed5a 100644 --- a/src/sidecar-bundle.ts +++ b/src/sidecar-bundle.ts @@ -19,6 +19,7 @@ import type { ToolCall, ToolDefinition, ToolResult } from "@intx/types/runtime"; import { type } from "arktype"; import { + DEFAULT_TIMEOUT_MS, mcpCallTool, mcpInitialize, McpToolSchema, @@ -69,6 +70,8 @@ export interface McpBoundServer { export interface McpServersConfig { readonly servers: readonly McpBoundServer[]; + /** Per-request bound on `initialize` and `tools/call`; defaults to 60s. */ + readonly timeoutMs?: number; } const BoundServerSchema = type({ @@ -78,7 +81,10 @@ const BoundServerSchema = type({ "allowWithoutAsk?": "string[]", }); -const ConfigSchema = type({ servers: BoundServerSchema.array() }); +const ConfigSchema = type({ + servers: BoundServerSchema.array(), + "timeoutMs?": "number > 0", +}); /** Validate at construction: a malformed deploy config is a deploy bug, not a * tool error the model should see. */ @@ -142,6 +148,7 @@ export function mcpServers( config: McpServersConfig, ): AnnotatedToolFactory { assertConfig(config); + const timeoutMs = config.timeoutMs ?? DEFAULT_TIMEOUT_MS; const entries = config.servers.flatMap((server) => server.tools.map((tool) => ({ @@ -196,6 +203,7 @@ export function mcpServers( credential.fetch(input, init); await mcpInitialize(requestPath(runtime.server.url), { fetch: mediated, + timeoutMs, }); return mediated; })().catch((cause: unknown) => { @@ -208,7 +216,7 @@ export function mcpServers( return { definitions: toolDefinitions, - async run(call: ToolCall, _signal: AbortSignal): Promise { + async run(call: ToolCall, signal: AbortSignal): Promise { const found = byName.get(call.name); const runtime = found === undefined ? undefined : runtimes.get(found.server.handle); @@ -225,7 +233,7 @@ export function mcpServers( requestPath(found.server.url), found.tool.name, call.arguments, - { fetch: mediated }, + { fetch: mediated, timeoutMs, signal }, ); return { callId: call.id, diff --git a/src/test-server.ts b/src/test-server.ts index 9082df8..c5ca103 100644 --- a/src/test-server.ts +++ b/src/test-server.ts @@ -21,7 +21,7 @@ const ToolCallParams = type({ }); export function startTestMcpServer( - opts: { requireAuth?: string } = {}, + opts: { requireAuth?: string; stallToolsCall?: boolean } = {}, ): TestServerHandle { const requestsSeen: Request[] = []; const server = Bun.serve({ @@ -77,6 +77,8 @@ export function startTestMcpServer( }); } if (body.method === "tools/call") { + if (opts.stallToolsCall === true) + return new Promise(() => {}); const params = ToolCallParams(body.params); if (params instanceof type.errors) { return new Response("bad request", { status: 400 }); diff --git a/src/tool.test.ts b/src/tool.test.ts index 38743f4..3127c20 100644 --- a/src/tool.test.ts +++ b/src/tool.test.ts @@ -98,6 +98,49 @@ describe("mcpTools discovers a live server and floors every tool at ask", () => }); }); +describe("mcpTools bounds a stalling server", () => { + const env = { + sources: [], + defaultSource: "x", + storage: {}, + workdir: "/tmp", + audit: {}, + authorize: () => Promise.resolve({ effect: "allow", matchingGrants: [] }), + directors: {}, + }; + + test("a stalling tools/call times out as a tool error", async () => { + handle = startTestMcpServer({ stallToolsCall: true }); + const factory = await mcpTools({ + servers: [{ name: "srv", url: handle.url }], + timeoutMs: 100, + }); + const bundle = factory(env as unknown as Parameters[0]); + const result = await bundle.run( + { id: "1", name: "srv.echo", arguments: { text: "hi" } }, + new AbortController().signal, + ); + expect(result.isError).toBe(true); + expect(result.content).toContain("within 100ms"); + }); + + test("aborting the tool call cancels a stalling tools/call", async () => { + handle = startTestMcpServer({ stallToolsCall: true }); + const factory = await mcpTools({ + servers: [{ name: "srv", url: handle.url }], + }); + const bundle = factory(env as unknown as Parameters[0]); + const controller = new AbortController(); + setTimeout(() => controller.abort(), 50); + const result = await bundle.run( + { id: "1", name: "srv.echo", arguments: { text: "hi" } }, + controller.signal, + ); + expect(result.isError).toBe(true); + expect(result.content).toContain("was cancelled"); + }); +}); + describe("env.credentials wiring: mcpTools resolves a server's fetch through the standard capability", () => { test("a bound credential's fetch is used for tools/call", async () => { handle = startTestMcpServer({ requireAuth: "Bearer secret" }); diff --git a/src/tool.ts b/src/tool.ts index cffc170..f972754 100644 --- a/src/tool.ts +++ b/src/tool.ts @@ -13,6 +13,7 @@ import type { FetchLike } from "@intx/harness"; import type { CredentialCapability } from "@intx/types"; import { + DEFAULT_TIMEOUT_MS, mcpCallTool, mcpInitialize, mcpListTools, @@ -47,6 +48,8 @@ export interface McpToolsOptions { * that flag wins no matter what the operator lists here. */ allowWithoutAsk?: string[]; + /** Per-request bound on every MCP round trip; defaults to 60s. */ + timeoutMs?: number; } async function resolveFetch( @@ -82,11 +85,15 @@ export async function mcpTools( discoveryEnv: { credentials?: CredentialCapability } = {}, ): Promise> { const allowWithoutAsk = options.allowWithoutAsk ?? []; + const timeoutMs = options.timeoutMs ?? DEFAULT_TIMEOUT_MS; const discovered: DiscoveredTool[] = []; for (const server of options.servers) { const clientFetch = await resolveFetch(server, discoveryEnv.credentials); - const clientOpts = clientFetch !== undefined ? { fetch: clientFetch } : {}; + const clientOpts = { + ...(clientFetch !== undefined ? { fetch: clientFetch } : {}), + timeoutMs, + }; await mcpInitialize(server.url, clientOpts); const tools = await mcpListTools(server.url, clientOpts); for (const tool of tools) { @@ -129,7 +136,7 @@ export async function mcpTools( return { definitions: toolDefinitions, - async run(call: ToolCall, _signal: AbortSignal): Promise { + async run(call: ToolCall, signal: AbortSignal): Promise { const found = byName.get(call.name); if (found === undefined) { return { @@ -144,7 +151,11 @@ export async function mcpTools( found.server.url, found.tool.name, call.arguments, - clientFetch !== undefined ? { fetch: clientFetch } : {}, + { + ...(clientFetch !== undefined ? { fetch: clientFetch } : {}), + timeoutMs, + signal, + }, ); return { callId: call.id,