fix(mcp): consume self-cancelling JSON-RPC batches

.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.

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<string, unknown>;
+  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<string, unknown>;
+  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<string, unknown>).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);

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"],