Skip to content

fix: bound discovery time and response size #16

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
TheGreatAxios merged 1 commit into from
Sep 27, 2026
Merged

fix: bound discovery time and response size #16

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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

Expand Down
131 changes: 131 additions & 0 deletions src/client.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
});
});
178 changes: 127 additions & 51 deletions src/client.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -58,30 +68,51 @@ async function sendRequest(
): Promise<unknown> {
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) {
Expand All @@ -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<unknown> {
signal: AbortSignal | undefined,
): AsyncGenerator<Uint8Array> {
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<string> {
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<unknown> {
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",
Expand Down
Loading
Loading