【LangGraph】MCP Adapters（模型上下文协议 适配器）：连接多个 MCP 服务器并加载 LangChain 兼容工具、提示和资源的类和函数
LangChain 模型上下文协议（MCP）适配器
MCP 适配器模块提供了用于连接多个 MCP 服务器并加载 LangChain 兼容工具、提示和资源的类和函数。
1. 类
1.1 类 MultiServerMCPClient
描述：
MultiServerMCPClient 是用于连接多个 MCP 服务器并从中加载 LangChain 兼容工具、提示和资源的客户端。
1.1.1 方法
| 名称 | 描述 |
|---|---|
__init__ | 使用 MCP 服务器连接初始化 MultiServerMCPClient。 |
session | 连接到 MCP 服务器并初始化会话。 |
get_tools | 获取所有连接服务器的所有工具列表。 |
get_prompt | 从指定 MCP 服务器获取提示。 |
get_resources | 从指定 MCP 服务器获取资源。 |
方法 __init__
__init__(
connections: dict[str, Connection] | None = None,
) -> None
描述：
使用 MCP 服务器连接初始化 MultiServerMCPClient。
参数：
| 名称 | 类型 | 描述 | 默认值 |
|---|---|---|---|
connections | dict[str, Connection] | None | 映射服务器名称到连接配置的字典。如果为 None，则不建立初始连接。 | None |
方法 session
async session(
server_name: str, *, auto_initialize: bool = True
) -> AsyncIterator[ClientSession]
描述：
连接到 MCP 服务器并初始化会话。
参数：
| 名称 | 类型 | 描述 | 默认值 |
|---|---|---|---|
server_name | str | 标识此服务器连接的名称。 | 必填 |
auto_initialize | bool | 是否自动初始化会话。 | True |
引发异常：
| 类型 | 描述 |
|---|---|
ValueError | 如果服务器名称未在连接中找到。 |
返回值：
| 类型 | 描述 |
|---|---|
AsyncIterator[ClientSession] | 已初始化的客户端会话异步迭代器。 |
方法 get_tools
async get_tools(
*, server_name: str | None = None
) -> list[BaseTool]
描述：
获取所有连接服务器的所有工具列表。
参数：
| 名称 | 类型 | 描述 | 默认值 |
|---|---|---|---|
server_name | str | None | 获取工具的可选服务器名称。如果为 None，返回所有服务器的所有工具（默认）。 | None |
注意：
每次工具调用将创建一个新会话。
返回值：
| 类型 | 描述 |
|---|---|
list[BaseTool] | LangChain 工具列表。 |
方法 get_prompt
async get_prompt(
server_name: str,
prompt_name: str,
*,
arguments: dict[str, Any] | None = None
) -> list[HumanMessage | AIMessage]
描述：
从指定 MCP 服务器获取提示。
参数：
| 名称 | 类型 | 描述 | 默认值 |
|---|---|---|---|
server_name | str | 获取提示的服务器名称。 | 必填 |
prompt_name | str | 提示的名称。 | 必填 |
arguments | dict[str, Any] | None | 提示的可选参数字典。 | None |
返回值：
| 类型 | 描述 |
|---|---|
list[HumanMessage | AIMessage] | LangChain 消息列表（人类消息或 AI 消息）。 |
方法 get_resources
async get_resources(
server_name: str, *, uris: str | list[str] | None = None
) -> list[Blob]
描述：
从指定 MCP 服务器获取资源。
参数：
| 名称 | 类型 | 描述 | 默认值 |
|---|---|---|---|
server_name | str | 获取资源的服务器名称。 | 必填 |
uris | `str | list[str] | None` |
返回值：
| 类型 | 描述 |
|---|---|
list[Blob] | LangChain Blob 列表。 |
1.1.2 示例
示例：基本用法（每次工具调用启动新会话）
from langchain_mcp_adapters.client import MultiServerMCPClient
client = MultiServerMCPClient(
{
"math": {
"command": "python",
# 请确保更新为您的 math_server.py 文件的完整绝对路径
"args": ["/path/to/math_server.py"],
"transport": "stdio",
},
"weather": {
# 请确保在端口 8000 上启动您的天气服务器
"url": "http://localhost:8000/mcp",
"transport": "streamable_http",
}
}
)
all_tools = await client.get_tools()
示例：显式启动会话
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)
2. 函数
2.1 函数 load_mcp_tools
async load_mcp_tools(
session: ClientSession | None,
*,
connection: Connection | None = None
) -> list[BaseTool]
描述：
加载所有可用的 MCP 工具并将其转换为 LangChain 工具。
参数：
| 名称 | 类型 | 描述 | 默认值 |
|---|---|---|---|
session | `ClientSession | None` | MCP 客户端会话。 |
connection | `Connection | None` | 可选连接配置，用于在未提供会话时创建新会话。 |
返回值：
| 类型 | 描述 |
|---|---|
list[BaseTool] | LangChain 工具列表。 |
2.2 函数 load_mcp_prompt
async load_mcp_prompt(
session: ClientSession,
name: str,
*,
arguments: dict[str, Any] | None = None
) -> list[HumanMessage | AIMessage]
描述：
加载 MCP 提示并将其转换为 LangChain 消息。
参数：
| 名称 | 类型 | 描述 | 默认值 |
|---|---|---|---|
session | ClientSession | MCP 客户端会话。 | 必填 |
name | str | 提示的名称。 | 必填 |
arguments | `dict[str, Any] | None` | 提示的可选参数字典。 |
返回值：
| 类型 | 描述 |
|---|---|
| `list[HumanMessage | AIMessage]` |
2.3 函数 load_mcp_resources
async load_mcp_resources(
session: ClientSession,
*,
uris: str | list[str] | None = None
) -> list[Blob]
描述：
加载 MCP 资源并将其转换为 LangChain Blob。
参数：
| 名称 | 类型 | 描述 | 默认值 |
|---|---|---|---|
session | ClientSession | MCP 客户端会话。 | 必填 |
uris | `str | list[str] | None` |
返回值：
| 类型 | 描述 |
|---|---|
list[Blob] | LangChain Blob 列表。 |
3. 总结
- MCP 适配器机制：LangChain 的 MCP 适配器模块提供了连接多个 MCP 服务器并加载工具、提示和资源的客户端，支持与 LangChain 生态系统的无缝集成。
- 关键类：
MultiServerMCPClient：客户端类，用于管理多个 MCP 服务器连接并加载资源。
- 关键函数：
load_mcp_tools：加载 MCP 工具并转换为 LangChain 工具。load_mcp_prompt：加载 MCP 提示并转换为 LangChain 消息。load_mcp_resources：加载 MCP 资源并转换为 LangChain Blob。
- 使用建议：
- 使用
MultiServerMCPClient连接多个 MCP 服务器，灵活管理工具和资源。 - 通过
session方法显式管理会话以优化性能，或使用get_tools自动创建会话。 - 指定
uris参数以加载特定资源，避免加载不需要的动态资源。 - 确保服务器连接配置（如路径、端口、传输方式）正确，以避免
ValueError。 - 在生产环境中，使用异步编程模式以提高效率。
- 使用
- 注意事项：
load_mcp_resources的uris=None不会加载动态资源，需显式指定 URI。- 确保 MCP 服务器运行并可访问，例如，天气服务器需在指定端口（如 8000）启动。
MCP 适配器模块为 LangChain 提供了强大的外部服务器资源整合能力，适合构建需要动态加载工具和提示的复杂工作流。
参考资料：
- https://langchain-ai.github.io/langgraph/reference/mcp/
