From da0eadf0433d29516287c7e3d7b3db325a446281 Mon Sep 17 00:00:00 2001 From: Armin Ronacher Date: Mon, 12 Oct 2026 15:33:15 +0200 Subject: [PATCH] fix(mcp): do not retry cancelled HTTP requests after an authorization refresh A tool call that was aborted or timed out while waiting for a shared OAuth refresh was still retried by StreamableHttpTransport once the refresh finished, so the server executed a call the caller had already given up on. McpClient now passes a per-request signal to McpTransport.send(), aborted once the request settles (cancelled, timed out, answered, or closed). StreamableHttpTransport checks it before each dispatch. The shared refresh itself is not aborted, so other requests waiting on it still succeed. fixes #10598 --- packages/mcp/CHANGELOG.md | 1 + packages/mcp/src/client.ts | 8 ++- packages/mcp/src/index.ts | 1 + .../mcp/src/transports/streamable-http.ts | 28 +++++++-- packages/mcp/src/transports/transport.ts | 11 +++- packages/mcp/test/streamable-http.test.ts | 60 +++++++++++++++++++ 6 files changed, 101 insertions(+), 8 deletions(-) diff --git a/packages/mcp/CHANGELOG.md b/packages/mcp/CHANGELOG.md index f46a5d781..0643fe652 100644 --- a/packages/mcp/CHANGELOG.md +++ b/packages/mcp/CHANGELOG.md @@ -9,6 +9,7 @@ ### Fixed - Fixed `authorizeMcp()` exchanging an authorization code whose `iss` parameter could not be checked because no authorization server metadata was discovered. It now throws `OAuthIssuerMismatchError`, whose `expected` is `undefined` in that case (RFC 9207). Reported by [@AizenvoltPrime](https://github.com/AizenvoltPrime) +- Fixed `StreamableHttpTransport` retrying a request after an authorization refresh even though the caller had already cancelled it or it had timed out, so the server still executed it. `McpTransport.send()` now takes an optional `signal`, which `McpClient` aborts once a request settles ([#10598](https://github.com/earendil-works/pi/issues/10598)) ## [1.1.0] - 2026-10-07 diff --git a/packages/mcp/src/client.ts b/packages/mcp/src/client.ts index e749f5cbf..90997aa4c 100644 --- a/packages/mcp/src/client.ts +++ b/packages/mcp/src/client.ts @@ -64,6 +64,8 @@ interface PendingRequest { timer: ReturnType | undefined; signal: AbortSignal | undefined; onAbort: () => void; + /** Aborted when the request settles, so the transport does not dispatch (or retry) it afterwards. */ + sendController: AbortController; cancellable: boolean; onProgress: ((progress: ProgressNotification) => void) | undefined; progressToken: JsonRpcId | undefined; @@ -426,6 +428,7 @@ export class McpClient { method !== "initialize", String(options.signal?.reason ?? "Aborted"), ), + sendController: new AbortController(), cancellable: method !== "initialize", onProgress: options.onProgress, progressToken, @@ -434,7 +437,9 @@ export class McpClient { if (progressToken !== undefined) this.progressRequests.set(progressToken, id); options.signal?.addEventListener("abort", entry.onAbort, { once: true }); this.armTimeout(id, entry); - transport.send(message).catch((error) => this.cancelPending(id, error, false)); + transport + .send(message, { signal: entry.sendController.signal }) + .catch((error) => this.cancelPending(id, error, false)); }); } @@ -574,6 +579,7 @@ export class McpClient { if (entry.timer) clearTimeout(entry.timer); if (entry.progressToken !== undefined) this.progressRequests.delete(entry.progressToken); entry.signal?.removeEventListener("abort", entry.onAbort); + entry.sendController.abort(); } private rejectPending(error: unknown): void { diff --git a/packages/mcp/src/index.ts b/packages/mcp/src/index.ts index 6b9235fae..3d0f8fc03 100644 --- a/packages/mcp/src/index.ts +++ b/packages/mcp/src/index.ts @@ -68,4 +68,5 @@ export type { McpTransportCloseListener, McpTransportErrorListener, McpTransportMessageListener, + McpTransportSendOptions, } from "./transports/transport.ts"; diff --git a/packages/mcp/src/transports/streamable-http.ts b/packages/mcp/src/transports/streamable-http.ts index 662465100..af9c922d2 100644 --- a/packages/mcp/src/transports/streamable-http.ts +++ b/packages/mcp/src/transports/streamable-http.ts @@ -5,11 +5,17 @@ import { JSON_RPC_ERROR_CODES, type JsonRpcId, type JsonRpcMessage, + McpAbortError, McpConnectionClosedError, parseJsonRpcMessage, toError, } from "../protocol/jsonrpc.ts"; -import { DEFAULT_MAX_MESSAGE_BYTES, type McpTransport, TransportEvents } from "./transport.ts"; +import { + DEFAULT_MAX_MESSAGE_BYTES, + type McpTransport, + type McpTransportSendOptions, + TransportEvents, +} from "./transport.ts"; const MAX_ERROR_BODY_BYTES = 8 * 1024; const ERROR_MESSAGE_BODY_CHARS = 500; @@ -222,12 +228,16 @@ export class StreamableHttpTransport extends TransportEvents implements McpTrans this.protocolVersion = version; } - async send(message: JsonRpcMessage): Promise { + async send(message: JsonRpcMessage, { signal }: McpTransportSendOptions = {}): Promise { if (!this.started || this.closed) throw new McpConnectionClosedError(); - const response = await this.authorizedFetch("POST", { - headers: { accept: "application/json, text/event-stream", "content-type": "application/json" }, - body: JSON.stringify(message), - }); + const response = await this.authorizedFetch( + "POST", + { + headers: { accept: "application/json, text/event-stream", "content-type": "application/json" }, + body: JSON.stringify(message), + }, + signal, + ); await this.checkResponse(response); this.captureSession(response); @@ -277,14 +287,20 @@ export class StreamableHttpTransport extends TransportEvents implements McpTrans /** * Fetch with auth headers. A 401 (or a 403 asking for more scope) is handed to the auth provider * once, and the request is retried with whatever credentials it left behind. + * + * `signal` only stops dispatch: getting a token or the authorization recovery can take a while, and a + * request its sender gave up on in the meantime must not reach the server. It does not abort the shared + * recovery, which other requests may wait for. */ private async authorizedFetch( method: "GET" | "POST", init: { headers: Record; body?: string }, + signal?: AbortSignal, ): Promise { const onUnauthorized = this.options.authProvider?.onUnauthorized?.bind(this.options.authProvider); for (let attempt = 0; ; attempt++) { const { headers, token } = await this.headers(init.headers); + if (signal?.aborted) throw new McpAbortError(); const response = await this.fetch(this.url, { method, headers, diff --git a/packages/mcp/src/transports/transport.ts b/packages/mcp/src/transports/transport.ts index 353486e60..702f041df 100644 --- a/packages/mcp/src/transports/transport.ts +++ b/packages/mcp/src/transports/transport.ts @@ -6,9 +6,18 @@ export type McpTransportMessageListener = (message: JsonRpcMessage) => void; export type McpTransportErrorListener = (error: Error) => void; export type McpTransportCloseListener = () => void; +export interface McpTransportSendOptions { + /** + * Aborted once the sender no longer waits for a reply (the request was cancelled, timed out, answered, + * or the client closed). A transport must not dispatch the message after that, for example when + * retrying it after an authorization refresh. It does not undo a message already dispatched. + */ + signal?: AbortSignal; +} + export interface McpTransport { start(): Promise; - send(message: JsonRpcMessage): Promise; + send(message: JsonRpcMessage, options?: McpTransportSendOptions): Promise; close(): Promise; onMessage(listener: McpTransportMessageListener): () => void; onError(listener: McpTransportErrorListener): () => void; diff --git a/packages/mcp/test/streamable-http.test.ts b/packages/mcp/test/streamable-http.test.ts index e0c0b25ad..837508e81 100644 --- a/packages/mcp/test/streamable-http.test.ts +++ b/packages/mcp/test/streamable-http.test.ts @@ -342,6 +342,66 @@ describe("StreamableHttpTransport", () => { await client.close(); }); + // #10598: a request cancelled or timed out while waiting for a shared refresh must not be retried. + it("does not retry requests that were cancelled while authorization was recovering", async () => { + let token = "old"; + const executed: unknown[] = []; + const { url } = await startServer(async (request, response, requests) => { + if (request.method !== "POST") return protocolHandler(request, response, requests); + const message = JSON.parse(await readBody(request)) as Record; + if (message.method === "tools/call" && request.headers.authorization === "Bearer old") { + response.writeHead(401, { "www-authenticate": "Bearer" }).end(); + return; + } + if (message.method === "tools/call") executed.push((message.params as { name: string }).name); + return protocolHandler(request, response, requests, message); + }); + let releaseRefresh = () => {}; + const refreshGate = new Promise((resolve) => { + releaseRefresh = resolve; + }); + let refresh: Promise | undefined; + let refreshes = 0; + let waiting = 0; + let allWaiting = () => {}; + const waitingGate = new Promise((resolve) => { + allWaiting = resolve; + }); + const client = new McpClient({ name: "http-test", version: "1.0.0" }); + await client.connect( + new StreamableHttpTransport({ + url, + openGetStream: false, + authProvider: { + token: async () => token, + onUnauthorized: async () => { + if (!refresh) { + refreshes++; + refresh = refreshGate.then(() => { + token = "new"; + }); + } + if (++waiting === 3) allWaiting(); + await refresh; + }, + }, + }), + ); + const controller = new AbortController(); + const aborted = client.callTool("aborted", {}, { signal: controller.signal }); + const timedOut = client.callTool("timed-out", {}, { timeoutMs: 200 }); + const kept = client.callTool("kept"); + await waitingGate; + controller.abort(); + await expect(aborted).rejects.toMatchObject({ name: "AbortError" }); + await expect(timedOut).rejects.toMatchObject({ name: "McpTimeoutError" }); + releaseRefresh(); + expect(await kept).toEqual({ content: [{ type: "text", text: "hello" }] }); + expect(executed).toEqual(["kept"]); + expect(refreshes).toBe(1); + await client.close(); + }); + // #10565: the auth provider may refresh over the network, which closing must not wait for. it("closes the session with the last request's token, without asking the auth provider", async () => { const { url, requests } = await startServer(protocolHandler);