MCPClientManager

agents/mcp/client 中的 MCPClientManager 可将 Agent 连接到 Model Context Protocol (MCP) 服务器。它会将你添加的服务器存储在 Durable Object 的 SQLite 数据库中，在对象启动时重新连接，并在你的 onRequest() 运行前处理 OAuth 回调。

安装 capability

安装 Agents SDK 和 MCP 客户端包：

npm i agents @modelcontextprotocol/[email protected]

yarn add agents @modelcontextprotocol/[email protected]

pnpm add agents @modelcontextprotocol/[email protected]

bun add agents @modelcontextprotocol/[email protected]

使用客户端名称和版本构造管理器；它会在握手期间将这些信息发送给服务器。然后将其安装到 Lifecycle 上：

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { MCPClientManager } from "agents/mcp/client";
import { routeAgentRequest } from "agents/routing";

export class ToolAgent extends DurableObject {
	mcp = new MCPClientManager("tool-agent", "1.0.0");
	lifecycle = Lifecycle.install(this).use(this.mcp);

	async onRequest(request) {
		const url = new URL(request.url);
		if (request.method !== "POST" || !url.pathname.endsWith("/connect")) {
			return Response.json({ tools: this.mcp.listTools() });
		}

		await this.mcp.registerServer("docs", {
			name: "Cloudflare Docs",
			url: "https://docs.mcp.cloudflare.com/mcp",
			callbackUrl: `${url.origin}/agents/tool-agent/${this.lifecycle.name}/callback`,
			transport: { type: "auto" },
		});

		const connection = await this.mcp.connectToServer("docs");
		if (connection.state === "authenticating") {
			return Response.redirect(connection.authUrl);
		}
		if (connection.state === "connected") {
			await this.mcp.discoverIfConnected("docs");
		}
		return Response.json({ state: connection.state });
	}
}

export default {
	async fetch(request, env) {
		return (
			(await routeAgentRequest(request, env)) ??
			new Response("Not found", { status: 404 })
		);
	},
};

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { MCPClientManager } from "agents/mcp/client";
import { routeAgentRequest } from "agents/routing";

export class ToolAgent extends DurableObject<Env> {
	readonly mcp = new MCPClientManager("tool-agent", "1.0.0");
	readonly lifecycle = Lifecycle.install(this).use(this.mcp);

	async onRequest(request: Request): Promise<Response> {
		const url = new URL(request.url);
		if (request.method !== "POST" || !url.pathname.endsWith("/connect")) {
			return Response.json({ tools: this.mcp.listTools() });
		}

		await this.mcp.registerServer("docs", {
			name: "Cloudflare Docs",
			url: "https://docs.mcp.cloudflare.com/mcp",
			callbackUrl: `${url.origin}/agents/tool-agent/${this.lifecycle.name}/callback`,
			transport: { type: "auto" },
		});

		const connection = await this.mcp.connectToServer("docs");
		if (connection.state === "authenticating") {
			return Response.redirect(connection.authUrl);
		}
		if (connection.state === "connected") {
			await this.mcp.discoverIfConnected("docs");
		}
		return Response.json({ state: connection.state });
	}
}

export default {
	async fetch(request: Request, env: Env): Promise<Response> {
		return (
			(await routeAgentRequest(request, env)) ??
			new Response("Not found", { status: 404 })
		);
	},
} satisfies ExportedHandler<Env>;

构造函数的第三个参数接受以下选项：

选项	说明
`env`	Worker 的绑定。若要恢复通过绑定访问的服务器，此项为必填。
`createAuthProvider`	`(callbackUrl) => AgentMcpOAuthProvider`。为已存储的 HTTP 服务器构建 OAuth 提供程序。

添加服务器

添加服务器需要调用三个方法：

registerServer(id, options) 会存储服务器，但不会连接。
connectToServer(id) 会打开传输连接。
discoverIfConnected(id) 会列出服务器的工具、提示和资源。

registerServer() 接受以下选项：

选项	说明
`url`	服务器 URL。系统会拒绝私有和内部地址。
`name`	显示名称
`callbackUrl`	OAuth 提供程序将用户重定向回来的地址。请将其路由到名称相同的对象。
`transport`	`type` 可以是 `"auto"`、`"streamable-http"` 或 `"sse"`。还接受 `authProvider` 和 MCP 传输选项，例如用于设置标头的 `requestInit`。
`client`	MCP 客户端选项，例如要公布的 `capabilities`。
`clientId`	预先注册的 OAuth 客户端 ID
`retry`	连接和重新连接的重试策略。默认重试三次，基础延迟为 500 毫秒，最大延迟为 5 秒。

connectToServer() 会返回以下状态之一：

状态	含义
`"connected"`	传输连接已打开。调用 `discoverIfConnected()` 加载工具。
`"authenticating"`	服务器需要 OAuth。将用户重定向到 `authUrl`。
`"failed"`	连接失败。结果中包含 `error` 消息。

discoverIfConnected(id, { timeoutMs? }) 默认最多等待 30 秒，并返回 { success, state, error? }。

对象从内存中移除后，服务器仍会保持注册状态。下次启动时，管理器会在后台重新连接所有已存储的服务器。

处理 OAuth

服务器需要 OAuth 时，提供程序会将用户重定向回你注册的 callbackUrl。管理器是一种选择性处理的能力：它会根据 state 参数匹配回调，完成令牌交换并建立连接，然后在你的 onRequest() 运行前作出响应。其他所有请求都会继续传递。

默认情况下，管理器会将用户重定向到回调 URL 的源站。若要选择响应方式，请在 onStart() 中调用 configureOAuthCallback()：

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { MCPClientManager } from "agents/mcp/client";

export class ToolAgent extends DurableObject {
	mcp = new MCPClientManager("tool-agent", "1.0.0");
	lifecycle = Lifecycle.install(this).use(this.mcp);

	onStart() {
		this.mcp.configureOAuthCallback({
			successRedirect: "/settings/integrations",
			errorRedirect: "/settings/integrations",
		});
	}
}

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { MCPClientManager } from "agents/mcp/client";

export class ToolAgent extends DurableObject<Env> {
	readonly mcp = new MCPClientManager("tool-agent", "1.0.0");
	readonly lifecycle = Lifecycle.install(this).use(this.mcp);

	onStart(): void {
		this.mcp.configureOAuthCallback({
			successRedirect: "/settings/integrations",
			errorRedirect: "/settings/integrations",
		});
	}
}

失败时，管理器会将错误作为 error 查询参数添加到 errorRedirect URL。你也可以传入 customHandler: (result) => Response 来返回自定义页面，例如关闭 OAuth 弹出窗口的页面。错误文本可能来自 OAuth 提供程序，因此在渲染前请先转义。

如需使用自己的 OAuth 存储，请将 createAuthProvider 传给构造函数，或将 transport.authProvider 传给 registerServer()。更多信息，请参阅 MCP 客户端的 OAuth。

使用工具、提示和资源

方法	返回值
`listTools(filter?)`	所有已连接服务器提供的 MCP 工具，每个工具都带有对应的 `serverId`。
`getAITools(filter?)`	转换为 AI SDK 工具集的相同工具，可直接用于 `generateText()` 或 `streamText()`。
`callTool({ serverId, name, arguments })`	一次工具调用的结果
`listPrompts(filter?)`	所有已连接服务器提供的提示
`getPrompt({ serverId, name, arguments })`	一个提示
`listResources(filter?)`	所有已连接服务器提供的资源
`listResourceTemplates(filter?)`	所有已连接服务器提供的资源模板
`readResource({ serverId, uri })`	一个资源的内容

filter 可按 serverId、serverName 或连接 state 限制结果。每个字段都接受单个值或数组。

对象唤醒后，管理器会在后台重新连接。读取工具前请调用 waitForConnections()，以便包含仍在重新连接的服务器。此方法不会拒绝 Promise。若未指定 timeout，它会一直等待，直到所有待处理连接都完成。

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { MCPClientManager } from "agents/mcp/client";
import { generateText, stepCountIs } from "ai";

export class ToolAgent extends DurableObject {
	mcp = new MCPClientManager("tool-agent", "1.0.0");
	lifecycle = Lifecycle.install(this).use(this.mcp);

	async onRequest(request) {
		await this.mcp.waitForConnections({ timeout: 10_000 });
		const { text } = await generateText({
			model,
			prompt: await request.text(),
			tools: this.mcp.getAITools(),
			stopWhen: stepCountIs(5),
		});
		return new Response(text);
	}

	// RPC methods do not start the Lifecycle on their own.
	async callDocsSearch(query) {
		await this.lifecycle.start();
		return this.mcp.callTool({
			serverId: "docs",
			name: "search",
			arguments: { query },
		});
	}
}

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { MCPClientManager } from "agents/mcp/client";
import { generateText, stepCountIs, type LanguageModel } from "ai";

declare const model: LanguageModel;

export class ToolAgent extends DurableObject<Env> {
	readonly mcp = new MCPClientManager("tool-agent", "1.0.0");
	readonly lifecycle = Lifecycle.install(this).use(this.mcp);

	async onRequest(request: Request): Promise<Response> {
		await this.mcp.waitForConnections({ timeout: 10_000 });
		const { text } = await generateText({
			model,
			prompt: await request.text(),
			tools: this.mcp.getAITools(),
			stopWhen: stepCountIs(5),
		});
		return new Response(text);
	}

	// RPC methods do not start the Lifecycle on their own.
	async callDocsSearch(query: string) {
		await this.lifecycle.start();
		return this.mcp.callTool({
			serverId: "docs",
			name: "search",
			arguments: { query },
		});
	}
}

响应交互请求

MCP 服务器可以在工具调用、提示或资源读取期间向客户端请求输入。请为 form 模式、url 模式或两者配置处理程序。连接只会公布你已配置处理程序的模式，因此不支持相应模式的服务器会使用自己的备用方案：

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { MCPClientManager } from "agents/mcp/client";

export class ToolAgent extends DurableObject {
	mcp = new MCPClientManager("tool-agent", "1.0.0");
	lifecycle = Lifecycle.install(this).use(this.mcp);

	onStart() {
		this.mcp.configureElicitationHandlers({
			form: async (request, serverId) => {
				console.log(`${serverId} asks:`, request.params.message);
				// Collect values that match request.params.requestedSchema.
				return { action: "decline" };
			},
		});
	}
}

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { MCPClientManager } from "agents/mcp/client";

export class ToolAgent extends DurableObject<Env> {
	readonly mcp = new MCPClientManager("tool-agent", "1.0.0");
	readonly lifecycle = Lifecycle.install(this).use(this.mcp);

	onStart(): void {
		this.mcp.configureElicitationHandlers({
			form: async (request, serverId) => {
				console.log(`${serverId} asks:`, request.params.message);
				// Collect values that match request.params.requestedSchema.
				return { action: "decline" };
			},
		});
	}
}

请在 onStart() 中配置处理程序，以确保每次唤醒后都已就绪。公布的模式会与每台服务器一起存储，因此恢复连接时会再次公布这些模式。处理程序和正在进行的调用只存在于内存中。如果对象在等待输入的调用期间被逐出，请在连接恢复后重试该调用。

管理服务器

方法	说明
`listServers()`	返回所有已存储的服务器
`mcpConnections[id]`	当前连接。`connectionState` 会报告其状态。
`closeConnection(id)`	关闭一个连接，但保留服务器记录
`closeAllConnections()`	关闭所有连接，但保留服务器记录
`removeServer(id)`	关闭连接并删除服务器
`onServerStateChanged(listener)`	添加、连接失败或移除服务器时调用 `listener`。返回一个可释放对象。
`onObservabilityEvent(listener)`	每当发生 `mcp:client:*` 事件时调用 `listener`。

例如，服务器列表发生变化时，将其推送给已连接的客户端：

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { MCPClientManager } from "agents/mcp/client";
import { WebSockets } from "agents/websockets";

export class ToolAgent extends DurableObject {
	mcp = new MCPClientManager("tool-agent", "1.0.0");
	webSockets = new WebSockets();
	lifecycle = Lifecycle.install(this).use(this.mcp).use(this.webSockets);

	constructor(ctx, env) {
		super(ctx, env);
		this.mcp.onServerStateChanged(() => {
			const servers = this.mcp.listServers().map(({ id, name }) => ({
				id,
				name,
				state: this.mcp.mcpConnections[id]?.connectionState,
			}));
			for (const connection of this.webSockets.getConnections()) {
				connection.send(JSON.stringify({ type: "mcp-servers", servers }));
			}
		});
	}
}

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { MCPClientManager } from "agents/mcp/client";
import { WebSockets } from "agents/websockets";

export class ToolAgent extends DurableObject<Env> {
	readonly mcp = new MCPClientManager("tool-agent", "1.0.0");
	readonly webSockets = new WebSockets();
	readonly lifecycle = Lifecycle.install(this)
		.use(this.mcp)
		.use(this.webSockets);

	constructor(ctx: DurableObjectState, env: Env) {
		super(ctx, env);
		this.mcp.onServerStateChanged(() => {
			const servers = this.mcp.listServers().map(({ id, name }) => ({
				id,
				name,
				state: this.mcp.mcpConnections[id]?.connectionState,
			}));
			for (const connection of this.webSockets.getConnections()) {
				connection.send(JSON.stringify({ type: "mcp-servers", servers }));
			}
		});
	}
}

对象销毁时，管理器会关闭所有连接。

通过 RPC 调用管理器

RPC 调用不会经过 fetch()，因此不会启动生命周期。在每个使用 capability 或使用 onStart() 所设置内容的 RPC 方法开头，调用 await this.lifecycle.start()。

前一个示例中的 callDocsSearch() 会在读取工具或服务器之前启动 Lifecycle。

存储

管理器会将服务器存储在 cf_agents_mcp_servers 表中。默认 OAuth 提供程序会将客户端注册信息和令牌存储在对象的键值存储中。

在 Agent 类上使用

Agent 类会将 MCPClientManager 安装为 this.mcp。其 addMcpServer() 方法会一次性完成注册、连接和发现步骤，并为你构建回调 URL。removeMcpServer() 和 getMcpServers() 也会封装管理器；此外，Agent 会将服务器列表发送给客户端，以供 useAgent({ onMcpUpdate }) 使用。更多信息，请参阅 McpClient。