进阶 agently.cn 2026-10-08 20:14:53 · 1 阅读

第3章 MCP 模型上下文协议详解

MCP 语言:英文 · 中文

MCP(Model Context Protocol,模型上下文协议)将外部工具暴露给 AI 智能体。Agently 通过 MCPActionExecutor 将 MCP 服务器接入动作运行时,使模型能够通过同一接口看到 MCP 工具和你自有的 @agent.action_func 动作。

在服务集成中,请使用 URL / Streamable HTTP MCP 端点;在本地开发、桌面客户端或单用户本地服务器场景中,请使用 stdio 命令配置。SSE 端点仅作为遗留的兼容性路径保留。

最小示例

python import os import asyncio from dotenv import load_dotenv, find_dotenv from agently import Agently

load_dotenv(find_dotenv())

Agently.set_settings("OpenAICompatible", { "base_url": "${ENV.OPENAI_BASE_URL}", "api_key": "${ENV.OPENAI_API_KEY}", "model": "${ENV.OPENAI_MODEL}", })

agent = Agently.create_agent()

async def main(): result = ( await agent.use_mcp(f"https://mcp.amap.com/mcp?key={os.environ.get('AMAP_API_KEY')}") .input("What's the weather like in Shanghai today?") .async_start() ) print(result)

asyncio.run(main())

use_mcp(url) 会注册 MCP 服务器暴露的所有工具。随后,代理将针对 {@agent.action_func, use_tool, use_mcp 工具} 的并集来规划工具调用,仿佛它们是一组统一的操作。

API

- 方法:await agent.use_mcp(url)。行为:连接服务器、列出工具并注册;返回 agent 以支持链式调用。 - 方法:await agent.use_mcp(url, headers={...})。行为:携带自定义 HTTP 头部(如认证令牌)。 - 方法:await agent.use_mcp({"mcpServers": {...}})。行为:使用包含一个或多个 HTTP 或 stdio 服务器的 MCP 配置。

对于默认执行器,带有 URL 的 headers= 参数会在 FastMCP 处理之前被标准化为 MCP 配置。

python await agent.use_mcp( "https://example.com/mcp", headers={"Authorization": f"Bearer {token}"}, )

对于本地 stdio 服务器,请直接传入 MCP 配置:

python await agent.use_mcp({ "mcpServers": { "filesystem": { "command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "./workspace"], } } })

混合使用 MCP 与自定义动作

python @agent.action_func async def lookup_internal(id: str): """Look up a record in the internal database.""" ...

await agent.use_mcp("https://example-mcp/server") agent.use_actions(lookup_internal)

# The model now sees MCP tools + lookup_internal in the same plan result = await agent.input(question).async_start()

MCP 提供的工具与本地定义的动作之间不存在优先级之分。模型根据名称、描述以及提示词上下文来做出选择。

检查调用详情

对于请求范围的回合,将回合提示传入动作循环,以检查模型实际调用的工具:

python turn = agent.input("Use the MCP server to answer this question.") records = agent.get_action_result(prompt=turn.prompt) for r in records: print(r)

动作记录也会写入 extra.action_logs(或在兼容层中写入 extra.tool_logs)。

常见陷阱

- 忘记 await:use_mcp(...) 是异步的,因为它需要从服务器列出工具。忘记 await 会返回协程,且注册过程会静默失败。 - 在 URL 中传递密钥:建议优先使用头部和环境变量。URL 查询参数最终会出现在日志中。 - 将 MCP 视为与本地动作完全相同:托管的 MCP 服务器可能存在延迟或速率限制。对于延迟敏感或高并发的调用,请优先使用本地动作函数。

相关参考

- 动作运行时 —— MCPActionExecutor 是内置执行器之一 - 工具 —— 在兼容层中,use_mcp(...) 的行为相同

评论 (0)