5.4 Model Context Protocol (MCP)
本节介绍 LangChain 对 MCP 协议的支持，实现标准化的工具和上下文提供。
什么是 MCP？
MCP（Model Context Protocol） 是一个开放协议，标准化应用程序向 LLM 提供工具和上下文的方式。
"MCP is an open protocol that standardizes how applications provide tools and context to LLMs."
LangChain Agent 可以通过 langchain-mcp-adapters 库使用 MCP 服务器。
安装
bash
pip install langchain-mcp-adapters如果需要创建 MCP 服务器：
bash
pip install mcp传输机制
MCP 支持三种通信方式：
| 传输方式 | 说明 | 适用场景 |
|---|---|---|
| stdio | 本地子进程，通过标准输入输出 | 简单本地设置 |
| Streamable HTTP | 独立服务器处理 HTTP 请求 | 远程连接 |
| SSE | HTTP 变体，优化实时流 | 实时流式场景 |
访问 MCP 工具
使用 MultiServerMCPClient 访问多个 MCP 服务器的工具：
python
from langchain_mcp_adapters.client import MultiServerMCPClient
# 配置多个 MCP 服务器
client = MultiServerMCPClient({
"math": {
"transport": "stdio",
"command": "python",
"args": ["/path/to/math_server.py"],
},
"weather": {
"transport": "streamable_http",
"url": "http://localhost:8000/mcp",
},
"database": {
"transport": "sse",
"url": "http://localhost:8001/sse",
}
})
# 获取所有工具
tools = await client.get_tools()
print(f"可用工具: {[t.name for t in tools]}")与 Agent 集成
python
from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
async def create_mcp_agent():
client = MultiServerMCPClient({
"math": {
"transport": "stdio",
"command": "python",
"args": ["math_server.py"],
}
})
tools = await client.get_tools()
agent = create_agent(
model="gpt-4o",
tools=tools,
)
return agent构建 MCP 服务器
数学服务器（stdio）
python
# math_server.py
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Math")
@mcp.tool()
def add(a: int, b: int) -> int:
"""两数相加"""
return a + b
@mcp.tool()
def subtract(a: int, b: int) -> int:
"""两数相减"""
return a - b
@mcp.tool()
def multiply(a: int, b: int) -> int:
"""两数相乘"""
return a * b
@mcp.tool()
def divide(a: int, b: int) -> float:
"""两数相除"""
if b == 0:
raise ValueError("除数不能为零")
return a / b
if __name__ == "__main__":
mcp.run(transport="stdio")天气服务器（HTTP）
python
# weather_server.py
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Weather")
@mcp.tool()
async def get_weather(location: str) -> str:
"""获取指定位置的天气"""
# 实际应用中调用天气 API
weather_data = {
"北京": "晴天，25°C",
"上海": "多云，28°C",
"广州": "小雨，30°C",
}
return weather_data.get(location, f"{location}: 天气数据不可用")
@mcp.tool()
async def get_forecast(location: str, days: int = 3) -> str:
"""获取天气预报"""
return f"{location} 未来 {days} 天预报: 晴转多云"
if __name__ == "__main__":
mcp.run(transport="streamable-http")数据库服务器
python
# database_server.py
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Database")
# 模拟数据库
users_db = {
"1": {"name": "张三", "email": "zhangsan@example.com"},
"2": {"name": "李四", "email": "lisi@example.com"},
}
@mcp.tool()
async def get_user(user_id: str) -> dict:
"""获取用户信息"""
user = users_db.get(user_id)
if user:
return user
raise ValueError(f"用户 {user_id} 不存在")
@mcp.tool()
async def list_users() -> list:
"""列出所有用户"""
return list(users_db.values())
@mcp.tool()
async def create_user(user_id: str, name: str, email: str) -> dict:
"""创建新用户"""
if user_id in users_db:
raise ValueError(f"用户 {user_id} 已存在")
users_db[user_id] = {"name": name, "email": email}
return users_db[user_id]
if __name__ == "__main__":
mcp.run(transport="streamable-http")有状态工具使用
对于需要保持会话状态的服务器，使用持久会话：
python
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.tools import load_mcp_tools
async def use_stateful_tools():
client = MultiServerMCPClient({
"stateful_service": {
"transport": "stdio",
"command": "python",
"args": ["stateful_server.py"],
}
})
# 使用持久会话
async with client.session("stateful_service") as session:
tools = await load_mcp_tools(session)
# 在同一会话中多次调用工具
# 状态会在调用之间保持
result1 = await tools[0].ainvoke({"input": "first"})
result2 = await tools[0].ainvoke({"input": "second"})完整示例
python
import asyncio
from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
async def main():
# 1. 配置 MCP 客户端
client = MultiServerMCPClient({
"calculator": {
"transport": "stdio",
"command": "python",
"args": ["math_server.py"],
},
"weather": {
"transport": "streamable_http",
"url": "http://localhost:8000/mcp",
}
})
# 2. 获取所有工具
tools = await client.get_tools()
print(f"加载了 {len(tools)} 个工具")
# 3. 创建 Agent
agent = create_agent(
model="gpt-4o",
tools=tools,
system_prompt="你是一个多功能助手，可以进行数学计算和查询天气。"
)
# 4. 使用 Agent
result = agent.invoke({
"messages": [{
"role": "user",
"content": "计算 15 + 27，然后告诉我北京的天气"
}]
})
print(result["messages"][-1].content)
if __name__ == "__main__":
asyncio.run(main())MCP 服务器配置选项
stdio 传输
python
{
"server_name": {
"transport": "stdio",
"command": "python", # 执行命令
"args": ["server.py"], # 命令参数
"env": { # 环境变量（可选）
"API_KEY": "xxx"
},
"cwd": "/path/to/dir" # 工作目录（可选）
}
}HTTP 传输
python
{
"server_name": {
"transport": "streamable_http",
"url": "http://localhost:8000/mcp",
"headers": { # 请求头（可选）
"Authorization": "Bearer xxx"
},
"timeout": 30 # 超时时间（可选）
}
}SSE 传输
python
{
"server_name": {
"transport": "sse",
"url": "http://localhost:8000/sse",
"headers": {}
}
}MCP 工具装饰器
python
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("MyServer")
# 基本工具
@mcp.tool()
def simple_tool(param: str) -> str:
"""简单工具"""
return f"处理: {param}"
# 异步工具
@mcp.tool()
async def async_tool(param: str) -> str:
"""异步工具"""
await asyncio.sleep(1)
return f"异步处理: {param}"
# 带类型注解的工具
@mcp.tool()
def typed_tool(
name: str,
count: int = 1,
enabled: bool = True
) -> dict:
"""带完整类型注解的工具"""
return {
"name": name,
"count": count,
"enabled": enabled
}架构图
┌─────────────────────────────────────────────────────────┐
│ LangChain Agent │
├─────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────┐ │
│ │ MultiServerMCPClient │ │
│ └─────────────────────────────────────────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ stdio │ │ HTTP │ │ SSE │ │
│ └──────────┘ └──────────┘ └──────────┘ │
│ │ │ │ │
└──────────│──────────────│──────────────│─────────────────┘
│ │ │
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ Math │ │ Weather │ │ Database │
│ Server │ │ Server │ │ Server │
└──────────┘ └──────────┘ └──────────┘最佳实践
| 实践 | 说明 |
|---|---|
| 选择合适传输 | 本地用 stdio，远程用 HTTP |
| 错误处理 | MCP 工具应返回清晰的错误信息 |
| 文档完善 | 工具 docstring 会被 LLM 使用 |
| 类型注解 | 完整的类型注解提高可靠性 |
| 会话管理 | 需要状态时使用持久会话 |