MCP与LangChain的集成
这是一个全面展示将模型上下文协议（MCP）服务器与LangChain和LangGraph集成的演示，展示了多服务器工具编排和智能代理工作流程。该项目实现了通过统一的AI代理接口访问的数学计算和天气服务。
目录
概述
这个项目展示了LangChain生态系统中先进的MCP（模型上下文协议）集成模式。它展示了：
- 多服务器架构同时连接到使用不同传输协议的多个MCP服务器
- ReAct 代理模式由LangGraph驱动的智能体，能够使用外部工具进行推理和行动
- 跨协议通信STDIO和HTTP传输协议协同工作
- 工具编排智能路由及数学和天气相关任务的执行
建筑学
┌─────────────────────┐ ┌─────────────────────┐ ┌─────────────────────┐
│ LangGraph Agent │◄──►│ MultiServerMCP │◄──►│ Math Server │
│ (ReAct Pattern) │ │ Client │ │ (STDIO) │
│ │ │ │ │ │
└─────────────────────┘ └─────────────────────┘ └─────────────────────┘
│ │
│ ▼
│ ┌─────────────────────┐
└──────────────────►│ Weather Server │
│ (HTTP/SSE) │
└─────────────────────┘
┌─────────────────────┐
│ Groq LLM │◄──── Tool execution results
│ (Llama 3.3 70B) │
└─────────────────────┘特点/功能
🧠 智能代理系统
- ReAct 模式使用外部工具进行推理和行动
- 多步骤规划复杂任务的分解与执行
- 上下文感知在工具调用之间保持对话状态
🔧 多服务器工具集成
- 数学运算通过STDIO传输进行加法和乘法运算
- 天气服务通过HTTP传输进行基于位置的天气查询
- 协议灵活性展示了同步和异步两种模式
🚀 高级LangChain集成
- LangGraph 工作流结构化的智能体执行模式
- 工具抽象MCP工具无缝集成到LangChain生态系统中
- 状态管理持久的对话和执行上下文
📡 传输协议支持
- STDIO传输数学运算的直接进程通信
- HTTP传输天气服务的RESTful API通信
- 统一界面单个客户端管理多种传输类型
先决条件
- python3.12或更高版本
- API密钥用于LLM访问的Groq API密钥
- 网络用于HTTP传输和API调用的互联网连接
安装
1. 环境设置
# Clone the project
git clone
cd mcp_langchain
# Create and activate virtual environment
python -m venv .venv
source .venv/bin/activate # On Windows: .venv\Scripts\activate2. 安装依赖项
使用紫外线（推荐）：
# Install UV package manager
curl -LsSf https://astral.sh/uv/install.sh | sh
# Install all dependencies
uv sync使用 pip：
# Install from requirements
pip install -r requirements.txt
# Or install from pyproject.toml
pip install -e .3. 环境配置
创建一个 .env 项目根目录中的文件：
GROQ_API_KEY=your_groq_api_key_here配置
多服务器客户端配置
这个（或：那个，具体根据上下文确定） MultiServerMCPClient 被配置为 client.py：
client = MultiServerMCPClient({
"math": {
"command": "python",
"args": ["mathserver.py"],
"transport": "stdio",
},
"weather": {
"url": "http://localhost:8000/mcp",
"transport": "streamable_http",
}
})服务器配置
数学服务器（STDIO）：
- 传输：标准输入/输出
- 操作：同步数学计算
- 协议：直接进程间通信
天气服务器（HTTP）:
- 传输：HTTP/SSE 流媒体
- 操作：异步天气查询
- 协议：RESTful API通信
使用
运行完整演示
执行主要客户端以展示两个服务器：
python client.py预期输出：
Math response: The result of (3 + 5) × 12 is 96.
Weather response: The weather in California shows that it's always raining.运行独立服务器
数学服务器（STDIO）
python mathserver.py天气服务器（HTTP）
python weather.py服务器将在 http://localhost:8000
交互式使用
您可以创建自定义的代理交互：
import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_groq import ChatGroq
async def custom_interaction():
# Initialize client and agent (as shown in client.py)
# Custom mathematical query
math_result = await agent.ainvoke({
"messages": [{"role": "user", "content": "Calculate 15 multiplied by 7, then add 23"}]
})
# Custom weather query
weather_result = await agent.ainvoke({
"messages": [{"role": "user", "content": "What's the weather like in New York?"}]
})组件
核心文件
client.py - 多服务器代理客户端
目的主要的编曲客户端，用于连接多个MCP服务器
主要特点:
- 多服务器MCP客户端初始化
- 带有工具集成的ReAct代理创建
- 数学和天气查询的演示
- 错误处理和响应处理
架构模式：
# Multi-server client setup
client = MultiServerMCPClient({server_configs})
# Tool extraction and agent creation
tools = await client.get_tools()
agent = create_react_agent(model, tools)
# Query execution with reasoning
response = await agent.ainvoke({"messages": [query]})mathserver.py - 数学运算服务器
目的MCP服务器提供数学计算工具
可用工具：
add(a: int, b: int) -> int加法运算multiple(a: int, b: int) -> int乘法运算
交通STDIO（标准输入/输出通信）
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers"""
return a + bweather.py - 天气信息服务器
目的提供天气相关信息的MCP服务器
可用工具：
get_weather(location: str) -> str天气信息检索
交通HTTP/SSE 流式传输
@mcp.tool()
async def get_weather(location: str) -> str:
"""Get the weather for a location."""
return f"Weather information for {location}"配置文件
pyproject.toml - 项目配置
依赖项：
langchain-groqGroq LLM集成langchain-mcp-adaptersMCP-LangChain 桥接器langgraph高级代理工作流mcp核心模型上下文协议
requirements.txt - 替代安装
基于 pip 的安装简化依赖列表。
API 参考文档
多服务器MCP客户端
构造函数参数
MultiServerMCPClient(server_configs: Dict[str, ServerConfig])服务器配置选项:
command(str): 用于STDIO传输的可执行命令args(列表\[str\])：命令参数transport(str): 传输协议（“stdio”，“streamable_http”）url(str): 用于HTTP传输的服务器URL
方法
get_tools() -> List[Tool]
从配置的服务器中检索所有可用工具。
回报兼容LangChain的工具对象列表
close() -> None
正确关闭所有服务器连接。
数学工具
add(a: int, b: int) -> int
执行两个整数的加法运算。
参数：
a(int): 第一个数字b(int): 第二个数字
退货两个数的和
multiple(a: int, b: int) -> int
执行两个整数的乘法运算。
参数:
a(int): 第一个数字b(int)：第二个数字
退货两个数的乘积
天气工具
get_weather(location: str) -> str
检索指定位置的天气信息。
参数：
location(str): 地理位置
回报天气描述字符串
发展
添加新工具
到数学服务器
@mcp.tool()
def subtract(a: int, b: int) -> int:
"""Subtract two numbers"""
return a - b连接到天气服务器
@mcp.tool()
async def get_forecast(location: str, days: int) -> str:
"""Get weather forecast for multiple days"""
return f"{days}-day forecast for {location}"创建新服务器
- 创建服务器文件：
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("YourServerName")
@mcp.tool()
def your_tool(param: str) -> str:
"""Your tool description"""
return f"Result for {param}"
if __name__ == "__main__":
mcp.run(transport="stdio") # or "streamable_http"- 更新客户端配置：
client = MultiServerMCPClient({
# ... existing servers ...
"your_server": {
"command": "python",
"args": ["your_server.py"],
"transport": "stdio",
}
})代码风格指南
- 类型提示所有函数都包含全面的类型注解
- 异步/等待为HTTP传输服务器使用异步模式
- 工具文档为所有MCP工具添加描述性文档字符串
- 错误处理实现强大的异常处理
测试
手动测试
# Test math server independently
python mathserver.py
# Test weather server independently
python weather.py
# Test integrated client
python client.py单元测试框架
import pytest
from mathserver import add, multiple
def test_addition():
assert add(3, 5) == 8
def test_multiplication():
assert multiple(4, 7) == 28故障排除
常见问题
1. 导入错误
# Ensure virtual environment is activated
source .venv/bin/activate # Linux/Mac
.venv\Scripts\activate # Windows
# Reinstall dependencies
uv sync # or pip install -r requirements.txt2. API密钥问题
- 验证
.env文件包含有效内容GROQ_API_KEY - 检查API密钥权限和配额
- 确保环境变量已正确加载
3. 服务器连接问题
STDIO传输问题:
- 验证客户端配置中的Python可执行文件路径
- 检查服务器文件是否可执行
- 确保服务器代码中没有语法错误
HTTP传输问题：
- 确认天气服务器正在正确的端口上运行
- 检查本地主机:8000的防火墙设置
- 验证HTTP端点的可访问性
4. 工具执行失败
# Debug tool availability
tools = await client.get_tools()
print(f"Available tools: {[tool.name for tool in tools]}")
# Debug agent responses
print(f"Agent response: {response}")5. LangGraph 代理问题
- 确保Groq API密钥有效且有配额
- 检查LangGraph版本兼容性
- 验证工具模式兼容性
调试模式
启用详细日志记录：
import logging
logging.basicConfig(level=logging.DEBUG)
# Add debug output to client
print(f"Tools loaded: {len(tools)}")
print(f"Tool names: {[t.name for t in tools]}")性能优化
连接池
# Reuse client connections
async with MultiServerMCPClient(config) as client:
# Multiple operations using the same client
pass批处理操作
# Execute multiple queries efficiently
queries = [
"Calculate 5 + 3",
"What's 7 × 9",
"Weather in Boston"
]
responses = await asyncio.gather(*[
agent.ainvoke({"messages": [{"role": "user", "content": q}]})
for q in queries
])高级主题
自定义传输协议
为特定使用场景实现自定义传输：
class CustomTransport:
async def connect(self, config):
# Custom connection logic
pass
async def send_request(self, request):
# Custom request handling
pass代理工作流定制
创建专门的代理工作流程：
from langgraph.graph import StateGraph, END
def create_custom_agent():
workflow = StateGraph(AgentState)
# Add custom nodes
workflow.add_node("math_operations", handle_math)
workflow.add_node("weather_queries", handle_weather)
# Define conditional edges
workflow.add_conditional_edges(
"router",
route_request,
{"math": "math_operations", "weather": "weather_queries"}
)
return workflow.compile()错误恢复模式
实现强大的错误处理机制：
async def resilient_query(agent, query, max_retries=3):
for attempt in range(max_retries):
try:
return await agent.ainvoke({"messages": [{"role": "user", "content": query}]})
except Exception as e:
if attempt == max_retries - 1:
raise
await asyncio.sleep(2 ** attempt) # Exponential backoff监控与可观测性
增加全面监控：
import time
from typing import Dict, Any
class AgentMonitor:
def __init__(self):
self.metrics = {"requests": 0, "errors": 0, "response_times": []}
async def monitored_invoke(self, agent, query):
start_time = time.time()
try:
result = await agent.ainvoke(query)
self.metrics["requests"] += 1
return result
except Exception as e:
self.metrics["errors"] += 1
raise
finally:
self.metrics["response_times"].append(time.time() - start_time)贡献；做出贡献
- 为仓库创建分支（或“克隆仓库”）
- 创建一个特性分支（
git checkout -b feature/amazing-feature) - 在进行适当测试后，做出您的更改
- 确保代码遵循风格指南
- 根据需要更新文档
- 提交一个拉取请求
支持
对于问题和疑问：
- 查看上面的故障排除部分
- 查阅LangChain和MCP的文档
- 在项目仓库中打开一个问题（或提交一个问题）
______________________________________________________________________
*这个项目展示了MCP与LangChain之间前沿的集成模式，为构建具备复杂推理能力的高级多工具AI应用奠定了基础。*