lastmile-ai/mcp-agent · 上手攻略
- 仓库:lastmile-ai/mcp-agent
- 链接:https://github.com/lastmile-ai/mcp-agent
- 分类:skill(Agent 框架 / MCP 工具链)
- 作者:spark
- 更新:2026-07-15
是什么
mcp-agent 是 Lastmile AI 开源的、专门面向 Model Context Protocol(MCP) 的 Python Agent 框架。它的基本立场是:"MCP 就够了,简单的可组合模式比花哨架构更稳"。它围绕一个 MCPApp 运行时来注册 Agent、MCP 服务器、工具和工作流,把 Anthropic 那篇「Building Effective Agents」里讲到的全部范式(Parallel / Router / Orchestrator / Evaluator-Optimizer / Swarm / Deep Research 等)都实现成了可复用的工作流。
技术上分三层:
- MCPApp:核心运行时,统一配置、日志、tracing 和执行引擎(asyncio 或 Temporal)。
- Agent / AugmentedLLM:Agent 把指令和允许调用的 MCP server 绑定;AugmentedLLM 把 OpenAI / Anthropic / Google / Bedrock / Azure 等 Provider SDK 包一层,加上工具、memory、结构化输出。
- Workflow decorator:
@app.workflow/@app.workflow_run/@app.workflow_task,同一份代码可在 asyncio 和 Temporal 之间切换。
协议层支持 MCP 全功能:Tools / Resources / Prompts / Notifications 全齐,OAuth、Sampling、Elicitation、Roots 这些高级能力也开了。
解决什么问题
- 把 MCP server 真正"用"起来:MCP 协议本身只是 JSON-RPC,需要你自己管 server 进程生命周期、连接池、tool schema 拼装、token 计量。
mcp-agent把这一切封装掉。 - 实现可组合的 Agent 模式:直接给你 Anthropic 那张图里的所有模式,每种都给一个
create_*_llm(...)factory helper,可以像乐高一样拼装。 - 从本地脚本无痛过渡到生产:把执行引擎从 asyncio 切到 Temporal,就能拿到 pause / resume / retry / human-in-the-loop / durable history,不用改业务代码。
- 不需要写胶水代码的 Agent-as-MCP-Server:内置 FastMCP 兼容 API,你写的 agent 可以直接对外暴露成 MCP server,被 ChatGPT、Claude Desktop 等再次调用。
快速安装
仓库需要 Python 3.10+(具体上限以 setup.py 为准,当前各 example 在 3.11/3.12 上跑)。推荐用 uv:
# 方式一:脚手架直接起项目
mkdir hello-mcp-agent && cd hello-mcp-agent
uvx mcp-agent init # 脚手架,生成 main.py + config 模板
uv init
uv add "mcp-agent[openai]" # 任选 provider extra:anthropic / google / azure / bedrock
# 把 OPENAI_API_KEY 写到 mcp_agent.secrets.yaml,或直接 export OPENAI_API_KEY=...
uv run main.py
# 方式二:装到已有项目
uv add "mcp-agent"
# 或者
pip install mcp-agent
CLI 单独装也可以:uvx mcp-agent 可用命令包含 init、deploy、cloud、check。
核心用法
1. 最小 hello-world:一个能读文件、读 URL 的 agent
import asyncio
from mcp_agent.app import MCPApp
from mcp_agent.agents.agent import Agent
from mcp_agent.workflows.llm.augmented_llm_openai import OpenAIAugmentedLLM
app = MCPApp(name="hello_world")
async def main():
async with app.run():
agent = Agent(
name="finder",
instruction="Use filesystem and fetch to answer questions.",
server_names=["filesystem", "fetch"],
)
async with agent:
llm = await agent.attach_llm(OpenAIAugmentedLLM)
answer = await llm.generate_str("Summarize README.md in two sentences.")
print(answer)
if __name__ == "__main__":
asyncio.run(main())
server_names 引用的 MCP server 在 mcp_agent.config.yaml 里注册:
# mcp_agent.config.yaml
execution_engine: asyncio
mcp:
servers:
fetch:
command: "uvx"
args: ["mcp-server-fetch"]
filesystem:
command: "npx"
args: ["-y", "@modelcontextprotocol/server-filesystem", "<your_dir>"]
openai:
default_model: gpt-4o
# mcp_agent.secrets.yaml(gitignore)
openai:
api_key: "${OPENAI_API_KEY}"
2. 让 LLM 吐结构化 JSON
from pydantic import BaseModel
from mcp_agent.workflows.llm.augmented_llm import RequestParams
class Summary(BaseModel):
title: str
verdict: str
async with agent:
llm = await agent.attach_llm(OpenAIAugmentedLLM)
report = await llm.generate_str(
message="Draft a 3-sentence release note from CHANGELOG.md",
request_params=RequestParams(maxTokens=400, temperature=0.2),
)
structured = await llm.generate_structured(
message="Return a JSON object with title and verdict summarising the README.",
response_model=Summary,
)
3. 直接组合 Anthropic 那张图里的模式
mcp_agent.workflows.factory 里给每个模式一个 factory:
| 模式 | Factory | 一句话 |
|---|---|---|
| Parallel (Map-Reduce) | create_parallel_llm(...) |
分发到专家 Agent 再聚合 |
| Router | create_router_llm(...) / create_router_embedding(...) |
路由到合适的 agent/server |
| Intent classifier | create_intent_classifier_llm(...) |
先分类再自动化 |
| Orchestrator-workers | create_orchestrator(...) |
planner 生成计划 + worker 执行 |
| Deep research | create_deep_orchestrator(...) |
长程研究 + 知识抽取 + 策略检查 |
| Evaluator-optimizer | create_evaluator_optimizer_llm(...) |
evaluator 收敛判断、迭代优化 |
| Swarm | create_swarm(...) |
兼容 OpenAI Swarm 的多 agent handoff |
from mcp_agent.workflows.factory import create_parallel_llm
# 假设你已经在 factories/researcher_finder.yaml 里定义好两个 Agent
parallel = await create_parallel_llm(
name="research-swarm",
fan_out_agents=["web_researcher", "filesystem_researcher"],
fan_in_agent="summariser",
context=app.context,
)
result = await parallel.generate_str(message="Research the impact of MCP on agent design.")
4. 用 Temporal 做 durable 执行
把 mcp_agent.config.yaml 里 execution_engine 改成 temporal,启动一个 worker:
from mcp_agent.executor.temporal import create_temporal_worker_for_app
async with create_temporal_worker_for_app(app) as worker:
await worker.run()
业务侧的 @app.workflow / @app.workflow_run 不需要改一行:
from datetime import timedelta
from mcp_agent.executor.workflow import Workflow, WorkflowResult
@app.workflow
class PublishArticle(Workflow[WorkflowResult[str]]):
@app.workflow_task(schedule_to_close_timeout=timedelta(minutes=5))
async def draft(self, topic: str) -> str:
return f"- intro to {topic}\n- highlights\n- next steps"
@app.workflow_run
async def run(self, topic: str) -> WorkflowResult[str]:
outline = await self.draft(topic)
return WorkflowResult(value=outline)
得到:pause / resume / 重试 / 人类审批 / 历史回放,且不用改 agent 代码。
5. 把 Agent 暴露成 MCP Server
同一个 Swift App 里加上 mcp-agent-server(详见 examples/mcp_agent_server),你定义的 agent 就能被别的 MCP 客户端(包括 ChatGPT、Claude Desktop、自定义 MCP 客户端)直接调。
6. Signals & Human Input
Signals API 让外部信号(人类审批、webhook、新数据)注入正在跑的 workflow;新版同时支持 Elicitation 和 Sampling,使得 LLM 可以在跑的过程中反过来向用户问问题或主动调子 LLM。
典型适用场景
- 研发 / 知识库检索:filesystem + fetch + 一个写手 agent 拼装成一个"读 GitHub / 读 Notion / 写摘要"工作流。
- 客服 / 工单自动化:router + intent-classifier 先识别意图,再把工单转给相应的 MCP server(Slack、Jira、Notion)。
- 多源深度调研:parallel + orchestrator,从 5 个 web 源同时拉数据,最后让一个 summarizer 把 RTTM 风格的结构化报告汇总出来。
- 长期跑的后台 Agent:用 Temporal backend,写一个"每天扫一次邮件、自动摘要、推到 Slack"的 agent,进程崩溃后能从中断点继续。
- 把现有 Agent 接到 ChatGPT / Claude:把 mcp-agent 启动成一个 MCP server,被 ChatGPT Apps SDK、Claude Desktop、Cursor 直接调用(这也是 Anthropic 推 MCP 的核心动机)。
坑与注意
- 协议版本要锁:MCP 还是个动得很快的协议。锁定
mcp-agent的小版本(~=而不是^)能减少和mcpSDK 升级之间的兼容问题。 - MCP server 进程管理:
mcp-agent会subprocess.Popen启每个 server(stdio 模式),macOS 上 npx/uvx 较慢,多 server 启动时建议加长 timeout。 - Secrets 文件必须 gitignore:
mcp_agent.secrets.yaml默认不该提交;CI 上统一用 env var,OPENAI_API_KEY/ANTHROPIC_API_KEY/GOOGLE_API_KEY都能直接读。 - asyncio 后台占满:默认
execution_engine: asyncio下,长任务会占据 event loop;要几十分钟或几小时级别、必须 pause / resume 的就用 Temporal backend。 - 多 Agent 上下文:swarm 和 orchestrator 里每个 agent 都有自己的 MCP server 列表,token 会被多个 agent 同时消费;建议给大模型挂 Anthropic / Google 的更便宜的 tier,或用 router 先过滤。
- Claude Sonnet 4.5 的 tool call 边界:MCP server 返回的工具列表太长(>100 个)会让 Anthropic 系列模型挑错工具;尽量拆 agent,或在 server 侧禁用部分 tools。
- 没有官方 TypeScript SDK:仓库本身只发 Python 包;如果要 Node 版本,社区里 MCP 周边项目的生态更成熟。
与同类对比
| 框架 | 协议原生度 | 易用度 | 生产化 | 主要场景 |
|---|---|---|---|---|
| mcp-agent | ✅ MCP-natively 围绕 MCP 设计 | 装饰器 + context manager,写得少 | Temporal backend,durability 第一梯队 | 想"严肃用 MCP"、要 durability |
| LangGraph | 中性(自己定义 tool 调用) | 学习曲线陡 | LangSmith 强大 | 图状复杂分支流 |
| CrewAI | 中性 | 上手最快,role-based 叙事强 | 较差(团队成员粒度的容错偏弱) | 角色化 prototype |
| AutoGen (Microsoft) | 中性 | 对话轮次控制直观 | 一般 | 跨 LLM 对话协作 |
| Smolagents / OpenAI Swarm | 中性 Swarm 设计 | 极简 | 弱 | 学习 / 轻量脚本 |
| Pydantic AI | 中性(无 MCP 强绑定) | 类型友好 | 一般 | 想锁 Pydantic 体验 |
核心差异:mcp-agent 是少数把"MCP 是一等公民"这件事当产品定位的框架。如果你已经在用 Claude Desktop / Cursor / ChatGPT 的 MCP 生态,从 mcp-agent 入手的摩擦最低;如果你想写的 Agent 完全不碰 MCP,更通用的 LangGraph / Pydantic AI 更合算。
一句话推荐结论
想用 MCP 协议搭 Agent,又想保留"asyncio 写起来快、Temporal 跑起来稳"两条腿 —— mcp-agent 是当下最不折腾的选择;先 uvx mcp-agent init 起一个 demo,再决定要不要把 execution_engine 切到 Temporal。
来源
- 仓库 README:https://github.com/lastmile-ai/mcp-agent
- 官方文档:https://docs.mcp-agent.com
- PyPI 包:https://pypi.org/project/mcp-agent/
- 例程:https://github.com/lastmile-ai/mcp-agent/tree/main/examples(basic / mcp / temporal / cloud / mcp_agent_server / workflows)
- 协议说明:https://modelcontextprotocol.io/introduction
不确定处
- mcp-agent 当前最新小版本号(写文档时 PyPI 页面数字未单独核对),示例代码的不变性以官方
examples/basic目录为准。 - Temporal backend 在生产环境对 worker 资源的要求(CPU/GIL 限制)文档未给硬数字,建议先在本地压测再上生产。
- "managed agent runtime mcp-c" 的 GA 时机和价格档位未在 README 披露,需查 Cloud 文档确认。