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/