# https://github.com/langchain-ai/langchain-mcp-adapters 项目说明书
生成时间：2026-05-11 16:35:38 UTC
## 目录
- [项目介绍](#page-project-introduction)
- [安装与配置](#page-installation)
- [核心模块结构](#page-core-modules)
- [数据流与转换机制](#page-data-flow)
- [MultiServerMCPClient 客户端](#page-multiserver-client)
- [工具转换机制](#page-tools-conversion)
- [传输协议类型](#page-transport-types)
- [工具调用拦截器](#page-interceptors)
- [回调机制](#page-callbacks)
- [提示词与资源转换](#page-prompts-resources)
## 项目介绍
### 相关页面
相关主题：[安装与配置](#page-installation), [核心模块结构](#page-core-modules)
相关源码文件
以下源码文件用于生成本页说明：
- [README.md](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/README.md)
- [langchain_mcp_adapters/__init__.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/__init__.py)
- [langchain_mcp_adapters/tools.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/tools.py)
- [langchain_mcp_adapters/resources.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/resources.py)
- [langchain_mcp_adapters/interceptors.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/interceptors.py)
- [langchain_mcp_adapters/client.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/client.py)
- [langchain_mcp_adapters/sessions.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/sessions.py)
# 项目介绍
## 项目概述
langchain-mcp-adapters 是一个轻量级适配器库，用于将 [Anthropic Model Context Protocol (MCP)](https://modelcontextprotocol.io/introduction) 工具转换为 [LangChain](https://github.com/langchain-ai/langchain) 和 [LangGraph](https://github.com/langchain-ai/langgraph) 兼容的格式。该项目使开发者能够在 LangChain/LangGraph 应用中无缝使用 MCP 协议定义的各种工具。
资料来源：[langchain_mcp_adapters/__init__.py:1-9]()
## 核心定位
该项目充当 MCP 协议与 LangChain 生态之间的桥梁，解决两个协议栈之间的互操作性问题。
```mermaid
graph LR
A[MCP 服务器] --> B[langchain-mcp-adapters]
B --> C[LangChain/LangGraph]
C --> D[AI Agent]
style B fill:#e1f5fe
```
资料来源：[README.md:1-1]()
## 核心功能特性
| 功能模块 | 说明 |
|---------|------|
| **工具适配** | 将 MCP 工具转换为 LangChain 工具，可与 LangGraph agents 配合使用 |
| **多服务器客户端** | 支持连接多个 MCP 服务器并从中加载工具 |
| **资源转换** | 将 MCP 资源转换为 LangChain Blob 对象 |
| **拦截器机制** | 提供工具调用生命周期管理能力 |
| **多传输协议支持** | 支持 stdio、HTTP、SSE、WebSocket 等多种传输方式 |
资料来源：[README.md:10-16]()
## 系统架构
### 核心组件
```mermaid
graph TD
subgraph "客户端层"
A[MultiServerMCPClient]
end
subgraph "会话管理层"
B[sessions.py]
C[Connection 抽象]
D[StdioConnection]
E[StreamableHttpConnection]
F[SSEConnection]
G[WebsocketConnection]
end
subgraph "适配器层"
H[tools.py]
I[resources.py]
J[prompts.py]
end
subgraph "拦截器层"
K[interceptors.py]
L[ToolCallInterceptor]
end
A --> B
B --> C
C --> D
C --> E
C --> F
C --> G
A --> H
A --> I
A --> J
H --> K
```
### 工具转换流程
```mermaid
sequenceDiagram
participant Client as MultiServerMCPClient
participant Session as MCP ClientSession
participant Adapter as tools.py
participant Interceptor as ToolCallInterceptor
participant LC as LangChain Tool
Client->>Session: list_tools()
Session-->>Client: List[MCPTool]
Client->>Adapter: convert_to_langchain_tool(MCPTool)
Adapter->>Interceptor: 包装拦截器链
Adapter->>LC: StructuredTool
Client-->>Client: 返回 LangChain 工具列表
```
资料来源：[langchain_mcp_adapters/tools.py:1-50]()
## 模块详解
### tools.py - 工具适配模块
该模块负责将 MCP 工具转换为 LangChain 工具，核心功能包括：
- **内容块转换**：支持 TextContent、ImageContent、FileContent 等多种内容类型转换
- **工具调用执行**：通过拦截器链执行工具调用并处理结果
- **错误处理**：将 MCP 错误转换为 ToolException
- **结构化内容包装**：通过 MCPToolArtifact 包装结构化返回内容
| 转换类型 | MCP 输入 | LangChain 输出 |
|---------|---------|---------------|
| 文本内容 | TextContent | {"type": "text", "text": ...} |
| 图片内容 | ImageContent | {"type": "image", "base64": ..., "mime_type": ...} |
| 图片链接 | ResourceLink(image/*) | {"type": "image", "url": ...} |
| 文件链接 | ResourceLink(other) | {"type": "file", "url": ...} |
| 内嵌资源 | EmbeddedResource | 同上映射规则 |
资料来源：[langchain_mcp_adapters/tools.py:100-150]()
### resources.py - 资源适配模块
该模块处理 MCP 资源的转换，将 MCP 资源内容转换为 LangChain Blob 对象：
- **文本资源**：直接提取 text 内容
- **二进制资源**：Base64 解码后封装为 Blob
- **动态资源**：支持按 URI 列表加载特定资源
资料来源：[langchain_mcp_adapters/resources.py:1-60]()
### interceptors.py - 拦截器模块
提供工具调用生命周期管理能力，采用洋葱模式构建拦截器链：
```mermaid
graph TD
A[请求入口] --> B[拦截器 1]
B --> C[拦截器 2]
C --> D[拦截器 N]
D --> E[实际工具执行]
E --> D
D --> C
C --> B
B --> A[响应返回]
style E fill:#c8e6c9
```
| 组件 | 说明 |
|-----|------|
| MCPToolCallRequest | 工具调用请求数据结构 |
| MCPToolCallResult | 工具调用结果，支持 Command 格式 |
| ToolCallInterceptor | 拦截器接口定义 |
资料来源：[langchain_mcp_adapters/interceptors.py:1-50]()
### client.py - 多服务器客户端
MultiServerMCPClient 是核心入口类，提供以下能力：
- 同时管理多个 MCP 服务器连接
- 支持按服务器名获取工具列表
- 支持显式会话管理和上下文管理器会话
- 集成回调和拦截器配置
| 配置项 | 类型 | 说明 |
|-------|------|------|
| connections | dict | 服务器连接配置映射 |
| callbacks | Callbacks | 事件回调处理器 |
| tool_interceptors | list | 工具调用拦截器列表 |
| tool_name_prefix | bool | 是否为工具名添加服务器前缀 |
资料来源：[langchain_mcp_adapters/client.py:1-100]()
### sessions.py - 会话管理层
负责创建和管理 MCP ClientSession，支持多种传输协议：
| 传输类型 | 适用场景 | 配置方式 |
|---------|---------|---------|
| StdioConnection | 本地进程通信 | command + args |
| StreamableHttpConnection | 无状态 HTTP 服务器 | url |
| SSEConnection | Server-Sent Events | url + headers |
| WebsocketConnection | WebSocket 通信 | url + headers |
资料来源：[langchain_mcp_adapters/sessions.py:1-100]()
## 快速开始
### 安装依赖
```bash
pip install langchain-mcp-adapters
```
资料来源：[README.md:22-24]()
### 创建 MCP 服务器
```python
# math_server.py
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Math")
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers"""
return a + b
@mcp.tool()
def multiply(a: int, b: int) -> int:
"""Multiply two numbers"""
return a * b
```
资料来源：[README.md:30-50]()
### 使用 MultiServerMCPClient
```python
from langchain_mcp_adapters.client import MultiServerMCPClient
client = MultiServerMCPClient(
{
"math": {
"command": "python",
"args": ["./math_server.py"],
"transport": "stdio",
}
}
)
tools = await client.get_tools()
```
资料来源：[README.md:70-85]()
### 集成 LangGraph Agent
```python
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition
builder = StateGraph(MessagesState)
builder.add_node(call_model)
builder.add_node(ToolNode(tools))
builder.add_edge(START, "call_model")
builder.add_conditional_edges(
"call_model",
tools_condition,
)
```
资料来源：[README.md:110-125]()
## 数据类型定义
### MCPToolArtifact
```python
class MCPToolArtifact(TypedDict):
"""MCP 工具调用返回的工件"""
structured_content: dict[str, Any]
```
用于包装 MCP 工具的结构化返回内容，支持未来扩展。
资料来源：[langchain_mcp_adapters/tools.py:75-90]()
### ConvertedToolResult
根据 langgraph 安装状态，返回类型有所不同：
| LangGraph 状态 | 返回类型 |
|---------------|---------|
| 已安装 | `list[ToolMessageContentBlock] \| ToolMessage \| Command` |
| 未安装 | `list[ToolMessageContentBlock] \| ToolMessage` |
资料来源：[langchain_mcp_adapters/tools.py:60-70]()
## 生态关联
该项目与以下项目形成完整的 MCP 工具生态：
```mermaid
graph TD
A[langchain-mcp-adapters] --> B[LangChain]
A --> C[LangGraph]
A --> D[MCP SDK]
E[langchainjs] --> A
E --> F[JavaScript/TypeScript 版本]
style A fill:#fff9c4
```
> 注意：该库的 JavaScript/TypeScript 版本也可在 [langchainjs](https://github.com/langchain-ai/langchainjs/tree/main/libs/langchain-mcp-adapters/) 获取。
资料来源：[README.md:20-21]()
## 总结
langchain-mcp-adapters 作为 MCP 协议与 LangChain 生态之间的关键适配层，提供：
1. **开箱即用的工具转换**：零配置将 MCP 工具转换为 LangChain 工具
2. **灵活的拦截机制**：支持在工具调用生命周期中注入自定义逻辑
3. **强大的多服务器支持**：单客户端管理多个 MCP 服务器
4. **完善的类型安全**：全面使用 Pydantic 和 TypedDict 确保类型安全
该库使开发者能够充分利用 MCP 协议丰富的工具生态，同时享受 LangChain/LangGraph 带来的强大 Agent 开发能力。
---
## 安装与配置
### 相关页面
相关主题：[项目介绍](#page-project-introduction), [MultiServerMCPClient 客户端](#page-multiserver-client)
相关源码文件
以下源码文件用于生成本页说明：
- [pyproject.toml](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/pyproject.toml)
- [README.md](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/README.md)
- [langchain_mcp_adapters/client.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/client.py)
- [langchain_mcp_adapters/tools.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/tools.py)
- [langchain_mcp_adapters/sessions.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/sessions.py)
- [langchain_mcp_adapters/resources.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/resources.py)
# 安装与配置
## 概述
`langchain-mcp-adapters` 是一个轻量级适配器库，用于将 Anthropic 的 Model Context Protocol (MCP) 工具转换为 LangChain 和 LangGraph 兼容的工具格式。该库的主要功能包括：
- 将 MCP 工具转换为 LangChain 工具，使其能够与 LangGraph 代理配合使用
- 提供客户端实现，支持连接多个 MCP 服务器并加载其工具
资料来源：[README.md:1-12]()
## 环境要求
### 系统要求
| 组件 | 最低版本要求 | 说明 |
|------|-------------|------|
| Python | 3.10+ | 推荐使用 Python 3.10 或更高版本 |
| pip | 最新版本 | 用于安装 Python 包 |
### 核心依赖
| 依赖包 | 版本要求 | 用途 |
|--------|---------|------|
| langchain-core | 最新稳定版 | LangChain 核心功能 |
| mcp | 最新稳定版 | MCP 协议客户端支持 |
| pydantic | 2.x | 数据验证和模型定义 |
### 可选依赖
| 依赖包 | 用途 |
|--------|------|
| langgraph | 用于 LangGraph StateGraph 集成 |
| langchain[openai] | OpenAI 模型支持 |
| httpx | HTTP 传输协议支持 |
## 安装方法
### 基础安装
使用 pip 安装 langchain-mcp-adapters 的基本功能：
```bash
pip install langchain-mcp-adapters
```
资料来源：[README.md:14]()
### 完整安装（包含所有依赖）
为了使用完整功能，包括与 LangGraph 和 OpenAI 的集成：
```bash
pip install langchain-mcp-adapters langgraph "langchain[openai]"
```
资料来源：[README.md:20-21]()
### 设置 API 密钥
安装完成后，需要设置相应的 API 密钥环境变量：
```bash
export OPENAI_API_KEY=
```
资料来源：[README.md:22]()
## 客户端配置
### MultiServerMCPClient 概述
`MultiServerMCPClient` 是连接多个 MCP 服务器的主要接口，允许你从多个服务器同时加载工具。
资料来源：[client.py:1-50]()
### 连接配置参数
#### 支持的传输协议
| 传输类型 | 配置方式 | 说明 |
|----------|----------|------|
| `stdio` | 通过 command 和 args 参数 | 适用于本地子进程通信 |
| `http` | 通过 url 参数 | 适用于远程 HTTP 服务器 |
| `sse` | 通过 url 参数 | 支持 Server-Sent Events |
资料来源：[sessions.py:1-200]()
#### stdio 传输配置
stdio 传输适用于本地 MCP 服务器进程：
```python
from langchain_mcp_adapters.client import MultiServerMCPClient
client = MultiServerMCPClient(
{
"math": {
"command": "python",
"args": ["/path/to/math_server.py"],
"transport": "stdio",
},
}
)
```
**配置参数说明：**
| 参数 | 类型 | 必填 | 说明 |
|------|------|------|------|
| `command` | str | 是 | 要执行的命令（如 `python`、`node`） |
| `args` | list[str] | 是 | 命令行参数列表 |
| `transport` | str | 是 | 必须设置为 `"stdio"` |
| `env` | dict | 否 | 环境变量字典，支持变量展开 |
| `cwd` | str | 否 | 命令执行的工作目录 |
| `encoding` | str | 否 | 字符编码方式 |
| `encoding_error_handler` | str | 否 | 编码错误处理策略 |
资料来源：[sessions.py:1-100]()
#### HTTP 传输配置
HTTP 传输适用于远程 MCP 服务器：
```python
client = MultiServerMCPClient(
{
"weather": {
"url": "http://localhost:8000/mcp",
"transport": "http",
"headers": {
"Authorization": "Bearer custom-value"
}
}
}
)
```
**HTTP 配置参数说明：**
| 参数 | 类型 | 必填 | 说明 |
|------|------|------|------|
| `url` | str | 是 | MCP 服务器的完整 URL |
| `transport` | str | 是 | 必须设置为 `"http"` 或 `"sse"` |
| `headers` | dict | 否 | HTTP 请求头，仅支持 `http` 和 `sse` 传输 |
资料来源：[client.py:50-100]()
### 工具名称前缀配置
当多个 MCP 服务器存在相同名称的工具时，可以使用 `tool_name_prefix` 参数为工具名称添加服务器前缀：
```python
client = MultiServerMCPClient(
{
"weather": {
"url": "http://localhost:8000/mcp",
"transport": "http",
}
},
tool_name_prefix=True # 启用工具名称前缀
)
```
启用后，工具名称将从 `"search"` 变为 `"weather_search"`。
资料来源：[client.py:40-45]()
## MCP 服务器配置
### 创建 MCP 服务器
使用 FastMCP 框架创建 MCP 服务器：
```python
# math_server.py
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Math")
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers"""
return a + b
@mcp.tool()
def multiply(a: int, b: int) -> int:
"""Multiply two numbers"""
return a * b
if __name__ == "__main__":
mcp.run(transport="stdio")
```
资料来源：[README.md:30-55]()
### HTTP 传输服务器
对于 HTTP 传输模式的服务器：
```python
# weather_server.py
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Weather")
@mcp.tool()
async def get_weather(location: str) -> str:
"""Get weather for location."""
return "It's always sunny in New York"
if __name__ == "__main__":
mcp.run(transport="http")
```
运行服务器：
```bash
python weather_server.py
```
资料来源：[README.md:60-75]()
### 无状态 HTTP 流式服务器
对于可扩展的 HTTP 服务，可以使用无状态流式传输：
```bash
cd examples/servers/streamable-http-stateless/
uv run mcp-simple-streamablehttp-stateless --port 3000
```
资料来源：[README.md:2-8]()
## 会话管理配置
### 显式会话管理
虽然 `MultiServerMCPClient` 默认在每次工具调用时自动创建新会话，但也支持显式会话管理：
```python
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.tools import load_mcp_tools
client = MultiServerMCPClient({...})
async with client.session("math") as session:
tools = await load_mcp_tools(session)
```
资料来源：[client.py:60-70]()
### 连接对象配置
`Connection` 对象用于配置与 MCP 服务器的连接细节：
```python
from langchain_mcp_adapters.sessions import create_session
async with create_session(
command="python",
args=["math_server.py"],
transport="stdio"
) as session:
tools = await load_mcp_tools(session)
```
## 工具加载配置
### load_mcp_tools 函数参数
| 参数 | 类型 | 必填 | 说明 |
|------|------|------|------|
| `session` | ClientSession | 否 | MCP 客户端会话，与 connection 二选一 |
| `connection` | Connection | 否 | 连接配置，用于创建新会话 |
| `callbacks` | Callbacks | 否 | 通知和事件处理回调 |
| `tool_interceptors` | list[ToolCallInterceptor] | 否 | 工具调用拦截器列表 |
| `server_name` | str | 否 | 服务器名称 |
| `tool_name_prefix` | bool | 否 | 是否为工具名称添加前缀 |
资料来源：[tools.py:100-150]()
## 架构流程图
### 客户端-服务器通信架构
```mermaid
graph TD
A[MultiServerMCPClient] --> B[Connection 配置]
B --> C[stdio 传输]
B --> D[HTTP 传输]
B --> E[SSE 传输]
C --> F[子进程 MCP 服务器]
D --> G[远程 HTTP 服务器]
E --> H[SSE 服务器]
F --> I[ClientSession]
G --> I
H --> I
I --> J[load_mcp_tools]
J --> K[LangChain StructuredTool]
K --> L[LangGraph Agent]
```
### 工具调用流程
```mermaid
sequenceDiagram
participant User as 用户代码
participant LC as LangChain Agent
participant MCP as MCP Adapter
participant Server as MCP Server
User->>LC: 调用工具
LC->>MCP: call_tool(args)
MCP->>MCP: 应用拦截器链
MCP->>Server: session.call_tool()
Server-->>MCP: CallToolResult
MCP->>MCP: 转换结果格式
MCP-->>LC: ToolMessage + Artifact
LC-->>User: 返回结果
```
## 环境变量与密钥管理
### 必需的环境变量
| 变量名 | 说明 | 示例 |
|--------|------|------|
| `OPENAI_API_KEY` | OpenAI API 密钥 | `sk-xxx...` |
### 传输配置中的环境变量
在 stdio 传输配置中，可以使用环境变量展开语法：
```python
client = MultiServerMCPClient(
{
"server": {
"command": "python",
"args": ["script.py"],
"env": {
"API_KEY": "${API_KEY}", # 从当前环境展开
"CUSTOM_VAR": "fixed-value"
},
"transport": "stdio"
}
}
)
```
> [!warning]
> 如果环境变量值包含 `${variable}` 语法但无法展开，库会发出警告。
资料来源：[sessions.py:80-95]()
## 完整配置示例
### 多服务器配置
```python
from langchain_mcp_adapters.client import MultiServerMCPClient
# 初始化客户端，连接多个 MCP 服务器
client = MultiServerMCPClient(
{
# 本地 stdio 服务器
"math": {
"command": "python",
"args": ["/absolute/path/to/math_server.py"],
"transport": "stdio",
},
# 远程 HTTP 服务器
"weather": {
"url": "http://localhost:8000/mcp",
"transport": "http",
"headers": {
"Authorization": "Bearer token123"
}
}
},
tool_name_prefix=True # 避免工具名冲突
)
# 获取所有工具
tools = await client.get_tools()
```
资料来源：[README.md:85-110]()
### 与 LangGraph 集成配置
```python
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition
from langchain.chat_models import init_chat_model
# 配置 MCP 客户端
client = MultiServerMCPClient(
{
"math": {
"command": "python",
"args": ["./examples/math_server.py"],
"transport": "stdio",
},
"weather": {
"url": "http://localhost:8000/mcp",
"transport": "http",
}
}
)
# 加载工具
tools = await client.get_tools()
# 初始化模型
model = init_chat_model("openai:gpt-4.1")
# 定义模型调用节点
def call_model(state: MessagesState):
response = model.bind_tools(tools).invoke(state["messages"])
return {"messages": response}
# 构建状态图
builder = StateGraph(MessagesState)
builder.add_node(call_model)
builder.add_node(ToolNode(tools))
builder.add_edge(START, "call_model")
builder.add_conditional_edges(
"call_model",
tools_condition,
)
```
资料来源：[README.md:120-165]()
## 故障排除
### 常见配置错误
| 错误 | 原因 | 解决方案 |
|------|------|----------|
| `ValueError: Either session or connection must be provided` | 未提供 session 或 connection 参数 | 确保传递有效的 session 或 connection 对象 |
| 工具名冲突 | 多服务器有同名工具 | 设置 `tool_name_prefix=True` |
| 连接超时 | HTTP 服务器未启动或地址错误 | 确认服务器运行状态和 URL |
| 环境变量未展开 | 变量语法错误 | 检查 `${variable}` 格式是否正确 |
### 调试建议
1. **验证 MCP 服务器**：确保服务器可以独立运行
2. **检查传输类型**：stdio 需要命令和参数，HTTP 需要 URL
3. **查看工具列表**：使用 `client.get_tools()` 确认工具加载成功
4. **环境变量检查**：确保必需的 API 密钥已设置
---
## 核心模块结构
### 相关页面
相关主题：[工具转换机制](#page-tools-conversion), [MultiServerMCPClient 客户端](#page-multiserver-client), [数据流与转换机制](#page-data-flow)
相关源码文件
以下源码文件用于生成本页说明：
- [langchain_mcp_adapters/tools.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/tools.py)
- [langchain_mcp_adapters/client.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/client.py)
- [langchain_mcp_adapters/sessions.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/sessions.py)
- [langchain_mcp_adapters/callbacks.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/callbacks.py)
- [langchain_mcp_adapters/interceptors.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/interceptors.py)
- [langchain_mcp_adapters/prompts.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/prompts.py)
- [langchain_mcp_adapters/resources.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/resources.py)
# 核心模块结构
## 概述
langchain-mcp-adapters 是一个轻量级适配器库，用于将 Anthropic 的 Model Context Protocol (MCP) 服务器与 LangChain/LangGraph 应用集成。该库的核心设计围绕**工具适配**、**会话管理**、**资源转换**和**拦截器机制**四大支柱展开，通过模块化架构实现 MCP 工具向 LangChain 工具的透明转换。
项目采用异步优先的设计理念，充分利用 Python 的 `asyncio` 能力处理 MCP 通信，同时通过 LangGraph 条件判断实现智能的工具路由。资料来源：[README.md]()
## 架构总览
```mermaid
graph TD
A[MultiServerMCPClient] --> B[Connection 管理]
A --> C[Session 管理]
B --> D[StdioConnection]
B --> E[StreamableHttpConnection]
B --> F[SSEConnection]
B --> G[WebsocketConnection]
C --> H[ClientSession]
H --> I[load_mcp_tools]
H --> J[load_mcp_resources]
H --> K[load_mcp_prompt]
I --> L[StructuredTool]
J --> M[Blob]
K --> N[ChatPromptValue]
L --> O[ToolCallInterceptor]
O --> P[MCPToolCallRequest]
P --> Q[MCPToolCallResult]
```
## 核心模块划分
| 模块 | 文件 | 主要职责 |
|------|------|----------|
| 客户端入口 | `client.py` | 多服务器连接管理、工具资源加载 |
| 会话管理 | `sessions.py` | 不同传输层连接创建 |
| 工具适配 | `tools.py` | MCP工具→LangChain工具转换 |
| 资源适配 | `resources.py` | MCP资源→LangChain Blob转换 |
| 提示加载 | `prompts.py` | MCP提示→LangChain提示转换 |
| 拦截器 | `interceptors.py` | 工具调用拦截与修改 |
| 回调 | `callbacks.py` | 进度通知与事件处理 |
资料来源：[langchain_mcp_adapters/client.py:1-50]()
## MultiServerMCPClient 客户端模块
### 功能定位
`MultiServerMCPClient` 是库的入口点，负责管理对多个 MCP 服务器的连接。它支持通过配置字典同时连接到多个不同类型的服务器（stdio、HTTP SSE、WebSocket 等），并提供统一的工具、资源和提示加载接口。
资料来源：[langchain_mcp_adapters/client.py:1-80]()
### 连接配置
客户端接受服务器配置字典，键为服务器名称，值为连接参数字典：
```python
client = MultiServerMCPClient(
{
"math": {
"command": "python",
"args": ["./math_server.py"],
"transport": "stdio",
},
"weather": {
"url": "http://localhost:8000/mcp",
"transport": "http",
}
}
)
```
| 参数 | 类型 | 说明 |
|------|------|------|
| `command` | `str` | 执行命令（stdio传输必需） |
| `args` | `list[str]` | 命令行参数 |
| `transport` | `str` | 传输类型：`stdio`、`http`、`sse`、`websocket` |
| `url` | `str` | 服务器URL（HTTP类传输必需） |
| `headers` | `dict` | HTTP请求头（仅sse/http传输支持） |
| `timeout` | `float` | HTTP超时时间 |
| `env` | `dict` | 环境变量 |
资料来源：[langchain_mcp_adapters/client.py:1-100]()
### 上下文管理器使用方式
版本 0.1.0 后，`MultiServerMCPClient` 不再支持作为异步上下文管理器直接使用：
```python
# 推荐方式一：直接调用
client = MultiServerMCPClient({...})
tools = await client.get_tools()
# 推荐方式二：显式会话管理
client = MultiServerMCPClient({...})
async with client.session("math") as session:
tools = await load_mcp_tools(session)
```
资料来源：[langchain_mcp_adapters/client.py:80-120]()
## sessions.py 会话管理模块
### 传输层抽象
`sessions.py` 模块定义了多种连接类型，每种类型对应不同的 MCP 传输协议：
```mermaid
graph LR
A[Connection 抽象] --> B[StdioConnection]
A --> C[StreamableHttpConnection]
A --> D[SSEConnection]
A --> E[WebsocketConnection]
```
资料来源：[langchain_mcp_adapters/sessions.py:1-100]()
### StdioConnection
通过标准输入/输出与本地进程通信，适用于本地 MCP 服务器：
```python
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
yield session
```
| 参数 | 说明 |
|------|------|
| `command` | 可执行命令路径 |
| `args` | 命令行参数列表 |
| `env` | 环境变量（支持`${...}`变量展开） |
| `cwd` | 工作目录 |
| `encoding` | 字符编码 |
资料来源：[langchain_mcp_adapters/sessions.py:100-200]()
### StreamableHttpConnection
支持流式 HTTP 传输的连接类型，适用于无状态的 HTTP MCP 服务器：
```python
async with streamablehttp_client(url, ...) as (read, write, _):
async with ClientSession(read, write) as session:
await session.initialize()
```
资料来源：[langchain_mcp_adapters/sessions.py:200-300]()
### SSEConnection
服务器发送事件（Server-Sent Events）传输连接，支持实时事件推送：
```python
async with sse_client(url, headers=headers) as sessions:
for session in sessions:
# 处理每个SSE会话
```
资料来源：[langchain_mcp_adapters/sessions.py:300-400]()
### WebsocketConnection
WebSocket 传输连接，支持双向通信：
```python
async with websocket_client(url) as (read, write):
async with ClientSession(read, write) as session:
yield session
```
资料来源：[langchain_mcp_adapters/sessions.py:400-500]()
## tools.py 工具适配模块
### 核心转换流程
`tools.py` 模块负责将 MCP 工具转换为 LangChain 的 `StructuredTool` 对象，支持工具执行、参数验证和结果转换：
```mermaid
graph TD
A[MCP Tool] --> B[load_mcp_tools]
B --> C[convert_mcp_tool_to_langchain_tool]
C --> D[StructuredTool]
E[工具调用] --> F[call_tool 函数]
F --> G[拦截器链处理]
G --> H[MCPToolCallRequest]
H --> I[session.call_tool]
I --> J[MCPToolCallResult]
J --> K[_convert_call_tool_result]
K --> L[Content + Artifact]
```
资料来源：[langchain_mcp_adapters/tools.py:1-100]()
### 工具加载函数
#### load_mcp_tools
加载 MCP 服务器上的所有工具：
```python
async def load_mcp_tools(
session: ClientSession | None,
*,
connection: Connection | None = None,
callbacks: Callbacks | None = None,
tool_interceptors: list[ToolCallInterceptor] | None = None,
server_name: str | None = None,
tool_name_prefix: bool = False,
) -> list[BaseTool]
```
| 参数 | 类型 | 说明 |
|------|------|------|
| `session` | `ClientSession` | MCP客户端会话 |
| `connection` | `Connection` | 连接配置（session为空时使用） |
| `callbacks` | `Callbacks` | 回调处理器 |
| `tool_interceptors` | `list[ToolCallInterceptor]` | 工具调用拦截器列表 |
| `server_name` | `str` | 服务器名称 |
| `tool_name_prefix` | `bool` | 是否将服务器名作为工具名前缀 |
资料来源：[langchain_mcp_adapters/tools.py:200-280]()
#### convert_mcp_tool_to_langchain_tool
转换单个 MCP 工具为 LangChain 工具：
```python
async def convert_mcp_tool_to_langchain_tool(
session: ClientSession | None,
tool: MCPTool,
*,
connection: Connection | None = None,
callbacks: Callbacks | None = None,
tool_interceptors: list[ToolCallInterceptor] | None = None,
server_name: str | None = None,
tool_name_prefix: bool = False,
) -> BaseTool
```
资料来源：[langchain_mcp_adapters/tools.py:150-200]()
### 内容块转换
`_convert_mcp_content_to_lc_block` 函数将 MCP 内容块转换为 LangChain 内容块格式：
| MCP 内容类型 | LangChain 输出 |
|-------------|----------------|
| `TextContent` | `{"type": "text", "text": ...}` |
| `ImageContent` | `{"type": "image", "base64": ..., "mime_type": ...}` |
| `ResourceLink` (图片) | `{"type": "image", "url": ...}` |
| `ResourceLink` (其他) | `{"type": "file", "url": ...}` |
| `EmbeddedResource` (文本) | `{"type": "text", "text": ...}` |
| `EmbeddedResource` (二进制) | `{"type": "image"}` 或 `{"type": "file"}` |
| `AudioContent` | 抛出 `NotImplementedError` |
资料来源：[langchain_mcp_adapters/tools.py:100-150]()
### 工具调用结果类型
```python
# 根据 LangGraph 可用性条件定义
if LANGGRAPH_PRESENT:
ConvertedToolResult = list[ToolMessageContentBlock] | ToolMessage | Command
else:
ConvertedToolResult = list[ToolMessageContentBlock] | ToolMessage
# MCP 工具工件类型
class MCPToolArtifact(TypedDict):
structured_content: dict[str, Any]
```
返回结果为元组 `(content, artifact)`，其中 `artifact` 包含 MCP 服务器返回的结构化内容。
资料来源：[langchain_mcp_adapters/tools.py:50-80]()
## interceptors.py 拦截器模块
### 拦截器模式
拦截器采用洋葱模型（Onion Pattern）包装工具调用处理器，允许在调用前后进行拦截和修改：
```mermaid
graph TD
A[调用者] --> B[拦截器1 包装]
B --> C[拦截器2 包装]
C --> D[拦截器N 包装]
D --> E[execute_tool 实际执行]
E --> D
D --> C
C --> B
B --> A
```
资料来源：[langchain_mcp_adapters/interceptors.py:1-50]()
### 核心数据结构
#### MCPToolCallRequest
工具调用请求数据结构：
```python
@dataclass
class MCPToolCallRequest:
name: str # 工具名称
args: dict[str, Any] # 工具参数
server_name: str # 服务器名称
headers: dict[str, Any] | None # HTTP请求头
runtime: Any # 运行时上下文
```
资料来源：[langchain_mcp_adapters/interceptors.py:50-100]()
#### MCPToolCallResult
工具调用结果类型，根据 LangGraph 可用性条件定义：
```python
if LANGGRAPH_PRESENT:
MCPToolCallResult = CallToolResult | ToolMessage | Command
else:
MCPToolCallResult = CallToolResult | ToolMessage
```
资料来源：[langchain_mcp_adapters/interceptors.py:20-40]()
### ToolCallInterceptor 接口
```python
@runtime_checkable
class ToolCallInterceptor(Protocol):
async def intercept(
self,
request: MCPToolCallRequest,
call_next: Callable[[MCPToolCallRequest], Awaitable[MCPToolCallResult]],
) -> MCPToolCallResult:
...
```
拦截器可以：
- 在调用前修改 `request`（通过 `request.override()` 方法）
- 跳过实际执行，直接返回结果
- 修改或增强返回结果
- 在调用前后执行副作用（如日志记录）
资料来源：[langchain_mcp_adapters/interceptors.py:100-150]()
### 请求覆盖机制
`MCPToolCallRequest` 提供 `override` 方法支持部分字段修改：
```python
class _MCPToolCallRequestOverrides(TypedDict, total=False):
name: NotRequired[str]
args: NotRequired[dict[str, Any]]
headers: NotRequired[dict[str, Any] | None]
def override(self, **overrides: Unpack[_MCPToolCallRequestOverrides]) -> MCPToolCallRequest:
"""返回修改后的请求副本"""
```
资料来源：[langchain_mcp_adapters/interceptors.py:40-50]()
## callbacks.py 回调模块
### 回调接口设计
回调系统用于处理 MCP 服务器的通知和进度事件：
```mermaid
graph LR
A[MCP Server] -->|进度通知| B[Callbacks]
A -->|日志消息| B
A -->|资源列表变更| B
B --> C[工具级别回调]
```
资料来源：[langchain_mcp_adapters/callbacks.py:1-50]()
### CallbackContext
回调上下文包含服务器信息：
```python
@dataclass
class CallbackContext:
server_name: str | None = None
tool_name: str | None = None
```
### Callbacks 类
核心回调接口定义，支持多种事件类型：
| 方法 | 触发时机 |
|------|----------|
| `progress_callback` | 服务器发送进度更新 |
| `logger` | 服务器发送日志消息 |
| `list_changed` | 资源/工具列表变更 |
资料来源：[langchain_mcp_adapters/callbacks.py:50-100]()
## resources.py 资源模块
### 资源转换流程
`resources.py` 模块将 MCP 资源转换为 LangChain 的 `Blob` 对象：
```mermaid
graph LR
A[MCP Resource URI] --> B[session.read_resource]
B --> C[ResourceContents]
C --> D[convert_mcp_resource_to_langchain_blob]
D --> E[Blob]
```
资料来源：[langchain_mcp_adapters/resources.py:1-50]()
### 内容类型支持
| MCP 内容类型 | Blob 处理方式 |
|-------------|---------------|
| `TextResourceContents` | 直接使用文本内容 |
| `BlobResourceContents` | Base64 解码后存储 |
资料来源：[langchain_mcp_adapters/resources.py:50-80]()
### 资源加载函数
#### load_mcp_resources
```python
async def load_mcp_resources(
session: ClientSession,
*,
uris: str | list[str] | None = None,
) -> list[Blob]
```
| 参数 | 说明 |
|------|------|
| `uris=None` | 加载所有资源 |
| `uris=str` | 加载指定 URI 的资源 |
| `uris=list[str]` | 批量加载指定 URI 的资源 |
> 注意：动态资源在 `uris=None` 时不会被加载，因为它们需要参数。
资料来源：[langchain_mcp_adapters/resources.py:80-120]()
## prompts.py 提示模块
### 提示加载功能
`prompts.py` 模块负责将 MCP 服务器上的提示模板加载为 LangChain 的聊天提示值对象，支持参数化提示生成。
资料来源：[langchain_mcp_adapters/prompts.py:1-50]()
## 数据流集成图
### 完整工具调用流程
```mermaid
sequenceDiagram
participant User as 用户/LangGraph
participant Client as MultiServerMCPClient
participant Interceptor as ToolCallInterceptor
participant Session as ClientSession
participant Server as MCP Server
User->>Client: get_tools()
Client->>Session: create_session(connection)
Session->>Server: list_tools()
Server-->>Session: List[McpTool]
Session-->>Client: List[BaseTool]
Client-->>User: List[StructuredTool]
User->>Client: tool.invoke(args)
Client->>Interceptor: intercept(request)
Interceptor->>Interceptor: request.override(...)
Interceptor->>Session: call_tool(name, args)
Session->>Server: call_tool()
Server-->>Session: CallToolResult
Session-->>Interceptor: MCPToolCallResult
Interceptor-->>Client: MCPToolCallResult
Client-->>User: (Content, Artifact)
```
## 配置与扩展
### 工具名称前缀
当连接多个服务器时，可使用 `tool_name_prefix=True` 避免工具名冲突：
```python
client = MultiServerMCPClient(
{"math": {...}, "weather": {...}},
tool_name_prefix=True
)
# 工具名变为: math_add, weather_get_weather
```
### 拦截器链配置
```python
async def my_interceptor(
request: MCPToolCallRequest,
call_next: Callable,
) -> MCPToolCallResult:
# 在调用前记录日志
logger.info(f"Calling {request.name}")
# 可以修改参数
if request.name == "add":
modified_request = request.override(args={"a": 1, "b": 2})
return await call_next(modified_request)
return await call_next(request)
client = MultiServerMCPClient(
{...},
tool_interceptors=[my_interceptor]
)
```
资料来源：[langchain_mcp_adapters/tools.py:250-300]()
### HTTP 运行时头
仅 `sse` 和 `http` 传输支持运行时头：
```python
client = MultiServerMCPClient(
{
"weather": {
"transport": "http",
"url": "http://localhost:8000/mcp",
"headers": {
"Authorization": "Bearer custom-value"
}
}
}
)
```
资料来源：[README.md:1-50]()
## 错误处理
### 工具执行异常
当 MCP 工具返回错误时，`_convert_call_tool_result` 会抛出 `ToolException`：
```python
if call_tool_result.isError:
error_parts = []
for item in tool_content:
if isinstance(item, dict) and item.get("type") == "text":
error_parts.append(item.get("text", ""))
error_msg = "\n".join(error_parts)
raise ToolException(error_msg)
```
### 不支持的内容类型
`AudioContent` 转换尚未支持，会抛出 `NotImplementedError`：
```python
if isinstance(content, AudioContent):
raise NotImplementedError(
f"AudioContent conversion not supported. "
f"Mime type: {content.mimeType}"
)
```
资料来源：[langchain_mcp_adapters/tools.py:100-130]()
## 依赖关系
| 依赖包 | 用途 |
|--------|------|
| `langchain-core` | LangChain 核心工具和消息类型 |
| `mcp` | Model Context Protocol SDK |
| `pydantic` | 数据模型验证 |
| `langgraph` (可选) | LangGraph 集成支持 |
可选依赖 `langgraph` 用于支持 `Command` 类型返回值，提供 LangGraph 特有的控制流能力。
资料来源：[langchain_mcp_adapters/tools.py:1-30]()
## 总结
langchain-mcp-adapters 的核心模块结构展现了清晰的分层架构设计：
1. **会话层** (`sessions.py`) 屏蔽了不同传输协议的差异
2. **适配层** (`tools.py`、`resources.py`、`prompts.py`) 处理 MCP 与 LangChain 类型系统间的转换
3. **控制层** (`interceptors.py`) 提供了灵活的扩展机制
4. **入口层** (`client.py`) 整合所有能力，提供统一的用户体验
这种设计使得库能够在保持简洁 API 的同时，支持复杂的 MCP 服务器集成场景。
---
## 数据流与转换机制
### 相关页面
相关主题：[核心模块结构](#page-core-modules), [工具转换机制](#page-tools-conversion)
相关源码文件
以下源码文件用于生成本页说明：
- [langchain_mcp_adapters/tools.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/tools.py)
- [langchain_mcp_adapters/resources.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/resources.py)
- [langchain_mcp_adapters/client.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/client.py)
- [langchain_mcp_adapters/sessions.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/sessions.py)
- [langchain_mcp_adapters/interceptors.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/interceptors.py)
# 数据流与转换机制
本页面详细说明 langchain-mcp-adapters 库中数据如何在 MCP（Model Context Protocol）与 LangChain 之间流转与转换。核心转换逻辑涵盖工具（Tools）、资源（Resources）和内容块（Content Blocks）三种数据类型。
---
## 1. 整体架构概览
该库采用分层架构，通过适配器模式将 MCP 服务器的功能桥接到 LangChain 生态。数据流遵循以下基本原则：
```mermaid
graph TD
subgraph MCP服务器层
A[MCP Server]
A1[Tools]
A2[Resources]
A3[Prompts]
end
subgraph 适配器层
B[ClientSession]
C[tools.py]
D[resources.py]
E[prompts.py]
end
subgraph LangChain层
F[BaseTool]
G[Blob]
H[Prompt]
end
A1 --> B --> C --> F
A2 --> B --> D --> G
A3 --> B --> E --> H
style C fill:#e1f5fe
style D fill:#e8f5e8
style E fill:#fff3e0
```
---
## 2. 内容块转换机制
### 2.1 核心转换函数
`_convert_mcp_content_to_lc_block()` 函数负责将 MCP 内容块转换为 LangChain 内容块格式 资料来源：[langchain_mcp_adapters/tools.py:60-102]()
```python
def _convert_mcp_content_to_lc_block(content: ContentBlock) -> ToolMessageContentBlock:
"""Convert any MCP content block to a LangChain content block."""
```
### 2.2 支持的内容类型映射
| MCP 内容类型 | LangChain 内容块类型 | 转换逻辑 |
|-------------|---------------------|---------|
| `TextContent` | `{"type": "text", "text": ...}` | 直接提取 `text` 字段 |
| `ImageContent` | `{"type": "image", "base64": ..., "mime_type": ...}` | 提取 `data` 和 `mimeType` |
| `AudioContent` | - | 抛出 `NotImplementedError` |
| `ResourceLink` (image/*) | `{"type": "image", "url": ...}` | 提取 URI 和 mimeType |
| `ResourceLink` (其他) | `{"type": "file", "url": ...}` | 提取 URI 和 mimeType |
| `EmbeddedResource` (text) | `{"type": "text", "text": ...}` | 提取内部 resource.text |
| `EmbeddedResource` (blob) | `{"type": "image"}` 或 `{"type": "file"}` | 根据 mimeType 判断 |
### 2.3 内容块转换流程图
```mermaid
graph LR
A[ContentBlock 输入] --> B{类型判断}
B -->|TextContent| C[create_text_block]
B -->|ImageContent| D[create_image_block]
B -->|AudioContent| E[NotImplementedError]
B -->|ResourceLink| F{image/* ?}
B -->|EmbeddedResource| G{resource类型}
F -->|是| H[create_image_block]
F -->|否| I[create_file_block]
G -->|TextResourceContents| J[create_text_block]
G -->|BlobResourceContents| K{image/* ?}
K -->|是| L[create_image_block]
K -->|否| M[create_file_block]
C --> N[ToolMessageContentBlock]
D --> N
H --> N
I --> N
J --> N
L --> N
M --> N
```
---
## 3. 工具转换机制
### 3.1 工具加载入口
`load_mcp_tools()` 函数是加载 MCP 工具的核心入口，负责列出并转换所有可用工具 资料来源：[langchain_mcp_adapters/tools.py:240-290]()
```python
async def load_mcp_tools(
session: ClientSession | None,
*,
connection: Connection | None = None,
callbacks: Callbacks | None = None,
tool_interceptors: list[ToolCallInterceptor] | None = None,
server_name: str | None = None,
tool_name_prefix: bool = False,
) -> list[BaseTool]:
```
### 3.2 单个工具转换
`convert_mcp_tool_to_langchain_tool()` 函数将单个 MCP 工具转换为 LangChain `StructuredTool` 资料来源：[langchain_mcp_adapters/tools.py:165-238]()
```python
def convert_mcp_tool_to_langchain_tool(
session: ClientSession | None,
tool: MCPTool,
*,
connection: Connection | None = None,
# ... 其他参数
) -> BaseTool:
```
**返回的 StructuredTool 配置：**
| 参数 | 来源 | 说明 |
|-----|------|-----|
| `name` | `tool.name` 或 `server_name + tool.name` | 可选前缀 |
| `description` | `tool.description` | 工具描述 |
| `args_schema` | `tool.inputSchema` | Pydantic 模型 |
| `coroutine` | `call_tool` 协程 | 异步执行函数 |
| `response_format` | `"content_and_artifact"` | 返回格式 |
| `metadata` | 工具注解 | 包含元数据 |
### 3.3 工具转换流程
```mermaid
graph TD
A[MCP Tool 输入] --> B[创建 MCP ClientSession]
B --> C[构建 MCPToolCallRequest]
C --> D{是否配置拦截器?}
D -->|是| E[执行拦截器链]
D -->|否| F[直接执行工具]
E --> G[洋葱模型执行]
F --> H[session.call_tool]
G --> H
H --> I[MCPToolCallResult]
I --> J[_convert_call_tool_result]
J --> K[ConvertedToolResult]
J --> L[MCPToolArtifact]
K --> M[StructuredTool.content]
L --> M
```
### 3.4 工具结果转换
`_convert_call_tool_result()` 处理工具调用结果，支持多种返回类型 资料来源：[langchain_mcp_adapters/tools.py:104-164]()
```python
def _convert_call_tool_result(
call_tool_result: MCPToolCallResult,
) -> tuple[ConvertedToolResult, MCPToolArtifact | None]:
```
**返回值类型定义：**
```python
if LANGGRAPH_PRESENT:
ConvertedToolResult = list[ToolMessageContentBlock] | ToolMessage | Command
else:
ConvertedToolResult = list[ToolMessageContentBlock] | ToolMessage
```
**特殊返回值处理：**
| 输入类型 | 行为 |
|---------|------|
| `ToolMessage` | 直接返回，不转换 |
| `Command` (LangGraph) | 直接返回，支持图执行 |
| `CallToolResult` + `isError=True` | 抛出 `ToolException` |
| `CallToolResult` + `structuredContent` | 包装为 `MCPToolArtifact` |
### 3.5 MCPToolArtifact 数据结构
```python
class MCPToolArtifact(TypedDict):
"""Artifact returned from MCP tool calls."""
structured_content: dict[str, Any]
```
---
## 4. 拦截器链机制
### 4.1 拦截器模式
拦截器采用洋葱模型（Onion Pattern）构建处理链，允许在工具调用前后插入自定义逻辑 资料来源：[langchain_mcp_adapters/interceptors.py]()
```python
def _build_interceptor_chain(
base_handler: Callable[[MCPToolCallRequest], Awaitable[MCPToolCallResult]],
tool_interceptors: list[ToolCallInterceptor] | None,
) -> Callable[[MCPToolCallRequest], Awaitable[MCPToolCallResult]]:
```
### 4.2 拦截器请求类型
```python
@dataclass
class MCPToolCallRequest:
"""Tool execution request passed to MCP tool call interceptors."""
name: str
args: dict[str, Any]
server_name: str
headers: dict[str, Any] | None
runtime: Any
```
### 4.3 拦截器链执行顺序
```mermaid
graph TD
subgraph 拦截器链
A[请求入口] --> B[拦截器 1 外层]
B --> C[拦截器 2]
C --> D[...]
D --> E[拦截器 N]
E --> F[base_handler]
end
subgraph 返回路径
F --> G[拦截器 N 处理]
G --> H[拦截器 2 处理]
H --> I[拦截器 1 处理]
I --> J[最终结果]
end
style F fill:#c8e6c9
```
---
## 5. 资源转换机制
### 5.1 资源到 Blob 的转换
`convert_mcp_resource_to_langchain_blob()` 将 MCP 资源内容转换为 LangChain Blob 对象 资料来源：[langchain_mcp_adapters/resources.py:19-41]()
```python
def convert_mcp_resource_to_langchain_blob(
resource_uri: str, contents: ResourceContents
) -> Blob:
"""Convert an MCP resource content to a LangChain Blob."""
```
**转换逻辑：**
| 资源类型 | 数据处理 |
|---------|---------|
| `TextResourceContents` | 直接使用 `text` 字段 |
| `BlobResourceContents` | `base64.b64decode(blob)` 解码 |
**返回的 Blob 配置：**
```python
return Blob.from_data(
data=data,
mime_type=contents.mimeType,
metadata={"uri": resource_uri}
)
```
### 5.2 资源加载函数
```mermaid
graph LR
A[load_mcp_resources] --> B{uris 参数?}
B -->|None| C[session.list_resources]
B -->|str| D[单个 URI]
B -->|list| E[多个 URI]
C --> F[获取 URI 列表]
D --> G[session.read_resource]
E --> G
F --> G
G --> H[convert_mcp_resource_to_langchain_blob]
H --> I[Blob 列表]
```
---
## 6. 会话与连接管理
### 6.1 连接类型支持
`MultiServerMCPClient` 支持多种传输协议连接 资料来源：[langchain_mcp_adapters/client.py]()：
| 传输类型 | 实现类 | 说明 |
|---------|--------|------|
| `stdio` | `StdioConnection` | 标准输入输出进程通信 |
| `sse` | `SSEConnection` | Server-Sent Events |
| `http` | `StreamableHttpConnection` | HTTP 流式传输 |
| `websocket` | `WebsocketConnection` | WebSocket 通信 |
### 6.2 会话创建流程
```mermaid
graph TD
A[MultiServerMCPClient.get_tools] --> B{连接类型}
B -->|Stdio| C[StdioServerParameters]
B -->|SSE| D[SSE 参数]
B -->|HTTP| E[HTTP 参数]
B -->|WebSocket| F[WebSocket 参数]
C --> G[stdio_client]
D --> H[mcp.client.sse]
E --> I[streamablehttp_client]
F --> J[websocket_client]
G --> K[ClientSession]
H --> K
I --> K
J --> K
K --> L[load_mcp_tools]
L --> M[LangChain Tools]
```
---
## 7. 完整数据流示例
### 7.1 工具调用完整流程
```mermaid
sequenceDiagram
participant User as 用户/Agent
participant LC as LangChain Tool
participant Adapter as langchain-mcp-adapters
participant MCP as MCP ClientSession
participant Server as MCP Server
User->>LC: 调用工具 (name, args)
LC->>Adapter: call_tool(request)
Adapter->>Adapter: 构建 MCPToolCallRequest
Adapter->>Adapter: 执行拦截器链
Adapter->>MCP: session.call_tool(name, args)
MCP->>Server: MCP 协议调用
Server-->>MCP: CallToolResult
MCP-->>Adapter: CallToolResult
Adapter->>Adapter: _convert_call_tool_result
Adapter-->>LC: (ConvertedToolResult, Artifact)
LC-->>User: ToolMessage / Command
```
### 7.2 资源访问完整流程
```mermaid
sequenceDiagram
participant User as 用户
participant Adapter as langchain-mcp-adapters
participant MCP as MCP ClientSession
participant Server as MCP Server
User->>Adapter: load_mcp_resources(session, uris)
Adapter->>MCP: session.list_resources()
MCP-->>Adapter: 资源列表
Adapter->>MCP: session.read_resource(uri)
MCP->>Server: 读取资源请求
Server-->>MCP: ResourceContents
MCP-->>Adapter: ResourceContents
Adapter->>Adapter: 转换内容块
Adapter-->>User: Blob 对象列表
```
---
## 8. 类型转换参考
### 8.1 导入的 MCP 类型
| 类型 | 来源 | 用途 |
|-----|------|-----|
| `Tool` | `mcp.types` | MCP 工具定义 |
| `CallToolResult` | `mcp.types` | 工具调用结果 |
| `ContentBlock` | `mcp.types` | 内容块基类 |
| `TextContent` | `mcp.types` | 文本内容 |
| `ImageContent` | `mcp.types` | 图片内容 |
| `AudioContent` | `mcp.types` | 音频内容 |
| `ResourceLink` | `mcp.types` | 资源链接 |
| `EmbeddedResource` | `mcp.types` | 嵌入式资源 |
### 8.2 导入的 LangChain 类型
| 类型 | 来源 | 用途 |
|-----|------|-----|
| `BaseTool` | `langchain_core.tools` | LangChain 工具基类 |
| `StructuredTool` | `langchain_core.tools` | 结构化工具 |
| `ToolMessage` | `langchain_core.messages` | 工具消息 |
| `ToolMessageContentBlock` | 自定义类型别名 | 内容块联合类型 |
| `Blob` | `langchain_core.documents.base` | LangChain Blob |
---
## 9. 错误处理
### 9.1 已实现的异常
| 异常类型 | 触发条件 | 处理位置 |
|---------|---------|---------|
| `ToolException` | 工具返回 `isError=True` | `_convert_call_tool_result()` |
| `NotImplementedError` | 遇到 `AudioContent` | `_convert_mcp_content_to_lc_block()` |
| `ValueError` | 未知内容类型 | `_convert_mcp_content_to_lc_block()` |
| `TypeError` | 不支持的资源类型 | `convert_mcp_resource_to_langchain_blob()` |
### 9.2 错误处理流程
```mermaid
graph TD
A[工具调用结果] --> B{isError?}
B -->|是| C[收集所有文本块]
C --> D[拼接错误消息]
D --> E[抛出 ToolException]
B -->|否| F{structuredContent?}
F -->|有| G[创建 MCPToolArtifact]
F -->|无| H[返回 None artifact]
G --> I[返回结果和 artifact]
H --> I
```
---
## 10. 关键配置参数
### 10.1 load_mcp_tools 参数说明
| 参数 | 类型 | 默认值 | 说明 |
|-----|------|--------|-----|
| `session` | `ClientSession` | 必填 | MCP 客户端会话 |
| `connection` | `Connection` | `None` | 连接配置（无 session 时使用） |
| `callbacks` | `Callbacks` | `None` | 回调处理器 |
| `tool_interceptors` | `list[ToolCallInterceptor]` | `None` | 拦截器列表 |
| `server_name` | `str` | `None` | 服务器名称 |
| `tool_name_prefix` | `bool` | `False` | 是否为工具名添加前缀 |
### 10.2 MultiServerMCPClient 配置结构
```python
client = MultiServerMCPClient({
"server_name": {
"transport": "stdio" | "sse" | "http" | "websocket",
"command": "python",
"args": ["/path/to/server.py"],
"env": {"KEY": "value"},
"url": "http://localhost:8000/mcp",
"headers": {"Authorization": "Bearer xxx"},
}
})
```
---
本页面涵盖了 langchain-mcp-adapters 库中数据流与转换机制的核心实现，理解这些机制有助于深度定制工具调用流程和构建复杂的多服务器 MCP 集成方案。
---
## MultiServerMCPClient 客户端
### 相关页面
相关主题：[传输协议类型](#page-transport-types), [工具转换机制](#page-tools-conversion)
相关源码文件
以下源码文件用于生成本页说明：
- [langchain_mcp_adapters/client.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/client.py)
- [langchain_mcp_adapters/tools.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/tools.py)
- [langchain_mcp_adapters/resources.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/resources.py)
- [langchain_mcp_adapters/sessions.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/sessions.py)
- [langchain_mcp_adapters/prompts.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/prompts.py)
- [langchain_mcp_adapters/callbacks.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/callbacks.py)
- [langchain_mcp_adapters/interceptors.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/interceptors.py)
- [README.md](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/README.md)
# MultiServerMCPClient 客户端
## 概述
`MultiServerMCPClient` 是 langchain-mcp-adapters 库的核心客户端类，用于连接多个 MCP（Model Context Protocol）服务器并从中加载 LangChain 兼容的工具、提示词和资源。
该客户端的主要作用是将 MCP 服务器提供的工具转换为 LangChain 工具格式，使其能够与 LangChain 和 LangGraph 代理无缝集成。`MultiServerMCPClient` 支持同时连接多个 MCP 服务器，并提供了灵活的配置选项来管理连接、回调和工具调用拦截。
资料来源：[langchain_mcp_adapters/client.py:40-50]()
## 架构设计
### 组件关系图
```mermaid
graph TD
Client[MultiServerMCPClient]
Conn[connections
dict[str, Connection]]
Callbacks[Callbacks]
Interceptors[tool_interceptors
list[ToolCallInterceptor]]
Prefix[tool_name_prefix
bool]
Client --> Conn
Client --> Callbacks
Client --> Interceptors
Client --> Prefix
Conn --> Stdio[StdioConnection]
Conn --> SSE[SSEConnection]
Conn --> HTTP[StreamableHttpConnection]
Conn --> WS[WebsocketConnection]
Client --> Tools[load_mcp_tools]
Client --> Resources[load_mcp_resources]
Client --> Prompts[load_mcp_prompt]
```
### 核心数据流
```mermaid
sequenceDiagram
participant User as 用户
participant Client as MultiServerMCPClient
participant Session as ClientSession
participant Interceptor as ToolCallInterceptor
participant Server as MCP Server
User->>Client: get_tools()
Client->>Client: create_session()
Client->>Session: 初始化连接
Client->>Session: list_tools()
Session->>Server: 请求工具列表
Server-->>Session: 返回工具定义
Session-->>Client: MCP Tools
Client->>Interceptor: 链式包装
Client-->>User: LangChain BaseTools
```
## 初始化参数
`MultiServerMCPClient` 的构造函数接受以下参数：
| 参数 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `connections` | `dict[str, Connection] \| None` | `None` | 服务器名称到连接配置的映射字典 |
| `callbacks` | `Callbacks \| None` | `None` | 处理通知和事件的回调函数 |
| `tool_interceptors` | `list[ToolCallInterceptor] \| None` | `None` | 工具调用拦截器列表 |
| `tool_name_prefix` | `bool` | `False` | 是否将工具名称前缀服务器名称 |
### connections 配置结构
```python
connections: dict[str, Connection] = {
"server_name": {
# 标准传输配置
"command": "python",
"args": ["/path/to/server.py"],
"transport": "stdio",
# 或 HTTP 传输配置
"url": "http://localhost:8000/mcp",
"transport": "http",
"headers": {"Authorization": "Bearer token"},
# 或其他连接选项...
}
}
```
资料来源：[langchain_mcp_adapters/client.py:50-90]()
## 连接类型
`sessions.py` 模块定义了多种连接类型，支持不同的 MCP 服务器部署方式：
| 连接类型 | 传输协议 | 说明 |
|----------|----------|------|
| `StdioConnection` | stdio | 通过标准输入/输出通信，适用于本地进程 |
| `SSEConnection` | SSE | Server-Sent Events，适用于长连接场景 |
| `StreamableHttpConnection` | HTTP | 可流式传输的 HTTP 连接 |
| `WebsocketConnection` | WebSocket | WebSocket 协议连接 |
### StdioConnection 配置
```python
{
"command": "python",
"args": ["/path/to/math_server.py"],
"transport": "stdio",
"env": {"KEY": "value"}, # 可选：传递给进程的环境变量
"cwd": "/working/directory", # 可选：工作目录
}
```
资料来源：[langchain_mcp_adapters/sessions.py]()
### HTTP/StreamableHttpConnection 配置
```python
{
"url": "http://localhost:8000/mcp",
"transport": "http", # 或 "streamable_http"
"headers": {
"Authorization": "Bearer YOUR_TOKEN",
"X-Custom-Header": "custom-value"
},
"timeout": 60.0, # HTTP 请求超时时间
}
```
> 仅 `sse` 和 `http` 传输支持运行时 headers
资料来源：[README.md]()
## 核心方法
### get_tools()
获取所有已连接 MCP 服务器的工具列表。
```python
async def get_tools(
self,
server_name: str | None = None
) -> list[BaseTool]
```
| 参数 | 类型 | 说明 |
|------|------|------|
| `server_name` | `str \| None` | 可选，指定服务器名称过滤工具 |
**返回值**: LangChain `BaseTool` 对象列表
资料来源：[langchain_mcp_adapters/client.py]()
### session()
建立与特定 MCP 服务器的会话连接。
```python
@asynccontextmanager
async def session(
self,
server_name: str
) -> AsyncIterator[ClientSession]
```
**用法示例**:
```python
client = MultiServerMCPClient({...})
async with client.session("math") as session:
tools = await load_mcp_tools(session)
```
资料来源：[langchain_mcp_adapters/client.py]()
## 使用示例
### 基础用法
```python
from langchain_mcp_adapters.client import MultiServerMCPClient
client = MultiServerMCPClient(
{
"math": {
"command": "python",
"args": ["/path/to/math_server.py"],
"transport": "stdio",
},
"weather": {
"url": "http://localhost:8000/mcp",
"transport": "http",
}
}
)
all_tools = await client.get_tools()
```
资料来源：[langchain_mcp_adapters/client.py:75-90]()
### 多服务器连接
```python
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
client = MultiServerMCPClient(
{
"math": {
"command": "python",
"args": ["/path/to/math_server.py"],
"transport": "stdio",
},
"weather": {
"url": "http://localhost:8000/mcp",
"transport": "http",
}
}
)
tools = await client.get_tools()
agent = create_agent("openai:gpt-4.1", tools)
response = await agent.ainvoke({"messages": "what's (3 + 5) x 12?"})
```
资料来源：[README.md]()
### 工具名称前缀
当多个服务器有相同名称的工具时，可使用 `tool_name_prefix` 避免冲突：
```python
client = MultiServerMCPClient(
{
"weather": {
"url": "http://localhost:8000/mcp",
"transport": "http",
}
},
tool_name_prefix=True # 工具名变为 "weather_search" 而非 "search"
)
tools = await client.get_tools()
```
资料来源：[langchain_mcp_adapters/client.py:61-65]()
## 与 LangGraph 集成
`MultiServerMCPClient` 可与 LangGraph 的 `StateGraph` 结合使用构建复杂代理：
```python
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition
from langchain.chat_models import init_chat_model
model = init_chat_model("openai:gpt-4.1")
client = MultiServerMCPClient(
{
"math": {
"command": "python",
"args": ["./examples/math_server.py"],
"transport": "stdio",
},
}
)
tools = await client.get_tools()
def call_model(state: MessagesState):
response = model.bind_tools(tools).invoke(state["messages"])
return {"messages": response}
builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "call_model")
builder.add_conditional_edges(
"call_model",
tools_condition,
)
builder.add_edge("tools", "call_model")
graph = builder.compile()
```
资料来源：[README.md]()
## 回调机制
`Callbacks` 类提供了处理 MCP 事件的接口：
```python
@dataclass
class Callbacks:
"""LangChain MCP 客户端回调"""
on_logging_message: LoggingMessageCallback | None = None
on_progress: ProgressCallback | None = None
on_elicitation: ElicitationCallback | None = None
```
| 回调类型 | 说明 |
|----------|------|
| `on_logging_message` | 日志消息回调 |
| `on_progress` | 进度通知回调 |
| `on_elicitation` | 追问/确认请求回调 |
### 回调使用示例
```python
from langchain_mcp_adapters.callbacks import Callbacks, CallbackContext
async def my_logging_callback(params, context: CallbackContext):
print(f"Log: {params}")
callbacks = Callbacks(on_logging_message=my_logging_callback)
client = MultiServerMCPClient(
{...},
callbacks=callbacks
)
```
资料来源：[langchain_mcp_adapters/callbacks.py]()
## 工具拦截器
`ToolCallInterceptor` 接口允许在工具调用前后进行拦截和处理：
```mermaid
graph TD
Request[MCPToolCallRequest] -->|洋葱模型| I1[Interceptor 1]
I1 --> I2[Interceptor 2]
I2 --> I3[Interceptor N]
I3 --> Handler[Base Handler]
Handler --> R1[Result 1]
R1 --> R2[Result 2]
R2 --> R3[Result N]
R3 --> Response[MCPToolCallResult]
```
### 拦截器链构建
```python
def _build_interceptor_chain(
base_handler: Callable[[MCPToolCallRequest], Awaitable[MCPToolCallResult]],
tool_interceptors: list[ToolCallInterceptor] | None,
) -> Callable[[MCPToolCallRequest], Awaitable[MCPToolCallResult]]
```
拦截器列表中的第一个拦截器成为最外层，以洋葱模式层层包裹。
### 使用拦截器
```python
from langchain_mcp_adapters.interceptors import (
MCPToolCallRequest,
MCPToolCallResult,
ToolCallInterceptor,
)
class LoggingInterceptor(ToolCallInterceptor):
async def intercept(
self, request: MCPToolCallRequest, handler
) -> MCPToolCallResult:
print(f"Calling tool: {request.name}")
result = await handler(request)
print(f"Tool result: {result}")
return result
client = MultiServerMCPClient(
{...},
tool_interceptors=[LoggingInterceptor()]
)
```
资料来源：[langchain_mcp_adapters/interceptors.py]()
资料来源：[langchain_mcp_adapters/tools.py:145-175]()
## 注意事项
### 上下文管理器使用限制
> [!warning]
> 从 langchain-mcp-adapters 0.1.0 版本开始，`MultiServerMCPClient` 不能作为异步上下文管理器使用。
**正确用法**:
```python
# 方法 1：直接获取工具
client = MultiServerMCPClient(...)
tools = await client.get_tools()
# 方法 2：显式创建会话
client = MultiServerMCPClient(...)
async with client.session(server_name) as session:
tools = await load_mcp_tools(session)
```
资料来源：[langchain_mcp_adapters/client.py:38-55]()
## 资源与提示词加载
除了工具，`MultiServerMCPClient` 还支持加载 MCP 资源（Blob）和提示词：
### 资源加载
```python
from langchain_mcp_adapters.resources import load_mcp_resources
async with client.session("server_name") as session:
blobs = await load_mcp_resources(session)
```
资料来源：[langchain_mcp_adapters/resources.py]()
### 提示词加载
```python
from langchain_mcp_adapters.prompts import load_mcp_prompt
async with client.session("server_name") as session:
messages = await load_mcp_prompt(session, "prompt_name", arguments)
```
资料来源：[langchain_mcp_adapters/prompts.py]()
## 配置参考
### 完整配置示例
```python
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.callbacks import Callbacks
client = MultiServerMCPClient(
connections={
"math": {
"command": "python",
"args": ["/path/to/math_server.py"],
"transport": "stdio",
"env": {"DEBUG": "true"},
},
"weather": {
"url": "http://localhost:8000/mcp",
"transport": "http",
"headers": {"Authorization": "Bearer token"},
},
},
callbacks=Callbacks(
on_logging_message=my_logging_callback,
on_progress=my_progress_callback,
),
tool_interceptors=[LoggingInterceptor()],
tool_name_prefix=True,
)
```
## 总结
`MultiServerMCPClient` 是连接 MCP 服务器与 LangChain 生态系统的桥梁，提供：
- 多服务器并发连接管理
- 统一的工具转换机制
- 灵活的回调和拦截器支持
- 多种传输协议兼容（stdio、HTTP、SSE、WebSocket）
- 与 LangGraph 的深度集成
---
## 工具转换机制
### 相关页面
相关主题：[MultiServerMCPClient 客户端](#page-multiserver-client), [数据流与转换机制](#page-data-flow), [工具调用拦截器](#page-interceptors)
相关源码文件
以下源码文件用于生成本页说明：
- [langchain_mcp_adapters/tools.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/tools.py)
- [langchain_mcp_adapters/resources.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/resources.py)
- [langchain_mcp_adapters/client.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/client.py)
- [langchain_mcp_adapters/interceptors.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/interceptors.py)
- [README.md](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/README.md)
# 工具转换机制
## 概述
工具转换机制是 langchain-mcp-adapters 库的核心功能，负责将 MCP（Model Context Protocol）工具转换为 LangChain 兼容的工具格式。该机制使得 MCP 服务器提供的工具能够无缝集成到 LangChain 和 LangGraph 智能体应用中。
主要职责包括：
- 将 MCP 工具声明转换为 LangChain StructuredTool
- 处理工具调用请求和响应的格式转换
- 支持内容块的类型转换（文本、图像、音频等）
- 提供拦截器链模式用于工具调用的扩展处理
资料来源：[tools.py:1-30]()
## 核心类型定义
### 内容块类型
工具转换机制定义了一系列类型别名用于处理不同格式的内容：
```python
# Type alias for LangChain content blocks used in ToolMessage
ToolMessageContentBlock = TextContentBlock | ImageContentBlock | FileContentBlock
```
### 转换结果类型
根据 langgraph 是否安装，返回类型有所不同：
```python
if LANGGRAPH_PRESENT:
ConvertedToolResult = list[ToolMessageContentBlock] | ToolMessage | Command
else:
ConvertedToolResult = list[ToolMessageContentBlock] | ToolMessage
```
### 工具调用产物
```python
class MCPToolArtifact(TypedDict):
"""Artifact returned from MCP tool calls."""
structured_content: dict[str, Any]
```
MCPToolArtifact 是一个 TypedDict，用于包装 MCP 工具调用返回的结构化内容，允许未来扩展更多字段。
资料来源：[tools.py:44-68]()
## 核心转换函数
### 内容块转换
`_convert_mcp_content_to_lc_block()` 函数负责将 MCP 内容块转换为 LangChain 内容块格式：
```mermaid
graph TD
A[MCP ContentBlock] --> B{内容类型判断}
B -->|TextContent| C[create_text_block]
B -->|ImageContent| D[create_image_block]
B -->|AudioContent| E[NotImplementedError]
B -->|ResourceLink| F{是否为图像类型}
B -->|EmbeddedResource| G{资源类型判断}
F -->|是| H[create_image_block]
F -->|否| I[create_file_block]
G -->|TextResourceContents| J[create_text_block]
G -->|BlobResourceContents| K{是否为图像}
K -->|是| L[create_image_block]
K -->|否| M[create_file_block]
```
支持的转换映射关系：
| MCP 内容类型 | LangChain 内容块 | 说明 |
|-------------|-----------------|------|
| TextContent | {"type": "text", "text": ...} | 文本内容直接转换 |
| ImageContent | {"type": "image", "base64": ..., "mime_type": ...} | 图像内容带 MIME 类型 |
| ResourceLink (image/*) | {"type": "image", "url": ..., "mime_type": ...} | 图像资源链接 |
| ResourceLink (other) | {"type": "file", "url": ..., "mime_type": ...} | 非图像资源链接 |
| EmbeddedResource (text) | {"type": "text", "text": ...} | 内嵌文本资源 |
| EmbeddedResource (blob) | {"type": "image/file", ...} | 内嵌二进制资源 |
| AudioContent | 抛出 NotImplementedError | 暂不支持 |
资料来源：[tools.py:71-133]()
### 工具调用结果转换
`_convert_call_tool_result()` 函数将 MCP 的 CallToolResult 转换为 LangChain 格式：
```python
def _convert_call_tool_result(
call_tool_result: MCPToolCallResult,
) -> tuple[ConvertedToolResult, MCPToolArtifact | None]:
```
处理逻辑：
1. 如果返回的是 ToolMessage，直接返回并设置 artifact 为 None
2. 如果返回的是 Command（LangGraph），直接返回并设置 artifact 为 None
3. 将所有 MCP 内容块转换为 LangChain 内容块
4. 如果工具调用出错，抛出 ToolException
5. 如果存在 structuredContent，包装为 MCPToolArtifact
资料来源：[tools.py:166-216]()
## 工具加载与转换
### 加载所有 MCP 工具
`load_mcp_tools()` 函数从 MCP 会话加载所有可用工具：
```python
async def load_mcp_tools(
session: ClientSession | None,
*,
connection: Connection | None = None,
callbacks: Callbacks | None = None,
tool_interceptors: list[ToolCallInterceptor] | None = None,
server_name: str | None = None,
tool_name_prefix: bool = False,
) -> list[BaseTool]:
```
参数说明：
| 参数 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| session | ClientSession | 必填 | MCP 客户端会话 |
| connection | Connection | None | 用于创建新会话的连接配置 |
| callbacks | Callbacks | None | 处理通知和事件的回调 |
| tool_interceptors | list[ToolCallInterceptor] | None | 工具调用拦截器列表 |
| server_name | str | None | 服务器名称 |
| tool_name_prefix | bool | False | 是否将工具名添加服务器前缀 |
该函数支持分页加载工具，处理最大 1000 次迭代的循环：
```python
MAX_ITERATIONS = 1000
```
资料来源：[tools.py:240-295]()
### 单个工具转换
`convert_mcp_tool_to_langchain_tool()` 函数将单个 MCP 工具转换为 LangChain StructuredTool：
```python
def convert_mcp_tool_to_langchain_tool(
session: ClientSession | None,
tool: MCPTool,
*,
connection: Connection | None = None,
callbacks: Callbacks | None = None,
tool_interceptors: list[ToolCallInterceptor] | None = None,
server_name: str | None = None,
tool_name_prefix: bool = False,
) -> BaseTool:
```
转换后的工具配置：
| 属性 | 来源 | 说明 |
|------|------|------|
| name | tool.name 或 f"{server_name}_{tool.name}" | 工具名称 |
| description | tool.description | 工具描述 |
| args_schema | tool.inputSchema | 参数模式定义 |
| coroutine | call_tool | 异步执行函数 |
| response_format | "content_and_artifact" | 响应格式 |
| metadata | tool.annotations.model_dump() | 元数据信息 |
资料来源：[tools.py:297-370]()
## 工具调用执行机制
### 拦截器链模式
工具调用采用洋葱模式（Onion Pattern）构建拦截器链：
```mermaid
graph LR
A[请求] --> B[拦截器 N]
B --> C[拦截器 N-1]
C --> D[...]
D --> E[拦截器 1]
E --> F[实际工具执行]
F --> G[拦截器 1]
G --> H[...]
H --> I[拦截器 N-1]
I --> J[响应]
```
`_build_interceptor_chain()` 函数实现该模式：
```python
def _build_interceptor_chain(
base_handler: Callable[[MCPToolCallRequest], Awaitable[MCPToolCallResult]],
tool_interceptors: list[ToolCallInterceptor] | None,
) -> Callable[[MCPToolCallRequest], Awaitable[MCPToolCallResult]]:
"""Build composed handler chain with interceptors in onion pattern."""
handler = base_handler
if tool_interceptors:
for interceptor in reversed(tool_interceptors):
current_handler = handler
# 包装逻辑
return handler
```
资料来源：[tools.py:136-160](), [interceptors.py:1-45]()
### 工具调用请求
MCPToolCallRequest 是拦截器系统中定义的请求类型：
```python
@dataclass
class MCPToolCallRequest:
"""Tool execution request passed to MCP tool call interceptors."""
name: str # 工具名称
args: dict[str, Any] # 工具参数
server_name: str # 服务器名称
headers: dict[str, Any] | None # HTTP 请求头
runtime: object | None # LangGraph 运行时
```
### 工具调用响应
MCPToolCallResult 的类型定义：
```python
if LANGGRAPH_PRESENT:
MCPToolCallResult = CallToolResult | ToolMessage | Command
else:
MCPToolCallResult = CallToolResult | ToolMessage
```
资料来源：[interceptors.py:40-55]()
## 错误处理机制
工具转换机制实现了多层错误处理：
### 工具调用错误
当 MCP 工具调用返回错误时：
```python
if call_tool_result.isError:
error_parts = []
for item in tool_content:
if isinstance(item, str):
error_parts.append(item)
elif isinstance(item, dict) and item.get("type") == "text":
error_parts.append(item.get("text", ""))
error_msg = "\n".join(error_parts) if error_parts else str(tool_content)
raise ToolException(error_msg)
```
### 内容类型错误
不支持的内容类型会抛出相应异常：
| 内容类型 | 异常类型 | 错误信息 |
|---------|---------|---------|
| AudioContent | NotImplementedError | "AudioContent conversion to LangChain content blocks is not yet supported" |
| 未知内容类型 | ValueError | "Unknown MCP content type: {type}" |
| 未知嵌入资源 | ValueError | "Unknown embedded resource type: {type}" |
资料来源：[tools.py:88-95](), [tools.py:125-133]()
## 使用示例
### 基础用法
```python
from mcp import ClientSession
from langchain_mcp_adapters.tools import load_mcp_tools
# 通过现有会话加载工具
tools = await load_mcp_tools(session)
```
### 完整客户端示例
```python
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
client = MultiServerMCPClient(
{
"math": {
"command": "python",
"args": ["/path/to/math_server.py"],
"transport": "stdio",
},
"weather": {
"url": "http://localhost:8000/mcp",
"transport": "http",
}
}
)
tools = await client.get_tools()
```
### 带拦截器的用法
```python
from langchain_mcp_adapters.interceptors import ToolCallInterceptor
from langchain_mcp_adapters.tools import load_mcp_tools
class LoggingInterceptor(ToolCallInterceptor):
async def intercept(
self,
request: MCPToolCallRequest,
handler: Callable,
) -> MCPToolCallResult:
print(f"Calling tool: {request.name}")
result = await handler(request)
print(f"Tool result: {result}")
return result
tools = await load_mcp_tools(
session,
tool_interceptors=[LoggingInterceptor()]
)
```
资料来源：[README.md:1-100]()
## 架构总结
```mermaid
graph TD
subgraph "MCP 服务器层"
A[MCP Tools]
B[MCP Resources]
C[MCP Prompts]
end
subgraph "转换适配层"
D[load_mcp_tools]
E[load_mcp_resources]
F[load_mcp_prompt]
end
subgraph "拦截器层"
G[ToolCallInterceptor]
H[MCPToolCallRequest]
I[MCPToolCallResult]
end
subgraph "LangChain/LangGraph 层"
J[StructuredTool]
K[BaseTool]
L[ToolNode]
end
A --> D
D --> G
G --> H
H --> I
I --> J
J --> K
J --> L
```
工具转换机制作为 MCP 与 LangChain 之间的桥梁，通过标准化的转换接口和灵活的拦截器模式，实现了两种协议生态的无缝集成。
---
## 传输协议类型
### 相关页面
相关主题：[MultiServerMCPClient 客户端](#page-multiserver-client)
相关源码文件
以下源码文件用于生成本页说明：
- [langchain_mcp_adapters/sessions.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/sessions.py)
- [langchain_mcp_adapters/client.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/client.py)
- [langchain_mcp_adapters/tools.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/tools.py)
- [examples/servers/streamable-http-stateless/server.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/examples/servers/streamable-http-stateless/server.py)
- [README.md](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/README.md)
# 传输协议类型
## 概述
`langchain-mcp-adapters` 库支持多种 MCP (Model Context Protocol) 传输协议，用于在不同场景下连接 MCP 服务器与 LangChain 应用。每种传输协议都有其特定的适用环境和配置方式，库通过统一的抽象接口隐藏了底层差异，使得开发者可以灵活选择最适合的传输方式。
传输协议在架构中扮演着客户端与服务器之间通信桥梁的角色，负责建立连接、传输请求和响应。库支持 stdio、HTTP、SSE (Server-Sent Events)、Streamable HTTP 和 WebSocket 五种传输类型。
资料来源：[langchain_mcp_adapters/sessions.py:1-50]()
## 传输协议架构
```mermaid
graph TD
A[MultiServerMCPClient] --> B[Connection 抽象层]
B --> C[StdioConnection]
B --> D[McpHttpClientFactory]
B --> E[SSEConnection]
B --> F[StreamableHttpConnection]
B --> G[WebsocketConnection]
C --> H[stdio transport]
D --> I[http transport]
E --> J[sse transport]
F --> K[streamable-http transport]
G --> L[websocket transport]
```
## 支持的传输协议类型
### 1. stdio 传输
**适用场景**：本地进程通信，适用于同一机器上的 MCP 服务器。
stdio 传输通过标准输入/输出流进行进程间通信，是最简单的连接方式。当 MCP 服务器作为子进程运行时，使用此传输协议。
```python
from langchain_mcp_adapters.client import MultiServerMCPClient
client = MultiServerMCPClient(
{
"math": {
"command": "python",
"args": ["/path/to/math_server.py"],
"transport": "stdio",
}
}
)
```
**配置参数**：
| 参数 | 类型 | 必需 | 说明 |
|------|------|------|------|
| `command` | `str` | 是 | 要执行的命令 |
| `args` | `list[str]` | 是 | 命令行参数列表 |
| `env` | `dict[str, str]` | 否 | 环境变量 |
| `cwd` | `str` | 否 | 工作目录 |
| `encoding` | `str` | 否 | 字符编码 |
资料来源：[README.md:85-95]()
### 2. HTTP 传输
**适用场景**：远程服务器通信，支持跨网络访问 MCP 服务。
HTTP 传输基于传统的请求/响应模式，适用于可以发起 HTTP 请求的环境。客户端通过 RESTful API 与 MCP 服务器通信。
```python
client = MultiServerMCPClient(
{
"weather": {
"url": "http://localhost:8000/mcp",
"transport": "http",
}
}
)
```
**配置参数**：
| 参数 | 类型 | 必需 | 说明 |
|------|------|------|------|
| `url` | `str` | 是 | MCP 服务器的 HTTP 端点 URL |
| `headers` | `dict[str, str]` | 否 | HTTP 请求头 |
资料来源：[langchain_mcp_adapters/client.py:60-75]()
### 3. SSE 传输
**适用场景**：需要服务器推送更新的场景，支持长连接。
SSE (Server-Sent Events) 传输允许服务器主动向客户端推送事件，实现单向实时通信。
```python
from langchain_mcp_adapters.sessions import SSEConnection
connection = SSEConnection(
url="http://localhost:8000/sse",
headers={"Authorization": "Bearer token"}
)
```
**配置参数**：
| 参数 | 类型 | 必需 | 说明 |
|------|------|------|------|
| `url` | `str` | 是 | SSE 端点 URL |
| `headers` | `dict[str, Any]` | 否 | 请求头 |
| `timeout` | `float` | 否 | 超时时间，默认 5.0 秒 |
资料来源：[langchain_mcp_adapters/sessions.py:180-200]()
### 4. Streamable HTTP 传输
**适用场景**：现代 MCP 应用，需要支持流式响应和会话管理。
Streamable HTTP 是 MCP 规范推荐的传输方式，支持流式响应和会话状态管理。它结合了 HTTP 的请求/响应模式和流式传输的能力。
**启动服务器示例**：
```bash
cd examples/servers/streamable-http-stateless/
uv run mcp-simple-streamablehttp-stateless --port 3000
```
**客户端使用方式**：
```python
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
from langchain_mcp_adapters.tools import load_mcp_tools
async with streamablehttp_client("http://localhost:3000/mcp") as (read, write, _):
async with ClientSession(read, write) as session:
await session.initialize()
tools = await load_mcp_tools(session)
```
资料来源：[examples/servers/streamable-http-stateless/server.py:1-30]()
**服务器端实现示例**：
```python
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("StreamableHTTP")
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers"""
return a + b
if __name__ == "__main__":
mcp.run(transport="http")
```
**配置参数**：
| 参数 | 类型 | 必需 | 说明 |
|------|------|------|------|
| `url` | `str` | 是 | Streamable HTTP 端点 URL |
| `headers` | `dict[str, Any]` | 否 | 请求头 |
| `timeout` | `float` | 否 | 超时时间 |
> 注意：SSE 和 HTTP 传输支持运行时 headers，这些 headers 会随每个 HTTP 请求一起发送到 MCP 服务器。
资料来源：[README.md:45-60]()
### 5. WebSocket 传输
**适用场景**：需要双向实时通信的低延迟应用。
WebSocket 传输提供全双工通信通道，适用于需要频繁双向交互的场景。
**配置参数**：
| 参数 | 类型 | 必需 | 说明 |
|------|------|------|------|
| `url` | `str` | 是 | WebSocket 服务器 URL |
| `headers` | `dict[str, Any]` | 否 | 连接头 |
资料来源：[langchain_mcp_adapters/sessions.py:220-240]()
## 连接类型映射表
| 传输协议 | Connection 类 | 内部传输类型 | 支持 Headers |
|---------|--------------|-------------|-------------|
| stdio | `StdioConnection` | `stdio` | ❌ |
| http | `McpHttpClientFactory` | `http` | ✅ |
| sse | `SSEConnection` | `sse` | ✅ |
| streamable-http | `StreamableHttpConnection` | `http` | ✅ |
| websocket | `WebsocketConnection` | `websocket` | ✅ |
资料来源：[langchain_mcp_adapters/sessions.py:50-150]()
## 多服务器传输配置
`MultiServerMCPClient` 支持同时连接多个使用不同传输协议的 MCP 服务器：
```python
from langchain_mcp_adapters.client import MultiServerMCPClient
client = MultiServerMCPClient(
{
"math": {
"command": "python",
"args": ["/path/to/math_server.py"],
"transport": "stdio",
},
"weather": {
"url": "http://localhost:8000/mcp",
"transport": "http",
},
"realtime": {
"url": "ws://localhost:9000/mcp",
"transport": "websocket",
}
}
)
tools = await client.get_tools()
```
资料来源：[README.md:95-115]()
## 传输协议选择指南
```mermaid
graph TD
A[选择传输协议] --> B{通信场景}
B --> C{本地进程}
C -->|是| D[stdio]
C -->|否| E{需要远程访问}
E -->|是| F{需要流式响应}
F -->|是| G[streamable-http]
F -->|否| H{需要实时推送}
H -->|是| I[SSE]
H -->|否| J{需要双向通信}
J -->|是| K[WebSocket]
J -->|否| L[HTTP]
```
**选择建议**：
- **本地开发测试**：使用 `stdio`，配置简单，无需启动独立服务器
- **生产环境部署**：使用 `streamable-http`，符合 MCP 规范推荐
- **需要实时更新**：使用 `SSE` 或 `WebSocket`
- **跨网络访问**：使用 `HTTP` 或 `streamable-http`
## 运行时 Headers 配置
对于支持 headers 的传输协议（HTTP、SSE、streamable-http），可以在配置中指定自定义请求头：
```python
client = MultiServerMCPClient(
{
"secure_api": {
"url": "http://localhost:8000/mcp",
"transport": "http",
"headers": {
"Authorization": "Bearer custom-value",
"X-Custom-Header": "custom-value"
},
}
}
)
```
> 仅 `sse` 和 `http` 传输支持运行时 headers。这些 headers 会随每个 HTTP 请求传递到 MCP 服务器。
资料来源：[README.md:35-50]()
## 会话管理
### 默认会话行为
默认情况下，每次工具调用都会启动一个新的 `ClientSession`。这简化了使用方式，但在高频调用场景下可能影响性能。
```python
tools = await client.get_tools()
response = await agent.ainvoke({"messages": "what's (3 + 5) x 12?"})
```
资料来源：[README.md:110-120]()
### 显式会话管理
对于需要显式控制会话生命周期的场景，可以使用上下文管理器：
```python
from langchain_mcp_adapters.tools import load_mcp_tools
client = MultiServerMCPClient({...})
async with client.session("math") as session:
tools = await load_mcp_tools(session)
# 在此上下文中复用同一会话
```
```python
async with client.session(server_name) as session:
resources = await load_mcp_resources(session, uris=uris)
```
资料来源：[README.md:120-130]()
## 工具名称前缀
当多个服务器具有相同名称的工具时，可以启用工具名称前缀来避免冲突：
```python
client = MultiServerMCPClient(
{
"math": {
"command": "python",
"args": ["/path/to/math_server.py"],
"transport": "stdio",
},
"weather": {
"url": "http://localhost:8000/mcp",
"transport": "http",
}
},
tool_name_prefix=True # 启用前缀
)
tools = await client.get_tools()
# 工具名称将变为 "math_add", "weather_search" 等
```
| 参数 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `tool_name_prefix` | `bool` | `False` | 是否在工具名称前添加服务器名称前缀 |
资料来源：[langchain_mcp_adapters/client.py:45-55]()
## 环境变量扩展
在 stdio 传输中，支持环境变量的展开功能：
```python
client = MultiServerMCPClient(
{
"server": {
"command": "python",
"args": ["${SCRIPT_PATH}"], # 将被展开
"transport": "stdio",
"env": {
"API_KEY": "${API_KEY}" # 引用环境变量
}
}
}
)
```
> 只有值（而非键）会被展开；`${command}` 和 `${args}` 保持不变。如果未指定 `env`，则会继承当前环境变量的一个子集。
资料来源：[langchain_mcp_adapters/sessions.py:100-130]()
## 错误处理
不同传输协议的错误处理遵循统一的模式：
```python
try:
async with client.session("server_name") as session:
tools = await load_mcp_tools(session)
except ConnectionError:
# 处理连接错误
pass
except ValueError as e:
# 处理配置错误
pass
```
工具调用错误会被包装为 `ToolException`：
```python
from langchain_core.tools import ToolException
try:
result = await tool.ainvoke({"arg": "value"})
except ToolException as e:
# 处理工具执行错误
pass
```
资料来源：[langchain_mcp_adapters/tools.py:80-120]()
---
## 工具调用拦截器
### 相关页面
相关主题：[回调机制](#page-callbacks), [工具转换机制](#page-tools-conversion)
相关源码文件
以下源码文件用于生成本页说明：
- [langchain_mcp_adapters/interceptors.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/interceptors.py)
- [langchain_mcp_adapters/tools.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/tools.py)
# 工具调用拦截器
## 概述
工具调用拦截器（Tool Call Interceptor）是 langchain-mcp-adapters 框架中用于拦截和控制 MCP 工具调用生命周期的核心机制。它采用**洋葱模式（Onion Pattern）**构建可组合的处理器链，允许开发者在工具调用的请求和响应阶段注入自定义逻辑。
拦截器的主要作用包括：
- 在工具执行前后添加日志记录和监控
- 修改工具调用的参数或请求头
- 验证或转换工具返回结果
- 实现重试逻辑和错误处理
- 支持 LangGraph Command 的特殊返回类型处理
资料来源：[langchain_mcp_adapters/interceptors.py:1-10]()
## 架构设计
### 洋葱模式处理器链
拦截器通过洋葱模式组织，形成多层嵌套的处理器链。当工具调用时，请求从外到内逐层传递，执行完成后结果再从内到外逐层返回。
```mermaid
graph TD
subgraph 拦截器链
A["拦截器 1 (最外层)"] --> B["拦截器 2"]
B --> C["拦截器 N"]
C --> D["实际工具执行器"]
end
E["MCPToolCallRequest"] --> A
D --> F["MCPToolCallResult"]
style A fill:#e1f5fe
style D fill:#c8e6c9
style F fill:#fff3e0
```
### 核心类型体系
| 类型 | 说明 | 资料来源 |
|------|------|----------|
| `MCPToolCallRequest` | 工具调用请求的数据类，包含名称、参数、服务器名、请求头等信息 | [interceptors.py:38-56]() |
| `MCPToolCallResult` | 工具调用结果的类型别名，支持 `CallToolResult`、`ToolMessage` 或 `Command` | [interceptors.py:28-33]() |
| `ToolCallInterceptor` | 拦截器接口协议，定义异步调用方法 | [interceptors.py:70-75]() |
| `_MCPToolCallRequestOverrides` | 可覆盖的请求字段类型定义 | [interceptors.py:35-37]() |
## MCPToolCallRequest 数据结构
`MCPToolCallRequest` 是拦截器接收到的工具调用请求对象，采用 dataclass 模式定义：
```python
@dataclass
class MCPToolCallRequest:
"""工具执行请求的请求对象"""
name: str # 工具名称
args: dict[str, Any] # 工具参数
server_name: str # 服务器名称
headers: dict[str, Any] | None # HTTP 请求头
runtime: object | None # LangGraph 运行时对象
```
### 请求覆盖机制
通过 `override()` 方法可以创建修改后的请求副本：
```python
def override(
self, **overrides: Unpack[_MCPToolCallRequestOverrides]
) -> MCPToolCallRequest:
"""创建带有指定覆盖字段的新请求实例"""
```
支持覆盖的字段：
| 字段名 | 类型 | 说明 |
|--------|------|------|
| `name` | `str` | 工具名称 |
| `args` | `dict[str, Any]` | 工具参数 |
| `headers` | `dict[str, Any] \| None` | HTTP 请求头 |
资料来源：[interceptors.py:56-66]()
## ToolCallInterceptor 接口
拦截器必须实现 `ToolCallInterceptor` 协议：
```python
@runtime_checkable
class ToolCallInterceptor(Protocol):
"""拦截器接口协议"""
async def __call__(
self,
request: MCPToolCallRequest,
handler: Callable[[MCPToolCallRequest], Awaitable[MCPToolCallResult]],
) -> MCPToolCallResult:
"""拦截并处理工具调用"""
...
```
### 接口参数说明
| 参数 | 类型 | 说明 |
|------|------|------|
| `request` | `MCPToolCallRequest` | 工具调用请求对象 |
| `handler` | `Callable[[MCPToolCallRequest], Awaitable[MCPToolCallResult]]` | 下一个处理器（调用链中的下一个拦截器或实际执行器） |
### 返回值
返回类型为 `MCPToolCallResult`，支持以下三种格式：
| 类型 | 说明 | 使用场景 |
|------|------|----------|
| `CallToolResult` | MCP 协议标准结果 | 标准工具调用返回 |
| `ToolMessage` | LangChain 消息格式 | LangChain 集成 |
| `Command` | LangGraph 命令 | LangGraph 工作流控制 |
资料来源：[interceptors.py:28-33]()
## 拦截器链构建
拦截器链通过 `_build_interceptor_chain()` 函数构建，采用反向迭代方式将拦截器层层包装：
```python
def _build_interceptor_chain(
base_handler: Callable[[MCPToolCallRequest], Awaitable[MCPToolCallResult]],
tool_interceptors: list[ToolCallInterceptor] | None,
) -> Callable[[MCPToolCallRequest], Awaitable[MCPToolCallResult]]:
"""构建带有拦截器的洋葱模式处理器链"""
handler = base_handler
if tool_interceptors:
for interceptor in reversed(tool_interceptors):
current_handler = handler
async def wrapped(request, next_handler=current_handler):
return await interceptor(request, next_handler)
handler = wrapped
return handler
```
### 构建流程说明
1. **基础处理器** (`base_handler`)：最内层的实际工具执行函数
2. **反向迭代**：从列表最后一个拦截器开始向前包装
3. **闭包捕获**：每个拦截器通过闭包持有对下一个处理器的引用
4. **返回组合处理器**：最终返回最外层的处理器
资料来源：[langchain_mcp_adapters/tools.py:75-95]()
## 拦截器在工具加载中的应用
### load_mcp_tools 函数集成
`load_mcp_tools()` 函数接受 `tool_interceptors` 参数，将拦截器链注入到工具调用流程中：
```python
async def load_mcp_tools(
session: ClientSession | None,
*,
connection: Connection | None = None,
callbacks: Callbacks | None = None,
tool_interceptors: list[ToolCallInterceptor] | None = None, # 拦截器列表
server_name: str | None = None,
tool_name_prefix: bool = False,
) -> list[BaseTool]:
```
### 工具调用执行流程
```mermaid
sequenceDiagram
participant 调用者
participant 工具 as LangChain 工具
participant 拦截器链 as 拦截器链
participant 执行器 as 实际执行器
participant MCP as MCP Session
调用者->>工具: 调用工具函数
工具->>拦截器链: MCPToolCallRequest
拦截器链->>执行器: 传递请求
执行器->>MCP: session.call_tool()
MCP-->>执行器: CallToolResult
执行器-->>拦截器链: MCPToolCallResult
拦截器链-->>工具: MCPToolCallResult
工具-->>调用者: (content, artifact)
```
### 拦截器链的执行环境
在工具的 `call_tool` 内部函数中，拦截器链被创建和应用：
```python
async def call_tool(
runtime: Annotated[object | None, InjectedToolArg()] = None,
**arguments: dict[str, Any],
) -> tuple[ConvertedToolResult, MCPToolArtifact | None]:
# ... 参数验证和回调处理 ...
# 创建最内层处理器 - 实际执行 MCP 工具调用
async def execute_tool(request: MCPToolCallRequest) -> MCPToolCallResult:
return await session.call_tool(
tool_name,
tool_args,
progress_callback=mcp_callbacks.progress_callback,
)
# 构建并执行拦截器链
handler = _build_interceptor_chain(execute_tool, tool_interceptors)
request = MCPToolCallRequest(
name=tool.name,
args=arguments,
server_name=server_name or "unknown",
headers=None,
runtime=runtime,
)
call_tool_result = await handler(request)
return _convert_call_tool_result(call_tool_result)
```
资料来源：[langchain_mcp_adapters/tools.py:98-130]()
## 使用示例
### 基本拦截器实现
```python
from langchain_mcp_adapters.interceptors import (
MCPToolCallRequest,
MCPToolCallResult,
ToolCallInterceptor,
)
class LoggingInterceptor:
"""日志记录拦截器"""
async def __call__(
self,
request: MCPToolCallRequest,
handler: Callable[[MCPToolCallRequest], Awaitable[MCPToolCallResult]],
) -> MCPToolCallResult:
print(f"调用工具: {request.name}, 参数: {request.args}")
result = await handler(request)
print(f"工具 {request.name} 执行完成")
return result
class ValidationInterceptor:
"""参数验证拦截器"""
async def __call__(
self,
request: MCPToolCallRequest,
handler: Callable[[MCPToolCallRequest], Awaitable[MCPToolCallResult]],
) -> MCPToolCallResult:
# 修改请求参数
if request.name == "search" and "query" in request.args:
modified_request = request.override(
args={**request.args, "query": request.args["query"].strip()}
)
return await handler(modified_request)
return await handler(request)
```
### 多拦截器组合使用
```python
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.tools import load_mcp_tools
client = MultiServerMCPClient(
{
"math": {
"command": "python",
"args": ["/path/to/math_server.py"],
"transport": "stdio",
},
}
)
# 按照配置顺序，最前面的拦截器成为最外层
interceptors = [
LoggingInterceptor(), # 最外层 - 首先执行
ValidationInterceptor(), # 次外层 - 第二执行
]
tools = await client.get_tools(tool_interceptors=interceptors)
```
### 拦截器在 LangGraph 中的使用
```python
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.graph import StateGraph, MessagesState
from langgraph.prebuilt import ToolNode
client = MultiServerMCPClient({...})
tools = await client.get_tools(
tool_interceptors=[LoggingInterceptor()],
server_name="math"
)
# 拦截器可以返回 Command 对象控制 LangGraph 流程
class ConditionalInterceptor:
async def __call__(
self,
request: MCPToolCallRequest,
handler: Callable,
) -> MCPToolCallResult:
result = await handler(request)
# 根据结果决定是否路由到其他节点
return result
```
## MultiServerMCPClient 中的拦截器支持
`MultiServerMCPClient` 类在初始化时接受全局拦截器配置：
```python
class MultiServerMCPClient:
def __init__(
self,
connections: dict[str, dict[str, Any]] | None = None,
*,
callbacks: Callbacks | None = None,
tool_interceptors: list[ToolCallInterceptor] | None = None, # 全局拦截器
tool_name_prefix: bool = False,
) -> None:
self.tool_interceptors = tool_interceptors or []
```
### 配置参数说明
| 参数 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `connections` | `dict[str, dict[str, Any]]` | `None` | 服务器连接配置映射 |
| `callbacks` | `Callbacks \| None` | `None` | 回调处理器 |
| `tool_interceptors` | `list[ToolCallInterceptor]` | `[]` | 全局拦截器列表 |
| `tool_name_prefix` | `bool` | `False` | 是否为工具名添加服务器前缀 |
资料来源：[langchain_mcp_adapters/client.py:50-60]()
## 最佳实践
### 1. 拦截器顺序
拦截器的执行顺序与其在列表中的顺序一致：
- **列表首位** → **最外层** → **最先收到请求，最后收到响应**
- **列表末位** → **最内层** → **最后收到请求，最先收到响应**
### 2. 错误处理
建议在最外层拦截器中实现统一的错误处理：
```python
class ErrorHandlingInterceptor:
async def __call__(
self,
request: MCPToolCallRequest,
handler: Callable,
) -> MCPToolCallResult:
try:
return await handler(request)
except Exception as e:
# 统一错误处理和日志记录
return CallToolResult(
content=[TextContent(text=f"Error: {str(e)}")],
isError=True,
)
```
### 3. 请求修改
使用 `override()` 方法进行不可变请求修改：
```python
# 正确：创建新实例
modified = request.override(args={"query": "new value"})
# 错误：直接修改（不应该这样做）
request.args["query"] = "new value"
```
### 4. 性能考量
- 避免在拦截器中执行耗时的同步操作
- 使用异步处理保持非阻塞执行
- 对于高频调用，考虑拦截器的性能开销
---
## 回调机制
### 相关页面
相关主题：[工具调用拦截器](#page-interceptors), [MultiServerMCPClient 客户端](#page-multiserver-client)
相关源码文件
以下源码文件用于生成本页说明：
- [langchain_mcp_adapters/tools.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/tools.py)
- [langchain_mcp_adapters/client.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/client.py)
- [langchain_mcp_adapters/interceptors.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/interceptors.py)
- [langchain_mcp_adapters/sessions.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/sessions.py)
# 回调机制
## 概述
回调机制是 langchain-mcp-adapters 中用于处理 MCP 工具调用生命周期事件的核心组件。该机制允许开发者在工具执行的不同阶段插入自定义逻辑，实现进度监控、事件通知、错误处理等高级功能。
回调系统在 MCP 客户端与 LangChain 工具之间建立了桥梁，使得 MCP 服务器发出的通知（notifications）能够被 LangChain 端正确处理和响应。
## 核心组件
### 组件架构
```mermaid
graph TD
A[MCP 服务器] -->|通知/进度| B[ClientSession]
B --> C[Callbacks]
C --> D[CallbackContext]
D --> E[工具执行]
E --> F[_MCPCallbacks]
F --> G[进度回调]
H[开发者代码] -->|自定义回调| C
```
### CallbackContext
`CallbackContext` 是回调上下文的载体，用于在回调处理过程中传递工具调用的关键信息。
| 属性 | 类型 | 说明 |
|------|------|------|
| `server_name` | `str` | 工具所属的 MCP 服务器名称 |
| `tool_name` | `str` | 被调用的工具名称 |
资料来源：[langchain_mcp_adapters/tools.py:1-1]()
### Callbacks
`Callbacks` 类是回调机制的主要入口点，封装了所有与 MCP 服务器通信的回调处理逻辑。
```python
class Callbacks:
"""用于处理 MCP 通知和事件的回调接口"""
```
开发者通过实例化 `Callbacks` 并传递给 `MultiServerMCPClient` 或 `load_mcp_tools` 来启用自定义回调处理。
资料来源：[langchain_mcp_adapters/client.py:1-1]()
### _MCPCallbacks
`_MCPCallbacks` 是内部默认回调实现类，用于将 LangChain 格式的回调转换为 MCP SDK 所需的格式。
```python
class _MCPCallbacks:
"""内部回调实现，桥接 LangChain 和 MCP SDK"""
```
资料来源：[langchain_mcp_adapters/tools.py:1-1]()
## 工作流程
### 回调初始化流程
```mermaid
sequenceDiagram
participant 用户 as 用户代码
participant 客户端 as MultiServerMCPClient
participant 回调 as Callbacks
participant 上下文 as CallbackContext
participant 工具 as load_mcp_tools
用户->>客户端: 传入 Callbacks 实例
客户端->>工具: 传递 callbacks 参数
工具->>回调: to_mcp_format(context)
回调->>上下文: 创建 CallbackContext
Note over 上下文: 包含 server_name, tool_name
```
### 工具调用中的回调处理
```mermaid
graph LR
A[工具调用请求] --> B[call_tool 函数]
B --> C{已有 Callbacks?}
C -->|是| D[to_mcp_format]
C -->|否| E[_MCPCallbacks]
D --> F[CallbackContext]
E --> F
F --> G[progress_callback]
G --> H[MCP Session.call_tool]
```
在 `call_tool` 函数中，回调按以下方式使用：
```python
mcp_callbacks = (
callbacks.to_mcp_format(
context=CallbackContext(server_name=server_name, tool_name=tool.name)
)
if callbacks is not None
else _MCPCallbacks()
)
```
资料来源：[langchain_mcp_adapters/tools.py:145-156]()
## API 参考
### load_mcp_tools 参数
| 参数 | 类型 | 必填 | 说明 |
|------|------|------|------|
| `session` | `ClientSession` | 否 | MCP 客户端会话 |
| `connection` | `Connection` | 否 | 连接配置 |
| `callbacks` | `Callbacks` | 否 | 回调处理器 |
| `tool_interceptors` | `list[ToolCallInterceptor]` | 否 | 工具调用拦截器 |
| `server_name` | `str` | 否 | 服务器名称 |
| `tool_name_prefix` | `bool` | 否 | 工具名称前缀 |
资料来源：[langchain_mcp_adapters/tools.py:199-230]()
### convert_mcp_tool_to_langchain_tool 参数
| 参数 | 类型 | 必填 | 说明 |
|------|------|------|------|
| `session` | `ClientSession` | 否 | MCP 客户端会话 |
| `tool` | `MCPTool` | 是 | 待转换的 MCP 工具 |
| `callbacks` | `Callbacks` | 否 | 回调处理器 |
| `tool_interceptors` | `list[ToolCallInterceptor]` | 否 | 拦截器列表 |
| `server_name` | `str` | 否 | 服务器名称 |
| `tool_name_prefix` | `bool` | 否 | 是否添加前缀 |
资料来源：[langchain_mcp_adapters/tools.py:232-290]()
## 使用示例
### 基本用法
```python
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.callbacks import Callbacks
# 定义自定义回调处理
class MyCallbacks(Callbacks):
def __init__(self):
super().__init__()
def to_mcp_format(self, context):
# 实现自定义转换逻辑
return super().to_mcp_format(context)
# 使用回调客户端
client = MultiServerMCPClient(
{
"math": {
"command": "python",
"args": ["/path/to/math_server.py"],
"transport": "stdio",
},
},
callbacks=MyCallbacks()
)
tools = await client.get_tools()
```
### 与拦截器配合使用
回调机制可以与拦截器（interceptors）配合使用，实现更复杂的工具调用控制：
```mermaid
graph TD
A[请求] --> B[拦截器1]
B --> C[拦截器2]
C --> D[拦截器N]
D --> E[execute_tool]
E --> D
D --> C
C --> B
B --> F[响应]
```
```python
from langchain_mcp_adapters.interceptors import ToolCallInterceptor
async def my_interceptor(
request: MCPToolCallRequest,
handler: Callable
) -> MCPToolCallResult:
# 前置处理
print(f"调用工具: {request.name}")
result = await handler(request)
# 后置处理
print(f"工具返回: {result}")
return result
tools = await load_mcp_tools(
session,
callbacks=custom_callbacks,
tool_interceptors=[my_interceptor]
)
```
资料来源：[langchain_mcp_adapters/tools.py:61-95]()
## 高级功能
### 进度回调
回调系统支持 MCP 协议的进度通知机制。当 MCP 服务器通过 `progress_callback` 发送进度更新时，系统会将其传递给 LangChain 端处理。
```python
call_tool_result = await session.call_tool(
tool_name,
tool_args,
progress_callback=mcp_callbacks.progress_callback,
)
```
资料来源：[langchain_mcp_adapters/tools.py:157-165]()
### 工具调用请求拦截
通过 `MCPToolCallRequest` 和 `MCPToolCallResult` 类型，回调系统支持对工具调用请求和响应进行完整的拦截处理：
```python
@dataclass
class MCPToolCallRequest:
"""工具执行请求，传递给拦截器"""
name: str
args: dict[str, Any]
server_name: str
headers: dict[str, Any] | None
runtime: Any # LangGraph 运行时上下文
```
资料来源：[langchain_mcp_adapters/interceptors.py:1-1]()
## 最佳实践
1. **始终传递 server_name**：确保回调上下文能够正确识别工具所属的服务器
2. **合理使用拦截器链**：多个拦截器按洋葱模型执行，避免过度嵌套
3. **处理异常情况**：在回调中实现适当的错误处理逻辑
4. **性能考虑**：避免在回调中执行耗时操作，以免阻塞工具调用
## 与 LangGraph 集成
当项目中安装了 `langgraph` 时，回调机制支持额外的 `runtime` 参数，用于传递 LangGraph 的执行上下文：
```python
async def call_tool(
runtime: Annotated[object | None, InjectedToolArg()] = None,
**arguments: dict[str, Any],
) -> tuple[ConvertedToolResult, MCPToolArtifact | None]:
request = MCPToolCallRequest(
name=tool.name,
args=arguments,
server_name=server_name or "unknown",
headers=None,
runtime=runtime, # LangGraph 运行时上下文
)
```
资料来源：[langchain_mcp_adapters/tools.py:126-142]()
## 相关模块
| 模块 | 功能 |
|------|------|
| `langchain_mcp_adapters.callbacks` | 回调接口定义和实现 |
| `langchain_mcp_adapters.interceptors` | 工具调用拦截器接口 |
| `langchain_mcp_adapters.tools` | 工具加载和转换，含回调集成 |
| `langchain_mcp_adapters.client` | 多服务器客户端，回调配置入口 |
---
## 提示词与资源转换
### 相关页面
相关主题：[数据流与转换机制](#page-data-flow), [核心模块结构](#page-core-modules)
相关源码文件
以下源码文件用于生成本页说明：
- [langchain_mcp_adapters/prompts.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/prompts.py)
- [langchain_mcp_adapters/resources.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/resources.py)
- [langchain_mcp_adapters/tools.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/tools.py)
- [langchain_mcp_adapters/client.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/client.py)
- [langchain_mcp_adapters/sessions.py](https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/langchain_mcp_adapters/sessions.py)
# 提示词与资源转换
## 概述
`langchain-mcp-adapters` 库提供了两大核心转换模块，用于将 MCP（Model Context Protocol）的提示词（Prompts）和资源（Resources）转换为 LangChain 兼容的格式。这些模块使得 MCP 服务器可以作为 LangChain 和 LangGraph 应用的外部工具和数据来源。
| 模块 | 源文件 | 功能描述 |
|------|--------|----------|
| 提示词转换 | `prompts.py` | 将 MCP 提示词消息转换为 LangChain 消息对象 |
| 资源转换 | `resources.py` | 将 MCP 资源内容转换为 LangChain Blob 对象 |
资料来源：[langchain_mcp_adapters/prompts.py:1-10]()，[langchain_mcp_adapters/resources.py:1-15]()
## 架构概览
```mermaid
graph TD
subgraph MCP服务器端
MCP[MCP Server]
Prompts[MCP Prompts]
Resources[MCP Resources]
end
subgraph langchain_mcp_adapters
prompts_module[prompts.py]
resources_module[resources.py]
end
subgraph LangChain端
LCMessages[LangChain Messages
HumanMessage/AIMessage]
LCBlobs[LangChain Blobs]
end
Prompts -->|get_prompt| prompts_module
Resources -->|read_resource| resources_module
prompts_module --> LCMessages
resources_module --> LCBlobs
style MCP服务器端 fill:#e1f5fe
style LangChain端 fill:#f3e5f5
style langchain_mcp_adapters fill:#fff3e0
```
## 提示词转换模块
### 模块定位
`prompts.py` 模块负责将 MCP 服务器提供的提示词（Prompts）转换为 LangChain 的消息对象。提示词是 MCP 协议中用于定义预定义对话模板的机制，允许服务器提供结构化的提示内容。
资料来源：[langchain_mcp_adapters/prompts.py:1-10]()
### 核心函数
#### `convert_mcp_prompt_message_to_langchain_message()`
该函数将单个 MCP 提示词消息转换为 LangChain 消息对象。
**函数签名：**
```python
def convert_mcp_prompt_message_to_langchain_message(
message: PromptMessage,
) -> HumanMessage | AIMessage:
```
**参数说明：**
| 参数 | 类型 | 描述 |
|------|------|------|
| `message` | `PromptMessage` | MCP 提示词消息对象 |
**支持的转换规则：**
| MCP 消息类型 | LangChain 消息类型 |
|-------------|-------------------|
| `role: user, content.type: text` | `HumanMessage` |
| `role: assistant, content.type: text` | `AIMessage` |
**异常处理：**
| 错误条件 | 抛出异常 |
|---------|----------|
| `content.type` 非 `text` | `ValueError` |
| `role` 非 `user` 或 `assistant` | `ValueError` |
资料来源：[langchain_mcp_adapters/prompts.py:12-35]()
#### `load_mcp_prompt()`
异步函数，用于从 MCP 会话加载提示词并转换为 LangChain 消息列表。
**函数签名：**
```python
async def load_mcp_prompt(
session: ClientSession,
name: str,
*,
arguments: dict[str, Any] | None = None,
) -> list[HumanMessage | AIMessage]:
```
**参数说明：**
| 参数 | 类型 | 必填 | 描述 |
|------|------|------|------|
| `session` | `ClientSession` | 是 | MCP 客户端会话 |
| `name` | `str` | 是 | 提示词名称 |
| `arguments` | `dict[str, Any] \| None` | 否 | 传递给提示词的参数 |
**返回值：**
返回转换后的 LangChain 消息列表，消息类型为 `HumanMessage` 或 `AIMessage`。
**工作流程：**
```mermaid
sequenceDiagram
participant Client as 客户端
participant Session as MCP ClientSession
participant Adapter as prompts.py
participant LC as LangChain Messages
Client->>Adapter: load_mcp_prompt(session, name, arguments)
Adapter->>Session: get_prompt(name, arguments)
Session-->>Adapter: PromptMessage[]
Loop 遍历每条消息
Adapter->>Adapter: convert_mcp_prompt_message_to_langchain_message()
end
Adapter-->>Client: list[HumanMessage | AIMessage]
```
资料来源：[langchain_mcp_adapters/prompts.py:38-60]()
## 资源转换模块
### 模块定位
`resources.py` 模块负责将 MCP 协议中的资源（Resources）转换为 LangChain 的 Blob 对象。资源可以是文本内容或二进制数据（如图片、文件等），Blob 对象提供了统一的数据包装接口。
资料来源：[langchain_mcp_adapters/resources.py:1-18]()
### 核心函数
#### `convert_mcp_resource_to_langchain_blob()`
将单个 MCP 资源内容转换为 LangChain Blob 对象。
**函数签名：**
```python
def convert_mcp_resource_to_langchain_blob(
resource_uri: str, contents: ResourceContents
) -> Blob:
```
**参数说明：**
| 参数 | 类型 | 描述 |
|------|------|------|
| `resource_uri` | `str` | 资源的 URI 标识符 |
| `contents` | `ResourceContents` | 资源内容（TextResourceContents 或 BlobResourceContents） |
**支持的资源类型：**
| MCP 资源类型 | LangChain 处理方式 |
|-------------|-------------------|
| `TextResourceContents` | 直接使用文本数据 |
| `BlobResourceContents` | Base64 解码后使用 |
**Blob 元数据：**
转换后的 Blob 对象包含以下元数据：
| 元数据键 | 值 |
|---------|-----|
| `uri` | 资源的 URI |
**异常处理：**
| 错误条件 | 抛出异常 |
|---------|----------|
| 不支持的资源内容类型 | `TypeError` |
资料来源：[langchain_mcp_adapters/resources.py:20-42]()
#### `get_mcp_resource()`
异步函数，用于获取单个 MCP 资源并转换为 LangChain Blob 对象列表。
**函数签名：**
```python
async def get_mcp_resource(session: ClientSession, uri: str) -> list[Blob]:
```
**参数说明：**
| 参数 | 类型 | 描述 |
|------|------|------|
| `session` | `ClientSession` | MCP 客户端会话 |
| `uri` | `str` | 要获取的资源 URI |
**返回值：**
返回 LangChain Blob 对象列表。如果资源内容为空，则返回空列表。
资料来源：[langchain_mcp_adapters/resources.py:44-61]()
#### `load_mcp_resources()`
异步函数，用于批量加载 MCP 资源。
**函数签名：**
```python
async def load_mcp_resources(
session: ClientSession,
*,
uris: str | list[str] | None = None,
) -> list[Blob]:
```
**参数说明：**
| 参数 | 类型 | 必填 | 描述 |
|------|------|------|------|
| `session` | `ClientSession` | 是 | MCP 客户端会话 |
| `uris` | `str \| list[str] \| None` | 否 | 要加载的 URI 列表，`None` 表示加载所有资源 |
**加载行为：**
| `uris` 值 | 行为 |
|----------|------|
| `None` | 调用 `session.list_resources()` 获取所有资源 URI |
| `str` | 仅加载指定的单个 URI |
| `list[str]` | 加载指定的 URI 列表 |
> [!note]
> 动态资源（Dynamic Resources）在 `uris` 为 `None` 时**不会**被加载，因为它们需要参数，而 MCP SDK 的 `session.list_resources()` 方法无法处理这类资源。
资料来源：[langchain_mcp_adapters/resources.py:64-90]()
## 数据模型
### MCP 类型引用
```mermaid
classDiagram
class PromptMessage {
+str role
+Content content
}
class ResourceContents {
<>
}
class TextResourceContents {
+str uri
+str text
+str mimeType
}
class BlobResourceContents {
+str uri
+str blob
+str mimeType
}
class TextContent {
+str type
+str text
}
class BlobResourceContents {
+str uri
+str blob
+str mimeType
}
ResourceContents <|-- TextResourceContents
ResourceContents <|-- BlobResourceContents
note for ResourceContents "MCP types from mcp.types"
```
### LangChain 类型对应
| MCP 类型 | LangChain 类型 | 说明 |
|---------|---------------|------|
| `PromptMessage` | `HumanMessage \| AIMessage` | 提示词消息 |
| `TextResourceContents` | `Blob` (文本数据) | 文本资源 |
| `BlobResourceContents` | `Blob` (二进制数据) | 二进制资源 |
## 客户端集成
`MultiServerMCPClient` 提供了对提示词和资源转换的统一访问接口。
资料来源：[langchain_mcp_adapters/client.py:1-60]()
### 核心方法
| 方法 | 功能 |
|------|------|
| `get_tools()` | 获取所有 MCP 工具并转换为 LangChain 工具 |
| `get_prompts()` | 获取所有 MCP 提示词 |
| `get_resources()` | 获取所有 MCP 资源 |
| `session(server_name)` | 获取指定服务器的会话上下文管理器 |
### 使用示例
```python
from langchain_mcp_adapters.client import MultiServerMCPClient
client = MultiServerMCPClient(
{
"math": {
"command": "python",
"args": ["./math_server.py"],
"transport": "stdio",
},
}
)
# 获取工具
tools = await client.get_tools()
# 获取资源
resources = await client.get_resources()
```
## 会话管理
提示词和资源的加载依赖于 `ClientSession` 对象，该对象通过 `sessions.py` 模块创建和管理。
资料来源：[langchain_mcp_adapters/sessions.py:1-100]()
### 支持的传输类型
| 传输类型 | 说明 | 支持状态 |
|---------|------|----------|
| `stdio` | 标准输入输出传输 | ✅ 完全支持 |
| `sse` | Server-Sent Events | ✅ 完全支持 |
| `http` | Streamable HTTP | ✅ 完全支持 |
| `websocket` | WebSocket | ✅ 完全支持 |
## 类型转换流程
### 提示词转换流程
```mermaid
flowchart LR
subgraph MCP层
MCP_Prompt[MCP get_prompt]
PM[PromptMessage]
end
subgraph 转换层
convert[convert_mcp_prompt_message_to_langchain_message]
end
subgraph LangChain层
HM[HumanMessage]
AM[AIMessage]
end
MCP_Prompt --> PM
PM --> convert
convert -->|role=user| HM
convert -->|role=assistant| AM
```
### 资源转换流程
```mermaid
flowchart LR
subgraph MCP层
MCP_Resource[MCP read_resource]
TRC[TextResourceContents]
BRC[BlobResourceContents]
end
subgraph 转换层
convert[convert_mcp_resource_to_langchain_blob]
decode[base64.b64decode]
end
subgraph LangChain层
Blob[Blob]
end
MCP_Resource --> TRC
MCP_Resource --> BRC
TRC --> convert
BRC --> decode
decode --> convert
convert --> Blob
```
## 错误处理
### 提示词转换错误
| 错误类型 | 条件 | 处理方式 |
|---------|------|----------|
| `ValueError` | 不支持的 `content.type` | 抛出异常 |
| `ValueError` | 不支持的 `role` 值 | 抛出异常 |
### 资源转换错误
| 错误类型 | 条件 | 处理方式 |
|---------|------|----------|
| `TypeError` | 不支持的资源内容类型 | 抛出异常 |
| `RuntimeError` | 获取资源时发生错误 | 由调用方处理 |
## 与工具转换的关系
提示词和资源转换与 `tools.py` 中的工具转换模块共同构成了 MCP 到 LangChain 的完整适配体系。
```mermaid
graph TD
subgraph MCP协议
MCP_Tools[Tools]
MCP_Prompts[Prompts]
MCP_Resources[Resources]
end
subgraph langchain_mcp_adapters
tools_module[tools.py]
prompts_module[prompts.py]
resources_module[resources.py]
end
subgraph LangChain
LC_Tools[StructuredTool]
LC_Messages[Messages]
LC_Blobs[Blobs]
end
MCP_Tools --> tools_module
MCP_Prompts --> prompts_module
MCP_Resources --> resources_module
tools_module --> LC_Tools
prompts_module --> LC_Messages
resources_module --> LC_Blobs
```
资料来源：[langchain_mcp_adapters/tools.py:1-100]()
## 最佳实践
### 1. 资源批量加载
当需要加载多个资源时，使用 `load_mcp_resources()` 的 `uris` 参数指定资源列表，避免多次调用 `get_mcp_resource()`：
```python
# 推荐方式
resources = await load_mcp_resources(session, uris=["uri1", "uri2", "uri3"])
# 不推荐方式（多次网络往返）
res1 = await get_mcp_resource(session, "uri1")
res2 = await get_mcp_resource(session, "uri2")
res3 = await get_mcp_resource(session, "uri3")
```
### 2. 提示词参数传递
使用 `arguments` 参数为提示词提供上下文信息：
```python
messages = await load_mcp_prompt(
session,
name="code_review",
arguments={"repo": "langchain-mcp-adapters", "pr": 123}
)
```
### 3. 错误处理
实现适当的错误处理以应对资源不可用或格式不支持的情况：
```python
try:
resources = await load_mcp_resources(session, uris=uri_list)
except TypeError as e:
logger.error(f"Unsupported resource type: {e}")
```
## 总结
`langchain-mcp-adapters` 的提示词和资源转换模块为 LangChain 应用提供了访问 MCP 服务器结构和数据的能力：
| 组件 | 文件 | 核心功能 |
|------|------|----------|
| 提示词转换 | `prompts.py` | 将 MCP Prompts 转换为 `HumanMessage`/`AIMessage` |
| 资源转换 | `resources.py` | 将 MCP Resources 转换为 LangChain `Blob` 对象 |
这两个模块与工具转换模块（`tools.py`）一起，构成了 MCP 到 LangChain 的完整适配层，使得开发者可以在 LangChain 和 LangGraph 应用中无缝使用 MCP 服务器提供的各种功能。
---
---
## Doramagic 踩坑日志
项目：langchain-ai/langchain-mcp-adapters
摘要：发现 17 个潜在踩坑项，其中 3 个为 high/blocking；最高优先级：安装坑 - 来源证据：Prompts and Resources auto-discovery。
## 1. 安装坑 · 来源证据：Prompts and Resources auto-discovery
- 严重度：high
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个安装相关的待验证问题：Prompts and Resources auto-discovery
- 对用户的影响：可能增加新用户试用和生产接入成本。
- 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_bf1812b74caa4e989767a9307a8ffc16 | https://github.com/langchain-ai/langchain-mcp-adapters/issues/62 | 来源讨论提到 python 相关条件，需在安装/试用前复核。
## 2. 安装坑 · 来源证据：`MultiServerMCPClient.get_tools()` silently returns no tools when any single server fails to connect
- 严重度：high
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个安装相关的待验证问题：`MultiServerMCPClient.get_tools()` silently returns no tools when any single server fails to connect
- 对用户的影响：可能增加新用户试用和生产接入成本。
- 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_a5093182914b4df0b7ad2cd560bacdf2 | https://github.com/langchain-ai/langchain-mcp-adapters/issues/492 | 来源讨论提到 python 相关条件，需在安装/试用前复核。
## 3. 运行坑 · 来源证据：Fix TypeError in resources.py and make __aexit__ an async coroutine in client.py
- 严重度：high
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个运行相关的待验证问题：Fix TypeError in resources.py and make __aexit__ an async coroutine in client.py
- 对用户的影响：可能增加新用户试用和生产接入成本。
- 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_ac102050dd4841d6954559a3413e0b92 | https://github.com/langchain-ai/langchain-mcp-adapters/issues/496 | 来源类型 github_issue 暴露的待验证使用条件。
## 4. 安装坑 · 来源证据：langchain-mcp-adapters==0.2.2
- 严重度：medium
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个安装相关的待验证问题：langchain-mcp-adapters==0.2.2
- 对用户的影响：可能增加新用户试用和生产接入成本。
- 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_0c6ca0722ab046379d28ecf30f8d2bcf | https://github.com/langchain-ai/langchain-mcp-adapters/releases/tag/langchain-mcp-adapters%3D%3D0.2.2 | 来源类型 github_release 暴露的待验证使用条件。
## 5. 配置坑 · 来源证据：langchain-mcp-adapters==0.1.10
- 严重度：medium
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个配置相关的待验证问题：langchain-mcp-adapters==0.1.10
- 对用户的影响：可能增加新用户试用和生产接入成本。
- 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_8b18dbf32ccd41e38b272a458f4040f5 | https://github.com/langchain-ai/langchain-mcp-adapters/releases/tag/langchain-mcp-adapters%3D%3D0.1.10 | 来源讨论提到 python 相关条件，需在安装/试用前复核。
## 6. 能力坑 · 来源证据：langchain-mcp-adapters==0.1.14
- 严重度：medium
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个能力理解相关的待验证问题：langchain-mcp-adapters==0.1.14
- 对用户的影响：可能增加新用户试用和生产接入成本。
- 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_6727e0d698e54fc38d7c60e262978ac2 | https://github.com/langchain-ai/langchain-mcp-adapters/releases/tag/langchain-mcp-adapters%3D%3D0.1.14 | 来源类型 github_release 暴露的待验证使用条件。
## 7. 能力坑 · 能力判断依赖假设
- 严重度：medium
- 证据强度：source_linked
- 发现：README/documentation is current enough for a first validation pass.
- 对用户的影响：假设不成立时，用户拿不到承诺的能力。
- 建议检查：将假设转成下游验证清单。
- 防护动作：假设必须转成验证项；没有验证结果前不能写成事实。
- 证据：capability.assumptions | github_repo:929158279 | https://github.com/langchain-ai/langchain-mcp-adapters | README/documentation is current enough for a first validation pass.
## 8. 运行坑 · 来源证据：langchain-mcp-adapters==0.1.12
- 严重度：medium
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个运行相关的待验证问题：langchain-mcp-adapters==0.1.12
- 对用户的影响：可能增加新用户试用和生产接入成本。
- 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_e71a46a9e0374d139555a78f229b0469 | https://github.com/langchain-ai/langchain-mcp-adapters/releases/tag/langchain-mcp-adapters%3D%3D0.1.12 | 来源类型 github_release 暴露的待验证使用条件。
## 9. 维护坑 · 来源证据：langchain-mcp-adapters==0.2.0
- 严重度：medium
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个维护/版本相关的待验证问题：langchain-mcp-adapters==0.2.0
- 对用户的影响：可能影响升级、迁移或版本选择。
- 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_59483f9a6a16414c9087b1751fba8efc | https://github.com/langchain-ai/langchain-mcp-adapters/releases/tag/langchain-mcp-adapters%3D%3D0.2.0 | 来源类型 github_release 暴露的待验证使用条件。
## 10. 维护坑 · 来源证据：langchain-mcp-adapters==0.2.0a1
- 严重度：medium
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个维护/版本相关的待验证问题：langchain-mcp-adapters==0.2.0a1
- 对用户的影响：可能影响升级、迁移或版本选择。
- 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_4e7fcda1716948898295279af95f8f96 | https://github.com/langchain-ai/langchain-mcp-adapters/releases/tag/langchain-mcp-adapters%3D%3D0.2.0a1 | 来源类型 github_release 暴露的待验证使用条件。
## 11. 维护坑 · 维护活跃度未知
- 严重度：medium
- 证据强度：source_linked
- 发现：未记录 last_activity_observed。
- 对用户的影响：新项目、停更项目和活跃项目会被混在一起，推荐信任度下降。
- 建议检查：补 GitHub 最近 commit、release、issue/PR 响应信号。
- 防护动作：维护活跃度未知时，推荐强度不能标为高信任。
- 证据：evidence.maintainer_signals | github_repo:929158279 | https://github.com/langchain-ai/langchain-mcp-adapters | last_activity_observed missing
## 12. 安全/权限坑 · 下游验证发现风险项
- 严重度：medium
- 证据强度：source_linked
- 发现：no_demo
- 对用户的影响：下游已经要求复核，不能在页面中弱化。
- 建议检查：进入安全/权限治理复核队列。
- 防护动作：下游风险存在时必须保持 review/recommendation 降级。
- 证据：downstream_validation.risk_items | github_repo:929158279 | https://github.com/langchain-ai/langchain-mcp-adapters | no_demo; severity=medium
## 13. 安全/权限坑 · 存在安全注意事项
- 严重度：medium
- 证据强度：source_linked
- 发现：No sandbox install has been executed yet; downstream must verify before user use.
- 对用户的影响：用户安装前需要知道权限边界和敏感操作。
- 建议检查：转成明确权限清单和安全审查提示。
- 防护动作：安全注意事项必须面向用户前置展示。
- 证据：risks.safety_notes | github_repo:929158279 | https://github.com/langchain-ai/langchain-mcp-adapters | No sandbox install has been executed yet; downstream must verify before user use.
## 14. 安全/权限坑 · 存在评分风险
- 严重度：medium
- 证据强度：source_linked
- 发现：no_demo
- 对用户的影响：风险会影响是否适合普通用户安装。
- 建议检查：把风险写入边界卡，并确认是否需要人工复核。
- 防护动作：评分风险必须进入边界卡，不能只作为内部分数。
- 证据：risks.scoring_risks | github_repo:929158279 | https://github.com/langchain-ai/langchain-mcp-adapters | no_demo; severity=medium
## 15. 安全/权限坑 · 来源证据：Feature Request: Support passing server-defined params extensions (e.g. LangGraph `context`) through tools/call
- 严重度：medium
- 证据强度：source_linked
- 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：Feature Request: Support passing server-defined params extensions (e.g. LangGraph `context`) through tools/call
- 对用户的影响：可能影响授权、密钥配置或安全边界。
- 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
- 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
- 证据：community_evidence:github | cevd_8c46dab4b6dd4a6e92c96af49ea47647 | https://github.com/langchain-ai/langchain-mcp-adapters/issues/502 | 来源讨论提到 python 相关条件，需在安装/试用前复核。
## 16. 维护坑 · issue/PR 响应质量未知
- 严重度：low
- 证据强度：source_linked
- 发现：issue_or_pr_quality=unknown。
- 对用户的影响：用户无法判断遇到问题后是否有人维护。
- 建议检查：抽样最近 issue/PR，判断是否长期无人处理。
- 防护动作：issue/PR 响应未知时，必须提示维护风险。
- 证据：evidence.maintainer_signals | github_repo:929158279 | https://github.com/langchain-ai/langchain-mcp-adapters | issue_or_pr_quality=unknown
## 17. 维护坑 · 发布节奏不明确
- 严重度：low
- 证据强度：source_linked
- 发现：release_recency=unknown。
- 对用户的影响：安装命令和文档可能落后于代码，用户踩坑概率升高。
- 建议检查：确认最近 release/tag 和 README 安装命令是否一致。
- 防护动作：发布节奏未知或过期时，安装说明必须标注可能漂移。
- 证据：evidence.maintainer_signals | github_repo:929158279 | https://github.com/langchain-ai/langchain-mcp-adapters | release_recency=unknown