fix(mcp): keep links from stale-session recovery and other review follow-ups

docs/agents/agent-class.md

@@ -288,7 +288,7 @@ Schedules are stored as jobs in the `cf_agents_jobs` SQL table, the Lifecycle-ow
 class MyAgent extends Agent {
   @callable()
   async addServer(url: string) {
-    // Calling this again for the same server reconnects it if it isn't
+    // Calling this again for the same server reconnects it if it is not
     // connected. A server stopped for a reason only a person can fix
     // (authorization refused, for example) needs reconnectMcpServer(id).
     return this.addMcpServer("GitHub", url, { id: "github" });

docs/agents/mcp-client.md

@@ -197,7 +197,7 @@ After 24 hours without a successful connection, the manager stops and records th
 
 #### Reconnecting on request
 
-Calling `addMcpServer()` for a server that is already registered reconnects it if it isn't connected, so calling it from `onStart`, `onConnect` or `onRequest` brings a server back once it can be reached:
+Calling `addMcpServer()` for a server that is already registered reconnects it if it is not connected, so calling it from `onStart`, `onConnect` or `onRequest` brings a server back once it can be reached:
 
 - `ready`, or a link still waiting for the user: returned as is.
 - `failed` from something that can pass (network, `5xx`, timeout) or a link that expired unused: connects now and returns the result, including a fresh `authUrl`. It tries at most once per first backoff step (30 seconds by default); a call sooner than that throws the current error.
@@ -297,7 +297,7 @@ type AddMcpServerResult =
 - **`ready`** - Server connected and tools discovered
 - **`authenticating`** - Server requires OAuth; redirect user to `authUrl`. This includes servers that accept the MCP handshake without a token but require one to list tools, prompts or resources.
 
-Calling `addMcpServer()` again for a server that is already registered returns its current state, and reconnects it when it isn't connected (see [Reconnecting on request](#reconnecting-on-request)). If the server is still connecting (for example while it is restored after the agent wakes), the call waits for that attempt and returns its result. Pass a stable `id` so the server keeps the same id everywhere.
+Calling `addMcpServer()` again for a server that is already registered returns its current state, and reconnects it when it is not connected (see [Reconnecting on request](#reconnecting-on-request)). If the server is still connecting (for example while it is restored after the agent wakes), the call waits for that attempt and returns its result. Pass a stable `id` so the server keeps the same id everywhere.
 
 ### Best practices
 

packages/agents/src/mcp/client/authorization-turns.ts

@@ -1,5 +1,11 @@
 import type { AgentMcpOAuthProvider } from "./do-oauth-client-provider";
 
+/** The `fetch` shape the SDK's HTTP transports accept. */
+export type TransportFetch = (
+  input: string | URL,
+  init?: RequestInit
+) => Promise<Response>;
+
 /** The part of the SDK's adapted auth provider a transport calls on a 401. */
 type UnauthorizedHandler = {
   onUnauthorized?: (context: unknown) => Promise<void>;
@@ -27,6 +33,24 @@ type UnauthorizedHandler = {
 export class AuthorizationTurns {
   /** Settles when every flow started so far has finished. */
   #tail: Promise<void> = Promise.resolve();
+  /** The access token each 401 response's request was sent with. */
+  readonly #sentTokens = new WeakMap<Response, string | undefined>();
+  #warnedUnwrapped = false;
+
+  /**
+   * A `fetch` for the transport that remembers which access token a request
+   * carried when it got a 401, so its authorization can tell whether a
+   * newer token already exists.
+   */
+  wrapFetch(base: TransportFetch | undefined): TransportFetch {
+    return async (input, init) => {
+      const response = await (base ?? fetch)(input, init);
+      if (response.status === 401) {
+        this.#sentTokens.set(response, bearerToken(init?.headers));
+      }
+      return response;
+    };
+  }
 
   /**
    * Take a transport's 401 handling in turns. The SDK keeps the adapted
@@ -38,13 +62,26 @@ export class AuthorizationTurns {
     const adapted = (transport as { _authProvider?: UnauthorizedHandler })
       ._authProvider;
     const onUnauthorized = adapted?.onUnauthorized?.bind(adapted);
-    if (!adapted || !onUnauthorized) return;
+    if (!adapted || !onUnauthorized) {
+      if (!this.#warnedUnwrapped) {
+        this.#warnedUnwrapped = true;
+        console.warn(
+          "[MCPClientManager] The MCP SDK transport no longer exposes its auth provider; concurrent 401s will not share one OAuth flow."
+        );
+      }
+      return;
+    }
 
     adapted.onUnauthorized = async (context) => {
-      const tokenBefore = await accessToken(provider);
+      // The token the failed request carried, else the one held now.
+      const response = responseOf(context);
+      const tokenSent =
+        response && this.#sentTokens.has(response)
+          ? this.#sentTokens.get(response)
+          : await accessToken(provider);
       await this.#run(async () => {
         const tokenNow = await accessToken(provider);
-        if (tokenNow && tokenNow !== tokenBefore) return;
+        if (tokenNow && tokenNow !== tokenSent) return;
         await onUnauthorized(context);
       });
     };
@@ -67,3 +104,16 @@ async function accessToken(
     return undefined;
   }
 }
+
+function bearerToken(headers: HeadersInit | undefined): string | undefined {
+  const value = new Headers(headers).get("authorization");
+  return value?.startsWith("Bearer ")
+    ? value.slice("Bearer ".length)
+    : undefined;
+}
+
+function responseOf(context: unknown): Response | undefined {
+  if (typeof context !== "object" || context === null) return undefined;
+  const { response } = context as { response?: unknown };
+  return response instanceof Response ? response : undefined;
+}

packages/agents/src/mcp/client/connection.ts

@@ -916,14 +916,18 @@ export class MCPClientConnection {
         return this.withAuthorizationTurns(
           new StreamableHTTPClientTransport(
             this.url,
-            this.options.transport as StreamableHTTPClientTransportOptions
+            this.authorizationTurnOptions(
+              this.options.transport as StreamableHTTPClientTransportOptions
+            )
           )
         );
       case "sse":
         return this.withAuthorizationTurns(
           new SSEClientTransport(
             this.url,
-            this.options.transport as SSEClientTransportOptions
+            this.authorizationTurnOptions(
+              this.options.transport as SSEClientTransportOptions
+            )
           )
         );
       case "rpc":
@@ -940,6 +944,17 @@ export class MCPClientConnection {
    * authorization an earlier one started, so concurrent requests share one
    * client registration and one pending link.
    */
+  /** Transport options whose `fetch` records the token each 401 carried. */
+  private authorizationTurnOptions<
+    T extends StreamableHTTPClientTransportOptions | SSEClientTransportOptions
+  >(options: T): T {
+    if (!this.options.transport.authProvider) return options;
+    return {
+      ...options,
+      fetch: this._authorizationTurns.wrapFetch(options.fetch)
+    };
+  }
+
   private withAuthorizationTurns<T extends object>(transport: T): T {
     const { authProvider } = this.options.transport;
     if (authProvider) {

packages/agents/src/mcp/client/jobs.ts

@@ -131,15 +131,16 @@ export function parseAuthExpiryJobPayload(
   if (
     !isRecord(value) ||
     typeof value.serverId !== "string" ||
-    typeof value.authUrl !== "string" ||
-    typeof value.savedAt !== "number"
+    typeof value.authUrl !== "string"
   ) {
     return undefined;
   }
   return {
     serverId: value.serverId,
     authUrl: value.authUrl,
-    savedAt: value.savedAt
+    // A job saved before links carried their age counts as old, so its
+    // link expires on the next run instead of never.
+    savedAt: typeof value.savedAt === "number" ? value.savedAt : 0
   };
 }
 

packages/agents/src/mcp/client/servers.ts

@@ -320,13 +320,20 @@ export class ServerController {
         state: conn.connectionState
       };
     }
+    if (connected.state === MCPConnectionState.AUTHENTICATING) {
+      // The fresh session needs the user: keep the link it produced.
+      return {
+        success: false,
+        error: "The server requires authorization",
+        state: conn.connectionState,
+        authUrl: connected.authUrl,
+        clientId: connected.clientId
+      };
+    }
     if (connected.state !== MCPConnectionState.CONNECTED) {
       return {
         success: false,
-        error:
-          connected.state === MCPConnectionState.FAILED
-            ? connected.error
-            : `Connection in ${connected.state} state after session re-initialization`,
+        error: connected.error,
         state: conn.connectionState
       };
     }
@@ -443,7 +450,11 @@ export class ServerController {
         "reconnect-exhausted",
         `Stopped reconnecting after ${gaveUpAfter} attempts: ${settlement.error}`
       );
+      return;
     }
+    // Show why while the reconnect waits, as a wake does, and tell clients.
+    this.ctx.connections[id].connectionError = settlement.error;
+    this.ctx.stateChanged();
   }
 
   private async connectWithRetry(

packages/agents/src/tests/capabilities/mcp-client.ts

@@ -97,6 +97,29 @@ export class PlainMcpClientObject extends DurableObject<Cloudflare.Env> {
       : { state: result.state };
   }
 
+  /**
+   * Register a server as if a session with it had been saved, so the next
+   * wake resumes that session instead of starting a new one.
+   */
+  async registerResumedServer(
+    id: string,
+    url: string,
+    callbackUrl: string,
+    sessionId: string
+  ): Promise<void> {
+    await this.lifecycle.start();
+    await this.mcp.registerServer(id, {
+      name: id,
+      url,
+      callbackUrl,
+      transport: {
+        type: "streamable-http",
+        sessionId,
+        protocolVersion: "2025-11-25"
+      }
+    });
+  }
+
   /** Wait for connection work started by callbacks, restores and jobs. */
   async settle(): Promise<void> {
     await this.lifecycle.start();

packages/agents/src/tests/mcp/client-capability.test.ts

@@ -159,4 +159,22 @@ describe("MCPClientManager capability", () => {
       });
     });
   });
+
+  it("keeps the authorization link when a forgotten session restarts into OAuth", async () => {
+    const name = crypto.randomUUID();
+    const stub = env.PlainMcpClientObject.getByName(name);
+    await stub.registerResumedServer("files", MCP_URL, CALLBACK_URL, "gone");
+
+    // The wake resumes the saved session, the server has forgotten it, and
+    // the fresh session it starts needs the user to authorize.
+    const restarted = await restart(stub, name);
+    await restarted.settle();
+
+    expect(await restarted.connectionStatus("files")).toMatchObject({
+      state: "authenticating"
+    });
+    expect(await restarted.mcpJobs()).toMatchObject([
+      { id: "auth-expiry:files" }
+    ]);
+  });
 });

packages/agents/src/tests/mcp/client-manager.test.ts

@@ -3056,8 +3056,11 @@ describe("MCPClientManager OAuth Integration", () => {
 
         await manager.establishConnection(id);
 
-        // Should fire 3 times: 1 from connectToServer (after init), 1 explicit in establishConnection, 1 from discoverIfConnected
-        expect(onStateChangedSpy).toHaveBeenCalledTimes(3);
+        // Fires 4 times: after init in connectToServer, from
+        // discoverIfConnected, when the scheduled reconnect sets the error,
+        // and explicitly in establishConnection.
+        expect(onStateChangedSpy).toHaveBeenCalledTimes(4);
+        expect(conn.connectionError).toBe("Discovery failed");
       })
     );
 

packages/agents/src/tests/mcp/integration/durable-jobs.test.ts

@@ -374,9 +374,13 @@ describe("MCP servers an app adds again", () => {
     server.options.up = true;
     const methodsBefore = server.mcpMethods.length;
 
-    expect(
-      await agent.connectServer(MCP_URL, { callbackHost: CLIENT_ORIGIN })
-    ).toMatchObject({ state: "error" });
+    const added = await agent.connectServer(MCP_URL, {
+      callbackHost: CLIENT_ORIGIN
+    });
+    expect(added).toMatchObject({ state: "error" });
+    if (added.state !== "error") throw new Error("unreachable");
+    // It reports why the server is down, not a generic failure.
+    expect(added.message).toContain("503");
     expect(server.mcpMethods.length).toBe(methodsBefore);
     expect((await agent.mcpJobs()).map((job) => job.id)).toEqual([
       `connect:${id}`

packages/agents/src/tests/mcp/integration/oauth-mcp-server.ts

@@ -283,6 +283,10 @@ export function createOAuthMcpServer({
     }
     await waitIfHeld("mcp");
     if (!options.up) return new Response("unavailable", { status: 503 });
+    // This server keeps no sessions, so any session id is one it forgot.
+    if (request.headers.has("mcp-session-id")) {
+      return new Response("Session not found", { status: 404 });
+    }
 
     const token = (request.headers.get("authorization") ?? "").replace(
       /^Bearer /,