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);