Fastmcp Example

MCP服务器+语言链/语言图集成

   

一个完整的示例，演示如何构建MCP（模型上下文协议）服务器并将其与LangChain和LangGraph集成，以构建AI代理工作流。

关键词： MCP、模型上下文协议、LangChain、LangGraph、AI代理、FastMCP、Tavily、OpenAI、Python、异步、工作流自动化、机器学习、LLM集成、代理框架

📚 目录

• 快速开始
• 文档
• 概述
• 特性
• 设置
• 项目结构
• 文件目的
• 添加新工具
• 例子
• 故障排除
• 贡献

快速开始

# Install dependencies
pip install -r requirements.txt

# Set up environment variables
cp .env.example .env  # Edit with your API keys

# Run the MCP server
python server.py

# In another terminal, run the LangGraph example
python langgraph_app.py

📖 文档

• 学习路径 -从初级到高级的分步指南
• 自定义工具指南 -创建自定义API工具的完整指南
• 架构概述 -系统设计和组件分解
• 快速参考 -快速查找常见任务

概述

该项目展示了 模型上下文协议（MCP） 随着 LangChain 和 LangGraph 用于构建生产就绪的AI代理工作流。它提供了一个工作示例，说明如何：

• 创建一个公开自定义工具的MCP服务器
• 将MCP工具与LangChain代理集成
• 使用LangGraph构建多步骤工作流
• 从JSON配置创建自定义工具
• 在AI应用程序中使用web搜索、HTTP请求和其他工具

什么是MCP？ 模型上下文协议是人工智能应用程序访问外部工具和数据源的标准化方式，可实现更强大、更灵活的人工智能代理。

项目组成部分：

• MCP服务器 (server.py)-通过HTTP上的模型上下文协议公开工具
• LangGraph应用程序 (langgraph_app.py)-演示使用MCP工具的多步骤工作流程
• 朗链客户端 (langchain_client.py)-演示如何将MCP工具与LangChain代理一起使用
• 自定义工具 (config/custom_tools.py)-从JSON配置创建工具
• 示例：自定义工具 (example_custom_tools.py)-演示自定义工具的使用方法

特性

• 🔧 MCP服务器 使用多种工具（网络搜索、HTTP请求、数学运算）
• 🔗 LangGraph集成 -多步骤代理工作流
• 🤖 LangChain代理商 -使用AI代理的工具
• 🛠️ 自定义工具 -从JSON配置创建工具（无需代码！）
• 🔍 网页搜索 -由Tavilly API提供动力
• 🌐 HTTP请求 -通用HTTP客户端工具
• ➕ 数学运算 -示例工具（加法、乘法）

设置

1.先决条件

• Python 3.8或更高版本
• 虚拟环境（推荐）

2.安装

# Clone the repository
git clone 
cd mcs-mcp

# Create and activate virtual environment
python -m venv venv
# On Windows:
venv\Scripts\activate
# On macOS/Linux:
source venv/bin/activate

# Install dependencies
pip install -r requirements.txt

3.环境配置

创建一个 .env 项目根目录中的文件：

# Required for web search functionality
TAVILY_API_KEY=your_tavily_api_key_here

# Optional: HTTP timeout in seconds (default: 15)
HTTP_TIMEOUT_SECONDS=15

# Optional: OpenAI API key (required for LangChain/LangGraph)
OPENAI_API_KEY=your_openai_api_key_here

获取API密钥：

• Tavilly API密钥：注册地址： tavily.com 获取免费的API密钥
• OpenAI API密钥：从以下位置获取一个 platform.openai.com

运行MCP服务器

MCP服务器通过HTTP传输公开工具。它可以通过两种方式运行：

选项1：独立服务器

python server.py

服务器将于启动 http://localhost:8000 默认情况下。它将：

• 通过MCP协议公开所有工具
• 接受来自MCP客户端的连接
• 处理工具调用

备注：服务器无限期运行，直到停止（Ctrl+C）。

选项2：自动（通过客户端）

使用时，MCP客户端会自动启动服务器 stdio 运输。但是，此项目使用HTTP传输，因此您需要单独运行服务器。

项目结构

mcs-mcp/
├── server.py                  # MCP server with tool definitions
├── langgraph_app.py           # LangGraph workflow example
├── langchain_client.py        # LangChain agent example (MCP tools)
├── example_custom_tools.py    # Custom tools example
├── custom_tools.json          # Custom tool configurations
├── helper.py                  # Shared utilities
├── config/
│   ├── config.py              # Environment configuration helpers
│   ├── mcp_client.py          # MCP client setup
│   └── custom_tools.py        # Custom tool factory and loader
├── docs/
│   ├── LEARNING_PATH.md       # Step-by-step learning guide
│   ├── CUSTOM_TOOLS.md        # Custom tools documentation
│   └── ARCHITECTURE.md        # Architecture overview
├── requirements.txt           # Python dependencies
└── README.md                  # This file

文件目的

server.py -MCP服务器

MCP服务器定义并公开了AI代理可以使用的工具。它使用FastMCP创建基于HTTP的MCP服务器。

可用工具：

• web_search -使用Tavily进行一般互联网搜索
• site_search -域限制搜索
• http_request -API调用的通用HTTP客户端
• add -简单添加工具（示例）
• multiply -简单乘法工具（示例）

主要特点：

• 工具定义使用 @mcp.tool() 装饰器
• 通过MCP协议自动发现刀具
• HTTP传输，便于集成

例子：

@mcp.tool()
def web_search(query: str, top_k: int = 5) -> List[Dict[str, Any]]:
    """General internet search via Tavily."""
    # Implementation...

langgraph_app.py -LangGraph工作流

使用LangGraph演示多步骤代理工作流。工作流程：

1. 规划师 -根据查询执行网络搜索
2. API取数器 -可选地从API端点获取数据
3. 总结者 -使用LLM总结结果

主要特点：

• 使用TypedDict进行状态管理
• 异步节点执行
• MCP服务器的工具集成
• LLM驱动的摘要

用途：

python langgraph_app.py

工作流程：

START → planner → api → summarize → END

该图处理：

• 输入： {"query": "latest AI news", "endpoint": "https://api.github.com"}
• 输出：合并搜索和API数据的摘要结果

langchain_client.py -朗链代理

演示如何将MCP工具与LangChain代理一起使用。它表明：

• 从MCP服务器发现工具
• 工具绑定到LLM
• 工具调用和结果处理
• 使用工具进行多回合对话

主要特点：

• 自动工具发现
• 工具调用处理
• 响应处理
• 错误处理

用途：

python langchain_client.py

example_custom_tools.py -自定义工具示例

演示如何使用从JSON配置加载的自定义工具。它显示：

• 从JSON加载自定义工具
• 将MCP工具与自定义工具合并
• 同时使用这两种工具类型
• 工具调用和结果处理

主要特点：

• 基于JSON的工具配置
• 动态刀具加载
• 工具合并
• 完整的工作示例

用途：

python example_custom_tools.py

自定义工具系统

从JSON配置创建可重复使用的API工具，而无需编写Python代码。

文件夹：

• custom_tools.json -工具配置文件
• config/custom_tools.py -工具厂和装载机
• docs/CUSTOM_TOOLS.md -完整的文档

快速示例：

{
  "name": "get_order_details",
  "description": "Fetches order details by ID",
  "base_url": "https://api.example.com/orders/{order_id}",
  "method": "GET",
  "headers": {"Authorization": "Bearer {api_token}"},
  "parameters": [
    {"name": "order_id", "type": "string", "required": true},
    {"name": "api_token", "type": "string", "required": true}
  ]
}

看 CUSTOM_TOOLS.md 完整的指南。

添加新工具

选项1：添加MCP工具（服务器端）

要向MCP服务器添加新工具，请执行以下操作：

1. 定义工具功能 在 server.py:

@mcp.tool()
def my_new_tool(param1: str, param2: int) -> dict:
    """Description of what the tool does."""
    # Your implementation
    return {"result": "value"}

1. 重新启动服务器 -工具通过MCP协议自动发现。

1. 在您的应用程序中使用:

from config.mcp_client import get_tool_by_name

tool = await get_tool_by_name("my_new_tool")
result = await tool.ainvoke({"param1": "value", "param2": 42})

选项2：添加自定义工具（JSON配置）

要从JSON添加自定义工具（无需代码）：

1. 添加工具配置 到 custom_tools.json:

{
  "name": "my_custom_tool",
  "description": "Description of what the tool does",
  "base_url": "https://api.example.com/endpoint/{param1}",
  "method": "GET",
  "parameters": [
    {"name": "param1", "type": "string", "required": true}
  ]
}

1. 无需重新启动！ 工具在运行时加载。

1. 在您的应用程序中使用:

from config.custom_tools import get_all_tools

tools = await get_all_tools()  # Includes MCP + custom tools
# Tool is automatically available

看 CUSTOM_TOOLS.md 完整的指南。

配置

MCP客户端配置

编辑 config/mcp_client.py 更改运输方式：

HTTP传输（默认）：

client = MultiServerMCPClient(
    connections={
        "mcs-mcp-server": {
            "transport": "streamable_http",
            "url": "http://localhost:8000/mcp"
        }
    }
)

标准运输（备选）：

client = MultiServerMCPClient(
    connections={
        "mcs-mcp-server": {
            "transport": "stdio",
            "command": "python",
            "args": ["server.py"]
        }
    }
)

例子

示例1：运行LangGraph工作流

# Terminal 1: Start the MCP server
python server.py

# Terminal 2: Run the LangGraph app
python langgraph_app.py

示例2：以编程方式使用MCP工具

import asyncio
from config.mcp_client import get_tool_by_name

async def main():
    # Get a tool
    web_search = await get_tool_by_name("web_search")
    
    # Use the tool
    results = await web_search.ainvoke({
        "query": "Python async programming",
        "top_k": 5
    })
    
    print(results)

asyncio.run(main())

示例3：使用自定义工具

import asyncio
from config.custom_tools import get_all_tools
from langchain_openai import ChatOpenAI

async def main():
    # Get all tools (MCP + custom)
    tools = await get_all_tools()
    
    # Bind to LLM
    model = ChatOpenAI(model="gpt-4o")
    model_with_tools = model.bind_tools(tools)
    
    # Use tools (LLM will automatically choose the right one)
    response = await model_with_tools.ainvoke([
        {"role": "user", "content": "Get info about langchain-ai/langchain repo"}
    ])
    
    print(response)

asyncio.run(main())

示例4：自定义LangGraph节点

from config.mcp_client import get_tool_by_name

async def my_custom_node(state: GraphState) -> Dict[str, Any]:
    # Get tool
    tool = await get_tool_by_name("http_request")
    
    # Use tool
    result = await tool.ainvoke({
        "url": "https://api.example.com/data",
        "method": "GET"
    })
    
    return {"custom_data": result}

示例5：运行自定义工具示例

# Terminal 1: Start the MCP server
python server.py

# Terminal 2: Run custom tools example
python example_custom_tools.py

故障排除

服务器无法启动

• 检查端口8000是否可用
• 验证是否已安装所有依赖项
• 检查 .env 文件存在并且具有必需的密钥

未找到工具

• 确保MCP服务器正在运行
• 检查 config/mcp_client.py 具有正确的服务器URL
• 验证工具名称是否完全匹配（区分大小写）

API密钥错误

• 验证 TAVILY_API_KEY 设定在 .env
• 检查 OPENAI_API_KEY 为LLM功能设置
• 确保 .env 文件位于项目根目录中

导入错误

• 激活虚拟环境
• 跑 pip install -r requirements.txt
• 检查Python版本（3.8+）

依赖项

• fastmcp -快速MCP服务器实施
• 请求： -HTTP客户端库
• 巨蟒 -Tavilly搜索API客户端
• python dotenv -环境变量管理
• 语言链 -LLM应用框架
• 兰开夏 -OpenAI集成
• langchain核心 -核心LangChain组件
• 兰格拉夫 -基于图的代理工作流
• langchain mcp适配器 -LangChain的MCP客户端适配器

许可证

此项目根据MIT许可证获得许可-请参阅 许可证 文件以获取详细信息。

MIT许可证摘要：

• ✅ 允许商业使用
• ✅ 允许修改
• ✅ 允许分发
• ✅ 允许私人使用
• ✅ 无责任
• ✅ 无担保

您可以自由地将此项目用于任何目的，包括商业应用程序。归因是值得赞赏的，但不是必需的。

贡献

我们欢迎并衷心感谢您的贡献！这个项目对所有人开放，我们鼓励您帮助改进它。

如何做出贡献

1. 分叉存储库

- 点击GitHub上的“Fork”按钮创建自己的副本

1. 创建要素分支

   git checkout -b feature/your-feature-name
   # or
   git checkout -b fix/your-bug-fix

1. 进行更改

- 编写干净、可读的代码 - 遵循现有的代码风格和约定 - 为复杂逻辑添加注释/docstring - 必要时更新文档

1. 测试您的更改

- 确保现有功能仍然有效 - 彻底测试新功能 - 检查是否有任何掉毛错误

1. 提交您的更改

   git commit -m "Add: description of your changes"

使用清晰、描述性的提交消息：

- Add: 对于新功能 - Fix: 用于修复错误 - Update: 改进 - Docs: 文档更改 - Refactor: 用于代码重构

1. 推到你的叉子

   git push origin feature/your-feature-name

1. 创建拉取请求

- 转到GitHub上的原始存储库 - 点击“新建拉取请求” - 选择您的叉子和树枝 - 填写PR模板： - 变更说明 - 为什么需要改变 - 任何重大变化 - 屏幕截图（如适用）

贡献指南

• 代码的风格：遵循PEP 8 Python风格指南
• 文档：如果添加新功能，请更新README.md
• 测试：在提交之前测试您的更改
• 尊重他人：在讨论中保持友善和建设性
• 提问：如果不确定，请先打开一个问题进行讨论

贡献类型

我们欢迎各种类型的捐款：

• 🐛 错误报告：发现一个bug？打开一个问题！
• 💡 功能请求：有主意吗？分享！
• 📝 文档：改进文档，修复拼写错误，添加示例
• 🧪 测试：添加测试，提高测试覆盖率
• 🎨 代码：修复错误、添加功能、重构代码
• 🌍 本地化：翻译文件
• 📢 促销：与他人分享项目

获取帮助

• 问题？ 打开A
• 发现Bug了吗？ 打开A
• 安全问题？ 请直接发送电子邮件（不要公开问题）

行为准则

• 尊重他人，包容他人
• 欢迎新来者并帮助他们学习
• 注重建设性反馈
• 尊重不同的观点和经验

感谢您的贡献！ 🎉