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/Word | Skills | 高频任务,有成熟流程 |
| 查询公司内部数据库 | MCP | 需要连接特定数据源 |
| 发送邮件通知 | MCP | 需要调用外部服务 |
| 按公司模板写周报 | Skills | 有固定格式和流程 |
| 查天气/查股票 | MCP | 需要实时外部数据 |
简单记忆:MCP 解决"能不能做",Skills 解决"怎么做好"。两者可以配合使用。
什么是 MCP?
MCP(Model Context Protocol,模型上下文协议)是 Anthropic 公司在 2024 年底推出的一项开放协议。它的核心目标是:让应用程序能够以标准化的方式向大语言模型提供工具调用能力。
简单来说,MCP 就像是 AI 世界的"USB 接口"——只要遵循这个协议,任何工具都可以被任何支持 MCP 的 AI 应用调用。
MCP 解决了什么问题?
在 MCP 出现之前,如果你想让 LLM 调用外部工具(比如查天气、搜索网页、操作数据库),你需要:
- 为每个工具编写适配代码
- 处理不同工具的认证方式
- 管理工具的输入输出格式
MCP 将这些繁琐的工作标准化了。现在,只要工具提供方按照 MCP 协议实现服务,客户端就可以"即插即用"。
MCP 的工作原理
核心概念:Function Call 的标准化封装
MCP 本质上是对大模型 Function Call 机制的协议层封装。让我们先回顾一下 Function Call 是如何工作的:
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)这个过程中有几个关键点:
- 工具声明:告诉模型有哪些工具可用
- 模型决策:模型自己判断是否需要调用工具
- 工具执行:客户端负责实际执行工具
- 结果整合:模型根据工具返回结果生成最终答案
MCP 就是把这套流程标准化,让工具的"声明"和"执行"可以通过统一的协议进行。
MCP 架构图
┌─────────────────┐
│ 大语言模型 │
│ (LLM) │
└────────┬────────┘
│
1. 注册工具 │ 4. 返回调用请求
┌───────────────────┼───────────────────┐
│ │ │
▼ ▼ ▼
┌─────────┐ ┌─────────┐ ┌─────────┐
│ MCP │ │ Client │ │ MCP │
│ Server │◄──────►│ App │◄──────►│ Server │
│ (天气) │ │ (Cline) │ │ (地图) │
└─────────┘ └─────────┘ └─────────┘
│ │ │
│ 5. 执行工具 │ 6. 返回结果 │
└───────────────────┴───────────────────┘MCP 的三种传输模式
MCP 协议支持三种传输方式:STDIO、SSE 和 Streamable HTTP。理解它们的区别对于正确使用 MCP 非常重要。
1. STDIO 模式(Standard Input/Output)
STDIO 模式通过标准输入输出进行通信,是最简单的本地通信方式:
- 本地执行:MCP Server 以子进程形式在客户端本地运行
- 进程通信:通过 stdin/stdout 传递消息
- 适用场景:本地工具、需要访问本地资源的场景
配置示例:
{
"mcpServers": {
"local-tools": {
"command": "python",
"args": ["./my_mcp_server.py"],
"transport": "stdio"
}
}
}2. SSE 模式(Server-Sent Events)
SSE 是一种基于 HTTP 的长连接协议,专为实时流式传输优化:
- 服务端部署:MCP Server 运行在远程服务器上
- 单向流式:服务端可以持续推送数据给客户端
- 适用场景:需要实时更新的场景,如进度通知
配置示例:
{
"mcpServers": {
"weather-service": {
"url": "https://api.example.com/mcp/sse",
"transport": "sse"
}
}
}3. Streamable HTTP 模式
这是较新的传输模式,处理标准的 HTTP 请求:
- 远程部署:MCP Server 作为 HTTP 服务运行
- 多客户端支持:可以同时服务多个客户端
- 适用场景:生产环境、公共 API 服务
配置示例:
{
"mcpServers": {
"api-service": {
"url": "http://localhost:8000/mcp",
"transport": "streamable_http"
}
}
}三种模式对比
| 特性 | STDIO | SSE | Streamable HTTP |
|---|---|---|---|
| 运行位置 | 本地机器 | 远程服务器 | 远程服务器 |
| 网络要求 | 无需网络 | 需要网络 | 需要网络 |
| 多客户端 | 不支持 | 支持 | 支持 |
| 实时推送 | 不支持 | 支持 | 按需 |
| 适用场景 | 本地开发 | 流式更新 | 生产部署 |
| 复杂度 | 最简单 | 中等 | 中等 |
安全提醒:STDIO 模式下,MCP Server 在你的本地机器执行,理论上可以访问本地文件、执行系统命令。使用第三方 MCP 服务时,请务必确认其来源可信。
LangGraph 集成 MCP
LangGraph 通过 langchain-mcp-adapters 模块提供了对 MCP 的原生支持。
安装依赖
pip install langchain-mcp-adapters langgraph基础用法:连接 MCP Server
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:
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 可以实现持久化会话:
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 工具
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
pip install mcp示例:天气查询服务
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")启动并测试
# 启动服务
python weather_server.py
# 服务将在 http://localhost:8080/sse 可用在 MCP 客户端配置中添加:
{
"mcpServers": {
"my-weather": {
"url": "http://localhost:8080/sse",
"transport": "sse"
}
}
}MCP 生态资源
常用 MCP 服务聚合平台
| 平台 | 地址 | 说明 |
|---|---|---|
| MCP Hub | https://mcp.so | 聚合数千个 MCP 服务 |
| Anthropic 官方 | https://modelcontextprotocol.io | 官方文档和 SDK |
热门 MCP 服务类型
- 地图导航:路线规划、地点搜索
- 文件操作:读写本地文件、目录管理
- 数据库:SQL 查询、数据管理
- 网页浏览:网页抓取、内容提取
- 代码执行:运行代码、调试程序
- 邮件服务:发送邮件、管理收件箱
冷静看待 MCP:局限性与未来
MCP 在 2024 年底发布时确实引发了巨大关注,但经过一年的实践检验,我们需要更理性地认识它的能力边界。
核心局限:上下文窗口的代价
MCP 最大的问题在于工具定义会占用模型的上下文窗口。
每个 MCP Server 的工具描述、参数定义、调用记录都需要放入 Context,模型才能"看到"并决策。这带来几个实际问题:
假设场景:
- 一个 GitHub MCP Server 定义了 90+ 个工具
- 光工具描述就消耗 50000+ Token
- 还没开始对话,上下文已经用掉一大半工具越多,模型越"笨"——这是一个真实存在的悖论。当上下文被大量工具定义填满时:
- 模型的注意力被稀释,推理能力下降
- 可能出现"幻觉":明明工具返回了正确结果,模型却用自己的"常识"覆盖
- 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,例如:
# SKILL.md 示例结构
---
name: "公司周报生成"
description: "按照公司标准模板生成周报"
triggers:
- "写周报"
- "生成周报"
- "本周总结"
---
## 执行步骤
1. 询问用户本周完成的主要工作
2. 按照以下模板组织内容:
- 本周完成
- 下周计划
- 需要协调的问题
3. 应用公司标准格式
4. 生成 Word 文档Skills vs MCP:技术对比
| 维度 | MCP | Skills |
|---|---|---|
| 加载时机 | 启动时全部加载 | 按需动态加载 |
| 上下文消耗 | 高(所有工具定义) | 低(渐进式披露) |
| 适用场景 | 连接外部服务和数据 | 执行标准化流程 |
| 可复用性 | 通用协议,跨平台 | Claude 生态内 |
| 开发门槛 | 较低 | 需要理解 Skill 架构 |
协同使用
Skills 和 MCP 并非互斥,而是互补:
场景:生成包含实时数据的销售报表
Skills 负责:
├── 报表的格式和结构
├── 图表的类型选择
└── 输出的样式规范
MCP 负责:
├── 连接数据库获取销售数据
├── 调用 API 获取汇率信息
└── 访问 CRM 系统获取客户信息行业影响
OpenAI、Google、AWS 等厂商陆续宣布支持 MCP,说明它作为一个开放标准正在被行业接受。但 Skills 的出现也表明:单纯依赖工具协议是不够的,AI 还需要"知识"来指导如何正确使用这些工具。
参考资料:
实用建议
基于以上分析,给出几点实用建议:
| 场景 | 建议 |
|---|---|
| 核心高频功能 | 优先使用原生集成或 Skills |
| 专业领域工具 | 精选 2-3 个高质量 MCP |
| 开发调试 | MCP 很方便,但注意控制数量 |
| 生产环境 | 严格评估安全性和稳定性 |
记住:MCP 是手段,不是目的。真正重要的是让 AI 完成任务,而不是堆叠更多工具。
最佳实践
1. 安全优先
# 不推荐:直接信任所有 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. 错误处理
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 工具的效果很大程度上取决于描述的质量:
# 较差的描述
@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-adapters | LangChain/LangGraph 的 MCP 集成库 |
| MultiServerMCPClient | 管理多个 MCP Server 连接的客户端类 |
| ClientSession | 有状态工具会话,用于需要跨调用保持状态的场景 |
| FastMCP | Python MCP SDK,用于快速创建 MCP Server |
关键要点
- MCP 是协议,不是魔法:它只是标准化了工具调用的方式,底层仍然是 Function Call
- 注意安全风险:特别是 STDIO 模式,第三方服务可能访问你的本地资源
- 工具描述很重要:好的描述能让模型更准确地调用工具
- LangGraph 原生支持:通过
langchain-mcp-adapters可以轻松集成 - 少即是多:MCP 工具数量要精简,过多反而会降低模型性能
- 理性看待生态:选择经过验证的高质量 MCP,而非盲目追求数量
思考题
- MCP 和传统的 REST API 有什么区别?它们各自适合什么场景?
- 如果一个 MCP Server 返回了错误的数据,模型会如何处理?
- 如何设计一个安全的 MCP 权限控制系统?
- 为什么"工具越多,模型越笨"?从技术角度分析这个现象。
- Skills 和 MCP 各自适合什么场景?如何在项目中合理选择?
- 什么情况下需要使用有状态会话(ClientSession)?举例说明。