LangGraph MCP 集成与工具扩展详细解读

📚 概述

Model Context Protocol (MCP) 是一个标准化协议，用于连接 LLM 应用与外部工具和数据源。本章讲解如何在 Research Agent 中集成 MCP，实现工具能力的灵活扩展。

核心价值：

🔌 标准化接口 - 无需为每个工具编写适配代码
🚀 动态工具发现 - 运行时查询可用工具
🌐 本地 + 远程 - 支持本地服务器和云端 API
🔧 易于扩展 - 插拔式添加新工具

🎯 核心概念：MCP 架构

MCP 是什么？

Model Context Protocol 是一个开放标准，定义了：

LLM 应用（Client）如何发现和调用工具
工具提供方（Server）如何暴露功能
双方如何通信（JSON-RPC 协议）

客户端-服务器模型

┌──────────────────────────────────┐
│   LangGraph Agent (Client)       │
│                                  │
│  ┌────────────────────────────┐  │
│  │ MultiServerMCPClient       │  │
│  │  - 管理多个 MCP server     │  │
│  │  - 查询可用工具            │  │
│  │  - 转发工具调用            │  │
│  └────────────────────────────┘  │
└──────────────────────────────────┘
         ↓ (stdio / HTTP)
┌──────────────────────────────────┐
│   MCP Server (Tool Provider)     │
│                                  │
│  示例:                           │
│  - Filesystem Server (文件操作)  │
│  - Database Server (数据库查询)  │
│  - API Server (远程 API 调用)    │
└──────────────────────────────────┘

🔧 核心技术：Transport 模式

stdio Transport（本地通信）

工作原理：

Client 启动 Server 作为子进程
通过标准输入/输出（stdin/stdout）通信
使用 JSON-RPC 格式交换消息

配置示例：

python

mcp_config = {
    "filesystem": {
        "command": "npx",  # 启动命令
        "args": [
            "-y",  # 自动安装
            "@modelcontextprotocol/server-filesystem",  # Server 包
            "/path/to/documents"  # Server 参数（允许访问的目录）
        ],
        "transport": "stdio"  # 使用 stdin/stdout
    }
}

适用场景：

✅ 本地文件系统访问
✅ 本地数据库查询
✅ 本地脚本执行
✅ 开发和测试

优点：

快速（无网络延迟）
安全（不暴露网络端口）
简单（无需认证配置）

缺点：

只能访问本地资源
需要安装 Server 软件

HTTP Transport（远程通信）

工作原理：

Client 连接到已运行的远程 Server
通过 HTTP/HTTPS 发送请求
Server 返回 JSON 响应

配置示例：

python

mcp_config = {
    "remote_api": {
        "url": "https://mcp.example.com/sse",  # Server URL
        "transport": "http",
        "headers": {
            "Authorization": "Bearer YOUR_TOKEN",  # 认证
            "X-API-Version": "v1"
        }
    }
}

适用场景：

✅ 第三方 API 集成（Asana, PayPal, Zapier）
✅ 团队共享的 MCP 服务
✅ 云端数据访问
✅ 生产部署

优点：

无需本地安装
可访问远程资源
便于集中管理

缺点：

网络延迟
需要认证管理
依赖外部服务可用性

🔍 关键技术细节：为什么必须异步？

MCP 协议的异步本质

python

# ❌ 这样不行
tools = client.get_tools()  # MCP 没有同步方法

# ✅ 必须这样
tools = await client.get_tools()  # 异步调用

三个原因：

1. 进程间通信（IPC）

Client (Python 进程)
    ↓ (stdin)
Server (Node.js 子进程)
    ↓ (stdout)
Client (Python 进程)

stdin/stdout 是 I/O 操作，天然异步
阻塞等待会冻结整个进程
异步允许并发处理多个请求

2. 网络请求（HTTP）

python

# HTTP transport 发送请求
await client.tools["read_file"].ainvoke({"path": "..."})

# 可能需要等待：
# - DNS 解析: 50ms
# - TLS 握手: 100ms
# - 服务器处理: 200ms
# - 数据传输: 50ms
# 总计: 400ms

# 同步会阻塞，异步可以同时发送多个请求

3. LangChain MCP Adapters 的设计

python

# LangChain MCP 工具强制异步
class MCPTool(StructuredTool):
    def invoke(self, *args, **kwargs):
        raise NotImplementedError(
            "MCP tools must be used asynchronously. Use ainvoke() instead."
        )

    async def ainvoke(self, *args, **kwargs):
        # 实际实现
        ...

设计理由：

确保一致的异步行为
避免阻塞事件循环
支持高并发场景

🛠️ 实战：集成 Filesystem MCP Server

1. 安装和配置

python

from langchain_mcp_adapters.client import MultiServerMCPClient

# MCP 配置
mcp_config = {
    "filesystem": {
        "command": "npx",
        "args": [
            "-y",
            "@modelcontextprotocol/server-filesystem",
            str(Path.cwd() / "research_docs")  # 允许访问的目录
        ],
        "transport": "stdio"
    }
}

# 创建 Client
client = MultiServerMCPClient(mcp_config)

关键点：

MultiServerMCPClient 可以管理多个 MCP server
Client 自动启动 server 子进程（stdio transport）
或连接到远程 server（HTTP transport）

2. 查询可用工具

python

# 获取所有工具（异步）
tools = await client.get_tools()

# 输出示例
for tool in tools:
    print(f"Tool: {tool.name}")
    print(f"Description: {tool.description}")

Filesystem Server 提供的工具：

工具名	功能	示例
`read_file`	读取文件内容	读取研究文档
`read_multiple_files`	批量读取文件	一次读取多个文档
`write_file`	写入文件	保存研究结果
`search_files`	搜索文件	按关键词查找
`list_directory`	列出目录	查看可用文档
`list_allowed_directories`	查看权限	确认访问范围

3. 在 Agent 中使用 MCP 工具

python

from deep_research_from_scratch.utils import think_tool

async def llm_call(state: ResearcherState):
    """
    Agent 决策节点（异步）

    关键：必须异步获取 MCP 工具
    """
    # 获取 MCP 工具
    mcp_tools = await client.get_tools()

    # 组合自定义工具
    all_tools = mcp_tools + [think_tool]

    # 绑定到模型
    model_with_tools = model.bind_tools(all_tools)

    # LLM 决策
    response = model_with_tools.invoke(
        [SystemMessage(content=research_agent_prompt_with_mcp)] +
        state["researcher_messages"]
    )

    return {"researcher_messages": [response]}

4. 工具执行节点（关键：异步处理）

python

async def tool_node(state: ResearcherState):
    """
    执行工具调用（异步）

    重要：MCP 工具必须用 ainvoke
    """
    tool_calls = state["researcher_messages"][-1].tool_calls

    # 获取最新工具引用
    mcp_tools = await client.get_tools()
    all_tools = mcp_tools + [think_tool]
    tools_by_name = {tool.name: tool for tool in all_tools}

    # 执行工具调用
    observations = []
    for tool_call in tool_calls:
        tool = tools_by_name[tool_call["name"]]

        if tool_call["name"] == "think_tool":
            # think_tool 是同步的
            observation = tool.invoke(tool_call["args"])
        else:
            # MCP 工具是异步的
            observation = await tool.ainvoke(tool_call["args"])

        observations.append(observation)

    # 格式化结果
    tool_outputs = [
        ToolMessage(
            content=observation,
            name=tool_call["name"],
            tool_call_id=tool_call["id"]
        )
        for observation, tool_call in zip(observations, tool_calls)
    ]

    return {"researcher_messages": tool_outputs}

关键区别：

python

# 自定义工具（同步）
result = tavily_search.invoke({"query": "..."})

# MCP 工具（异步）
result = await read_file.ainvoke({"path": "..."})

🎭 实战示例：从本地文档研究

研究流程

python

研究主题: "总结本地文档中关于 SF 咖啡店的信息"

┌─────────────────────────────────────────┐
│ Round 1: 探索可用文档                    │
├─────────────────────────────────────────┤
│ Tool: list_allowed_directories()        │
│ 结果: ["/path/to/research_docs"]        │
│                                         │
│ Tool: list_directory("/path/to/...")   │
│ 结果: ["coffee_shops_sf.md",           │
│        "sf_restaurants.md"]             │
│                                         │
│ Tool: think_tool()                      │
│ 反思: "找到相关文档，需要读取内容"       │
└─────────────────────────────────────────┘
    ↓
┌─────────────────────────────────────────┐
│ Round 2: 读取文档                        │
├─────────────────────────────────────────┤
│ Tool: read_file("coffee_shops_sf.md")   │
│ 结果: Blue Bottle, Philz, Sightglass... │
│                                         │
│ Tool: think_tool()                      │
│ 反思: "已获得足够信息，可以总结"         │
└─────────────────────────────────────────┘
    ↓
┌─────────────────────────────────────────┐
│ 生成答案                                 │
│ "基于本地文档，SF 知名咖啡店包括..."     │
└─────────────────────────────────────────┘

⚠️ 重要注意事项

1. LangGraph Platform 部署问题

问题： 在 LangGraph Platform 部署时，MCP client 初始化可能失败

python

# ❌ 直接初始化（部署时可能失败）
client = MultiServerMCPClient(mcp_config)

# ✅ 懒加载初始化
_client = None

def get_mcp_client():
    global _client
    if _client is None:
        _client = MultiServerMCPClient(mcp_config)
    return _client

# 在节点中使用
async def llm_call(state):
    client = get_mcp_client()  # 延迟初始化
    tools = await client.get_tools()
    ...

原因：

LangGraph Platform 可能在导入时扫描模块
过早初始化会启动子进程，导致资源问题
懒加载确保只在实际使用时初始化

2. 工具引用更新

python

# ❌ 错误：重用旧的工具引用
tools = await client.get_tools()  # 在初始化时获取

async def tool_node(state):
    # 使用旧引用 - 可能已失效
    tool.ainvoke(...)

# ✅ 正确：每次重新获取
async def tool_node(state):
    tools = await client.get_tools()  # 重新获取
    tool = tools_by_name[name]
    await tool.ainvoke(...)

原因：

MCP server 可能重启
工具列表可能更新
重新获取确保引用有效

3. Jupyter 环境的异步兼容

python

# Jupyter 已经有事件循环，asyncio.gather 可能冲突
try:
    import nest_asyncio
    from IPython import get_ipython

    if get_ipython() is not None:
        nest_asyncio.apply()  # 允许嵌套事件循环
except ImportError:
    pass  # 不在 Jupyter 中，无需处理

🌐 远程 MCP Server 示例

配置第三方 MCP 服务

python

# 示例：连接到 Asana MCP Server
mcp_config = {
    "asana": {
        "url": "https://mcp.asana.com/sse",
        "transport": "http",
        "headers": {
            "Authorization": f"Bearer {os.getenv('ASANA_TOKEN')}",
            "Content-Type": "application/json"
        }
    }
}

client = MultiServerMCPClient(mcp_config)

# 获取工具
tools = await client.get_tools()
# 可能包括: create_task, list_projects, update_task 等

可用的第三方 MCP Server（截至 2025）：

Asana - 任务管理
Cloudflare - 云服务管理
PayPal - 支付处理
Zapier - 自动化集成

💡 最佳实践

1. 何时使用 MCP vs 自定义工具

场景	推荐方案
简单的 API 调用	自定义工具（@tool 装饰器）
复杂的文件操作	MCP Filesystem Server
第三方服务集成	MCP HTTP Server
需要频繁更新的工具	MCP（动态发现）
高性能要求	自定义工具（避免 IPC 开销）

2. 安全考虑

python

# ❌ 不安全：允许访问整个文件系统
mcp_config = {
    "filesystem": {
        "args": [..., "/"]  # 根目录
    }
}

# ✅ 安全：限制访问范围
mcp_config = {
    "filesystem": {
        "args": [..., "/path/to/research_docs_only"]
    }
}

3. 错误处理

python

async def tool_node(state):
    try:
        client = get_mcp_client()
        tools = await client.get_tools()
        # 执行工具
        ...
    except Exception as e:
        # 降级：使用缓存的工具或返回错误消息
        return {
            "researcher_messages": [
                ToolMessage(
                    content=f"MCP 工具执行失败: {e}",
                    tool_call_id=tool_call["id"]
                )
            ]
        }

📊 MCP vs 自定义工具对比

维度	MCP 工具	自定义工具
实现复杂度	低（使用现成 server）	中（需要编写代码）
性能	中（IPC/HTTP 开销）	高（直接调用）
灵活性	高（动态发现）	中（静态定义）
维护性	低（server 更新自动生效）	中（需要手动更新）
适用场景	标准化操作	定制化逻辑

🎓 核心知识点总结

MCP 架构

Client (LangGraph Agent)
    ↓ (查询工具)
Server (MCP Server)
    ↓ (返回工具列表)
Client 绑定工具
    ↓ (LLM 调用工具)
Client 转发请求
    ↓ (stdio/HTTP)
Server 执行
    ↓ (返回结果)
Client 接收

异步必要性

IPC 通信 - stdin/stdout 是 I/O 操作
网络请求 - HTTP 调用需要等待
LangChain 设计 - MCP 适配器强制异步

stdio vs HTTP

stdio	HTTP
本地	远程
快速	较慢
安全	需认证
需安装	无需安装

🚀 下一步

完成本节，你已经掌握了 MCP 集成的核心技术。

下一章：9.4 多智能体协同研究 - 学习如何使用 Supervisor 模式协调多个 Research Agent，实现并行研究和上下文隔离！

LangGraph MCP 集成与工具扩展详细解读 ​

📚 概述 ​

🎯 核心概念：MCP 架构 ​

MCP 是什么？ ​

客户端-服务器模型 ​

🔧 核心技术：Transport 模式 ​

stdio Transport（本地通信） ​

HTTP Transport（远程通信） ​

🔍 关键技术细节：为什么必须异步？ ​

MCP 协议的异步本质 ​

1. 进程间通信（IPC） ​

2. 网络请求（HTTP） ​

3. LangChain MCP Adapters 的设计 ​

🛠️ 实战：集成 Filesystem MCP Server ​

1. 安装和配置 ​

2. 查询可用工具 ​

3. 在 Agent 中使用 MCP 工具 ​

4. 工具执行节点（关键：异步处理） ​

🎭 实战示例：从本地文档研究 ​

研究流程 ​

⚠️ 重要注意事项 ​

1. LangGraph Platform 部署问题 ​

2. 工具引用更新 ​

3. Jupyter 环境的异步兼容 ​

🌐 远程 MCP Server 示例 ​

配置第三方 MCP 服务 ​

💡 最佳实践 ​

1. 何时使用 MCP vs 自定义工具 ​

2. 安全考虑 ​

3. 错误处理 ​

📊 MCP vs 自定义工具对比 ​

🎓 核心知识点总结 ​

MCP 架构 ​

异步必要性 ​

stdio vs HTTP ​

🚀 下一步 ​

LangGraph MCP 集成与工具扩展详细解读

📚 概述

🎯 核心概念：MCP 架构

MCP 是什么？

客户端-服务器模型

🔧 核心技术：Transport 模式

stdio Transport（本地通信）

HTTP Transport（远程通信）

🔍 关键技术细节：为什么必须异步？

MCP 协议的异步本质

1. 进程间通信（IPC）

2. 网络请求（HTTP）

3. LangChain MCP Adapters 的设计

🛠️ 实战：集成 Filesystem MCP Server

1. 安装和配置

2. 查询可用工具

3. 在 Agent 中使用 MCP 工具

4. 工具执行节点（关键：异步处理）

🎭 实战示例：从本地文档研究

研究流程

⚠️ 重要注意事项

1. LangGraph Platform 部署问题

2. 工具引用更新

3. Jupyter 环境的异步兼容

🌐 远程 MCP Server 示例

配置第三方 MCP 服务

💡 最佳实践

1. 何时使用 MCP vs 自定义工具

2. 安全考虑

3. 错误处理

📊 MCP vs 自定义工具对比

🎓 核心知识点总结

MCP 架构

异步必要性

stdio vs HTTP

🚀 下一步