MCP 协议开发实战：从零搭建 AI Agent 工具链
系列前文： GitHub Actions 深度实践：零运维搭建 CI/CD 流水线
本文定位：在「流水线能自动构建/发布」之后，解决下一层问题—— AI Agent 如何标准化连接外部工具。
技术栈：Python 3.10+ · 官方mcpSDK（FastMCP）·langchain-mcp-adapters·（可选）GitHub Actions 做门禁
MCP（Model Context Protocol）正在成为 AI Agent 连接外部工具的事实标准。它解决了传统 AI 集成的 m×n 困境：m 个大模型 × n 个工具，往往要做 m×n 次适配；MCP 将其降维为 m+n——模型侧实现一次 Client，工具侧实现一次 Server，即可全互联。
本文将通过可运行的实战代码，从零构建一套完整的 MCP 工具链，并说明每个关键技术的应用价值。
一、理解 MCP 的核心架构
MCP 采用 Host / Client / Server 三层：
┌─────────────────────────────────────────────┐
│ Host（宿主） │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Client A │ │ Client B │ │ Client C │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
└────────┼─────────────┼─────────────┼────────┘
│ │ │
┌─────▼─────┐ ┌─────▼─────┐ ┌─────▼─────┐
│ MCP Server│ │ MCP Server│ │ MCP Server│
│ （数据库） │ │ （文件系统）│ │ （天气 API）│
└───────────┘ └───────────┘ └───────────┘|
角色 |
职责 |
类比 |
|
Host |
运行 AI 的应用（Claude Desktop、自研 Agent） |
操作系统 |
|
Client |
Host 内组件，维持与某个 Server 的独立会话 |
驱动 |
|
Server |
暴露 Tools / Resources / Prompts |
外设与服务 |
底层通信基于 JSON-RPC 2.0。本地开发常用 stdio（进程管道）；远程/多客户端场景常用 Streamable HTTP / SSE，便于 Server → Client 推送。
应用价值： 工具开发与 Agent 开发彻底解耦。业务方只维护一个 MCP Server；Cursor、Claude Desktop、LangChain Agent 都能按同一协议接入，不必为每个宿主重写插件。
二、环境准备
2.1 Python 环境
# 创建虚拟环境
python -m venv mcp-agent-env
# Windows
mcp-agent-env\Scripts\activate
# macOS / Linux
# source mcp-agent-env/bin/activate
pip install "mcp[cli]" httpx pydantic langchain-openai langchain-mcp-adapters langgraphWindows 下命令一般是python，不是python3。下文示例统一写python，按本机解释器调整即可。
2.2 项目结构
mcp-toolchain/
├── mcp_server.py # MCP Server：暴露工具
├── agent_host.py # 最小 Host：发现/调用工具
├── langchain_integration.py # LangChain Agent 集成
├── .github/workflows/ci.yml # （可选）用 Actions 做门禁
└── requirements.txt与系列前文的对应关系：
|
前文（Actions） |
本文（MCP） |
|
Workflow = 发布过程代码化 |
Server = 工具能力代码化 |
|
Secrets 管密钥 |
Token / 权限在 Server 层校验 |
|
失败可观测 + Artifact |
Tool 调用审计日志 + 限流 |
三、构建 MCP Server：暴露工具能力
用官方 FastMCP 实现天气查询、知识库检索、网页摘要三个工具。
# mcp_server.py
import json
import httpx
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("ResearchAssistantTools")
# ==================== Tool 1: 天气查询 ====================
@mcp.tool()
async def get_weather(city: str) -> str:
"""查询指定城市的实时天气信息。
Args:
city: 城市名称，例如 Beijing、Shanghai
"""
async with httpx.AsyncClient() as client:
response = await client.get(
f"https://wttr.in/{city}",
params={"format": "j1"},
timeout=10.0,
)
response.raise_for_status()
data = response.json()
current = data["current_condition"][0]
return json.dumps(
{
"city": city,
"temperature_c": current["temp_C"],
"feels_like_c": current["FeelsLikeC"],
"humidity": current["humidity"],
"description": current["weatherDesc"][0]["value"],
},
ensure_ascii=False,
)
# ==================== Tool 2: 知识库检索 ====================
KNOWLEDGE_BASE = {
"MCP协议": (
"MCP（Model Context Protocol）是开放标准协议，"
"用于让 AI 应用以统一方式连接外部工具与数据源。"
),
"AI Agent": (
"AI Agent（智能体）能感知环境、决策并调用工具完成任务，"
"工具调用通常通过 Function Calling 或 MCP 完成。"
),
}
@mcp.tool()
async def search_knowledge(query: str) -> str:
"""从本地知识库中检索相关信息。"""
results = []
q = query.lower()
for key, value in KNOWLEDGE_BASE.items():
if q in key.lower() or q in value.lower():
results.append({"topic": key, "content": value})
if not results:
return json.dumps(
{"status": "no_result", "message": f"未找到与'{query}'相关的知识"},
ensure_ascii=False,
)
return json.dumps({"status": "success", "results": results}, ensure_ascii=False)
# ==================== Tool 3: 网页摘要（预览） ====================
@mcp.tool()
async def summarize_url(url: str) -> str:
"""获取指定 URL 的网页内容预览（生产环境应再接摘要模型）。"""
async with httpx.AsyncClient() as client:
response = await client.get(url, timeout=15.0, follow_redirects=True)
content = response.text[:3000]
return json.dumps(
{
"url": url,
"status_code": response.status_code,
"content_preview": content[:1500],
"char_count": len(response.text),
},
ensure_ascii=False,
)
if __name__ == "__main__":
# stdio：适合本地 Host 拉起子进程；远程场景可改为 streamable-http / sse
mcp.run(transport="stdio")关键技术与价值
|
技术点 |
作用 |
应用价值 |
|
|
把函数注册为可发现工具 |
docstring / 类型注解自动变成 Agent 可理解的 schema |
|
|
非阻塞外呼 |
多工具并发时不堵死事件循环 |
|
|
标准输入输出通信 |
零端口、零 CORS，本地调试最快 |
|
返回 JSON 字符串 |
统一结果载体 |
Host / 多模型侧解析成本低 |
对比「每个平台写一套 Plugin」：同等三个工具，传统适配往往要几百行胶水；FastMCP 声明式注册后，维护点收敛到一个 Server 进程。
本地自检（可选）：
# 若已安装 mcp CLI
mcp dev mcp_server.py四、构建 MCP Host：连接 Server 并驱动调用
Host 是「能力消费者」：通过 MCP Client 发现工具、发起 tools/call。
# agent_host.py
import asyncio
import logging
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - HOST - %(levelname)s - %(message)s",
)
async def main():
server_params = StdioServerParameters(
command="python",
args=["mcp_server.py"],
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
logging.info("Host 已连接到 MCP Server")
# 1. 发现工具
listed = await session.list_tools()
print("可用工具:", [t.name for t in listed.tools])
# 2. 调用天气
weather = await session.call_tool("get_weather", {"city": "Beijing"})
print("天气:", weather.content)
# 3. 调用知识库
knowledge = await session.call_tool(
"search_knowledge", {"query": "MCP协议"}
)
print("知识:", knowledge.content)
if __name__ == "__main__":
asyncio.run(main())运行：
python agent_host.py关键技术与价值
|
技术点 |
作用 |
应用价值 |
|
|
声明如何拉起 Server 子进程 |
Host 不关心工具内部实现，只关心命令与参数 |
|
|
运行时能力发现 |
新增 |
|
|
JSON-RPC |
调用面统一，便于做审计、重试、熔断 |
|
双层 |
正确关闭管道与会话 |
避免僵尸子进程占满文件句柄 |
这一层还不接大模型——先保证「发现 → 调用 → 拿结果」稳定，再接到 Agent，排障成本低一个数量级。
五、LangChain 集成：可治理的 Agent
原则：LangChain / LangGraph 是「大脑」，MCP 是「手脚」。
所有外呼走 MCP Tool，权限与限流放在 Server 侧，避免模型「直接摸」任意 HTTP。
使用官方适配库 langchain-mcp-adapters（不要再用虚构的 langchain.tools.mcp API）。
# langchain_integration.py
import asyncio
import logging
import time
from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.tools import load_mcp_tools
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("audit")
async def main():
client = MultiServerMCPClient(
{
"research": {
"command": "python",
"args": ["mcp_server.py"],
"transport": "stdio",
}
}
)
# 有状态会话：同一 Server 连接上复用，适合多步工具编排
async with client.session("research") as session:
tools = await load_mcp_tools(session)
logger.info("已加载工具: %s", [t.name for t in tools])
agent = create_agent(
model="openai:gpt-4o",
tools=tools,
system_prompt=(
"你是智能研究助手，只能通过已提供的工具查询天气、"
"检索知识库、获取网页预览。不要编造工具结果。"
),
)
result = await agent.ainvoke(
{
"messages": [
{
"role": "user",
"content": "北京今天天气怎么样？另外简要解释 MCP 协议是什么。",
}
]
}
)
print("最终回答:", result["messages"][-1].content)
if __name__ == "__main__":
asyncio.run(main())需要 OpenAI 兼容密钥时：
# Windows PowerShell
$env:OPENAI_API_KEY="sk-..."
python langchain_integration.py5.1 审计回调（可观测）
若你使用带 callback 的 Executor / 自定义图节点，建议输出结构化步骤日志：
class AuditCallbackHandler:
"""示意：记录工具名与入参，便于审计与排障。"""
def on_tool_start(self, tool_name: str, tool_input: dict):
logging.info(
"agent_step",
extra={
"tool": tool_name,
"input": tool_input,
"timestamp": time.time(),
},
)关键技术与价值
|
技术点 |
作用 |
应用价值 |
|
|
同时挂多个 MCP Server |
天气、知识库、内部 API 可分进程拆分，互不影响 |
|
|
MCP Tool → LangChain Tool |
Agent 框架与工具协议解耦，换大脑不必重写手脚 |
|
|
工具调用 Agent |
多步任务由模型编排，人只写工具与策略 |
|
审计日志 |
记录每步 tool / input |
满足内控「谁在何时调了什么」；线上事故可回放 |
六、进阶：JSON-RPC 与 SSE / Streamable HTTP
6.1 JSON-RPC 2.0 请求
{
"jsonrpc": "2.0",
"id": "1",
"method": "tools/call",
"params": {
"name": "get_weather",
"arguments": { "city": "Beijing" }
}
}响应：
{
"jsonrpc": "2.0",
"id": "1",
"result": {
"content": [{ "type": "text", "text": "..." }]
}
}应用价值： 抓包/日志按 id 对齐请求与响应；自研 Host 时不必发明私有 RPC。
6.2 SSE / Streamable HTTP（远程场景）
远程部署时，Server 可改为 HTTP 传输（具体路径以当前 SDK 版本文档为准），典型事件流形态：
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive
event: message
data: {"jsonrpc":"2.0","id":"1","result":{"tools":[...]}}应用价值： 多 Host 共享同一工具集群；与 K8s Service / 网关鉴权更好接。本地仍推荐 stdio 降低复杂度。
把 Server 改成 HTTP 传输后，Client 配置示例：
client = MultiServerMCPClient(
{
"research": {
"url": "http://127.0.0.1:8000/mcp",
"transport": "http", # 或文档推荐的 streamable-http / sse
}
}
)七、与 GitHub Actions 衔接：给 MCP 工具链加门禁
前文解决「代码怎么自动验证与发布」；MCP Server 同样需要 CI，否则 Agent 一上线就踩坏工具。
# .github/workflows/mcp-ci.yml
name: MCP Toolchain CI
on:
push:
branches: [main]
pull_request:
jobs:
smoke:
runs-on: ubuntu-latest
timeout-minutes: 10
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.12"
cache: pip
- name: Install
run: |
python -m pip install -U pip
pip install "mcp[cli]" httpx pytest pytest-asyncio
- name: Import & list tools smoke
run: |
python - <<'PY'
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def smoke():
params = StdioServerParameters(command="python", args=["mcp_server.py"])
async with stdio_client(params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
tools = await session.list_tools()
names = {t.name for t in tools.tools}
assert {"get_weather", "search_knowledge", "summarize_url"} <= names
print("smoke ok:", sorted(names))
asyncio.run(smoke())
PY应用价值： PR 合入前证明「工具可发现、进程可拉起」；避免只测 Python import、不测 MCP 握手的假绿。
八、生产级最佳实践
8.1 权限与安全（在 Server 层）
def validate_token(token: str | None) -> str:
if not token or token != "demo-token":
raise PermissionError("无效令牌")
return "user-demo"
def has_permission(user_id: str, action: str) -> bool:
return action in {"weather", "knowledge"}
@mcp.tool()
async def get_weather(city: str, token: str | None = None) -> str:
"""带权限校验的天气查询（示意）。"""
user_id = validate_token(token)
if not has_permission(user_id, "weather"):
raise PermissionError("无权访问天气服务")
# ... 原业务逻辑
return json.dumps({"city": city, "ok": True}, ensure_ascii=False) 更稳妥的做法是把身份放在 HTTP Header / 网关，而不是让模型自由填写 token 参数——上面仅作「Server 必须有权控」的示意。
8.2 异常兜底
|
手段 |
做法 |
价值 |
|
超时 |
|
外部 API 挂了不拖死 Agent |
|
步数上限 |
Agent / Graph 限制迭代次数 |
防止工具互调死循环烧 Token |
|
敏感操作 |
返回 |
高风险动作人工确认后再执行 |
|
幂等 |
写操作带 |
重试不产生重复工单/重复扣费 |
8.3 可观测性
- 每次
tools/call打结构化日志：tool/latency_ms/ok/error_type - 与前文 Actions 的失败通知打通：Server 健康检查失败 → Webhook 告警
- 对外部依赖做熔断：连续失败则短时拒绝，避免雪崩
九、总结对比
|
对比维度 |
传统方式 |
MCP 方式 |
|
适配成本 |
m×n |
m+n |
|
代码维护点 |
各平台各一套 Plugin |
一个 Server，多 Host 复用 |
|
扩展性 |
新工具要重写适配 |
新 |
|
可治理性 |
依赖各宿主特性 |
统一在 Server 做权限 / 审计 / 限流 |
|
与 CI/CD |
难统一回归 |
stdio 冒烟可直接挂进 Actions |
落地顺序建议：
1. FastMCP 暴露 1～3 个只读工具（stdio）
2. agent_host 验证 list_tools / call_tool
3. langchain-mcp-adapters 接到 Agent
4. 加权限、超时、审计
5. CI 冒烟 +（可选）HTTP 远程部署MCP 的核心价值是零适配：工具开发与 Agent 开发解耦，「一次编写，到处调用」。叠加上文的 零运维 流水线，整条链路变成：
代码推送 → Actions 自动验证/发布 → MCP Server 对外暴露工具 → Agent 按协议发现并调用这才是「深度实践」想拼完整的一环，而不是又一个只能 Demo 的脚本。
系列续篇：工具接上之后，必须打开多步推理黑盒——见 可观测性深度实践：零盲盒追踪 Agent 多步推理。
参考
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
8
0- 0
已为社区贡献1条内容
所有评论(0)