diff --git a/packages/mcp/src/transports/streamable-http.ts b/packages/mcp/src/transports/streamable-http.ts index 5ea76682e..b55cb2084 100644 --- a/packages/mcp/src/transports/streamable-http.ts +++ b/packages/mcp/src/transports/streamable-http.ts @@ -103,6 +103,10 @@ export async function consumeSseStream(stream: ReadableStream, optio buffered += decoder.decode(); if (buffered) processLine(buffered); dispatch(); + } catch (error) { + // Releasing the lock alone leaves the response open; the original error wins over a failed cancel. + await reader.cancel().catch(() => {}); + throw error; } finally { reader.releaseLock(); } diff --git a/packages/mcp/test/streamable-http.test.ts b/packages/mcp/test/streamable-http.test.ts index 837508e81..81ae1d33d 100644 --- a/packages/mcp/test/streamable-http.test.ts +++ b/packages/mcp/test/streamable-http.test.ts @@ -118,6 +118,25 @@ describe("consumeSseStream", () => { ); expect(sent).toBeLessThan(100); }); + + // Regression test for #11729: a failed stream must be cancelled, not left open. + it("cancels the stream when an event exceeds the limit", async () => { + const encoder = new TextEncoder(); + let cancelled = false; + const stream = new ReadableStream({ + start(controller) { + // The server keeps the response open after the oversized event. + controller.enqueue(encoder.encode(`data: ${"x".repeat(512)}\n\n`)); + }, + cancel() { + cancelled = true; + }, + }); + await expect(consumeSseStream(stream, { maxEventBytes: 128, onEvent: () => {} })).rejects.toThrow( + "MCP SSE event exceeds 128 bytes", + ); + expect(cancelled).toBe(true); + }); }); describe("StreamableHttpTransport", () => {