强强联合:轻松构建企业级AI工作流,效率翻倍!
本项目展示了 LangGraph 与 FastMCP 2.0 的两种典型集成方式:
- 客户端集成:将 FastMCP 服务器暴露的工具作为 LangGraph 工作流中的可调用节点
- 服务端封装:把 LangGraph Agent 封装为 FastMCP 服务器,对外以标准协议提供能力
git clone <repository-url>
cd fastmcp_demo使用 pip:
pip install -r requirements.txt使用 uv (推荐):
uv pip install -r requirements.txt使用 poetry:
poetry install创建 .env 文件并配置你的 API 密钥:
# 复制环境变量模板
cp .env.example .env
# 编辑 .env 文件,添加你的 API 密钥
# DEEPSEEK_API_KEY=sk-your-deepseek-api-key-here# 启动 MCP 服务器
python mcp_server_sample.py
# 在新终端运行 LangGraph 客户端
python langgraph_use_mcp_as_client.py
# 或运行 LangGraph 服务器
python langgraph_use_mcp_as_server.pyfastmcp_demo/
├── mcp_server_sample.py # MCP 服务器示例(提供数学工具)
├── langgraph_use_mcp_as_client.py # LangGraph 调用 MCP 服务
├── langgraph_use_mcp_as_server.py # LangGraph 封装为 MCP 服务器
├── requirements.txt # Python 依赖
├── pyproject.toml # 项目配置
├── .env.example # 环境变量模板
└── README.md # 本文档
- FastMCP >= 2.12.0 - MCP 服务器框架
- LangGraph >= 0.2.0 - 工作流编排框架
- LangChain >= 0.3.0 - LLM 应用框架
- langchain-mcp-adapters >= 0.1.0 - MCP 适配器
pip install -r requirements.txt# 安装 uv
pip install uv
# 使用 uv 安装依赖
uv pip install -r requirements.txt# 安装 poetry
pip install poetry
# 安装依赖
poetry install# DeepSeek API配置
DEEPSEEK_API_KEY=sk-your-deepseek-api-key-here
# OpenAI API配置 (可选)
# OPENAI_API_KEY=sk-your-openai-api-key-here- DeepSeek: 访问 DeepSeek 官网 注册获取 API 密钥
- OpenAI (可选): 访问 OpenAI Platform 获取
启动一个提供数学计算工具的 MCP 服务器:
python mcp_server_sample.py服务器将在 http://127.0.0.1:8000/mcp 启动,提供以下工具:
add(a, b)- 加法multiply(a, b)- 乘法subtract(a, b)- 减法divide(a, b)- 除法
在另一个终端窗口运行:
# 确保 MCP 服务器已经在运行
python langgraph_use_mcp_as_client.py这个示例将:
- 连接到 MCP 服务器
- 获取可用工具
- 构建 LangGraph 工作流
- 使用工具解决数学问题
示例输出:
=== 获取MCP工具 ===
可用工具: ['add', 'multiply', 'subtract', 'divide']
=== 测试数学计算 ===
测试问题: what's (3 + 5) x 12?
最终答案: The result of (3 + 5) × 12 is 96.
运行 LangGraph 封装的 MCP 服务器:
python langgraph_use_mcp_as_server.py服务器将在 http://127.0.0.1:8004/mcp 启动,提供以下工具:
process_text_with_langgraph- 使用 LangGraph 处理文本analyze_sentiment- 情感分析summarize_text- 文本摘要
# 运行内置测试
python langgraph_use_mcp_as_server.py --testMCP Inspector 是一个可视化工具,用于测试和调试 MCP 服务器。
# 使用 npm 安装
npm install -g @modelcontextprotocol/inspector
# 或使用 npx 直接运行(无需安装)
npx @modelcontextprotocol/inspector- 启动你的 MCP 服务器
- 运行 Inspector:
npx @modelcontextprotocol/inspector
- 在浏览器中打开
http://localhost:6274 - 连接到你的 MCP 服务器 URL(如
http://127.0.0.1:8000/mcp) - 测试工具调用
┌─────────────┐ HTTP/WS ┌──────────────┐
│ LangGraph │ ───────────────> │ MCP Server │
│ Client │ │ (Tools) │
└─────────────┘ └──────────────┘
│ │
v v
┌─────────────┐ ┌──────────────┐
│ LLM │ │ Execute │
│ (DeepSeek) │ │ Function │
└─────────────┘ └──────────────┘
┌─────────────┐ HTTP/WS ┌──────────────┐
│ MCP Client │ ───────────────> │ FastMCP │
│ │ │ Server │
└─────────────┘ └──────────────┘
│
v
┌──────────────┐
│ LangGraph │
│ Workflow │
└──────────────┘
│
v
┌──────────────┐
│ LLM │
│ (DeepSeek) │
└──────────────┘
在 mcp_server_sample.py 中添加:
@mcp.tool()
def your_tool(param1: str, param2: int) -> str:
"""Your tool description"""
# 实现你的工具逻辑
return resultfrom langgraph.graph import StateGraph, START, END
def create_custom_workflow():
workflow = StateGraph(YourStateType)
# 添加节点
workflow.add_node("node1", node1_function)
workflow.add_node("node2", node2_function)
# 添加边
workflow.add_edge(START, "node1")
workflow.add_edge("node1", "node2")
workflow.add_edge("node2", END)
return workflow.compile()问题: 连接失败: Connection refused
解决方案:
- 确保 MCP 服务器已经启动
- 检查端口是否被占用:
lsof -i :8000 - 更换端口号
问题: Invalid API key
解决方案:
- 检查
.env文件中的 API 密钥是否正确 - 确保已经加载环境变量:
load_dotenv()
问题: No matching distribution found
解决方案:
- 升级 pip:
pip install --upgrade pip - 使用国内镜像:
pip install -i https://pypi.tuna.tsinghua.edu.cn/simple -r requirements.txt
问题: Tool execution timeout
解决方案:
- 在 MCP Inspector 中调整超时时间
- 优化工具实现,减少处理时间
- 考虑异步处理
欢迎提交 Pull Request 或创建 Issue!
- Fork 项目
- 创建特性分支:
git checkout -b feature/your-feature - 提交更改:
git commit -m 'Add your feature' - 推送分支:
git push origin feature/your-feature - 创建 Pull Request
MIT License
- 使用
uv可以显著提升依赖安装速度 - 开发时可以使用
--test参数快速测试功能 - MCP Inspector 是调试 MCP 服务器的最佳工具
- 建议在虚拟环境中运行项目