项目概述

langchain-mcp-adapters 是一个轻量级适配器库，用于将 Anthropic Model Context Protocol (MCP) 工具转换为 LangChain 和 LangGraph 兼容的格式。该项目使开发者能够在 LangChain/LangGraph 应用中无缝使用 MCP 协议定义的各种工具。

资料来源：langchain_mcp_adapters/__init__.py:1-9

核心定位

该项目充当 MCP 协议与 LangChain 生态之间的桥梁，解决两个协议栈之间的互操作性问题。

graph LR
    A[MCP 服务器] --> B[langchain-mcp-adapters]
    B --> C[LangChain/LangGraph]
    C --> D[AI Agent]
    
    style B fill:#e1f5fe

资料来源：README.md:1-1

核心功能特性

资料来源：README.md:10-16

系统架构

核心组件

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

工具转换流程

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 包装结构化返回内容

资料来源：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 - 拦截器模块

提供工具调用生命周期管理能力，采用洋葱模式构建拦截器链：

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

资料来源：langchain_mcp_adapters/interceptors.py:1-50

client.py - 多服务器客户端

MultiServerMCPClient 是核心入口类，提供以下能力：

• 同时管理多个 MCP 服务器连接
• 支持按服务器名获取工具列表
• 支持显式会话管理和上下文管理器会话
• 集成回调和拦截器配置

资料来源：langchain_mcp_adapters/client.py:1-100

sessions.py - 会话管理层

负责创建和管理 MCP ClientSession，支持多种传输协议：

资料来源：langchain_mcp_adapters/sessions.py:1-100

快速开始

安装依赖

pip install langchain-mcp-adapters

资料来源：README.md:22-24

创建 MCP 服务器

# 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

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

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

class MCPToolArtifact(TypedDict):
    """MCP 工具调用返回的工件"""
    structured_content: dict[str, Any]

用于包装 MCP 工具的结构化返回内容，支持未来扩展。

资料来源：langchain_mcp_adapters/tools.py:75-90

ConvertedToolResult

根据 langgraph 安装状态，返回类型有所不同：

资料来源：langchain_mcp_adapters/tools.py:60-70

生态关联

该项目与以下项目形成完整的 MCP 工具生态：

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 获取。

资料来源：README.md:20-21

总结

langchain-mcp-adapters 作为 MCP 协议与 LangChain 生态之间的关键适配层，提供：

1. 开箱即用的工具转换：零配置将 MCP 工具转换为 LangChain 工具
2. 灵活的拦截机制：支持在工具调用生命周期中注入自定义逻辑
3. 强大的多服务器支持：单客户端管理多个 MCP 服务器
4. 完善的类型安全：全面使用 Pydantic 和 TypedDict 确保类型安全

该库使开发者能够充分利用 MCP 协议丰富的工具生态，同时享受 LangChain/LangGraph 带来的强大 Agent 开发能力。

概述

langchain-mcp-adapters 是一个轻量级适配器库，用于将 Anthropic 的 Model Context Protocol (MCP) 工具转换为 LangChain 和 LangGraph 兼容的工具格式。该库的主要功能包括：

• 将 MCP 工具转换为 LangChain 工具，使其能够与 LangGraph 代理配合使用
• 提供客户端实现，支持连接多个 MCP 服务器并加载其工具

资料来源：README.md:1-12

环境要求

系统要求

核心依赖

可选依赖

安装方法

基础安装

使用 pip 安装 langchain-mcp-adapters 的基本功能：

pip install langchain-mcp-adapters

资料来源：README.md:14

完整安装（包含所有依赖）

为了使用完整功能，包括与 LangGraph 和 OpenAI 的集成：

pip install langchain-mcp-adapters langgraph "langchain[openai]"

资料来源：README.md:20-21

设置 API 密钥

安装完成后，需要设置相应的 API 密钥环境变量：

export OPENAI_API_KEY=<your_api_key>

资料来源：README.md:22

客户端配置

MultiServerMCPClient 概述

MultiServerMCPClient 是连接多个 MCP 服务器的主要接口，允许你从多个服务器同时加载工具。

资料来源：client.py:1-50

连接配置参数

#### 支持的传输协议

资料来源：sessions.py:1-200

#### stdio 传输配置

stdio 传输适用于本地 MCP 服务器进程：

from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient(
    {
        "math": {
            "command": "python",
            "args": ["/path/to/math_server.py"],
            "transport": "stdio",
        },
    }
)

配置参数说明：

资料来源：sessions.py:1-100

#### HTTP 传输配置

HTTP 传输适用于远程 MCP 服务器：

client = MultiServerMCPClient(
    {
        "weather": {
            "url": "http://localhost:8000/mcp",
            "transport": "http",
            "headers": {
                "Authorization": "Bearer custom-value"
            }
        }
    }
)

HTTP 配置参数说明：

资料来源：client.py:50-100

工具名称前缀配置

当多个 MCP 服务器存在相同名称的工具时，可以使用 tool_name_prefix 参数为工具名称添加服务器前缀：

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 服务器：

# 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 传输模式的服务器：

# 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")

运行服务器：

python weather_server.py

资料来源：README.md:60-75

无状态 HTTP 流式服务器

对于可扩展的 HTTP 服务，可以使用无状态流式传输：

cd examples/servers/streamable-http-stateless/
uv run mcp-simple-streamablehttp-stateless --port 3000

资料来源：README.md:2-8

会话管理配置

显式会话管理

虽然 MultiServerMCPClient 默认在每次工具调用时自动创建新会话，但也支持显式会话管理：

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 服务器的连接细节：

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 函数参数

资料来源：tools.py:100-150

架构流程图

客户端-服务器通信架构

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]

工具调用流程

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: 返回结果

环境变量与密钥管理

必需的环境变量

传输配置中的环境变量

在 stdio 传输配置中，可以使用环境变量展开语法：

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

完整配置示例

多服务器配置

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 集成配置

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

故障排除

常见配置错误

调试建议

1. 验证 MCP 服务器：确保服务器可以独立运行
2. 检查传输类型：stdio 需要命令和参数，HTTP 需要 URL
3. 查看工具列表：使用 client.get_tools() 确认工具加载成功
4. 环境变量检查：确保必需的 API 密钥已设置

概述

langchain-mcp-adapters 是一个轻量级适配器库，用于将 Anthropic 的 Model Context Protocol (MCP) 服务器与 LangChain/LangGraph 应用集成。该库的核心设计围绕工具适配、会话管理、资源转换和拦截器机制四大支柱展开，通过模块化架构实现 MCP 工具向 LangChain 工具的透明转换。

项目采用异步优先的设计理念，充分利用 Python 的 asyncio 能力处理 MCP 通信，同时通过 LangGraph 条件判断实现智能的工具路由。资料来源：README.md

架构总览

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]

核心模块划分

资料来源：langchain_mcp_adapters/client.py:1-50

MultiServerMCPClient 客户端模块

功能定位

MultiServerMCPClient 是库的入口点，负责管理对多个 MCP 服务器的连接。它支持通过配置字典同时连接到多个不同类型的服务器（stdio、HTTP SSE、WebSocket 等），并提供统一的工具、资源和提示加载接口。

资料来源：langchain_mcp_adapters/client.py:1-80

连接配置

客户端接受服务器配置字典，键为服务器名称，值为连接参数字典：

client = MultiServerMCPClient(
    {
        "math": {
            "command": "python",
            "args": ["./math_server.py"],
            "transport": "stdio",
        },
        "weather": {
            "url": "http://localhost:8000/mcp",
            "transport": "http",
        }
    }
)

资料来源：langchain_mcp_adapters/client.py:1-100

上下文管理器使用方式

版本 0.1.0 后，MultiServerMCPClient 不再支持作为异步上下文管理器直接使用：

# 推荐方式一：直接调用
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 传输协议：

graph LR
    A[Connection 抽象] --> B[StdioConnection]
    A --> C[StreamableHttpConnection]
    A --> D[SSEConnection]
    A --> E[WebsocketConnection]

资料来源：langchain_mcp_adapters/sessions.py:1-100

StdioConnection

通过标准输入/输出与本地进程通信，适用于本地 MCP 服务器：

async with stdio_client(server_params) as (read, write):
    async with ClientSession(read, write) as session:
        yield session

资料来源：langchain_mcp_adapters/sessions.py:100-200

StreamableHttpConnection

支持流式 HTTP 传输的连接类型，适用于无状态的 HTTP MCP 服务器：

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）传输连接，支持实时事件推送：

async with sse_client(url, headers=headers) as sessions:
    for session in sessions:
        # 处理每个SSE会话

资料来源：langchain_mcp_adapters/sessions.py:300-400

WebsocketConnection

WebSocket 传输连接，支持双向通信：

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 对象，支持工具执行、参数验证和结果转换：

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 服务器上的所有工具：

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]

资料来源：langchain_mcp_adapters/tools.py:200-280

#### convert_mcp_tool_to_langchain_tool

转换单个 MCP 工具为 LangChain 工具：

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 内容块格式：

资料来源：langchain_mcp_adapters/tools.py:100-150

工具调用结果类型

# 根据 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）包装工具调用处理器，允许在调用前后进行拦截和修改：

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

工具调用请求数据结构：

@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 可用性条件定义：

if LANGGRAPH_PRESENT:
    MCPToolCallResult = CallToolResult | ToolMessage | Command
else:
    MCPToolCallResult = CallToolResult | ToolMessage

资料来源：langchain_mcp_adapters/interceptors.py:20-40

ToolCallInterceptor 接口

@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 方法支持部分字段修改：

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 服务器的通知和进度事件：

graph LR
    A[MCP Server] -->|进度通知| B[Callbacks]
    A -->|日志消息| B
    A -->|资源列表变更| B
    B --> C[工具级别回调]

资料来源：langchain_mcp_adapters/callbacks.py:1-50

CallbackContext

回调上下文包含服务器信息：

@dataclass
class CallbackContext:
    server_name: str | None = None
    tool_name: str | None = None

Callbacks 类

核心回调接口定义，支持多种事件类型：

资料来源：langchain_mcp_adapters/callbacks.py:50-100

resources.py 资源模块

资源转换流程

resources.py 模块将 MCP 资源转换为 LangChain 的 Blob 对象：

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

内容类型支持

资料来源：langchain_mcp_adapters/resources.py:50-80

资源加载函数

#### load_mcp_resources

async def load_mcp_resources(
    session: ClientSession,
    *,
    uris: str | list[str] | None = None,
) -> list[Blob]

注意：动态资源在 uris=None 时不会被加载，因为它们需要参数。

资料来源：langchain_mcp_adapters/resources.py:80-120

prompts.py 提示模块

提示加载功能

prompts.py 模块负责将 MCP 服务器上的提示模板加载为 LangChain 的聊天提示值对象，支持参数化提示生成。

资料来源：langchain_mcp_adapters/prompts.py:1-50

数据流集成图

完整工具调用流程

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 避免工具名冲突：

client = MultiServerMCPClient(
    {"math": {...}, "weather": {...}},
    tool_name_prefix=True
)
# 工具名变为: math_add, weather_get_weather

拦截器链配置

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 传输支持运行时头：

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：

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：

if isinstance(content, AudioContent):
    raise NotImplementedError(
        f"AudioContent conversion not supported. "
        f"Mime type: {content.mimeType}"
    )

资料来源：langchain_mcp_adapters/tools.py:100-130

依赖关系

可选依赖 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 服务器集成场景。

概述

MultiServerMCPClient 是 langchain-mcp-adapters 库的核心客户端类，用于连接多个 MCP（Model Context Protocol）服务器并从中加载 LangChain 兼容的工具、提示词和资源。

该客户端的主要作用是将 MCP 服务器提供的工具转换为 LangChain 工具格式，使其能够与 LangChain 和 LangGraph 代理无缝集成。MultiServerMCPClient 支持同时连接多个 MCP 服务器，并提供了灵活的配置选项来管理连接、回调和工具调用拦截。

资料来源：langchain_mcp_adapters/client.py:40-50

架构设计

组件关系图

graph TD
    Client[MultiServerMCPClient]
    Conn[connections<br/>dict[str, Connection]]
    Callbacks[Callbacks]
    Interceptors[tool_interceptors<br/>list[ToolCallInterceptor]]
    Prefix[tool_name_prefix<br/>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]

核心数据流

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 配置结构

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 配置

{
    "command": "python",
    "args": ["/path/to/math_server.py"],
    "transport": "stdio",
    "env": {"KEY": "value"},  # 可选：传递给进程的环境变量
    "cwd": "/working/directory",  # 可选：工作目录
}

资料来源：langchain_mcp_adapters/sessions.py

HTTP/StreamableHttpConnection 配置

{
    "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 服务器的工具列表。

async def get_tools(
    self,
    server_name: str | None = None
) -> list[BaseTool]

返回值: LangChain BaseTool 对象列表

资料来源：langchain_mcp_adapters/client.py

session()

建立与特定 MCP 服务器的会话连接。

@asynccontextmanager
async def session(
    self,
    server_name: str
) -> AsyncIterator[ClientSession]

用法示例:

client = MultiServerMCPClient({...})
async with client.session("math") as session:
    tools = await load_mcp_tools(session)

资料来源：langchain_mcp_adapters/client.py

使用示例

基础用法

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

多服务器连接

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 避免冲突：

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 结合使用构建复杂代理：

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 事件的接口：

@dataclass
class Callbacks:
    """LangChain MCP 客户端回调"""
    
    on_logging_message: LoggingMessageCallback | None = None
    on_progress: ProgressCallback | None = None
    on_elicitation: ElicitationCallback | None = None

回调使用示例

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 接口允许在工具调用前后进行拦截和处理：

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]

拦截器链构建

def _build_interceptor_chain(
    base_handler: Callable[[MCPToolCallRequest], Awaitable[MCPToolCallResult]],
    tool_interceptors: list[ToolCallInterceptor] | None,
) -> Callable[[MCPToolCallRequest], Awaitable[MCPToolCallResult]]

拦截器列表中的第一个拦截器成为最外层，以洋葱模式层层包裹。

使用拦截器

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 不能作为异步上下文管理器使用。

正确用法:

# 方法 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）和提示词：

资源加载

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

提示词加载

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

配置参考

完整配置示例

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 的深度集成

概述

工具转换机制是 langchain-mcp-adapters 库的核心功能，负责将 MCP（Model Context Protocol）工具转换为 LangChain 兼容的工具格式。该机制使得 MCP 服务器提供的工具能够无缝集成到 LangChain 和 LangGraph 智能体应用中。

主要职责包括：

• 将 MCP 工具声明转换为 LangChain StructuredTool
• 处理工具调用请求和响应的格式转换
• 支持内容块的类型转换（文本、图像、音频等）
• 提供拦截器链模式用于工具调用的扩展处理

资料来源：tools.py:1-30

核心类型定义

内容块类型

工具转换机制定义了一系列类型别名用于处理不同格式的内容：

# Type alias for LangChain content blocks used in ToolMessage
ToolMessageContentBlock = TextContentBlock | ImageContentBlock | FileContentBlock

转换结果类型

根据 langgraph 是否安装，返回类型有所不同：

if LANGGRAPH_PRESENT:
    ConvertedToolResult = list[ToolMessageContentBlock] | ToolMessage | Command
else:
    ConvertedToolResult = list[ToolMessageContentBlock] | ToolMessage

工具调用产物

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 内容块格式：

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]

支持的转换映射关系：

资料来源：tools.py:71-133

工具调用结果转换

_convert_call_tool_result() 函数将 MCP 的 CallToolResult 转换为 LangChain 格式：

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 会话加载所有可用工具：

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]:

参数说明：

该函数支持分页加载工具，处理最大 1000 次迭代的循环：

MAX_ITERATIONS = 1000

资料来源：tools.py:240-295

单个工具转换

convert_mcp_tool_to_langchain_tool() 函数将单个 MCP 工具转换为 LangChain StructuredTool：

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:

转换后的工具配置：

资料来源：tools.py:297-370

工具调用执行机制

拦截器链模式

工具调用采用洋葱模式（Onion Pattern）构建拦截器链：

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() 函数实现该模式：

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 是拦截器系统中定义的请求类型：

@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 的类型定义：

if LANGGRAPH_PRESENT:
    MCPToolCallResult = CallToolResult | ToolMessage | Command
else:
    MCPToolCallResult = CallToolResult | ToolMessage

资料来源：interceptors.py:40-55

错误处理机制

工具转换机制实现了多层错误处理：

工具调用错误

当 MCP 工具调用返回错误时：

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)

内容类型错误

不支持的内容类型会抛出相应异常：

资料来源：tools.py:88-95, tools.py:125-133

使用示例

基础用法

from mcp import ClientSession
from langchain_mcp_adapters.tools import load_mcp_tools

# 通过现有会话加载工具
tools = await load_mcp_tools(session)

完整客户端示例

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()

带拦截器的用法

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

架构总结

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 之间的桥梁，通过标准化的转换接口和灵活的拦截器模式，实现了两种协议生态的无缝集成。

概述

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

传输协议架构

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 服务器作为子进程运行时，使用此传输协议。

from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient(
    {
        "math": {
            "command": "python",
            "args": ["/path/to/math_server.py"],
            "transport": "stdio",
        }
    }
)

配置参数：

资料来源：README.md:85-95

2. HTTP 传输

适用场景：远程服务器通信，支持跨网络访问 MCP 服务。

HTTP 传输基于传统的请求/响应模式，适用于可以发起 HTTP 请求的环境。客户端通过 RESTful API 与 MCP 服务器通信。

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

配置参数：

资料来源：langchain_mcp_adapters/client.py:60-75

3. SSE 传输

适用场景：需要服务器推送更新的场景，支持长连接。

SSE (Server-Sent Events) 传输允许服务器主动向客户端推送事件，实现单向实时通信。

from langchain_mcp_adapters.sessions import SSEConnection

connection = SSEConnection(
    url="http://localhost:8000/sse",
    headers={"Authorization": "Bearer token"}
)

配置参数：

资料来源：langchain_mcp_adapters/sessions.py:180-200

4. Streamable HTTP 传输

适用场景：现代 MCP 应用，需要支持流式响应和会话管理。

Streamable HTTP 是 MCP 规范推荐的传输方式，支持流式响应和会话状态管理。它结合了 HTTP 的请求/响应模式和流式传输的能力。

启动服务器示例：

cd examples/servers/streamable-http-stateless/
uv run mcp-simple-streamablehttp-stateless --port 3000

客户端使用方式：

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

服务器端实现示例：

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")

配置参数：

注意：SSE 和 HTTP 传输支持运行时 headers，这些 headers 会随每个 HTTP 请求一起发送到 MCP 服务器。

资料来源：README.md:45-60

5. WebSocket 传输

适用场景：需要双向实时通信的低延迟应用。

WebSocket 传输提供全双工通信通道，适用于需要频繁双向交互的场景。

配置参数：

资料来源：langchain_mcp_adapters/sessions.py:220-240

连接类型映射表

资料来源：langchain_mcp_adapters/sessions.py:50-150

多服务器传输配置

MultiServerMCPClient 支持同时连接多个使用不同传输协议的 MCP 服务器：

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

传输协议选择指南

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），可以在配置中指定自定义请求头：

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。这简化了使用方式，但在高频调用场景下可能影响性能。

tools = await client.get_tools()
response = await agent.ainvoke({"messages": "what's (3 + 5) x 12?"})

资料来源：README.md:110-120

显式会话管理

对于需要显式控制会话生命周期的场景，可以使用上下文管理器：

from langchain_mcp_adapters.tools import load_mcp_tools

client = MultiServerMCPClient({...})
async with client.session("math") as session:
    tools = await load_mcp_tools(session)
    # 在此上下文中复用同一会话

async with client.session(server_name) as session:
    resources = await load_mcp_resources(session, uris=uris)

资料来源：README.md:120-130

工具名称前缀

当多个服务器具有相同名称的工具时，可以启用工具名称前缀来避免冲突：

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" 等

资料来源：langchain_mcp_adapters/client.py:45-55

环境变量扩展

在 stdio 传输中，支持环境变量的展开功能：

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

错误处理

不同传输协议的错误处理遵循统一的模式：

try:
    async with client.session("server_name") as session:
        tools = await load_mcp_tools(session)
except ConnectionError:
    # 处理连接错误
    pass
except ValueError as e:
    # 处理配置错误
    pass

工具调用错误会被包装为 ToolException：

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

概述

工具调用拦截器（Tool Call Interceptor）是 langchain-mcp-adapters 框架中用于拦截和控制 MCP 工具调用生命周期的核心机制。它采用洋葱模式（Onion Pattern）构建可组合的处理器链，允许开发者在工具调用的请求和响应阶段注入自定义逻辑。

拦截器的主要作用包括：

• 在工具执行前后添加日志记录和监控
• 修改工具调用的参数或请求头
• 验证或转换工具返回结果
• 实现重试逻辑和错误处理
• 支持 LangGraph Command 的特殊返回类型处理

资料来源：langchain_mcp_adapters/interceptors.py:1-10

架构设计

洋葱模式处理器链

拦截器通过洋葱模式组织，形成多层嵌套的处理器链。当工具调用时，请求从外到内逐层传递，执行完成后结果再从内到外逐层返回。

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 数据结构

MCPToolCallRequest 是拦截器接收到的工具调用请求对象，采用 dataclass 模式定义：

@dataclass
class MCPToolCallRequest:
    """工具执行请求的请求对象"""
    name: str                           # 工具名称
    args: dict[str, Any]               # 工具参数
    server_name: str                    # 服务器名称
    headers: dict[str, Any] | None      # HTTP 请求头
    runtime: object | None             # LangGraph 运行时对象

请求覆盖机制

通过 override() 方法可以创建修改后的请求副本：

def override(
    self, **overrides: Unpack[_MCPToolCallRequestOverrides]
) -> MCPToolCallRequest:
    """创建带有指定覆盖字段的新请求实例"""

支持覆盖的字段：

资料来源：interceptors.py:56-66

ToolCallInterceptor 接口

拦截器必须实现 ToolCallInterceptor 协议：

@runtime_checkable
class ToolCallInterceptor(Protocol):
    """拦截器接口协议"""
    
    async def __call__(
        self,
        request: MCPToolCallRequest,
        handler: Callable[[MCPToolCallRequest], Awaitable[MCPToolCallResult]],
    ) -> MCPToolCallResult:
        """拦截并处理工具调用"""
        ...

接口参数说明

返回值

返回类型为 MCPToolCallResult，支持以下三种格式：

资料来源：interceptors.py:28-33

拦截器链构建

拦截器链通过 _build_interceptor_chain() 函数构建，采用反向迭代方式将拦截器层层包装：

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 参数，将拦截器链注入到工具调用流程中：

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]:

工具调用执行流程

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 内部函数中，拦截器链被创建和应用：

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

使用示例

基本拦截器实现

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)

多拦截器组合使用

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 中的使用

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 类在初始化时接受全局拦截器配置：

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 []

配置参数说明

资料来源：langchain_mcp_adapters/client.py:50-60

最佳实践

1. 拦截器顺序

拦截器的执行顺序与其在列表中的顺序一致：

• 列表首位 → 最外层 → 最先收到请求，最后收到响应
• 列表末位 → 最内层 → 最后收到请求，最先收到响应

2. 错误处理

建议在最外层拦截器中实现统一的错误处理：

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() 方法进行不可变请求修改：

# 正确：创建新实例
modified = request.override(args={"query": "new value"})

# 错误：直接修改（不应该这样做）
request.args["query"] = "new value"

4. 性能考量

• 避免在拦截器中执行耗时的同步操作
• 使用异步处理保持非阻塞执行
• 对于高频调用，考虑拦截器的性能开销

概述

回调机制是 langchain-mcp-adapters 中用于处理 MCP 工具调用生命周期事件的核心组件。该机制允许开发者在工具执行的不同阶段插入自定义逻辑，实现进度监控、事件通知、错误处理等高级功能。

回调系统在 MCP 客户端与 LangChain 工具之间建立了桥梁，使得 MCP 服务器发出的通知（notifications）能够被 LangChain 端正确处理和响应。

核心组件

组件架构

graph TD
    A[MCP 服务器] -->|通知/进度| B[ClientSession]
    B --> C[Callbacks]
    C --> D[CallbackContext]
    D --> E[工具执行]
    E --> F[_MCPCallbacks]
    F --> G[进度回调]
    
    H[开发者代码] -->|自定义回调| C

CallbackContext

CallbackContext 是回调上下文的载体，用于在回调处理过程中传递工具调用的关键信息。

资料来源：langchain_mcp_adapters/tools.py:1-1

Callbacks

Callbacks 类是回调机制的主要入口点，封装了所有与 MCP 服务器通信的回调处理逻辑。

class Callbacks:
    """用于处理 MCP 通知和事件的回调接口"""

开发者通过实例化 Callbacks 并传递给 MultiServerMCPClient 或 load_mcp_tools 来启用自定义回调处理。

资料来源：langchain_mcp_adapters/client.py:1-1

_MCPCallbacks

_MCPCallbacks 是内部默认回调实现类，用于将 LangChain 格式的回调转换为 MCP SDK 所需的格式。

class _MCPCallbacks:
    """内部回调实现，桥接 LangChain 和 MCP SDK"""

资料来源：langchain_mcp_adapters/tools.py:1-1

工作流程

回调初始化流程

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

工具调用中的回调处理

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 函数中，回调按以下方式使用：

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 参数

资料来源：langchain_mcp_adapters/tools.py:199-230

convert_mcp_tool_to_langchain_tool 参数

资料来源：langchain_mcp_adapters/tools.py:232-290

使用示例

基本用法

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）配合使用，实现更复杂的工具调用控制：

graph TD
    A[请求] --> B[拦截器1]
    B --> C[拦截器2]
    C --> D[拦截器N]
    D --> E[execute_tool]
    E --> D
    D --> C
    C --> B
    B --> F[响应]

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 端处理。

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 类型，回调系统支持对工具调用请求和响应进行完整的拦截处理：

@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 的执行上下文：

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 库提供了两大核心转换模块，用于将 MCP（Model Context Protocol）的提示词（Prompts）和资源（Resources）转换为 LangChain 兼容的格式。这些模块使得 MCP 服务器可以作为 LangChain 和 LangGraph 应用的外部工具和数据来源。

资料来源：langchain_mcp_adapters/prompts.py:1-10，langchain_mcp_adapters/resources.py:1-15

架构概览

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<br/>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 消息对象。

函数签名：

def convert_mcp_prompt_message_to_langchain_message(
    message: PromptMessage,
) -> HumanMessage | AIMessage:

参数说明：

支持的转换规则：

异常处理：

资料来源：langchain_mcp_adapters/prompts.py:12-35

#### load_mcp_prompt()

异步函数，用于从 MCP 会话加载提示词并转换为 LangChain 消息列表。

函数签名：

async def load_mcp_prompt(
    session: ClientSession,
    name: str,
    *,
    arguments: dict[str, Any] | None = None,
) -> list[HumanMessage | AIMessage]:

参数说明：

返回值：

返回转换后的 LangChain 消息列表，消息类型为 HumanMessage 或 AIMessage。

工作流程：

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 对象。

函数签名：

def convert_mcp_resource_to_langchain_blob(
    resource_uri: str, contents: ResourceContents
) -> Blob:

参数说明：

支持的资源类型：

Blob 元数据：

转换后的 Blob 对象包含以下元数据：

异常处理：

资料来源：langchain_mcp_adapters/resources.py:20-42

#### get_mcp_resource()

异步函数，用于获取单个 MCP 资源并转换为 LangChain Blob 对象列表。

函数签名：

async def get_mcp_resource(session: ClientSession, uri: str) -> list[Blob]:

参数说明：

返回值：

返回 LangChain Blob 对象列表。如果资源内容为空，则返回空列表。

资料来源：langchain_mcp_adapters/resources.py:44-61

#### load_mcp_resources()

异步函数，用于批量加载 MCP 资源。

函数签名：

async def load_mcp_resources(
    session: ClientSession,
    *,
    uris: str | list[str] | None = None,
) -> list[Blob]:

参数说明：

加载行为：

[!note]

动态资源（Dynamic Resources）在 uris 为 None 时不会被加载，因为它们需要参数，而 MCP SDK 的 session.list_resources() 方法无法处理这类资源。

资料来源：langchain_mcp_adapters/resources.py:64-90

数据模型

MCP 类型引用

classDiagram
    class PromptMessage {
        +str role
        +Content content
    }

    class ResourceContents {
        <<abstract>>
    }

    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 类型对应

客户端集成

MultiServerMCPClient 提供了对提示词和资源转换的统一访问接口。

资料来源：langchain_mcp_adapters/client.py:1-60

核心方法

使用示例

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

支持的传输类型

类型转换流程

提示词转换流程

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

资源转换流程

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

错误处理

提示词转换错误

资源转换错误

与工具转换的关系

提示词和资源转换与 tools.py 中的工具转换模块共同构成了 MCP 到 LangChain 的完整适配体系。

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()：

# 推荐方式
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 参数为提示词提供上下文信息：

messages = await load_mcp_prompt(
    session,
    name="code_review",
    arguments={"repo": "langchain-mcp-adapters", "pr": 123}
)

3. 错误处理

实现适当的错误处理以应对资源不可用或格式不支持的情况：

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 服务器结构和数据的能力：

这两个模块与工具转换模块（tools.py）一起，构成了 MCP 到 LangChain 的完整适配层，使得开发者可以在 LangChain 和 LangGraph 应用中无缝使用 MCP 服务器提供的各种功能。

Pitfall Log / 踩坑日志

项目：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