LangChain 模型上下文协议 (MCP) 适配器

类

名称	描述
MultiServerMCPClient	用于连接多个 MCP 服务器并从中加载与 LangChain 兼容的工具的客户端。

MultiServerMCPClient

用于连接多个 MCP 服务器并从中加载与 LangChain 兼容的工具的客户端。

方法

名称	描述
__init__	使用 MCP 服务器连接初始化 MultiServerMCPClient。
connect_to_server	连接到 MCP 服务器。
connect_to_server_via_stdio	使用 stdio 连接到特定的 MCP 服务器
connect_to_server_via_sse	使用 SSE 连接到特定的 MCP 服务器
connect_to_server_via_websocket	使用 Websockets 连接到特定的 MCP 服务器
get_tools	获取所有已连接服务器的工具列表。
get_prompt	从给定的 MCP 服务器获取提示。
get_resources	从给定的 MCP 服务器获取资源。

__init__

__init__(
    connections: (
        dict[
            str,
            StdioConnection
            | SSEConnection
            | WebsocketConnection,
        ]
        | None
    ) = None,
) -> None

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

参数

名称	类型	描述	默认值
connections	dict[str, StdioConnection | SSEConnection | WebsocketConnection] | None	一个字典，将服务器名称映射到连接配置。每个配置可以是 StdioConnection、SSEConnection 或 WebsocketConnection。如果为 None，则不建立初始连接。	None

示例

async with 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": "http://localhost:8000/sse",
            "transport": "sse",
        }
    }
) as client:
    all_tools = client.get_tools()
    ...

connect_to_server async

connect_to_server(
    server_name: str,
    *,
    transport: Literal[
        "stdio", "sse", "websocket"
    ] = "stdio",
    **kwargs: dict[str, Any]
) -> None

连接到 MCP 服务器。

这是一个通用方法，根据提供的 transport 参数调用 connect_to_server_via_stdio / connect_to_server_via_sse / connect_to_server_via_websocket 中的任一方法。

参数

名称	类型	描述	默认值
server_name	str	用于标识此服务器连接的名称	必需
transport	Literal['stdio', 'sse', 'websocket']	要使用的 transport 类型（"stdio"、"sse" 或 "websocket"），默认为 "stdio"	'stdio'
**kwargs	dict[str, Any]	传递给特定连接方法的附加参数	{}

引发

类型	描述
ValueError	如果 transport 未被识别
ValueError	如果指定的 transport 缺少必需参数

connect_to_server_via_stdio async

connect_to_server_via_stdio(
    server_name: str,
    *,
    command: str,
    args: list[str],
    env: dict[str, str] | None = None,
    encoding: str = DEFAULT_ENCODING,
    encoding_error_handler: Literal[
        "strict", "ignore", "replace"
    ] = DEFAULT_ENCODING_ERROR_HANDLER,
    session_kwargs: dict[str, Any] | None = None
) -> None

使用 stdio 连接到特定的 MCP 服务器

参数

名称	类型	描述	默认值
server_name	str	用于标识此服务器连接的名称	必需
command	str	要执行的命令	必需
args	list[str]	命令的参数	必需
env	dict[str, str] | None	命令的环境变量	None
encoding	str	字符编码	DEFAULT_ENCODING
encoding_error_handler	Literal['strict', 'ignore', 'replace']	如何处理编码错误	DEFAULT_ENCODING_ERROR_HANDLER
session_kwargs	dict[str, Any] | None	传递给 ClientSession 的附加关键字参数	None

connect_to_server_via_sse async

connect_to_server_via_sse(
    server_name: str,
    *,
    url: str,
    headers: dict[str, Any] | None = None,
    timeout: float = DEFAULT_HTTP_TIMEOUT,
    sse_read_timeout: float = DEFAULT_SSE_READ_TIMEOUT,
    session_kwargs: dict[str, Any] | None = None
) -> None

使用 SSE 连接到特定的 MCP 服务器

参数

名称	类型	描述	默认值
server_name	str	用于标识此服务器连接的名称	必需
url	str	SSE 服务器的 URL	必需
headers	dict[str, Any] | None	发送到 SSE 端点的 HTTP 头部	None
timeout	float	HTTP 超时	DEFAULT_HTTP_TIMEOUT
sse_read_timeout	float	SSE 读取超时	DEFAULT_SSE_READ_TIMEOUT
session_kwargs	dict[str, Any] | None	传递给 ClientSession 的附加关键字参数	None

connect_to_server_via_websocket async

connect_to_server_via_websocket(
    server_name: str,
    *,
    url: str,
    session_kwargs: dict[str, Any] | None = None
)

使用 Websockets 连接到特定的 MCP 服务器

参数

名称	类型	描述	默认值
server_name	str	用于标识此服务器连接的名称	必需
url	str	Websocket 端点的 URL	必需
session_kwargs	dict[str, Any] | None	传递给 ClientSession 的附加关键字参数	None

引发

类型	描述
ImportError	如果未安装 websockets 包

get_tools

get_tools() -> list[BaseTool]

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

get_prompt async

get_prompt(
    server_name: str,
    prompt_name: str,
    arguments: Optional[dict[str, Any]],
) -> 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	str | list[str] | None	要加载的可选资源 URI 或 URI 列表。如果未提供，将加载所有资源。	None

返回

类型	描述
list[Blob]	LangChain Blob 列表

函数

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

load_mcp_tools async

load_mcp_tools(session: ClientSession) -> list[BaseTool]

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

函数

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

load_mcp_prompt async

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

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

函数

名称	描述
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	ClientSession	MCP 客户端会话	必需
uris	str | list[str] | None	要加载的 URI 列表。如果为 None，将加载所有资源。注意：如果您指定 None，动态资源将不会被加载，因为它们需要提供参数，并且会被 MCP SDK 的 session.list_resources() 方法忽略。	None

返回

类型	描述
list[Blob]	LangChain Blob 列表

评论