Skip to content

LangGraph 与 MCP 集成

一句话理解 MCP 和 Skills

在深入技术细节之前,先用最简单的方式理解这两个概念:

概念一句话解释生活类比
MCP让 AI 能够调用外部工具的标准接口像手机的 USB 接口,插上就能用各种配件
Skills教会 AI 如何完成特定任务的说明书像菜谱,告诉厨师怎么做一道菜

真实例子:让 AI 帮你做 Excel 报表

假设你要让 AI 帮你做一份销售报表:

用 MCP 的方式:

你:帮我做个销售报表
AI:好的,让我调用 Excel MCP...
   [调用 create_sheet 工具]
   [调用 write_cell 工具 x 100 次]
   [调用 add_formula 工具 x 20 次]
   ...
结果:消耗大量 Token,过程繁琐,AI 可能中途出错

用 Skills 的方式:

你:帮我做个销售报表
AI:检测到报表任务,加载 Excel Skill...
   [读取预设的报表制作流程]
   [执行优化过的 Python 脚本]
结果:快速完成,格式规范,成本低

核心区别:

  • MCP = 给 AI 一堆工具,让它自己摸索怎么用
  • Skills = 给 AI 一套经过验证的流程,按步骤执行

什么时候用哪个?

场景推荐方案原因
生成 Excel/PPT/WordSkills高频任务,有成熟流程
查询公司内部数据库MCP需要连接特定数据源
发送邮件通知MCP需要调用外部服务
按公司模板写周报Skills有固定格式和流程
查天气/查股票MCP需要实时外部数据

简单记忆:MCP 解决"能不能做",Skills 解决"怎么做好"。两者可以配合使用。


什么是 MCP?

MCP(Model Context Protocol,模型上下文协议)是 Anthropic 公司在 2024 年底推出的一项开放协议。它的核心目标是:让应用程序能够以标准化的方式向大语言模型提供工具调用能力

简单来说,MCP 就像是 AI 世界的"USB 接口"——只要遵循这个协议,任何工具都可以被任何支持 MCP 的 AI 应用调用。

MCP 解决了什么问题?

在 MCP 出现之前,如果你想让 LLM 调用外部工具(比如查天气、搜索网页、操作数据库),你需要:

  1. 为每个工具编写适配代码
  2. 处理不同工具的认证方式
  3. 管理工具的输入输出格式

MCP 将这些繁琐的工作标准化了。现在,只要工具提供方按照 MCP 协议实现服务,客户端就可以"即插即用"。


MCP 的工作原理

核心概念:Function Call 的标准化封装

MCP 本质上是对大模型 Function Call 机制的协议层封装。让我们先回顾一下 Function Call 是如何工作的:

python
from langchain_openai import ChatOpenAI
from langchain.tools import tool

# 初始化模型
llm = ChatOpenAI(model="gpt-4o-mini")

# 定义工具函数
@tool(description="查询城市天气信息")
def query_weather(city_name: str):
    """查询指定城市的天气
    
    Args:
        city_name: 城市名称
    """
    return f"{city_name}今天晴朗,气温 25 度,适合户外活动"

# 将工具绑定到模型
llm_with_tools = llm.bind_tools([query_weather])

# 准备对话
user_question = "北京今天天气怎么样?"
messages = [{"role": "user", "content": user_question}]

# 第一次调用:模型判断需要调用工具
response = llm_with_tools.invoke(messages)
print(response.tool_calls)

# 执行工具调用
if response.tool_calls:
    tool_result = query_weather.invoke(response.tool_calls[0])
    messages.append(response)
    messages.append(tool_result)
    
    # 第二次调用:模型整合结果生成最终回答
    final_response = llm_with_tools.invoke(messages)
    print(final_response.content)

这个过程中有几个关键点:

  1. 工具声明:告诉模型有哪些工具可用
  2. 模型决策:模型自己判断是否需要调用工具
  3. 工具执行:客户端负责实际执行工具
  4. 结果整合:模型根据工具返回结果生成最终答案

MCP 就是把这套流程标准化,让工具的"声明"和"执行"可以通过统一的协议进行。

MCP 架构图

                    ┌─────────────────┐
                    │   大语言模型    │
                    │     (LLM)       │
                    └────────┬────────┘

         1. 注册工具         │         4. 返回调用请求
         ┌───────────────────┼───────────────────┐
         │                   │                   │
         ▼                   ▼                   ▼
    ┌─────────┐        ┌─────────┐        ┌─────────┐
    │   MCP   │        │  Client │        │   MCP   │
    │ Server  │◄──────►│   App   │◄──────►│ Server  │
    │ (天气)  │        │ (Cline) │        │ (地图)  │
    └─────────┘        └─────────┘        └─────────┘
         │                   │                   │
         │    5. 执行工具    │    6. 返回结果    │
         └───────────────────┴───────────────────┘

MCP 的三种传输模式

MCP 协议支持三种传输方式:STDIOSSEStreamable HTTP。理解它们的区别对于正确使用 MCP 非常重要。

1. STDIO 模式(Standard Input/Output)

STDIO 模式通过标准输入输出进行通信,是最简单的本地通信方式:

  • 本地执行:MCP Server 以子进程形式在客户端本地运行
  • 进程通信:通过 stdin/stdout 传递消息
  • 适用场景:本地工具、需要访问本地资源的场景

配置示例:

json
{
  "mcpServers": {
    "local-tools": {
      "command": "python",
      "args": ["./my_mcp_server.py"],
      "transport": "stdio"
    }
  }
}

2. SSE 模式(Server-Sent Events)

SSE 是一种基于 HTTP 的长连接协议,专为实时流式传输优化:

  • 服务端部署:MCP Server 运行在远程服务器上
  • 单向流式:服务端可以持续推送数据给客户端
  • 适用场景:需要实时更新的场景,如进度通知

配置示例:

json
{
  "mcpServers": {
    "weather-service": {
      "url": "https://api.example.com/mcp/sse",
      "transport": "sse"
    }
  }
}

3. Streamable HTTP 模式

这是较新的传输模式,处理标准的 HTTP 请求:

  • 远程部署:MCP Server 作为 HTTP 服务运行
  • 多客户端支持:可以同时服务多个客户端
  • 适用场景:生产环境、公共 API 服务

配置示例:

json
{
  "mcpServers": {
    "api-service": {
      "url": "http://localhost:8000/mcp",
      "transport": "streamable_http"
    }
  }
}

三种模式对比

特性STDIOSSEStreamable HTTP
运行位置本地机器远程服务器远程服务器
网络要求无需网络需要网络需要网络
多客户端不支持支持支持
实时推送不支持支持按需
适用场景本地开发流式更新生产部署
复杂度最简单中等中等

安全提醒:STDIO 模式下,MCP Server 在你的本地机器执行,理论上可以访问本地文件、执行系统命令。使用第三方 MCP 服务时,请务必确认其来源可信。


LangGraph 集成 MCP

LangGraph 通过 langchain-mcp-adapters 模块提供了对 MCP 的原生支持。

安装依赖

bash
pip install langchain-mcp-adapters langgraph

基础用法:连接 MCP Server

python
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

# 初始化模型
model = ChatOpenAI(model="gpt-4o-mini")

# 配置 MCP 客户端(SSE 模式)
mcp_client = MultiServerMCPClient({
    "map-service": {
        "url": "https://your-mcp-server.com/sse",
        "transport": "sse"
    }
})

async def main():
    # 从 MCP Server 获取工具列表
    tools = await mcp_client.get_tools()
    print(f"已加载 {len(tools)} 个工具")
    
    # 创建 ReAct Agent
    agent = create_react_agent(model=model, tools=tools)
    
    # 执行查询
    result = await agent.ainvoke({
        "messages": [{"role": "user", "content": "帮我规划从上海到杭州的驾车路线"}]
    })
    
    print(result["messages"][-1].content)

# 运行
import asyncio
asyncio.run(main())

多服务器配置

LangGraph 支持同时连接多个 MCP Server:

python
mcp_client = MultiServerMCPClient({
    # SSE 模式的远程服务
    "weather-api": {
        "url": "https://weather.example.com/mcp/sse",
        "transport": "sse"
    },
    # STDIO 模式的本地服务
    "file-manager": {
        "command": "python",
        "args": ["./file_mcp_server.py"],
        "transport": "stdio"
    },
    # Streamable HTTP 模式
    "search-engine": {
        "url": "http://localhost:8000/mcp",
        "transport": "streamable_http"
    }
})

# 获取所有服务器的工具
all_tools = await mcp_client.get_tools()

有状态的工具会话

默认情况下,MultiServerMCPClient无状态的——每次工具调用都会创建新的会话。这对于简单的无状态工具很方便,但有些工具需要跨调用保持状态(比如数据库连接、文件句柄等)。

使用 ClientSession 可以实现持久化会话:

python
from langchain_mcp_adapters.tools import load_mcp_tools

# 创建持久化会话
async with mcp_client.session("file-manager") as session:
    # 在同一个会话中加载工具
    tools = await load_mcp_tools(session)

    # 这些工具调用会共享同一个会话状态
    # 比如:打开文件 → 读取内容 → 关闭文件
    # 文件句柄会在整个会话期间保持有效

使用场景

场景是否需要状态会话
查询天气否,每次独立查询
数据库事务是,需要保持连接
文件编辑是,需要保持文件句柄
简单计算否,无状态即可

在自定义 Graph 中使用 MCP 工具

python
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI

class AgentState(TypedDict):
    messages: list

# 初始化
model = ChatOpenAI(model="gpt-4o-mini")

async def build_graph_with_mcp():
    # 获取 MCP 工具
    mcp_client = MultiServerMCPClient({
        "my-service": {"url": "https://example.com/mcp/sse", "transport": "sse"}
    })
    mcp_tools = await mcp_client.get_tools()
    
    # 绑定工具到模型
    model_with_tools = model.bind_tools(mcp_tools)
    
    # 定义节点
    def call_model(state: AgentState):
        response = model_with_tools.invoke(state["messages"])
        return {"messages": state["messages"] + [response]}
    
    def should_continue(state: AgentState):
        last_message = state["messages"][-1]
        if hasattr(last_message, "tool_calls") and last_message.tool_calls:
            return "tools"
        return "end"
    
    # 构建图
    builder = StateGraph(AgentState)
    builder.add_node("agent", call_model)
    builder.add_node("tools", ToolNode(mcp_tools))
    
    builder.add_edge(START, "agent")
    builder.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END})
    builder.add_edge("tools", "agent")
    
    return builder.compile()

实战:创建自定义 MCP Server

了解如何创建 MCP Server 有助于更深入理解协议原理。

使用 Python MCP SDK

bash
pip install mcp

示例:天气查询服务

python
from mcp.server.fastmcp import FastMCP

# 创建 MCP 服务实例
app = FastMCP("weather-demo")

@app.tool()
def get_temperature(location: str) -> str:
    """获取指定地点的温度信息
    
    Args:
        location: 地点名称,如"北京"、"上海"
    """
    mock_data = {
        "北京": "18",
        "上海": "22",
        "广州": "28",
        "深圳": "27"
    }
    temp = mock_data.get(location, "20")
    return f"{location}当前温度:{temp} 摄氏度"

@app.tool()
def get_forecast(location: str, days: int = 3) -> str:
    """获取天气预报
    
    Args:
        location: 地点名称
        days: 预报天数,默认3天
    """
    return f"{location}未来{days}天:晴转多云,气温15-25度"

@app.resource("config://settings")
def get_settings() -> str:
    """返回服务配置信息"""
    return "Weather Service v1.0 - Demo Mode"

if __name__ == "__main__":
    # SSE 模式启动
    app.run(transport="sse", port=8080)
    
    # 或者 STDIO 模式
    # app.run(transport="stdio")

启动并测试

bash
# 启动服务
python weather_server.py

# 服务将在 http://localhost:8080/sse 可用

在 MCP 客户端配置中添加:

json
{
  "mcpServers": {
    "my-weather": {
      "url": "http://localhost:8080/sse",
      "transport": "sse"
    }
  }
}

MCP 生态资源

常用 MCP 服务聚合平台

平台地址说明
MCP Hubhttps://mcp.so聚合数千个 MCP 服务
Anthropic 官方https://modelcontextprotocol.io官方文档和 SDK

热门 MCP 服务类型

  • 地图导航:路线规划、地点搜索
  • 文件操作:读写本地文件、目录管理
  • 数据库:SQL 查询、数据管理
  • 网页浏览:网页抓取、内容提取
  • 代码执行:运行代码、调试程序
  • 邮件服务:发送邮件、管理收件箱

冷静看待 MCP:局限性与未来

MCP 在 2024 年底发布时确实引发了巨大关注,但经过一年的实践检验,我们需要更理性地认识它的能力边界。

核心局限:上下文窗口的代价

MCP 最大的问题在于工具定义会占用模型的上下文窗口

每个 MCP Server 的工具描述、参数定义、调用记录都需要放入 Context,模型才能"看到"并决策。这带来几个实际问题:

假设场景:
- 一个 GitHub MCP Server 定义了 90+ 个工具
- 光工具描述就消耗 50000+ Token
- 还没开始对话,上下文已经用掉一大半

工具越多,模型越"笨"——这是一个真实存在的悖论。当上下文被大量工具定义填满时:

  1. 模型的注意力被稀释,推理能力下降
  2. 可能出现"幻觉":明明工具返回了正确结果,模型却用自己的"常识"覆盖
  3. Token 成本急剧上升,几轮对话就可能消耗大量费用

生态质量参差不齐

MCP 的低门槛是双刃剑。几十行代码就能发布一个 Server,导致:

  • 大量重复轮子:查天气的 MCP 可能有几十个,但真正维护良好的寥寥无几
  • 质量难以保证:有研究指出,相当数量的 MCP Server 存在凭证暴露、缺乏维护等问题
  • 筛选成本高:开发者反而要花更多时间评估哪个 MCP 靠谱

行业演进:Skills 的崛起

2025 年 10 月,Anthropic 推出了 Agent Skills 系统,这被认为是对 MCP 局限性的一次重要补充。

Skills 是什么?

Skills 是一套动态加载的指令和脚本集合,用于教会 AI 如何完成特定任务。它的核心设计理念是"渐进式披露"(Progressive Disclosure):

传统 MCP 方式:
┌─────────────────────────────────────┐
│  启动时加载所有工具定义到上下文      │
│  (消耗大量 Token,无论是否使用)    │
└─────────────────────────────────────┘

Skills 方式:
┌─────────────────────────────────────┐
│  启动时只加载 Skill 名称和描述       │  ← 第一层:轻量元数据
├─────────────────────────────────────┤
│  检测到相关任务时,加载完整指令       │  ← 第二层:按需加载
├─────────────────────────────────────┤
│  需要时才加载附加资源和脚本           │  ← 第三层:深度资源
└─────────────────────────────────────┘

Anthropic 官方 Skills

目前 Anthropic 提供了几个预置 Skills:

Skill功能特点
Excel Skill生成专业电子表格支持公式、图表、格式化
PowerPoint Skill制作演示文稿自动排版、主题应用
Word Skill创建文档格式规范、样式一致
PDF Skill处理 PDF 表单自动提取和填充字段

自定义 Skills

企业可以创建自己的 Skills,例如:

yaml
# SKILL.md 示例结构
---
name: "公司周报生成"
description: "按照公司标准模板生成周报"
triggers:
  - "写周报"
  - "生成周报"
  - "本周总结"
---

## 执行步骤

1. 询问用户本周完成的主要工作
2. 按照以下模板组织内容:
   - 本周完成
   - 下周计划
   - 需要协调的问题
3. 应用公司标准格式
4. 生成 Word 文档

Skills vs MCP:技术对比

维度MCPSkills
加载时机启动时全部加载按需动态加载
上下文消耗高(所有工具定义)低(渐进式披露)
适用场景连接外部服务和数据执行标准化流程
可复用性通用协议,跨平台Claude 生态内
开发门槛较低需要理解 Skill 架构

协同使用

Skills 和 MCP 并非互斥,而是互补:

场景:生成包含实时数据的销售报表

Skills 负责:
├── 报表的格式和结构
├── 图表的类型选择
└── 输出的样式规范

MCP 负责:
├── 连接数据库获取销售数据
├── 调用 API 获取汇率信息
└── 访问 CRM 系统获取客户信息

行业影响

OpenAI、Google、AWS 等厂商陆续宣布支持 MCP,说明它作为一个开放标准正在被行业接受。但 Skills 的出现也表明:单纯依赖工具协议是不够的,AI 还需要"知识"来指导如何正确使用这些工具。

参考资料

实用建议

基于以上分析,给出几点实用建议:

场景建议
核心高频功能优先使用原生集成或 Skills
专业领域工具精选 2-3 个高质量 MCP
开发调试MCP 很方便,但注意控制数量
生产环境严格评估安全性和稳定性

记住:MCP 是手段,不是目的。真正重要的是让 AI 完成任务,而不是堆叠更多工具。


最佳实践

1. 安全优先

python
# 不推荐:直接信任所有 MCP 工具
tools = await mcp_client.get_tools()
agent = create_react_agent(model, tools)

# 推荐:审核并过滤工具
tools = await mcp_client.get_tools()
safe_tools = [t for t in tools if t.name in ALLOWED_TOOL_NAMES]
agent = create_react_agent(model, safe_tools)

2. 错误处理

python
from langchain_mcp_adapters.client import MultiServerMCPClient

async def safe_get_tools(config: dict):
    try:
        client = MultiServerMCPClient(config)
        tools = await client.get_tools()
        return tools
    except ConnectionError:
        print("MCP 服务连接失败,使用备用工具")
        return get_fallback_tools()
    except TimeoutError:
        print("MCP 服务响应超时")
        return []

3. 工具描述优化

MCP 工具的效果很大程度上取决于描述的质量:

python
# 较差的描述
@app.tool()
def search(q: str):
    """搜索"""
    pass

# 优秀的描述
@app.tool()
def search_products(
    query: str,
    category: str = None,
    max_price: float = None
) -> list:
    """在商品库中搜索产品
    
    根据关键词搜索商品,支持按类别和价格过滤。
    返回匹配的商品列表,包含名称、价格、库存等信息。
    
    Args:
        query: 搜索关键词,如"蓝牙耳机"、"机械键盘"
        category: 商品类别,可选值:电子产品、服装、食品
        max_price: 最高价格限制,单位为元
    
    Returns:
        商品列表,每个商品包含 id, name, price, stock 字段
    """
    pass

本章小结

概念说明
MCP 协议标准化的工具调用协议,让 AI 应用可以即插即用各种外部服务
STDIO 模式基于进程通信,适合本地工具,最简单
SSE 模式基于 HTTP 长连接,适合需要实时流式更新的场景
Streamable HTTP 模式基于标准 HTTP,适合生产环境多客户端部署
langchain-mcp-adaptersLangChain/LangGraph 的 MCP 集成库
MultiServerMCPClient管理多个 MCP Server 连接的客户端类
ClientSession有状态工具会话,用于需要跨调用保持状态的场景
FastMCPPython MCP SDK,用于快速创建 MCP Server

关键要点

  1. MCP 是协议,不是魔法:它只是标准化了工具调用的方式,底层仍然是 Function Call
  2. 注意安全风险:特别是 STDIO 模式,第三方服务可能访问你的本地资源
  3. 工具描述很重要:好的描述能让模型更准确地调用工具
  4. LangGraph 原生支持:通过 langchain-mcp-adapters 可以轻松集成
  5. 少即是多:MCP 工具数量要精简,过多反而会降低模型性能
  6. 理性看待生态:选择经过验证的高质量 MCP,而非盲目追求数量

思考题

  1. MCP 和传统的 REST API 有什么区别?它们各自适合什么场景?
  2. 如果一个 MCP Server 返回了错误的数据,模型会如何处理?
  3. 如何设计一个安全的 MCP 权限控制系统?
  4. 为什么"工具越多,模型越笨"?从技术角度分析这个现象。
  5. Skills 和 MCP 各自适合什么场景?如何在项目中合理选择?
  6. 什么情况下需要使用有状态会话(ClientSession)?举例说明。

参考资源

基于 MIT 许可证发布。内容版权归作者所有。