DeepAgents + MCP Tools：从零搭建智能体“外挂工具箱”

码以致用

439人浏览 · 2026-05-11 17:26:47

码以致用 · 2026-05-11 17:26:47 发布

在大模型应用开发中，让 LLM 调用外部工具始终是架构设计的核心命题。从早期的 Function Calling，到 LangChain 的 Tool 体系，再到 Anthropic 提出的 MCP (Model Context Protocol)，工具调用的标准化和可扩展性不断被推向前沿。

本文从一个实际项目出发，完整展示如何在 DeepAgents 框架 (v0.5.3) 中集成 MCP 协议，构建一个既能调用本地工具、又能通过 MCP 协议调用远程服务的 ReAct Agent。

已开源在ai-demo项目，欢迎star~

一、项目全景

目录结构

langgraph_deepagents/              # 项目根目录
├── common.py                      # 公共依赖：模型工厂、消息打印、流式追踪
├── .env                           # 环境变量（DeepSeek API Key 等）
│
├── mcp/                           # MCP 模块
│   ├── mcp_agent.py               # 主入口：基于 DeepAgents 的 MCP Agent
│   ├── tools.py                   # 工具注册中心：本地工具 + MCP Client 配置
│   │
│   └── mcp_servers/               # MCP 服务端
│       ├── __init__.py
│       └── math_server.py         # 数学计算 MCP Server（FastMCP 实现）
│
├── skills/                        # Skills 模块（其他项目功能）
├── data/                          # 数据目录
├── research_agent.py              # 研究 Agent 示例
└── subagents.py                   # 子 Agent 示例

核心依赖

包名	版本	用途
`deepagents`	0.5.3	基于 LangGraph 的深度智能体框架
`langchain`	1.2.15	LangChain 核心
`langchain-core`	1.3.1	Tool、Message 等核心抽象
`mcp`	1.27.0	MCP 协议 Python SDK
`langchain_mcp_adapters`	—	LangChain ↔ MCP 桥接层
`fastmcp`	—	零配置构建 MCP Server 的高阶框架
`langchain-deepseek`	—	DeepSeek 模型 LangChain 集成

二、MCP 协议：为什么需要它？

2.1 传统工具调用的痛点

在没有 MCP 之前，我们通常这样组织工具调用：

LLM → Function Calling → 与 Agent 同进程的硬编码函数

这种方式的局限：

痛点	表现
强耦合	每增加一个工具，都要修改 Agent 代码，甚至重新部署
语言绑定	工具实现语言必须与 Agent 一致（Python Agent 只能调 Python 工具）
无标准化	每个框架的 Tool 定义、Schema、调用方式各不相同
难以复用	团队 A 写了个"数据查询工具"，团队 B 无法直接对接使用

2.2 MCP 的核心思想

MCP (Model Context Protocol) 由 Anthropic 提出，旨在标准化 LLM 与外部系统之间的交互方式。它的核心概念是 客户端 - 服务端架构：

┌─────────────────────────────┐
│   DeepAgents Agent          │
│   （MCP 客户端）            │
└──────────┬──────────────────┘
           │
   MultiServerMCPClient
           │
     ┌─────┴──────┐
     ▼            ▼
  MCP Server A  MCP Server B
  （数学计算）  （数据库查询）
   stdio/HTTP     stdio/HTTP

关键特性：

MCP Server：暴露一组工具，提供 Tool Schema 和实际执行逻辑
MCP Client：连接 Server，获取工具列表，转发调用请求
传输层：支持 stdio（本地进程间通信）和 HTTP + SSE（远程服务）
协议标准化：任何语言实现的 MCP Server 都能被任何 MCP Client 消费

💡 你可以把 MCP 理解为"AI 世界的 USB-C 接口"——一个通用的连接标准，让工具"即插即用"。

三、代码逐层解析

3.1 第一层：MCP Server（数学计算服务）

文件：mcp/mcp_servers/math_server.py

from typing import Any, Literal
from fastmcp.server import FastMCP

server = FastMCP[Any](name="math-mcp-server")

@server.tool
def calculate(
    a: float,
    b: float,
    operation: Literal["add", "subtract", "multiply", "divide"]
) -> float:
    """
    Performs a calculation on two numbers.
    """
    if operation == "add":
        return a + b
    elif operation == "subtract":
        return a - b
    elif operation == "multiply":
        return a * b
    elif operation == "divide":
        return a / b

if __name__ == "__main__":
    # 方式一：stdio 传输（本地开发）
    server.run(transport="stdio")

    # 方式二：HTTP + SSE 传输（远程部署）
    # server.run(transport="http", port=8000)

关键设计点：

特性	说明
FastMCP	基于 `mcp` SDK 的高阶封装，函数装饰器即可注册 Tool
类型推导自生成 Schema	自动将 Python 类型注解 (`float`, `Literal[...]`) 转为 MCP Tool JSON Schema
dual transport	同一份代码，通过参数切换 `stdio` 或 `http` 传输模式

3.2 深入理解 stdio 与 HTTP 两种传输模式

MCP 协议的传输层是整个架构的基石。math_server.py 只用一行代码切换传输方式：

# stdio：适合本地开发和单机部署
server.run(transport="stdio")

# HTTP + SSE：适合远程服务、多机协作、微服务化
server.run(transport="http", port=8000)

下面逐一深入剖析这两种模式。

3.3 stdio 传输模式：本地进程间通信

工作原理

┌──────────────────┐        stdin/stdout        ┌──────────────────────┐
│  DeepAgents      │  ──── 发送Tool调用请求 ───▶ │  MCP Server (子进程) │
│  Agent           │◀──── 返回Tool执行结果 ──────│  math_server.py     │
│  (父进程)        │                             │                      │
└──────────────────┘                             └──────────────────────┘

父进程：Agent 主进程，通过 MultiServerMCPClient 发起请求
子进程：math_server.py，由 Client 自动启动
通信协议：请求和响应通过标准输入/输出传递，格式为 MCP 协议定义的 JSON-RPC 消息

stdio 的生命周期

Agent 启动
  │
  ▼
首次调用工具 → MultiServerMCPClient 自动 fork 子进程
  │
  ├── 写入 stdin：{"jsonrpc":"2.0","method":"tools/call","params":{...}}
  ├── 读取 stdout：{"jsonrpc":"2.0","result":{"content":[...]}}
  │
  ▼
Agent 运行结束 → Client 自动关闭子进程

优缺点

优势	劣势
零配置，无需管理端口和网络	子进程与主进程耦合，无法独立部署
延迟最低（进程间管道通信）	只能单机运行
无端口冲突、防火墙等问题	子进程故障会导致 Agent 异常
适合开发调试	无法给其他 Agent 共享工具

3.4 HTTP 传输模式：远程服务化

这是 MCP 真正发挥去中心化威力的传输方式。

工作原理：SSE（Server-Sent Events）

MCP 的 HTTP 传输建立在 SSE (Server-Sent Events) 协议之上。SSE 是一种服务器向客户端单向推送事件的技术，与 WebSocket 不同：

对比维度	SSE (Server-Sent Events)	WebSocket
方向	服务器 → 客户端（单向推送）	全双工
协议	基于 HTTP	独立协议 (ws://)
连接方式	复用 HTTP 长连接	需要升级握手
自动重连	原生支持	需手动实现
适用场景	服务端主动推送	实时双向通信

为什么 MCP 选 SSE 而不是 WebSocket？因为 MCP 的核心模式是 客户端发送请求 → 服务端处理并推送结果，这个模式天然适配 SSE——客户端通过 HTTP POST 发起请求，服务端通过 SSE 推送响应事件。

MCP over HTTP 的架构

┌──────────────────────┐
│  DeepAgents Agent    │
│  (MCP Client)        │
└───────┬──────────────┘
        │
        │  ① POST /mcp（调用工具）
        │  ② SSE 事件流（返回结果）
        ▼
┌──────────────────────┐
│  MCP HTTP Server     │  ← 监听 http://host:8000
│  math_server.py      │
│   - /mcp 端点        │
│   - SSE 事件流       │
└──────────────────────┘
        │
        │  可以跨机器、跨网络
        ▼
┌──────────────────────┐
│  其他 Agent / 服务   │  ← 多个客户端可同时连接
└──────────────────────┘

启动 HTTP MCP Server

# 方式一：直接修改 math_server.py，取消注释 http 模式
python langgraph_deepagents/mcp/mcp_servers/math_server.py
# 此时服务器将在 http://localhost:8000 上监听 MCP 请求

启动后，你可以用 curl 验证服务是否正常：

# MCP 服务没有传统的 REST 端点，但可以通过 MCP 客户端验证
curl http://localhost:8000/health  # 如果有健康检查端点

客户端连接 HTTP MCP Server

在 tools.py 中，将 stdio 配置切换为 http：

client = MultiServerMCPClient({
    "math": {
        "transport": "http",
        "url": "http://localhost:8000/mcp"
    }
})

跨机器调用的网络架构

当 MCP Server 部署在远程服务器上时：

┌── 开发机 ──────────────────┐     ┌── 服务器 192.168.1.100:8000 ──┐
│                             │     │                               │
│  DeepAgents Agent          │     │  MCP Math Server              │
│  tools.py:                  │     │  math_server.py               │
│    "url": "http://192.168.1.100:8000/mcp"  │
│                             │     │                               │
└─────────────────────────────┘     └───────────────────────────────┘

优缺点

优势	劣势
独立部署，与 Agent 解耦	需要管理端口、网络、防火墙
多 Agent 可共享同一个工具服务	网络延迟高于 stdio（但通常 < 5ms 局域网）
支持跨机器、跨数据中心调用	需要额外的运维保障（进程守护、日志等）
服务可用其他语言实现（TS/Go/Java）	部署复杂度中等
易于集成到现有微服务体系	端口可能冲突，需规划

3.5 第二层：工具注册中心 + MCP Client

文件：mcp/tools.py

import asyncio
from typing import Literal

from langchain.tools import tool
from langchain_mcp_adapters.client import MultiServerMCPClient

"""
    tools 工具有两种形态：
    - 本地工具（Local Tool）：同进程直接调用
    - 远程工具（MCP Tool）：通过 MCP Server 调用
"""

# ═══════════════════════════════════
# 1. 本地工具（同进程直接调用）
# ═══════════════════════════════════

@tool
def calculate(
    a: float,
    b: float,
    operation: Literal["add", "subtract", "multiply", "divide"]
) -> float:
    """Performs a calculation on two numbers."""
    # ... 实现同上 ...


# ═══════════════════════════════════
# 2. MCP 客户端（连接远程 MCP Server）
# ═══════════════════════════════════

client = MultiServerMCPClient({
    # 方式一：stdio 传输
    "math": {
        "transport": "stdio",
        "command": "python",
        "args": ["langgraph_deepagents/mcp/mcp_servers/math_server.py"]
    },
    # 方式二：HTTP 传输
    # "math": {
    #     "transport": "http",
    #     "url": "http://localhost:8000/mcp"
    # }
})

def get_tools():
    """
    注意：MCP Client 的 get_tools() 是异步方法，
    需通过 asyncio.run() 或 await 调用。
    """
    # 当前使用本地工具（同步调用）
    return [calculate]

    # 若切换为 MCP 远程工具（异步调用）：
    # return asyncio.run(client.get_tools())

关于 `MultiServerMCPClient`

这个客户端类的核心能力：

多 Server 管理：通过字典 key 区分不同的 MCP Server，可同时连接多个
进程生命周期：stdio 模式下，自动管理子进程的启动和关闭
工具发现：自动从各个 MCP Server 获取 Tool 列表，统一返回
LangChain 兼容：返回的 Tool 对象直接兼容 LangChain 的 Tool 接口

3.6 HTTP 的高级用法：多 Server 集群

HTTP 传输模式下，MCP 的真正价值在于 一个 Agent 对接多个独立部署的 MCP Server。

同时连接多个 MCP Server

client = MultiServerMCPClient({
    "math": {
        "transport": "http",
        "url": "http://192.168.1.100:8000/mcp"
    },
    "database": {
        "transport": "http",
        "url": "http://192.168.1.101:8000/mcp"
    },
    "file_storage": {
        "transport": "http",
        "url": "http://192.168.1.102:8000/mcp"
    }
})

LLM 会根据任务自动选择合适的工具——数学问题找 math Server，数据查询找 database Server。

混合模式：stdio + HTTP 混用

你甚至可以同时使用两种传输方式：

client = MultiServerMCPClient({
    # stdio：本地快速工具
    "local_math": {
        "transport": "stdio",
        "command": "python",
        "args": ["langgraph_deepagents/mcp/mcp_servers/math_server.py"]
    },
    # HTTP：远程业务工具
    "remote_order": {
        "transport": "http",
        "url": "http://api.company.com/mcp/order"
    },
    "remote_payment": {
        "transport": "http",
        "url": "http://api.company.com/mcp/payment"
    }
})

3.7 第三层：DeepAgents Agent 主入口

文件：mcp/mcp_agent.py

import asyncio
import sys, os
from deepagents import create_deep_agent
from langchain_core.messages import HumanMessage
from tools import get_tools
from common import get_deepseek_model, print_messages


def create_mcp_agent():
    # 1. 获取工具列表
    mcp_tools = get_tools()

    # 2. 初始化 DeepSeek 模型
    model = get_deepseek_model()

    # 3. 创建 ReAct Agent
    agent = create_deep_agent(
        model=model,
        tools=mcp_tools,
        system_prompt="You are a helpful assistant, please response in Chinese."
    )
    return agent


async def run_agent(agent):
    response = await agent.ainvoke({
        "messages": [
            {
                "role": "user",
                "content": "请帮我计算 15 和 28 的和，以及它们的乘积"
            }
        ],
    })
    print_messages(response["messages"])


if __name__ == "__main__":
    # 在事件循环外创建 agent，避免嵌套 asyncio.run()
    agent = create_mcp_agent()
    asyncio.run(run_agent(agent))

DeepAgents 的 `create_deep_agent` 做了什么？

根据 DeepAgents v0.5.3 的源码，create_deep_agent 本质上是一个高阶工厂函数，内部自动构建了一个 LangGraph ReAct 图：

输入 (messages)
    │
    ▼
┌──────────────────────────┐
│  LLM (DeepSeek)          │
│  判断：是否需要调用工具？ │
└──────────┬───────────────┘
           │
   ┌───────┴───────┐
   ▼               ▼
 需要工具         不需要工具
   │               │
   ▼               ▼
 调用工具 → 获取结果 → 继续推理
   │
   ▼
  回到 LLM 继续判断

它仅暴露三个核心参数：

参数	说明
`model`	底层 LLM（支持 DeepSeek、OpenAI、Anthropic 等）
`tools`	工具列表（本地工具 + MCP 工具均可）
`system_prompt`	系统提示词，控制 Agent 行为风格

3.8 公共层：模型工厂与调试工具

文件：common.py（核心片段）

from langchain_deepseek import ChatDeepSeek

def get_deepseek_model():
    load_dotenv()
    deepseek_model = ChatDeepSeek(
        model="deepseek-chat",        # DeepSeek-V3
        api_key=os.environ["DEEPSEEK_API_KEY"],
        temperature=0.1,
    )
    return deepseek_model

配套的调试打印函数：

print_messages()：按角色（User / AI / Tool）格式化打印消息历史
run_and_print_stream()：流式追踪 Agent 的每一步执行，包括子任务启动、工具调用、LLM 推理过程

四、完整调用链路

用 “计算 15 和 28 的和，以及它们的乘积” 这个问题来走通整条链路：

Step 1：用户输入

用户 → Agent
"请帮我计算 15 和 28 的和，以及它们的乘积"

Step 2：Agent 推理（LLM 收到后自主决策）

LLM 拿到问题 + 系统提示 + 工具描述（MCP Server 返回的 Tool Schema），判断：

“这是一个数学计算问题，我需要先计算和，再计算乘积。用户提了两个要求，我可以调用两次 calculate 工具。”

Step 3：执行工具调用

Agent → MultiServerMCPClient → stdio → math_server.py (子进程)

MCP Server 依次执行两种运算：

calculate(a=15, b=28, operation="add")       → 43
calculate(a=15, b=28, operation="multiply")  → 420

Step 4：结果返回与整合

math_server.py → stdout → MultiServerMCPClient → Agent

Agent 综合两次结果，生成最终回复：

“15 和 28 的和是 43，它们的乘积是 420。”

五、Local Tool vs MCP Tool：选型决策

对比维度	本地工具 (Local Tool)	MCP 远程工具
注册方式	`@tool` 装饰器	`@server.tool` 装饰器
调用方式	同进程直接函数调用	跨进程 RPC（stdio/HTTP）
部署模型	内嵌于 Agent 进程	独立进程 / 独立服务
语言约束	必须与 Agent 同语言	无限制（Python/TS/Go/Java…）
延迟	零（同进程）	进程间通信开销（1~10ms）
典型场景	纯计算、字符串处理、简单逻辑	数据库查询、第三方 API、文件 I/O
调试难度	低	中（需管理进程生命周期）
可复用性	低（绑定 Agent 代码）	高（独立部署，多 Agent 共享）

建议：

简单内部逻辑 → 本地工具
涉及外部系统 / 需要独立部署 → MCP Tool
工具给多个 Agent 共用 → MCP Tool
工具用其他语言实现 → MCP Tool（必须）

六、实战踩坑与最佳实践

6.1 异步陷阱

# ❌ 错误：get_tools() 是 coroutine，需要 await
tools = client.get_tools()

# ✅ 正确方式一（同步上下文）
tools = asyncio.run(client.get_tools())

# ✅ 正确方式二（异步上下文）
tools = await client.get_tools()

6.2 stdio 传输的路径

args 中的脚本路径是相对于工作目录 (cwd) 的，而非相对于 Python 文件目录：

client = MultiServerMCPClient({
    "math": {
        "transport": "stdio",
        "command": "python",
        # 以下路径相对的是启动 Python 进程的 cwd
        "args": ["langgraph_deepagents/mcp/mcp_servers/math_server.py"]
    }
})

6.3 生命周期管理

在 stdio 传输模式下：

首次调用工具时，MCP Client 自动启动子进程
Agent 执行完毕后，子进程由 MCP Client 自动清理
多次请求可复用同一 Agent → 避免重复 fork 子进程

6.4 避免嵌套 `asyncio.run()`

# ✅ 正确
agent = create_mcp_agent()           # 在事件循环外创建
asyncio.run(run_agent(agent))        # 再启动事件循环

# ❌ 错误（如果 create 内部也调了 asyncio.run）
asyncio.run(create_mcp_agent())      # 可能导致嵌套 RuntimeError

6.5 工具函数名冲突

注意 tools.py（本地 calculate）和 math_server.py（MCP Server 的 calculate）有同名函数。这是允许的，因为 @tool 装饰器会生成唯一的 Tool Schema name，而 MCP Server 的 calculate 运行在独立子进程中，两者不会冲突。