fix(mcp): apply maxMessageBytes to Streamable HTTP JSON responses (#11749)

The Streamable HTTP transport checked `maxMessageBytes` on SSE replies
but not on `application/json` replies. The JSON branch called
`response.json()`, which buffered and parsed the whole body with no size
check. A server could send an oversized message as plain JSON, and the
same content sent as SSE would have been rejected.

A new helper, `readBodyText(response, maxBytes)`, reads a body through a
reader and decodes each chunk as it arrives. Once the body goes over
`maxBytes`, it cancels the rest and returns only the first `maxBytes`
bytes, marked as truncated. Counting during the read also covers chunked
replies that have no `Content-Length`. `TextDecoder` strips a leading
BOM, as `response.json()` does.

Two places use it:
- JSON replies: if the body is truncated, `send()` throws `MCP JSON
response exceeds N bytes` (default 16 MiB). Bodies within the limit are
parsed as before.
- HTTP error bodies (`checkResponse`): before, `response.text()` read
the whole error body and then cut it to 8 KiB. A huge or endless error
body was fully buffered, or never finished. Now only the 8 KiB excerpt
is read and the rest is cancelled. The excerpt is decoded from the kept
bytes only, so it does not keep the whole body alive (#9265).

Testing (`packages/mcp/test/streamable-http.test.ts`):
- `applies maxMessageBytes to JSON responses`: with a 512-byte limit, a
small reply is accepted and a 2 KB reply is rejected. Fails without the
fix.
- `decodes JSON responses split inside a BOM and a multi-byte
character`: the body arrives one byte per chunk.
- `reads only an excerpt of HTTP error bodies and cancels the rest`: an
endless 500 body yields an 8 KiB excerpt and is cancelled. Without the
fix it never finishes.
- All MCP tests pass; `npm run check` passes.

Fixes #11335

Requested by @petrroll

---------

Co-authored-by: triagent <triagent@localhost>
Co-authored-by: Petr Houška <houskape@gmail.com>

packages/mcp/CHANGELOG.md

@@ -18,6 +18,8 @@
 
 - OAuth metadata parsing and `startAuthorization()` now accept only `http:` and `https:` URLs. Before, a server could supply a `file:` or custom protocol authorization endpoint (for example `search-ms:`), which the platform URL handler would launch on sign-in.
 - Fixed the excerpt of an HTTP error body in Streamable HTTP errors keeping the whole response body alive ([#9265](https://github.com/earendil-works/pi/issues/9265))
+- Fixed Streamable HTTP `application/json` responses bypassing `maxMessageBytes`; oversized JSON responses are now cancelled and rejected like oversized SSE events ([#11335](https://github.com/earendil-works/pi/issues/11335))
+- Fixed Streamable HTTP errors reading the whole HTTP error body; only the first 8 KiB kept in the error is read, and the rest is cancelled
 
 ## [1.4.0] - 2026-10-21
 

packages/mcp/src/transports/streamable-http.ts

@@ -182,6 +182,33 @@ function discard(response: Response): Promise<void> {
 	return response.body?.cancel().catch(() => {}) ?? Promise.resolve();
 }
 
+/**
+ * Reads a response body as text, decoding chunks as they arrive. Once the body exceeds `maxBytes`, the rest is
+ * cancelled and the text holds only the first `maxBytes` bytes.
+ */
+async function readBodyText(response: Response, maxBytes: number): Promise<{ text: string; truncated: boolean }> {
+	if (!response.body) return { text: "", truncated: false };
+	const reader = response.body.getReader();
+	const decoder = new TextDecoder();
+	let text = "";
+	let remaining = maxBytes;
+	try {
+		while (true) {
+			const { done, value } = await reader.read();
+			if (done) return { text: text + decoder.decode(), truncated: false };
+			if (value.byteLength > remaining) {
+				await reader.cancel().catch(() => {});
+				// Decoding a subarray copies only the kept bytes, so the result never references the dropped rest.
+				return { text: text + decoder.decode(value.subarray(0, remaining), { stream: true }), truncated: true };
+			}
+			remaining -= value.byteLength;
+			text += decoder.decode(value, { stream: true });
+		}
+	} finally {
+		reader.releaseLock();
+	}
+}
+
 function describeHttpFailure(status: number, body: string): string {
 	const text = body.trim();
 	const snippet = text.length > ERROR_MESSAGE_BODY_CHARS ? `${text.slice(0, ERROR_MESSAGE_BODY_CHARS - 3)}...` : text;
@@ -257,7 +284,10 @@ export class StreamableHttpTransport extends TransportEvents implements McpTrans
 		}
 		const type = contentType(response);
 		if (type === "application/json") {
-			const body: unknown = await response.json();
+			const maxBytes = this.options.maxMessageBytes ?? DEFAULT_MAX_MESSAGE_BYTES;
+			const { text, truncated } = await readBodyText(response, maxBytes);
+			if (truncated) throw new Error(`MCP JSON response exceeds ${maxBytes} bytes`);
+			const body: unknown = JSON.parse(text);
 			for (const item of Array.isArray(body) ? body : [body]) this.emitMessage(parseJsonRpcMessage(item));
 			return;
 		}
@@ -342,8 +372,8 @@ export class StreamableHttpTransport extends TransportEvents implements McpTrans
 
 	private async checkResponse(response: Response): Promise<void> {
 		if (response.ok) return;
-		// A copy, not a slice: V8 slices keep the whole body alive, and the error keeps this excerpt.
-		const body = `${(await response.text().catch(() => "")).slice(0, MAX_ERROR_BODY_BYTES)} `.slice(0, -1);
+		// Reads only the excerpt the error keeps and cancels the rest, so a huge error body is never buffered.
+		const body = (await readBodyText(response, MAX_ERROR_BODY_BYTES).catch(() => ({ text: "" }))).text;
 		if (response.status === 401) throw new McpAuthRequiredError(response, body);
 		if (response.status === 404 && this.sessionIdValue) throw new McpSessionExpiredError(body);
 		throw new McpHttpError(response.status, describeHttpFailure(response.status, body), body);

packages/mcp/test/streamable-http.test.ts

@@ -5,6 +5,7 @@ import {
 	type McpAuthRequiredError,
 	McpClient,
 	type McpFetch,
+	McpHttpError,
 	McpSessionExpiredError,
 	StreamableHttpTransport,
 } from "../src/index.ts";
@@ -310,6 +311,80 @@ describe("StreamableHttpTransport", () => {
 		await client.close();
 	});
 
+	// #11335
+	it("applies maxMessageBytes to JSON responses", async () => {
+		const send = async (descriptionBytes: number) => {
+			const body = JSON.stringify({
+				jsonrpc: "2.0",
+				id: 1,
+				result: { tools: [{ name: "echo", description: "x".repeat(descriptionBytes), inputSchema: {} }] },
+			});
+			const transport = new StreamableHttpTransport({
+				url: "http://127.0.0.1/mcp",
+				maxMessageBytes: 512,
+				openGetStream: false,
+				fetch: async () => new Response(body, { headers: { "content-type": "application/json" } }),
+			});
+			const messages: unknown[] = [];
+			transport.onMessage((message) => messages.push(message));
+			await transport.start();
+			try {
+				await transport.send({ jsonrpc: "2.0", id: 1, method: "tools/list" });
+				return messages;
+			} finally {
+				await transport.close();
+			}
+		};
+
+		expect(await send(32)).toHaveLength(1);
+		await expect(send(2048)).rejects.toThrow("MCP JSON response exceeds 512 bytes");
+	});
+
+	it("decodes JSON responses split inside a BOM and a multi-byte character", async () => {
+		const bytes = new TextEncoder().encode(
+			`\uFEFF${JSON.stringify({ jsonrpc: "2.0", id: 1, result: { text: "é€" } })}`,
+		);
+		const body = new ReadableStream<Uint8Array>({
+			start(controller) {
+				for (const byte of bytes) controller.enqueue(new Uint8Array([byte]));
+				controller.close();
+			},
+		});
+		const transport = new StreamableHttpTransport({
+			url: "http://127.0.0.1/mcp",
+			openGetStream: false,
+			fetch: async () => new Response(body, { headers: { "content-type": "application/json" } }),
+		});
+		const messages: unknown[] = [];
+		transport.onMessage((message) => messages.push(message));
+		await transport.start();
+		await transport.send({ jsonrpc: "2.0", id: 1, method: "tools/list" });
+		await transport.close();
+		expect(messages).toEqual([{ jsonrpc: "2.0", id: 1, result: { text: "é€" } }]);
+	});
+
+	it("reads only an excerpt of HTTP error bodies and cancels the rest", async () => {
+		let cancelled = false;
+		// An endless error body: reading it whole would never finish.
+		const body = new ReadableStream<Uint8Array>({
+			pull: (controller) => controller.enqueue(new TextEncoder().encode("x".repeat(1024))),
+			cancel: () => {
+				cancelled = true;
+			},
+		});
+		const transport = new StreamableHttpTransport({
+			url: "http://127.0.0.1/mcp",
+			openGetStream: false,
+			fetch: async () => new Response(body, { status: 500 }),
+		});
+		await transport.start();
+		const error = await transport.send({ jsonrpc: "2.0", id: 1, method: "tools/list" }).catch((e: unknown) => e);
+		await transport.close();
+		expect(error).toBeInstanceOf(McpHttpError);
+		expect((error as McpHttpError).body).toHaveLength(8 * 1024);
+		expect(cancelled).toBe(true);
+	});
+
 	it("includes the response body in HTTP errors", async () => {
 		const { url } = await startServer(async (request, response) => {
 			await readBody(request);