【LangGraph】MCP Adapters（模型上下文协议 适配器）：连接多个 MCP 服务器并加载 LangChain 兼容工具、提示和资源的类和函数_multiservermcpclient

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 MultiServerMCPClientclient = 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 MultiServerMCPClientfrom langchain_mcp_adapters.tools import load_mcp_toolsclient = 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/