Skip to content

LangChain Model Context Protocol (MCP) Adapters

Classes:

  • MultiServerMCPClient –

    Client for connecting to multiple MCP servers and loading LangChain-compatible tools from them.

MultiServerMCPClient

Client for connecting to multiple MCP servers and loading LangChain-compatible tools from them.

Methods:

__init__ 

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

Initialize a MultiServerMCPClient with MCP servers connections.

Parameters:

  • connections (dict[str, StdioConnection | SSEConnection | WebsocketConnection] | None, default: None ) –

    A dictionary mapping server names to connection configurations. Each configuration can be a StdioConnection, SSEConnection or a WebsocketConnection. If None, no initial connections are established.

Example:

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

Connect to an MCP server.

This is a generic method that calls either of connect_to_server_via_stdio / connect_to_server_via_sse / connect_to_server_via_websocket based on the provided transport parameter.

Parameters:

  • server_name (str) –

    Name to identify this server connection

  • transport (Literal['stdio', 'sse', 'websocket'], default: 'stdio' ) –

    Type of transport to use ("stdio" or "sse" or "websocket"), defaults to "stdio"

  • **kwargs (dict[str, Any], default: {} ) –

    Additional arguments to pass to the specific connection method

Raises:

  • ValueError –

    If transport is not recognized

  • ValueError –

    If required parameters for the specified transport are missing

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

Connect to a specific MCP server using stdio

Parameters:

  • server_name (str) –

    Name to identify this server connection

  • command (str) –

    Command to execute

  • args (list[str]) –

    Arguments for the command

  • env (dict[str, str] | None, default: None ) –

    Environment variables for the command

  • encoding (str, default: DEFAULT_ENCODING ) –

    Character encoding

  • encoding_error_handler (Literal['strict', 'ignore', 'replace'], default: DEFAULT_ENCODING_ERROR_HANDLER ) –

    How to handle encoding errors

  • session_kwargs (dict[str, Any] | None, default: None ) –

    Additional keyword arguments to pass to the ClientSession

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

Connect to a specific MCP server using SSE

Parameters:

  • server_name (str) –

    Name to identify this server connection

  • url (str) –

    URL of the SSE server

  • headers (dict[str, Any] | None, default: None ) –

    HTTP headers to send to the SSE endpoint

  • timeout (float, default: DEFAULT_HTTP_TIMEOUT ) –

    HTTP timeout

  • sse_read_timeout (float, default: DEFAULT_SSE_READ_TIMEOUT ) –

    SSE read timeout

  • session_kwargs (dict[str, Any] | None, default: None ) –

    Additional keyword arguments to pass to the ClientSession

connect_to_server_via_websocket async 

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

Connect to a specific MCP server using Websockets

Parameters:

  • server_name (str) –

    Name to identify this server connection

  • url (str) –

    URL of the Websocket endpoint

  • session_kwargs (dict[str, Any] | None, default: None ) –

    Additional keyword arguments to pass to the ClientSession

Raises:

  • ImportError –

    If websockets package is not installed

get_tools 

get_tools() -> list[BaseTool]

Get a list of all tools from all connected servers.

get_prompt async 

get_prompt(
    server_name: str,
    prompt_name: str,
    arguments: Optional[dict[str, Any]],
) -> list[HumanMessage | AIMessage]

Get a prompt from a given MCP server.

get_resources async 

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

Get resources from a given MCP server.

Parameters:

  • server_name (str) –

    Name of the server to get resources from

  • uris (str | list[str] | None, default: None ) –

    Optional resource URI or list of URIs to load. If not provided, all resources will be loaded.

Returns:

  • list[Blob] –

    A list of LangChain Blobs

Functions:

  • load_mcp_tools –

    Load all available MCP tools and convert them to LangChain tools.

load_mcp_tools async 

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

Load all available MCP tools and convert them to LangChain tools.

Functions:

load_mcp_prompt async 

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

Load MCP prompt and convert to LangChain messages.

Functions:

load_mcp_resources async 

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

Load MCP resources and convert them to LangChain Blobs.

Parameters:

  • session (ClientSession) –

    MCP client session

  • uris (str | list[str] | None, default: None ) –

    List of URIs to load. If None, all resources will be loaded. NOTE: if you specify None, dynamic resources will NOT be loaded, as they need the parameters to be provided, and are ignored by MCP SDK's session.list_resources() method.

Returns:

  • list[Blob] –

    A list of LangChain Blobs

Comments