langchain-mcp-adapters

langchain-mcp-adapters 软件包的参考文档。

langchain_mcp_adapters.client

用于连接到多个 MCP 服务器并加载 LangChain 工具/资源的客户端。

该模块提供了 MultiServerMCPClient 类，用于管理到多个 MCP 服务器的连接，并从中加载工具、提示和资源。

MultiServerMCPClient

用于连接到多个 MCP 服务器的客户端。

从 MCP 服务器加载与 LangChain 兼容的工具、提示和资源。

方法	描述
__init__	使用 MCP 服务器连接初始化一个 MultiServerMCPClient。
session	连接到一个 MCP 服务器并初始化一个会话。
get_tools	从所有已连接的服务器获取所有工具的列表。
get_prompt	从给定的 MCP 服务器获取一个提示。
get_resources	从给定的 MCP 服务器获取资源。
__aenter__	异步上下文管理器的入口点。
__aexit__	异步上下文管理器的退出点。

__init__

__init__(
    connections: dict[str, Connection] | None = None,
    *,
    callbacks: Callbacks | None = None,
    tool_interceptors: list[ToolCallInterceptor] | None = None,
) -> None

使用 MCP 服务器连接初始化一个 MultiServerMCPClient。

参数	描述
connections	一个将服务器名称映射到连接配置的 `dict`。如果为 `None`，则不建立初始连接。 类型： dict[str, Connection] | None 默认值： None
回调	用于处理通知和事件的可选回调。 类型： Callbacks | None 默认值： None
tool_interceptors	可选的工具调用拦截器列表，用于修改请求和响应。 类型： list[ToolCallInterceptor] | None 默认值： None

基本用法（在每次工具调用时启动一个新会话）

from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient(
    {
        "math": {
            "command": "python",
            # Make sure to update to the full absolute path to your
            # math_server.py file
            "args": ["/path/to/math_server.py"],
            "transport": "stdio",
        },
        "weather": {
            # Make sure you start your weather server on port 8000
            "url": "https://: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)

session async

session(
    server_name: str, *, auto_initialize: bool = True
) -> AsyncIterator[ClientSession]

连接到一个 MCP 服务器并初始化一个会话。

参数	描述
server_name	用于标识此服务器连接的名称 类型： str
auto_initialize	是否自动初始化会话 类型: bool 默认值: True

引发	描述
ValueError	如果服务器名称在连接中未找到

YIELDS	描述
AsyncIterator[ClientSession]	一个已初始化的 `ClientSession`

get_tools async

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

从所有已连接的服务器获取所有工具的列表。

参数	描述
server_name	要从中获取工具的服务器的可选名称。如果为 `None`，将返回所有服务器的所有工具。 类型: str | 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 服务器获取一个提示。

get_resources async

get_resources(server_name: str, *, uris: str | list[str] | None = None) -> list[Blob]

从给定的 MCP 服务器获取资源。

参数	描述
server_name	要从中获取资源的服务器名称 类型： str
uris	要加载的可选资源 URI 或 URI 列表。如果未提供，将加载所有资源。 类型： str | list[str] | None 默认值： None

返回	描述
list[Blob]	LangChain Blob 对象列表。

__aenter__ async

__aenter__() -> MultiServerMCPClient

异步上下文管理器的入口点。

引发	描述
NotImplementedError	上下文管理器支持已被移除。

__aexit__

__aexit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None

异步上下文管理器的退出点。

参数	描述
exc_type	如果发生异常，则为异常类型。 类型： type[BaseException] | None
exc_val	如果发生异常，则为异常值。 类型： BaseException | None
exc_tb	如果发生异常，则为异常回溯信息。 类型： TracebackType | None

引发	描述
NotImplementedError	上下文管理器支持已被移除。

langchain_mcp_adapters.tools

用于将 MCP 工具转换为 LangChain 工具的工具适配器。

该模块提供了将 MCP 工具转换为 LangChain 兼容工具、处理工具执行以及管理两种格式之间工具转换的功能。

函数	描述
load_mcp_tools	加载所有可用的 MCP 工具，并将其转换为 LangChain 工具。

load_mcp_tools async

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,
) -> list[BaseTool]

加载所有可用的 MCP 工具，并将其转换为 LangChain 工具。

参数	描述
session	MCP 客户端会话。如果为 `None`，则必须提供连接。 类型： ClientSession | None
connection	如果会话为 `None`，则用于创建新会话的连接配置。 类型： Connection | None 默认值： None
回调	用于处理通知和事件的可选 `Callbacks`。 类型： Callbacks | None 默认值： None
tool_interceptors	用于工具调用处理的可选拦截器列表。 类型： list[ToolCallInterceptor] | None 默认值： None
server_name	这些工具所属的服务器名称。 类型: str | None 默认值: None

返回	描述
list[BaseTool]	LangChain 工具列表。工具注解将作为工具元数据对象的一部分返回。

引发	描述
ValueError	如果会话和连接都未提供。

langchain_mcp_adapters.prompts

用于将 MCP 提示转换为 LangChain 消息的提示适配器。

该模块提供了将 MCP 提示消息转换为 LangChain 消息对象的功能，处理用户和助手两种消息类型。

函数	描述
load_mcp_prompt	加载 MCP 提示并转换为 LangChain 消息。

load_mcp_prompt async

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

加载 MCP 提示并转换为 LangChain 消息。

参数	描述
session	MCP 客户端会话。 类型： ClientSession
name	要加载的提示名称。 类型： str
arguments	传递给提示的可选参数。 类型: dict[str, Any] | None 默认值: None

返回	描述
list[HumanMessage | AIMessage]	从 MCP 提示转换而来的 LangChain 消息列表。

langchain_mcp_adapters.resources

用于将 MCP 资源转换为 LangChain Blob 对象的资源适配器。

该模块提供了将 MCP 资源转换为 LangChain Blob 对象的功能，处理文本和二进制两种资源内容类型。

函数	描述
load_mcp_resources	加载 MCP 资源并将其转换为 LangChain Blob 对象。

load_mcp_resources async

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

加载 MCP 资源并将其转换为 LangChain Blob 对象。

参数	描述
session	MCP 客户端会话。 类型： ClientSession
uris	要加载的 URI 列表。如果为 `None`，将加载所有资源。 注意 当指定为 `None` 时，动态资源将不会被加载，因为它们需要参数，并且会被 MCP SDK 的 `session.list_resources()` 方法忽略。 类型： str | list[str] | None 默认值： None

返回	描述
list[Blob]	LangChain Blob 对象列表。

引发	描述
RuntimeError	如果在获取资源时发生错误。