---
title: "5分钟让你的AI Agent接入MCP工具：Python实现"
date: 2026-05-26
description: "从零到一，用Python让你的AI Agent快速接入MCP协议，调用任意工具"
tags: [Python, AI Agent, MCP, 工具调用, LLM]
---
> 本文同步发布在 Agent评测站
---
为什么要关注MCP？
MCP（Model Context Protocol）是由 Anthropic 提出的开放协议，它规范了 LLM 应用与外部工具之间的通信方式。简单说：你写一个 MCP Server 暴露工具，任何兼容 MCP 的 Agent 都能直接调用。不再需要为每个 Agent 单独写工具适配层。
今天的实战目标：用 Python 实现一个 MCP Client，让它调用一个“天气查询”MCP Server，整个过程控制在 5 分钟内可复现。
准备工作
pip install mcp httpx
确保 Python >= 3.10。
第一步：写一个 MCP Server
我们先定义一个简单的天气查询工具。MCP Server 负责注册工具并处理调用请求。
创建 weather_server.py：
# weather_server.py
import json
import random
from mcp.server import Server, stdio_server
app = Server("weather-agent")
@app.tool()
async def get_weather(city: str) -> str:
"""查询指定城市的天气情况"""
# 模拟天气数据，实际项目可对接真实API
weathers = ["晴天", "多云", "小雨", "阴天", "大风"]
temps = random.randint(5, 38)
weather = random.choice(weathers)
return json.dumps({
"city": city,
"weather": weather,
"temperature": f"{temps}°C",
"humidity": f"{random.randint(30, 90)}%",
"advice": "出门记得看天气"
}, ensure_ascii=False)
@app.tool()
async def get_air_quality(city: str) -> str:
"""查询空气质量指数"""
aqi = random.randint(20, 200)
level = "优" if aqi < 50 else "良" if aqi < 100 else "轻度污染" if aqi < 150 else "中度污染"
return json.dumps({
"city": city,
"aqi": aqi,
"level": level
}, ensure_ascii=False)
if __name__ == "__main__":
stdio_server.run(app)
关键点：
@app.tool()装饰器将函数暴露为 MCP 工具- 函数签名中的类型注解会被 MCP 自动解析为工具参数 schema
- 返回 JSON 字符串即可，Client 端会解析
第二步：写 MCP Client 连接 Server
MCP 支持 stdio 模式和 SSE 模式。本地开发最方便的是 stdio——你的 Agent 启动 Server 子进程，通过 stdin/stdout 通信。
创建 mcp_client.py：
# mcp_client.py
import asyncio
import json
import subprocess
from mcp.client import stdio_client
async def main():
# 启动 MCP Server 子进程
server_process = subprocess.Popen(
["python", "weather_server.py"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
# 通过 stdio 建立 MCP 连接
async with stdio_client(server_process) as session:
# 获取 Server 提供的工具列表
tools = await session.list_tools()
print("=== 可用工具 ===")
for tool in tools:
print(f" - {tool.name}: {tool.description}")
print(f" 参数: {tool.input_schema}")
print("\n=== 调用天气查询 ===")
result = await session.call_tool(
"get_weather",
arguments={"city": "北京"}
)
print(json.loads(result.content[0].text))
print("\n=== 调用空气质量 ===")
result = await session.call_tool(
"get_air_quality",
arguments={"city": "上海"}
)
print(json.loads(result.content[0].text))
if __name__ == "__main__":
asyncio.run(main())
运行：
python mcp_client.py
输出示例：
=== 可用工具 ===
- get_weather: 查询指定城市的天气情况
参数: {'type': 'object', 'properties': {'city': {'type': 'string'}}}
- get_air_quality: 查询空气质量指数
参数: {'type': 'object', 'properties': {'city': {'type': 'string'}}}
=== 调用天气查询 ===
{'city': '北京', 'weather': '多云', 'temperature': '24°C', 'humidity': '65%', 'advice': '出门记得看天气'}
=== 调用空气质量 ===
{'city': '上海', 'aqi': 85, 'level': '良'}
第三步：接入 LLM Agent，让模型自己决定调用哪个工具
手动调工具有什么意思？真正的 Agent 是让 LLM 自主决策调用哪个工具。
我们需要一个函数调用（Function Calling） 环节：把 MCP Server 暴露的工具列表转换成 LLM 可理解的 tools 格式，模型返回的 tool_call 再转成 MCP 调用。
创建 llm_agent.py：
# llm_agent.py
import asyncio
import json
import subprocess
from openai import AsyncOpenAI
from mcp.client import stdio_client
client = AsyncOpenAI(
api_key="your-api-key",
base_url="https://api.openai.com/v1"
)
async def main():
server_process = subprocess.Popen(
["python", "weather_server.py"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
async with stdio_client(server_process) as session:
# 1. 获取 MCP tools 并转换成 OpenAI function calling 格式
mcp_tools = await session.list_tools()
openai_tools = []
for tool in mcp_tools:
openai_tools.append({
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.input_schema
}
})
# 2. 用户提问
messages = [
{"role": "user", "content": "北京今天天气怎么样？空气质量如何？"}
]
# 3. 调用 LLM
response = await client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=openai_tools,
tool_choice="auto"
)
msg = response.choices[0].message
messages.append(msg)
# 4. 处理 tool_calls
if msg.tool_calls:
for tc in msg.tool_calls:
func_name = tc.function.name
args = json.loads(tc.function.arguments)
print(f"[Agent] 调用工具: {func_name}({args})")
result = await session.call_tool(func_name, arguments=args)
tool_result = result.content[0].text
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": tool_result
})
# 5. 将工具结果传给 LLM 生成最终回答
final_response = await client.chat.completions.create(
model="gpt-4o",
messages=messages
)
print(final_response.choices[0].message.content)
else:
print(msg.content)
if __name__ == "__main__":
asyncio.run(main())
输出：
[Agent] 调用工具: get_weather({'city': '北京'})
[Agent] 调用工具: get_air_quality({'city': '北京'})
北京今天天气多云，气温约24°C，湿度65%。
空气质量指数为85，属于"良"级别，适合户外活动。
出门记得看天气哦！
流程拆解：
用户提问
→ LLM 返回 tool_calls（自主决策调用哪些工具）
→ Agent 转发给 MCP Server 执行
→ 结果返回给 LLM 生成自然语言回复
→ 最终回答呈现给用户
第四步：用 SSE 模式连接远程 MCP Server
如果 MCP Server 部署在远程服务器上，需要用 SSE（Server-Sent Events）模式通信。
Server 端（用 FastAPI 或直接 mcp 包提供的 SSE 支持）：
# weather_sse_server.py
from mcp.server import Server, sse_server
app = Server("weather-agent")
@app.tool()
async def get_weather(city: str) -> str:
# ... 同上
pass
if __name__ == "__main__":
sse_server.run(app, host="0.0.0.0", port=8000)
Client 端通过 HTTP 连接：
# sse_client.py
from mcp.client import sse_client
import asyncio
async def main():
async with sse_client("http://your-server:8000/mcp") as session:
tools = await session.list_tools()
print(tools)
result = await session.call_tool("get_weather", arguments={"city": "深圳"})
print(result.content[0].text)
asyncio.run(main())
进阶技巧
1. 工具调用超时控制
# 为每个工具调用设置超时
result = await asyncio.wait_for(
session.call_tool("get_weather", arguments={"city": "广州"}),
timeout=10.0 # 10秒超时
)
2. 批量工具注册
如果你有大量工具，可以用目录扫描自动注册：
import os
import importlib
@app.startup()
async def register_tools():
tools_dir = "tools"
for f in os.listdir(tools_dir):
if f.endswith(".py") and not f.startswith("_"):
mod = importlib.import_module(f"tools.{f[:-3]}")
for attr in dir(mod):
obj = getattr(mod, attr)
if callable(obj) and hasattr(obj, "_mcp_tool"):
app.add_tool(obj)
3. 带上下文的工具（依赖注入）
MCP 支持在工具函数中注入上下文，比如获取当前对话历史：
@app.tool()
async def query_database(sql: str, ctx: dict) -> str:
# ctx 包含 session_id, user_id 等信息
user_id = ctx.get("user_id")
# 按用户权限执行查询
...
4. 配合 LangChain / CrewAI / AutoGen 使用
MCP 社区已经有很多集成。例如 LangChain 的 MCPAdapter：
from langchain_mcp import MCPAdapter
from langchain.agents import create_openai_functions_agent
# 将 MCP tools 包装成 LangChain tools
adapter = MCPAdapter(session)
langchain_tools = await adapter.get_tools()
agent = create_openai_functions_agent(
llm=llm,
tools=langchain_tools,
prompt=prompt
)
项目结构与生产建议
my-mcp-agent/
├── servers/ # MCP Server 实现
│ ├── weather.py
│ ├── database.py
│ └── file_search.py
├── agent/ # Agent 逻辑
│ ├── mcp_client.py
│ └── llm_agent.py
├── config.yaml # Server 连接配置
└── main.py
生产环境建议：
- 使用进程池管理多个 MCP Server：每个 Server 一个子进程，通过 stdio 通信，出错后自动重启
- 工具注册中心：启动时扫描所有 Server，构建统一工具列表
- 安全沙箱：MCP Server 应该运行在受限环境中，避免任意命令执行
- 日志与监控：记录每个工具调用的入参、出参、耗时，方便调试和审计
踩坑记录
- subprocess 挂死：如果你用
subprocess.Popen但没正确处理 stdout/stderr 的缓冲区，进程可能会阻塞。解决方案是用asyncio.create_subprocess_exec代替，或设置bufsize=1（行缓冲）。
- Tool Call 格式不匹配：不同 LLM 厂商的 tool_call 格式有微妙差异。OpenAI 返回
function.arguments是字符串，而 Gemini 可能直接返回 dict。做一层适配器很必要。
- 并发调用冲突：如果多个 Agent 共享一个 MCP Server 进程，注意加锁或使用独立进程。
- Server 崩溃：MCP stdio 模式下 Server 进程挂了 Client 并不知道。建议加心跳检测，5 秒无响应则重启。
生态工具推荐
| 工具 | 说明 |
|------|------|
| mcp-python-sdk | Anthropic 官方 Python SDK |
| mcp-cli | 命令行测试工具，快速验证 Server 是否正常 |
| fastmcp | 社区轻量框架，简化的装饰器 API |
| mcp-proxy | 将 stdio Server 暴露为 SSE 服务 |
| langchain-mcp | LangChain 的 MCP 集成包 |
小结
MCP 的价值在于标准化：Server 只写一次，任何 MCP 兼容的 Client、框架、IDE 都能用。Python 生态中实现起来非常直接——装饰器定义工具，stdio/sse 通信，几行代码就能让你的 Agent 拥有调用外部工具的能力。
如果还手动为每个工具写 if-else 分支，不妨试试 MCP，写完一次你就回不去了。
你遇到过什么坑？评论区聊聊。
---
本文同步发布在 Agent评测站