Skip to content

refactor(mcp): split MCPClientManager into a thin facade and three controllers #2759

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
mattzcarey wants to merge 1 commit into from
Closed

refactor(mcp): split MCPClientManager into a thin facade and three controllers #2759

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 21 additions & 11 deletions design/mcp.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,20 +85,30 @@ supporting modules live in `src/mcp/client/`; `types.ts`, `rpc.ts`, and

The SDK v2 client negotiates Stateless first and falls back to Legacy on the same connection. Agents persists the selected protocol and discovery result alongside resumable HTTP state. An in-flight elicitation handler and continuation remain memory-only; isolate restart rejects the operation and the caller retries it.

`MCPClientManager` (`src/mcp/client/index.ts`) is the capability facade. It keeps one `MCPClientConnection` per server and delegates to:

| Module | Owns |
| ------------------- | ------------------------------------------------------------------------------------ |
| `server-store.ts` | Every SQL statement over `cf_agents_mcp_servers` |
| `server-lanes.ts` | One queue of connection work per server; re-entrant for nested calls |
| `oauth-callback.ts` | Callback recognition and parsing, state checks, code exchange, response rendering |
| `ai-tools.ts` | AI SDK tool conversion and its per-connection schema cache |
| `url-policy.ts` | Stable server ids and the SSRF policy |
| `connection.ts` | Transport selection, the SDK client, catalogs and the connection-level state machine |
`MCPClientManager` (`src/mcp/client/manager.ts`) is the public API and the Lifecycle hooks; it decides nothing itself. It keeps one `MCPClientConnection` per server and hands every controller one `McpClientContext` (`context.ts`): the connections, the store, the lanes, storage, and the manager's events. Dependencies point one way, top to bottom:

| Module | Owns |
| --------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `servers.ts` | Each server's life: register, connect, discover, restore, settle, reconnect, close, remove, migrate. `afterFailure` is the one place that decides retry or stop. |
| `authorization.ts` | Links and OAuth: entering `authenticating` (round limit, spent links), callbacks, request recovery, link expiry, provider creation and storage |
| `status.ts` | The transitions both share: stop with a recorded failure, ready, save the session |
| `jobs.ts` | The durable `connect` and `auth-expiry` jobs and the reconnect policy; schedules, moves and cancels, decides nothing |
| `elicitation.ts` | Elicitation handlers and the capabilities they advertise |
| `legacy-connect.ts` | The deprecated `connect()` |
| `server-store.ts` | Every SQL statement over `cf_agents_mcp_servers` |
| `server-lanes.ts` | One queue of connection work per server; re-entrant for nested calls |
| `authorization-turns.ts` | One OAuth flow at a time per connection, so concurrent 401s share one client and one link |
| `authorization-rounds.ts` | The step-up round count, as pure functions |
| `oauth-callback.ts` | Callback recognition and parsing, state checks, code exchange, response rendering |
| `catalog.ts`, `ai-tools.ts` | Paginated catalogs, namespaced listings and filters; AI SDK tool conversion |
| `url-policy.ts` | Stable server ids and the SSRF policy |
| `connection.ts` | Transport selection, the SDK client, catalogs and the connection-level state machine |

`servers.ts` reaches `connectToServer` and `discoverIfConnected` through the manager, so a subclass override (or a test spy) sees internal calls too.

Connect, discovery, establish, OAuth callback completion and restore for one server run through its lane, so they never interleave on one connection.

Background work that must happen even if the isolate dies lives in the Lifecycle job queue (`reconnect-policy.ts`), not in memory:
Background work that must happen even if the isolate dies lives in the Lifecycle job queue (`jobs.ts`), not in memory:

- `connect:<serverId>`: after a failed connect or discovery, the next attempt on a 30s → 30m backoff. Each run is one bounded attempt. Restore leaves a server with a pending job `failed` (showing the job's last error) instead of connecting early. Ready or `authenticating` cancels it; 24h without success records a `reconnect-exhausted` failure. A refused authorization is never retried: a 401 no link can answer, a 403 left after any step-up, or an OAuth error other than `server_error` / `temporarily_unavailable` records an `authorization-failed` failure instead of scheduling the job. `migrateServerId` moves both jobs to the new id.
- `auth-expiry:<serverId>`: armed with every persisted `auth_url`. If the link's `state` can no longer be redeemed when it fires, the server records an `authorization-expired` failure instead of minting a new link. The OAuth callback clears `auth_url` before it redeems the state, so a dead link found on wake was never completed and is recorded the same way.
Expand Down
8 changes: 6 additions & 2 deletions packages/agents/AGENTS.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,7 +109,11 @@ src/
rpc.ts # RPC transports shared by client + server
abort.ts # Shared abort/race helper
client/ # MCP client submodule (agents/mcp/client)
index.ts # MCPClientManager for connecting to remote MCP servers
index.ts # Public exports for agents/mcp/client
manager.ts # MCPClientManager: public API and Lifecycle hooks, delegates to:
servers.ts # each server's connection life; decides retry or stop after a failure
authorization.ts # links, OAuth callbacks, re-authorization during requests
jobs.ts # durable reconnect and link-expiry jobs
connection.ts
rpc.ts # Persisted RPC binding restoration
storage.ts
Expand Down Expand Up @@ -403,7 +407,7 @@ AI evaluation suite (scheduling accuracy, etc.). Requires API keys in `.env`.
- **Sub-agents are facets** — `subAgent(Cls, name)` creates or resolves a child DO colocated on the same machine. Clients reach a child via `/agents/{parent}/{name}/sub/{child}/{name}` and `useAgent({ sub: [...] })`. Parents gate access with `onBeforeSubAgent`; children reach their parent with `parentAgent(Cls)` or `parentPath`.
- **Lifecycle owns the job queue and the physical alarm** — capabilities and the host push jobs (`capability` + `fn` + due time + payload) into the `cf_agents_jobs` table; Lifecycle drives due jobs as an alarm event loop with retry, deferral, and the memory-limit circuit breaker, and derives the physical alarm from queue state. Queue mutations re-arm automatically. Each capability keeps its own durable state rather than depending on Scheduler.
- **Scheduling uses cron-schedule** — `Scheduler` is the vocabulary over the job queue: it validates schedules, resolves named callbacks, and pushes jobs whose `fn` is the callback name. `Agent` installs the same primitive and delegates `this.schedule()` and related APIs to it. Schedules persist in SQLite and survive hibernation.
- **MCP has separate package boundaries** — `mcp/server/index.ts` is the Stateless Worker wrapper; `mcp/client/index.ts` connects Agents to external servers; `mcp/index.ts` is a compatibility barrel for retained Legacy APIs whose implementation lives in `mcp/server/legacy-agent.ts`.
- **MCP has separate package boundaries** — `mcp/server/index.ts` is the Stateless Worker wrapper; `mcp/client/manager.ts` (exported from `mcp/client/index.ts`) connects Agents to external servers; `mcp/index.ts` is a compatibility barrel for retained Legacy APIs whose implementation lives in `mcp/server/legacy-agent.ts`.
- **Telemetry has an independent schema version** — `instrumentation_scope.version` is hardcoded to `"1"`; it is not the package version. Notify Workers Observability and any other downstream consumers before bumping it.

## Boundaries
Expand Down
10 changes: 5 additions & 5 deletions packages/agents/src/mcp/client/authorization-rounds.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,13 +15,13 @@
*/

/** Completed authorization rounds allowed for one scope set. */
export const MAX_AUTHORIZATION_ROUNDS = 3;
const MAX_AUTHORIZATION_ROUNDS = 3;

/** A scope set's rounds are forgotten this long after its latest round. */
export const AUTHORIZATION_ROUNDS_WINDOW_MS = 60 * 60 * 1000;
const AUTHORIZATION_ROUNDS_WINDOW_MS = 60 * 60 * 1000;

/** Rounds recorded for one requested scope set. */
export type ScopeRounds = {
type ScopeRounds = {
/** Rounds started for this scope set within the window. */
readonly rounds: number;
/** Whether this scope set's latest round's callback exchanged a code. */
Expand All @@ -31,15 +31,15 @@ export type ScopeRounds = {
};

/** Authorization rounds recorded for one server. */
export type AuthorizationRounds = {
type AuthorizationRounds = {
/** The scope set of the round started most recently. */
readonly latest: string;
/** Rounds per normalized scope set; `""` when no scope was requested. */
readonly byScope: Readonly<Record<string, ScopeRounds>>;
};

/** Whether the next authorization link may be shown. */
export type RoundDecision =
type RoundDecision =
| { readonly _tag: "start"; readonly rounds: AuthorizationRounds }
| {
readonly _tag: "exhausted";
Expand Down
Loading
Loading