fix: cancel a Streamable HTTP call by closing its stream (#2140)
The Cancel button POSTed `notifications/cancelled` on every transport. On a 2026-07-28 Streamable HTTP connection that is the wrong signal — the spec makes closing the request's SSE response stream the cancellation signal there and says the notification is neither required nor expected — so a spec-compliant server acknowledged it 202 and dropped it while the tool ran to completion. The SDK already implements the fork, off `transport.hasPerRequestStream`. Every Inspector connection is wrapped in `MessageTrackingTransport`, which did not forward that flag, so the SDK read `undefined` and took the stdio branch on every client — the CLI and TUI included, whose real `StreamableHTTPClientTransport` advertises it correctly. The web client needed three more links: its browser transport answers for the upstream transport that lives on the backend, so it advertises the flag for `streamable-http` and applies the SDK's `requestSignal` to its `POST /api/mcp/send`; and `/api/mcp/send`, held open for the whole call, forwards the browser's disconnect to the upstream `transport.send`. The test fixture's `toWebRequest` built a Web Request with no `signal`, so no fixture server could observe a disconnect at all — without that, a test asserting the cancel reaches the server passes vacuously. stdio and SSE keep `notifications/cancelled`: they multiplex every request over one shared channel and have no per-request stream to close. So does the legacy era, which predates the mechanism. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013hzwzS8UvBsr4yZm7o7sev Signed-off-by: cliffhall <cliff@futurescale.com>
This commit is contained in:
parent
baff393675
commit
85c5206895
4 changed files with 654 additions and 20 deletions
|
|
@ -12,6 +12,8 @@ class FakeTransport implements Transport {
|
|||
sessionId?: string;
|
||||
// Optional on the SDK Transport interface; stdio omits it, HTTP defines it.
|
||||
setProtocolVersion?: (version: string) => void;
|
||||
// Optional too: Streamable HTTP advertises it, stdio and SSE say nothing.
|
||||
hasPerRequestStream?: boolean;
|
||||
async start(): Promise<void> {}
|
||||
async send(message: JSONRPCMessage): Promise<void> {
|
||||
this.sent.push(message);
|
||||
|
|
@ -272,4 +274,19 @@ describe("MessageTrackingTransport.setProtocolVersion", () => {
|
|||
expect(() => tracked.setProtocolVersion("2025-06-18")).not.toThrow();
|
||||
expect(tracked.protocolVersion).toBe("2025-06-18");
|
||||
});
|
||||
|
||||
// #2140: the SDK reads `hasPerRequestStream` off the transport it was handed
|
||||
// — always this wrapper — to choose between aborting a request's own stream
|
||||
// and POSTing `notifications/cancelled`. A wrapper that answers for the base
|
||||
// transport instead of forwarding its answer breaks Cancel on every client.
|
||||
it("forwards the base transport's per-request-stream capability", () => {
|
||||
const { tracked, base } = makeTracked();
|
||||
expect(tracked.hasPerRequestStream).toBeUndefined();
|
||||
|
||||
base.hasPerRequestStream = true;
|
||||
expect(tracked.hasPerRequestStream).toBe(true);
|
||||
|
||||
base.hasPerRequestStream = false;
|
||||
expect(tracked.hasPerRequestStream).toBe(false);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -569,4 +569,123 @@ describe("RemoteClientTransport", () => {
|
|||
|
||||
await transport.close();
|
||||
});
|
||||
|
||||
// #2140: the Cancel button's wire signal is chosen by the SDK from the
|
||||
// transport's `hasPerRequestStream`. The real upstream transport lives on the
|
||||
// backend, so this transport answers for it — and must answer per server
|
||||
// type, since only Streamable HTTP gives a request a stream of its own to
|
||||
// close.
|
||||
describe("per-request-stream cancellation (#2140)", () => {
|
||||
it("advertises hasPerRequestStream only for streamable-http", () => {
|
||||
const forType = (c: MCPServerConfig) =>
|
||||
new RemoteClientTransport(
|
||||
{ baseUrl, fetchFn: vi.fn() as unknown as typeof fetch },
|
||||
c,
|
||||
).hasPerRequestStream;
|
||||
|
||||
expect(
|
||||
forType({ type: "streamable-http", url: "http://server.example/mcp" }),
|
||||
).toBe(true);
|
||||
// stdio and SSE multiplex every request over one shared channel, so
|
||||
// aborting anything would cancel the whole session rather than the call.
|
||||
// The SDK must keep sending `notifications/cancelled` for them.
|
||||
expect(forType(config)).toBe(false);
|
||||
expect(forType({ type: "sse", url: "http://server.example/sse" })).toBe(
|
||||
false,
|
||||
);
|
||||
});
|
||||
|
||||
it("applies the SDK's requestSignal to the browser-to-backend send fetch", async () => {
|
||||
const seenInits: (RequestInit | undefined)[] = [];
|
||||
let sse = createPushableSseStream();
|
||||
const fetchFn = vi
|
||||
.fn<typeof fetch>()
|
||||
.mockImplementation(async (input, init) => {
|
||||
const url = String(input);
|
||||
if (url.endsWith("/api/mcp/connect")) {
|
||||
return new Response(JSON.stringify({ sessionId: "abc" }), {
|
||||
status: 200,
|
||||
});
|
||||
}
|
||||
if (url.includes("/api/mcp/events")) {
|
||||
sse = createPushableSseStream();
|
||||
return sse.response;
|
||||
}
|
||||
if (url.endsWith("/api/mcp/send")) {
|
||||
seenInits.push(init);
|
||||
return new Response(JSON.stringify({ ok: true }), { status: 200 });
|
||||
}
|
||||
return new Response("not found", { status: 404 });
|
||||
});
|
||||
|
||||
const transport = new RemoteClientTransport({ baseUrl, fetchFn }, config);
|
||||
await transport.start();
|
||||
|
||||
const controller = new AbortController();
|
||||
await transport.send(
|
||||
{ jsonrpc: "2.0", method: "notifications/initialized" },
|
||||
{ requestSignal: controller.signal },
|
||||
);
|
||||
expect(seenInits[0]?.signal).toBe(controller.signal);
|
||||
|
||||
// No signal supplied (stdio/SSE, where the SDK never creates one) must
|
||||
// leave the fetch unaborted rather than passing `undefined` through as a
|
||||
// deliberate value.
|
||||
await transport.send({
|
||||
jsonrpc: "2.0",
|
||||
method: "notifications/roots/list_changed",
|
||||
});
|
||||
expect(seenInits[1]).not.toHaveProperty("signal");
|
||||
|
||||
await transport.close();
|
||||
});
|
||||
|
||||
it("aborting the requestSignal rejects the in-flight send and releases its response wait", async () => {
|
||||
let sse = createPushableSseStream();
|
||||
const fetchFn = vi
|
||||
.fn<typeof fetch>()
|
||||
.mockImplementation(async (input, init) => {
|
||||
const url = String(input);
|
||||
if (url.endsWith("/api/mcp/connect")) {
|
||||
return new Response(JSON.stringify({ sessionId: "abc" }), {
|
||||
status: 200,
|
||||
});
|
||||
}
|
||||
if (url.includes("/api/mcp/events")) {
|
||||
sse = createPushableSseStream();
|
||||
return sse.response;
|
||||
}
|
||||
if (url.endsWith("/api/mcp/send")) {
|
||||
// The backend holds this route open for the whole call, so model it
|
||||
// as a fetch that only settles when the caller aborts.
|
||||
return new Promise<Response>((_resolve, reject) => {
|
||||
init?.signal?.addEventListener("abort", () =>
|
||||
reject(new DOMException("Aborted", "AbortError")),
|
||||
);
|
||||
});
|
||||
}
|
||||
return new Response("not found", { status: 404 });
|
||||
});
|
||||
|
||||
const transport = new RemoteClientTransport({ baseUrl, fetchFn }, config);
|
||||
await transport.start();
|
||||
|
||||
const controller = new AbortController();
|
||||
const sent = transport.send(
|
||||
{ jsonrpc: "2.0", id: 7, method: "tools/call" },
|
||||
{ requestSignal: controller.signal },
|
||||
);
|
||||
const settled = vi.fn();
|
||||
void sent.then(settled, settled);
|
||||
|
||||
controller.abort();
|
||||
await expect(sent).rejects.toThrow(/Aborted/);
|
||||
// The response wait registered for id 7 must have been cancelled too —
|
||||
// otherwise this send would hang past the abort waiting for a JSON-RPC
|
||||
// response the aborted request can never produce.
|
||||
expect(settled).toHaveBeenCalledTimes(1);
|
||||
|
||||
await transport.close();
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -437,6 +437,63 @@ describe("modern-era negotiation (2026-07-28)", () => {
|
|||
expect(result.result!.isError).toBeFalsy();
|
||||
});
|
||||
|
||||
// #2140: on the 2026-07-28 era, closing the request's own response stream IS
|
||||
// the cancellation signal for Streamable HTTP; `notifications/cancelled` is
|
||||
// the stdio mechanism and the spec says it is neither required nor expected
|
||||
// here. The SDK implements that fork off `transport.hasPerRequestStream` —
|
||||
// but every Inspector connection is wrapped in `MessageTrackingTransport`,
|
||||
// which was not forwarding it, so all three clients POSTed the notification
|
||||
// and a spec-compliant server acknowledged it `202` and kept running.
|
||||
//
|
||||
// This asserts at the far end, on the server's own request signal: nothing on
|
||||
// the client side distinguishes a cancel that reached the server from one
|
||||
// that was dropped, which is exactly how this shipped.
|
||||
it("cancels by aborting the request's stream, which the server observes", async () => {
|
||||
let sawAbort!: (value: boolean) => void;
|
||||
const aborted = new Promise<boolean>((resolve) => {
|
||||
sawAbort = resolve;
|
||||
});
|
||||
let started!: () => void;
|
||||
const running = new Promise<void>((resolve) => {
|
||||
started = resolve;
|
||||
});
|
||||
|
||||
const slowServer = createTestServerHttp({
|
||||
serverInfo: createTestServerInfo("cancel-era-test", "1.0.0"),
|
||||
tools: [
|
||||
{
|
||||
name: "slow_task",
|
||||
description: "Runs until the client cancels it",
|
||||
handler: async (_params, _context, extra) => {
|
||||
extra?.signal?.addEventListener("abort", () => sawAbort(true), {
|
||||
once: true,
|
||||
});
|
||||
started();
|
||||
// Never settles on its own, so a regression is a timeout rather
|
||||
// than a pass.
|
||||
return new Promise(() => {});
|
||||
},
|
||||
},
|
||||
],
|
||||
modern: {},
|
||||
});
|
||||
await slowServer.start();
|
||||
server = slowServer;
|
||||
|
||||
const connected = await connectWithEra(slowServer.url, "modern");
|
||||
const { tools } = await connected.listTools();
|
||||
const tool = tools.find((t) => t.name === "slow_task");
|
||||
|
||||
// Hold the rejection immediately so it is never seen as unhandled while we
|
||||
// wait on the server side.
|
||||
const settled = connected.callTool(tool!, {}).catch((err: unknown) => err);
|
||||
await running;
|
||||
expect(connected.cancelToolCall()).toBe(true);
|
||||
|
||||
await expect(settled).resolves.toBeInstanceOf(ToolCallCancelledError);
|
||||
await expect(aborted).resolves.toBe(true);
|
||||
}, 30_000);
|
||||
|
||||
it("cancels an in-flight MRTR call while its embedded request is pending", async () => {
|
||||
const started = await startMrtrServer(createMrtrTool());
|
||||
const connected = await connectWithEra(started.url, "modern");
|
||||
|
|
|
|||
|
|
@ -0,0 +1,199 @@
|
|||
/**
|
||||
* #2140 — the Cancel button must send the transport's cancellation signal.
|
||||
*
|
||||
* On a 2026-07-28 Streamable HTTP connection the spec makes *closing the
|
||||
* request's SSE response stream* the cancellation signal, and says a
|
||||
* `notifications/cancelled` is neither required nor expected. The SDK
|
||||
* implements that fork itself: `Protocol.request` aborts a per-request
|
||||
* `AbortController` — handed to the transport as
|
||||
* `TransportSendOptions.requestSignal` — when the connection is modern **and**
|
||||
* the transport advertises `hasPerRequestStream`; otherwise it POSTs the
|
||||
* notification.
|
||||
*
|
||||
* The web client lost it at the proxy boundary. The real upstream transport
|
||||
* lives on the Node backend, and the browser's `RemoteClientTransport` neither
|
||||
* advertised the flag nor applied a `requestSignal` to its
|
||||
* `POST /api/mcp/send` — so every web cancel took the stdio branch, the server
|
||||
* answered the notification `202 Accepted` and dropped it, and the tool kept
|
||||
* running to completion. Only a manual disconnect actually cancelled anything.
|
||||
*
|
||||
* This drives the whole chain the browser drives — InspectorClient ->
|
||||
* RemoteClientTransport -> the Hono backend -> a real upstream Streamable HTTP
|
||||
* transport -> a modern test server — and asserts at the far end, on the
|
||||
* server's own request abort signal. Nothing shallower reaches it: each of the
|
||||
* three hops can drop the signal on its own, and a unit test of any one of them
|
||||
* passes while the chain stays broken.
|
||||
*/
|
||||
|
||||
import { describe, it, expect, afterEach } from "vitest";
|
||||
import { serve } from "@hono/node-server";
|
||||
import type { ServerType } from "@hono/node-server";
|
||||
import { InspectorClient } from "@inspector/core/mcp/inspectorClient.js";
|
||||
import { createRemoteTransport } from "@inspector/core/mcp/remote/createRemoteTransport.js";
|
||||
import { createRemoteApp } from "@inspector/core/mcp/remote/node/server.js";
|
||||
import { ToolCallCancelledError } from "@inspector/core/mcp/toolCallCancelledError.js";
|
||||
import { eraToVersionNegotiation } from "@inspector/core/mcp/types.js";
|
||||
import {
|
||||
createTestServerHttp,
|
||||
createTestServerInfo,
|
||||
type TestServerHttp,
|
||||
type ToolDefinition,
|
||||
} from "@modelcontextprotocol/inspector-test-server";
|
||||
|
||||
function deferred<T>(): { promise: Promise<T>; resolve: (value: T) => void } {
|
||||
let resolve!: (value: T) => void;
|
||||
const promise = new Promise<T>((r) => {
|
||||
resolve = r;
|
||||
});
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
/**
|
||||
* A tool that never returns on its own. It reports when it started, and whether
|
||||
* the server-side request signal — which the SDK aborts when the client closes
|
||||
* this request's response stream — ever fired.
|
||||
*/
|
||||
function createSlowTool(): {
|
||||
tool: ToolDefinition;
|
||||
started: Promise<void>;
|
||||
aborted: Promise<boolean>;
|
||||
} {
|
||||
const started = deferred<void>();
|
||||
const aborted = deferred<boolean>();
|
||||
const tool: ToolDefinition = {
|
||||
name: "slow_task",
|
||||
description: "Runs until the client cancels it",
|
||||
handler: async (_params, _context, extra) => {
|
||||
extra?.signal?.addEventListener("abort", () => aborted.resolve(true), {
|
||||
once: true,
|
||||
});
|
||||
started.resolve();
|
||||
// Never settles: this call ends by cancellation or not at all, so a
|
||||
// regression shows up as the test's own timeout rather than as a pass.
|
||||
return new Promise(() => {});
|
||||
},
|
||||
};
|
||||
return { tool, started: started.promise, aborted: aborted.promise };
|
||||
}
|
||||
|
||||
async function startRemoteBackend(): Promise<{
|
||||
baseUrl: string;
|
||||
server: ServerType;
|
||||
authToken: string;
|
||||
}> {
|
||||
const { app, authToken } = createRemoteApp({
|
||||
initialConfig: { defaultEnvironment: {} },
|
||||
});
|
||||
return new Promise((resolve, reject) => {
|
||||
const server = serve(
|
||||
{ fetch: app.fetch, port: 0, hostname: "127.0.0.1" },
|
||||
(info) => {
|
||||
const port =
|
||||
info && typeof info === "object" && "port" in info
|
||||
? (info as { port: number }).port
|
||||
: 0;
|
||||
resolve({ baseUrl: `http://127.0.0.1:${port}`, server, authToken });
|
||||
},
|
||||
);
|
||||
server.on("error", reject);
|
||||
});
|
||||
}
|
||||
|
||||
describe("web-client cancellation over a per-request stream (#2140)", () => {
|
||||
let client: InspectorClient | null = null;
|
||||
let backend: ServerType | null = null;
|
||||
let upstream: TestServerHttp | null = null;
|
||||
|
||||
afterEach(async () => {
|
||||
if (client) {
|
||||
try {
|
||||
await client.disconnect();
|
||||
} catch {
|
||||
// Ignore disconnect errors.
|
||||
}
|
||||
client = null;
|
||||
}
|
||||
if (backend) {
|
||||
const server = backend;
|
||||
backend = null;
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
server.close((err) => (err ? reject(err) : resolve()));
|
||||
});
|
||||
}
|
||||
if (upstream) {
|
||||
try {
|
||||
await upstream.stop();
|
||||
} catch {
|
||||
// Ignore stop errors.
|
||||
}
|
||||
upstream = null;
|
||||
}
|
||||
});
|
||||
|
||||
it("aborts the upstream request stream, which the server observes as cancellation", async () => {
|
||||
const slow = createSlowTool();
|
||||
upstream = createTestServerHttp({
|
||||
serverInfo: createTestServerInfo("cancel-test", "1.0.0"),
|
||||
tools: [slow.tool],
|
||||
modern: {},
|
||||
});
|
||||
await upstream.start();
|
||||
|
||||
const { baseUrl, server, authToken } = await startRemoteBackend();
|
||||
backend = server;
|
||||
|
||||
const connected = new InspectorClient(
|
||||
{ type: "streamable-http", url: upstream.url },
|
||||
{
|
||||
environment: {
|
||||
transport: createRemoteTransport({ baseUrl, authToken }),
|
||||
},
|
||||
versionNegotiation: eraToVersionNegotiation("modern"),
|
||||
},
|
||||
);
|
||||
await connected.connect();
|
||||
client = connected;
|
||||
expect(connected.getProtocolEra()).toBe("modern");
|
||||
|
||||
const { tools } = await connected.listTools();
|
||||
const tool = tools.find((t) => t.name === "slow_task");
|
||||
expect(tool).toBeDefined();
|
||||
|
||||
// Hold the rejection from the first turn so the runner never sees it as
|
||||
// unhandled while we wait on the server side.
|
||||
const settled = connected.callTool(tool!, {}).catch((err: unknown) => err);
|
||||
|
||||
await slow.started;
|
||||
expect(connected.cancelToolCall()).toBe(true);
|
||||
|
||||
// The user-facing contract is unchanged: a deliberate cancel still surfaces
|
||||
// as ToolCallCancelledError, whichever wire signal carried it.
|
||||
await expect(settled).resolves.toBeInstanceOf(ToolCallCancelledError);
|
||||
|
||||
// The assertion that matters. Before the fix the server saw only a
|
||||
// `notifications/cancelled` it was free to ignore (and did), its request
|
||||
// signal never fired, and this promise never resolved.
|
||||
await expect(slow.aborted).resolves.toBe(true);
|
||||
}, 30_000);
|
||||
|
||||
it("advertises the per-request stream only for streamable-http", () => {
|
||||
// The flag is what the SDK forks on, and the factory is where a server's
|
||||
// configured type reaches the transport. stdio and SSE multiplex every
|
||||
// request over one shared channel: there is no per-request stream to close,
|
||||
// and aborting anything would take the whole session down — so they must
|
||||
// keep the `notifications/cancelled` mechanism.
|
||||
const create = createRemoteTransport({ baseUrl: "http://unused.example" });
|
||||
expect(
|
||||
create({ type: "stdio", command: "echo", args: [] }, {}).transport
|
||||
.hasPerRequestStream,
|
||||
).toBe(false);
|
||||
expect(
|
||||
create({ type: "sse", url: "http://unused.example/sse" }, {}).transport
|
||||
.hasPerRequestStream,
|
||||
).toBe(false);
|
||||
expect(
|
||||
create({ type: "streamable-http", url: "http://unused.example/mcp" }, {})
|
||||
.transport.hasPerRequestStream,
|
||||
).toBe(true);
|
||||
});
|
||||
});
|
||||
Loading…
Reference in a new issue