From 2b8f72723ac9142ad9676ff511d00ac83c8fa3c3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fahreddin=20=C3=96zcan?= Date: Tue, 25 Aug 2026 12:03:16 +0300 Subject: [PATCH] fix(mcp): consume self-cancelling batches --- .../mcp-reject-self-cancelling-batches.md | 5 + packages/mcp/src/index.ts | 76 +++++++++++- packages/mcp/test/integration.test.ts | 111 ++++++++++++++++++ 3 files changed, 191 insertions(+), 1 deletion(-) create mode 100644 .changeset/mcp-reject-self-cancelling-batches.md diff --git a/.changeset/mcp-reject-self-cancelling-batches.md b/.changeset/mcp-reject-self-cancelling-batches.md new file mode 100644 index 000000000..721619bf9 --- /dev/null +++ b/.changeset/mcp-reject-self-cancelling-batches.md @@ -0,0 +1,5 @@ +--- +"@upstash/context7-mcp": patch +--- + +Consume request/cancellation pairs contained in the same JSON-RPC batch before dispatch. This prevents the legacy stateless HTTP transport from leaving the response stream open when cancellation suppresses the terminal protocol response. diff --git a/packages/mcp/src/index.ts b/packages/mcp/src/index.ts index 4f8347614..720dc2993 100644 --- a/packages/mcp/src/index.ts +++ b/packages/mcp/src/index.ts @@ -123,6 +123,65 @@ const QUERY_DOCS_ALIASES: AliasMap = { libraryId: ["context7CompatibleLibraryID", "libraryID", "libraryName"], }; +type JsonRpcId = string | number; + +function jsonRpcRequestId(message: unknown): JsonRpcId | undefined { + if (!message || typeof message !== "object") return undefined; + + const candidate = message as Record; + if ( + candidate.jsonrpc === "2.0" && + typeof candidate.method === "string" && + (typeof candidate.id === "string" || typeof candidate.id === "number") + ) { + return candidate.id; + } +} + +function cancellationRequestId(message: unknown): JsonRpcId | undefined { + if (!message || typeof message !== "object") return undefined; + + const candidate = message as Record; + if ( + candidate.jsonrpc !== "2.0" || + candidate.method !== "notifications/cancelled" || + "id" in candidate || + !candidate.params || + typeof candidate.params !== "object" + ) { + return undefined; + } + + const requestId = (candidate.params as Record).requestId; + if (typeof requestId === "string" || typeof requestId === "number") return requestId; +} + +function filterBatchSelfCancellations(body: unknown): { + body: unknown; + removedSelfCancellation: boolean; +} { + if (!Array.isArray(body)) return { body, removedSelfCancellation: false }; + + const requestIds = new Set(body.map(jsonRpcRequestId).filter((id) => id !== undefined)); + const selfCancelledIds = new Set( + body + .map(cancellationRequestId) + .filter((id): id is JsonRpcId => id !== undefined && requestIds.has(id)) + ); + if (selfCancelledIds.size === 0) return { body, removedSelfCancellation: false }; + + return { + body: body.filter((message) => { + const requestId = jsonRpcRequestId(message); + if (requestId !== undefined && selfCancelledIds.has(requestId)) return false; + + const cancelledId = cancellationRequestId(message); + return cancelledId === undefined || !selfCancelledIds.has(cancelledId); + }), + removedSelfCancellation: true, + }; +} + // z.preprocess step that rewrites aliased arg names before validation. Living // in the schema keeps aliasing transport-agnostic: the SDK parses the wire // message (any transport, any protocol era) and runs this on validation. @@ -431,6 +490,21 @@ async function main() { } } + // A cancelled request deliberately produces no protocol response, but + // the SDK's legacy stateless transport still waits for one before it + // closes the batch's POST stream. Consume request/cancellation pairs + // that occur in the same batch before SDK dispatch. This applies the + // cancellation without starting the tool and without leaving an ID in + // the transport's response accounting. + const filteredBody = filterBatchSelfCancellations(req.body); + if ( + filteredBody.removedSelfCancellation && + Array.isArray(filteredBody.body) && + filteredBody.body.length === 0 + ) { + return res.status(202).end(); + } + const context: ClientContext = { clientIp: getClientIp(req), apiKey: apiKey, @@ -439,7 +513,7 @@ async function main() { }; await requestContext.run(context, async () => { - await nodeHandler(req, res, req.body); + await nodeHandler(req, res, filteredBody.body); }); } catch (error) { console.error("Error handling MCP request:", error); diff --git a/packages/mcp/test/integration.test.ts b/packages/mcp/test/integration.test.ts index 06d2a9ccf..cca1f7f09 100644 --- a/packages/mcp/test/integration.test.ts +++ b/packages/mcp/test/integration.test.ts @@ -135,6 +135,117 @@ describe("OAuth discovery", () => { }); }); +describe("HTTP batch cancellation", () => { + beforeEach(() => { + requests.length = 0; + }); + + test("consumes a request cancelled by the same batch and closes promptly", async () => { + const response = await fetch(httpUrl, { + method: "POST", + headers: { + "content-type": "application/json", + accept: "application/json, text/event-stream", + }, + body: JSON.stringify([ + { + jsonrpc: "2.0", + id: 102, + method: "tools/call", + params: { + name: "resolve-library-id", + arguments: { query: "framework documentation", libraryName: "Next.js" }, + }, + }, + { + jsonrpc: "2.0", + method: "notifications/cancelled", + params: { requestId: 102, reason: "regression-test" }, + }, + ]), + signal: AbortSignal.timeout(1_000), + }); + + expect(response.status).toBe(202); + expect(await response.text()).toBe(""); + expect(requests).toHaveLength(0); + }); + + test("returns responses for the uncancelled requests in a mixed batch", async () => { + const response = await fetch(httpUrl, { + method: "POST", + headers: { + "content-type": "application/json", + accept: "application/json, text/event-stream", + }, + body: JSON.stringify([ + { + jsonrpc: "2.0", + id: 104, + method: "tools/call", + params: { + name: "resolve-library-id", + arguments: { query: "cancelled", libraryName: "Cancelled.js" }, + }, + }, + { + jsonrpc: "2.0", + id: 105, + method: "tools/call", + params: { + name: "resolve-library-id", + arguments: { query: "framework documentation", libraryName: "Next.js" }, + }, + }, + { + jsonrpc: "2.0", + method: "notifications/cancelled", + params: { requestId: 104, reason: "regression-test" }, + }, + ]), + signal: AbortSignal.timeout(1_000), + }); + + expect(response.status).toBe(200); + const responseText = await response.text(); + expect(responseText).not.toContain('"id":104'); + expect(responseText).toContain('"id":105'); + expect(requests.filter((request) => request.path === "/v2/libs/search")).toHaveLength(1); + expect(requests[0].query.get("libraryName")).toBe("Next.js"); + }); + + test("allows a cancellation for an ID outside the batch", async () => { + const response = await fetch(httpUrl, { + method: "POST", + headers: { + "content-type": "application/json", + accept: "application/json, text/event-stream", + }, + body: JSON.stringify([ + { + jsonrpc: "2.0", + id: 103, + method: "tools/call", + params: { + name: "resolve-library-id", + arguments: { query: "framework documentation", libraryName: "Next.js" }, + }, + }, + { + jsonrpc: "2.0", + method: "notifications/cancelled", + params: { requestId: 999, reason: "different-request" }, + }, + ]), + signal: AbortSignal.timeout(1_000), + }); + + expect(response.status).toBe(200); + expect(await response.text()).toContain('"id":103'); + expect(requests.filter((request) => request.path === "/v2/libs/search")).toHaveLength(1); + }); +}); + describe.each([ ["http", "modern"], ["http", "legacy"],