CAMEL Agent 接入 MCP 生态：用 MCPToolkit 将多 MCP 服务器工具接入智能体实战指南-平芜编程栈

CAMEL Agent 接入 MCP 生态：用 MCPToolkit 将多 MCP 服务器工具接入智能体实战指南

【免费下载链接】camel🐫 CAMEL: The first and the best multi-agent framework. Finding the Scaling Law of Agents. https://www.camel-ai.org项目地址: https://gitcode.com/GitHub_Trending/ca/camel

本篇技术指南围绕 CAMEL 多智能体框架中的MCP（Model Context Protocol）客户端能力展开，讲解如何通过配置文件声明 MCP 服务器、使用MCPToolkit统一管理连接、再把服务器工具直接挂载到ChatAgent上使用。读完本文，你将掌握本地 stdio 服务器、云端 sse / streamable-http 服务器的完整接入流程，理解MCPToolkit、MCPClient、MCPAgent三层实现的协作原理，并能借助 ACI.dev 注册、PulseMCP 搜索快速扩展自己的 MCP 工具库。

什么是「CAMEL Agent 作为 MCP 客户端」

MCP（Model Context Protocol）为 LLM 应用定义了统一的「工具服务器」接入协议。在 CAMEL 中，智能体既可以作为 MCP服务器对外暴露工具（见 camel_toolkits_as_an_mcp_server.md），也可以作为 MCP客户端去消费外部服务器提供的工具。本文聚焦后者：让 CAMEL Agent 连接一个或多个 MCP 服务器，把 GitHub、Gmail、Notion、ArXiv、文件系统等外部能力直接变成智能体可以调用的函数工具。

整个接入过程只有三个步骤：

创建配置文件：告诉 CAMEL 要连接哪些 MCP 服务器（本地或远程，每种服务器带一种传输方式）；
使用MCPToolkit连接：加载配置文件建立与服务器的连接；
在 CAMEL Agent 中启用工具：把服务器工具列表传给ChatAgent使用。

对应的核心实现分布在三个文件中：工具聚合层 camel/toolkits/mcp_toolkit.py（MCPToolkit）、连接层 camel/utils/mcp_client.py（MCPClient）以及开箱即用的智能体封装 camel/agents/mcp_agent.py（MCPAgent）。

第一步：配置 MCP 服务器

配置文件采用 MCP 社区标准的 JSON 格式，顶层为mcpServers字典，每个键是服务器名，值是该服务器的启动参数。CAMEL 支持在同一份配置中混合定义本地与远程服务器。

本地服务器（stdio 传输）

本地服务器通过command+args启动子进程，使用标准输入输出与客户端通信，适合本机测试：

{ "mcpServers": { "time_server": { "command": "python", "args": ["time_server.py"], "transport": "stdio" } } }

仓库自带的 examples/agents/mcp_agent/mcp_servers_config.json 就是一个可直接运行的本地示例，它通过python examples/agents/mcp_agent/calculator_server.py启动一个计算器 MCP 服务器，对应的服务器实现可参考 examples/agents/mcp_agent/calculator_server.py。

远程服务器（streamable-http / sse）

远程服务器通过 URL 暴露服务。以下是典型的云端服务（以 Composio 的 Notion 集成服务为例）配置，用npx启动连接器并指定传输方式为streamable-http，API Key 通过env字段注入：

{ "mcpServers": { "composio-notion": { "command": "npx", "args": ["composio-core@rc", "mcp", "https://mcp.composio.dev/notion/your-server-id", "--client", "camel"], "env": { "COMPOSIO_API_KEY": "your-api-key-here" }, "transport": "streamable-http" } } }

ACI.dev 服务器（可配置传输方式）

ACI.dev 的服务同样通过uvx启动，--apps参数可以声明需要启用的应用集合（如 BRAVE_SEARCH、GITHUB、ARXIV），transport字段在sse与streamable-http之间按需选择：

{ "mcpServers": { "aci_apps": { "command": "uvx", "args": [ "aci-mcp", "apps-server", "--apps=BRAVE_SEARCH,GITHUB,ARXIV", "--linked-account-owner-id", "<your_linked_acc_owner_id>" ], "env": { "ACI_API_KEY": "your_aci_api_key" }, "transport": "sse" } } }

提示：ACI.dev 同时支持sse与streamable-http，选择哪种取决于你的 Agent/服务器支持情况。

配置字段的完整取值

从 camel/utils/mcp_client.py 中ServerConfig的定义可以梳理出每个字段的含义与默认值：

字段	类型	默认值	说明
`command`	str	无	本地服务器的启动命令，与`url`二选一，同时提供会报错
`args`	list[str]	无	传给`command`的参数
`env`	dict	无	注入子进程的环境变量，务必用它存放 API Key
`cwd`	str/Path	无	子进程工作目录
`url`	str	无	远程服务器地址（http/https/ws/wss）
`headers`	dict	无	请求头，可携带`Authorization: Bearer ...`等鉴权信息，用于受保护端点
`timeout`	float	30.0	连接/读写超时（秒）
`encoding`	str	`utf-8`	stdio 子进程的编解码
`sse_read_timeout`	float	300.0	SSE 读取超时（5 分钟）
`terminate_on_close`	bool	True	关闭时是否终止底层连接
`transport`/`type`	str	自动检测	显式指定传输方式：`stdio`、`sse`、`streamable_http`、`websocket`
`prefer_sse`	bool	False	旧版参数（已弃用），URL 场景下优先 SSE

配置解析由MCPToolkit._load_clients_from_config/_load_clients_from_dict完成：读取 JSON 后遍历mcpServers，为每个服务器生成一个MCPClient实例；如果某个服务器配置非法，会抛出带服务器名的ValueError帮助定位问题（见 camel/toolkits/mcp_toolkit.py 第 696-759 行）。

第二步：用 MCPToolkit 连接并构建 Agent

配置文件就绪后，用MCPToolkit建立连接，再通过get_tools()取出所有服务器工具并传入ChatAgent：

import asyncio from camel.agents import ChatAgent from camel.models import ModelFactory from camel.toolkits import MCPToolkit from camel.types import ModelPlatformType, ModelType model = ModelFactory.create( model_platform=ModelPlatformType.OPENAI, model_type=ModelType.GPT_4O, ) async def main(): async with MCPToolkit(config_path="config/time.json") as toolkit: agent = ChatAgent(model=model, tools=toolkit.get_tools()) response = await agent.astep("What time is it now?") print(response.msgs[0].content) asyncio.run(main())

注意MCPToolkit需要作为异步上下文管理器使用（async with），这样在退出代码块时会自动断开所有服务器连接。

连接生命周期的三种管理方式

从 camel/toolkits/mcp_toolkit.py 的类文档可以看到，除了推荐的async with写法外，还有两种等价方式：

方式一：异步上下文管理器（推荐）

async with MCPToolkit(config_path="config.json") as toolkit: tools = toolkit.get_tools() # 退出后自动 disconnect

方式二：工厂方法

toolkit = await MCPToolkit.create(config_path="config.json") tools = toolkit.get_tools() await toolkit.disconnect() # 记得手动断开

方式三：显式 connect / disconnect

toolkit = MCPToolkit(config_path="config.json") await toolkit.connect() tools = toolkit.get_tools() await toolkit.disconnect()

如果项目是同步代码，也可以使用配套的同步入口：MCPToolkit.create_sync()、connect_sync()、disconnect_sync()，以及__enter__/__exit__（with MCPToolkit(...) as toolkit:）。

MCPToolkit 核心参数

MCPToolkit.__init__的完整参数（含默认值）如下：

参数	默认值	作用
`clients`	None	直接传入`MCPClient`实例列表
`config_path`	None	配置文件路径（标准 MCP JSON 格式）
`config_dict`	None	与配置文件等价的 Python 字典，免去文件 IO，适合程序化配置
`timeout`	None	整体连接超时（秒）
`skip_failed`	True	某个服务器连接失败时仅记录警告，不拖垮整个 toolkit
`per_client_timeout`	None	单个客户端独立超时，默认取`timeout`，否则 60.0
`max_retries`	2	每个失败客户端的重试次数（首次尝试之外的重试）
`retry_delay`	3.0	重试间隔（秒）

三个配置来源（clients、config_path、config_dict）至少提供一个，且可以叠加——同时提供时客户端会被合并。多个服务器采用并发连接，并发度由环境变量MCP_CONNECT_CONCURRENCY控制（默认 4），避免大量 stdio 服务器首次启动时同时执行包构建而压垮系统。

工具聚合与 Schema 严格化

get_tools()会遍历所有已连接的客户端并聚合工具，同时做两件事（camel/toolkits/mcp_toolkit.py 第 910-978 行）：

去重：按函数名去重，重复工具名会被跳过并记录警告；
Schema 严格化：通过_ensure_strict_tool_schema将每个工具的 JSON Schema 转换为兼容 OpenAI strict mode 的格式——为 object 类型补充additionalProperties: false、将properties的所有键写入required、展开$ref、剔除default为None的字段等；若 Schema 含 strict 模式不兼容特性（如数组 items 中出现allOf、超大anyOf联合），则自动降级为strict: false，保证工具始终可用。

此外还可以调用toolkit.call_tool(name, args)/call_tool_sync(name, args)直接在 toolkit 层按名字跨客户端查找并调用工具，或用list_available_tools()查看每个客户端各提供了哪些工具。

第三步：添加更多工具与调试

连接建立后，可以按需扩展更多服务器：本地验证阶段建议用简单的 stdio 服务器（如仓库示例中的计算器服务器），熟悉后再接入 ACI.dev、Composio 或npx生态的云端工具。三个实用建议：

传输方式选择：本地测试用stdio，云端工具用sse或streamable-http；
API Key 安全：密钥一律放在配置文件的env字段中，绝不写进代码；
问题排查：使用 MCP 官方调试工具 MCP Inspector（npx @modelcontextprotocol/inspector）单独验证服务器本身是否工作正常。

尝试接入 GitHub、Notion、ArXiv 等服务器后，可以直接看到 CAMEL Agent 使用新工具完成真实任务。

传输方式深入：自动检测与自动回退

MCPClient支持四种传输协议：stdio、sse、streamable_http、websocket。如果你不在配置中显式写transport/type，ServerConfig.transport_type会自动推断：

提供了command→stdio；
url以ws:///wss://开头 →websocket；
url以http:///https://开头 → 默认streamable_http（除非prefer_sse=True）。

值得一提的容错设计（camel/utils/mcp_client.py 第 302-359 行）：只有自动检测到 HTTP 的场景，如果streamable_http连接失败或超时，客户端会自动回退尝试sse（MCP 2024 规范 / supergateway 兼容）；一旦你显式指定了type，则严格按指定方式连接、不做回退，保证行为可预期。连接期间的initialize()与list_tools()操作都施加了硬超时（默认取配置timeout），避免服务器挂起导致客户端无限阻塞。

连接失败时错误信息也会被_simplify_connection_error翻译成可读性更强的提示，例如：

命令启动失败 →Failed to start MCP server command '...'. The command may have exited unexpectedly.
超时 →Connection timeout after Xs. The MCP server may be taking too long to respond.
包不存在 →MCP server package not found. Check if '...' is correct.

MCPAgent：注册中心与无函数调用模式

如果不想手动组装MCPToolkit+ChatAgent，可以直接使用MCPAgent（camel/agents/mcp_agent.py）。它继承自ChatAgent，自动完成「解析注册配置 → 构建 MCPToolkit → 连接 → 挂载工具」的全流程，并支持async with agent:上下文管理。

连接 ACI.dev 注册中心

把 Agent 注册到 ACI.dev 等 MCP 注册中心后，你的 Agent 可以被生态内其他客户端发现：

import os from camel.agents import MCPAgent from camel.models import ModelFactory from camel.types import ACIRegistryConfig, ModelPlatformType, ModelType aci_config = ACIRegistryConfig( api_key=os.getenv("ACI_API_KEY"), linked_account_owner_id=os.getenv("ACI_LINKED_ACCOUNT_OWNER_ID"), ) model = ModelFactory.create( model_platform=ModelPlatformType.OPENAI, model_type=ModelType.GPT_4O, ) agent = MCPAgent( model=model, registry_configs=[aci_config], )

ACIRegistryConfig定义在 camel/types/mcp_registries.py 中，其get_config()会生成uvx aci-mcp unified-server --linked-account-owner-id ...的启动配置，API Key 优先取构造参数、缺省时回退读ACI_API_KEY环境变量；Windows 平台会自动包装为cmd /c形式。同文件中还提供了SmitheryRegistryConfig（基于npx @smithery/cli@latest run @smithery/toolbox）与通用的BaseMCPRegistryConfig，三者都通过MCPRegistryType枚举区分。仓库示例 examples/agents/mcp_agent/mcp_agent_using_registry.py 展示了完整的调用方式。

MCPAgent还支持local_config（配置字典）与local_config_path（配置文件）直接提供本地服务器配置，以及add_registry()动态追加注册配置并自动重连。

无函数调用（function-calling）的最小化模式

如果你使用的模型不支持函数调用，MCPAgent提供了function_calling_available=False的降级方案：系统提示词引导模型输出固定格式的 JSON（包含server_idx、tool_name、tool_args三个字段），工具名和描述以纯文本拼进提示词；astep()内部用 camel/parsers/mcp_tool_call_parser.py 的extract_tool_calls_from_text解析出工具调用、逐一执行，再把结果回填给模型生成最终答复。完整的轻量示例见 examples/agents/mcp_agent/mcp_agent_without_function_calling.py，适合不依赖高级工具调用的场景。

用 PulseMCP 快速发现 MCP 服务器

面对海量 MCP 服务器，可以用PulseMCPSearchToolkit（camel/toolkits/pulse_mcp_search_toolkit.py）在代码中直接搜索，而不必手工猜测服务器清单：

from camel.toolkits import PulseMCPSearchToolkit search_toolkit = PulseMCPSearchToolkit() results = search_toolkit.search_mcp_servers(query="Slack", top_k=1) print(results)

search_mcp_servers的核心参数：query（搜索词）、top_k（返回前 N 个，默认 5）、package_registry（按包注册源过滤）、count_per_page（单页数量，上限 5000）、offset（分页偏移）。搜索结果按综合评分排序：名称命中 +5 分、描述命中 +3 分、GitHub Star 数按千分之一加分，无搜索词时则按 Star 数排序。搜索到的服务器信息可用于拼装MCPToolkit的配置，实现「搜索 → 连接 → 使用」的闭环。

常见问题与最佳实践

从本地起步：先用一个简单的 stdio 服务器（如仓库的 calculator_server.py）跑通全流程，再切换到云端 sse / streamable-http 服务器，排查问题最省力；
密钥安全：API Key 只放配置文件的env字段，永远不要硬编码在代码中；
调试工具：服务器侧的问题优先用 MCP Inspector（npx @modelcontextprotocol/inspector）独立验证；
受保护端点：远程服务器需要鉴权时，在配置中通过headers字段携带Authorization等请求头；
失败容忍：生产环境保持skip_failed=True，单个服务器异常不会让整个 Agent 不可用；多个服务器并发连接可通过MCP_CONNECT_CONCURRENCY调节并发度。

相关测试用例可参考 test/toolkits/test_mcp_toolkit.py 与 test/utils/test_mcp_client.py，它们覆盖了配置解析、连接管理、工具聚合等核心行为，也是理解各组件契约的补充材料。

结语

通过MCPToolkit+ChatAgent或开箱即用的MCPAgent，CAMEL 智能体可以在数分钟内接入任意 MCP 生态工具：本地 stdio 服务器适合快速验证，sse/streamable-http打通云端服务，ACI.dev 注册让 Agent 被生态发现，PulseMCP 则解决了「工具从哪找」的问题。从配置文件声明、连接生命周期管理到工具 Schema 严格化，这一整套实现都沉淀在 camel/toolkits/mcp_toolkit.py 与 camel/utils/mcp_client.py 中，值得作为接入其他 MCP 客户端时的参考范式。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考