langchain-mcp-adapters

LangChain MCP Adapters — connect LangChain agents to MCP (Model Context Protocol) servers. Load MCP tools, prompts, and resources as LangChain-compatible objects. Supports stdio, SSE, StreamableHTTP, and WebSocket transports. Includes interceptors, callbacks, and multi-server management.

Jump to install

Repository: enuno/claude-command-and-control
Last source activity: May 7, 2026 at 02:49
Detected SKILL.md language: English
Stars: 15
Forks: 3

Install options

The review-first prompt is selected by default. You can switch to a direct command or download a local copy.

Review the source files

Read SKILL.md and any companion files shown by SkillsMP before deciding whether to install.

Back to search Back to installation

load_mcp_resources()

get_mcp_resource()

load_mcp_prompt()

from langchain_mcp_adapters.client import MultiServerMCPClient

async with MultiServerMCPClient(
    connections={
        "filesystem": {
            "transport": "stdio",
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
        },
        "weather": {
            "transport": "streamable_http",
            "url": "https://weather-mcp.example.com/mcp",
        },
    },
    tool_name_prefix=True,   # tools become "filesystem_read_file", "weather_search"
) as client:
    tools = await client.get_tools()
    # Use tools with any LangChain agent
    agent = create_react_agent(llm, tools)

from langchain_mcp_adapters.sessions import (
    StdioConnection, SSEConnection, StreamableHttpConnection, WebsocketConnection
)

# stdio — local process (most common for CLI tools)
stdio: StdioConnection = {
    "transport": "stdio",
    "command": "python",
    "args": ["-m", "my_mcp_server"],
    "env": {"MY_API_KEY": "..."},
    "cwd": "/path/to/server",
}

# StreamableHTTP — remote server (recommended for network)
http: StreamableHttpConnection = {
    "transport": "streamable_http",
    "url": "https://my-mcp-server.example.com/mcp",
    "headers": {"Authorization": "Bearer my-token"},
    "timeout": timedelta(seconds=30),
}

# SSE — legacy remote (use streamable_http for new servers)
sse: SSEConnection = {
    "transport": "sse",
    "url": "https://my-mcp-server.example.com/sse",
}

from langchain_mcp_adapters.sessions import create_session
from langchain_mcp_adapters.tools import load_mcp_tools

async with create_session(connection) as session:
    tools: list[BaseTool] = await load_mcp_tools(
        session=session,
        server_name="my_server",
        tool_name_prefix=True,   # prefix tool names with server_name
    )

from langchain_mcp_adapters.interceptors import ToolCallInterceptor, MCPToolCallRequest, MCPToolCallResult

class RateLimitInterceptor(ToolCallInterceptor):
    async def intercept(self, request: MCPToolCallRequest, handler) -> MCPToolCallResult:
        await rate_limiter.acquire()
        return await handler(request)   # call next interceptor or actual tool

class RetryInterceptor(ToolCallInterceptor):
    async def intercept(self, request: MCPToolCallRequest, handler) -> MCPToolCallResult:
        for attempt in range(3):
            try:
                return await handler(request)
            except Exception:
                if attempt == 2:
                    raise

# First interceptor = outermost layer
client = MultiServerMCPClient(
    connections={...},
    tool_interceptors=[RateLimitInterceptor(), RetryInterceptor()],
)

from langchain_mcp_adapters.callbacks import LoggingMessageCallback, ProgressCallback

# Log all MCP messages
logging_cb = LoggingMessageCallback()

# Track progress of long-running operations
progress_cb = ProgressCallback(on_progress=lambda p: print(f"Progress: {p}%"))

client = MultiServerMCPClient(
    connections={...},
    callbacks=[logging_cb, progress_cb],
)

from langchain_mcp_adapters.resources import load_mcp_resources, get_mcp_resource
from langchain_mcp_adapters.prompts import load_mcp_prompt

async with create_session(connection) as session:
    # Load all resources as LangChain Blobs
    resources = await load_mcp_resources(session)

    # Get a specific resource
    blob = await get_mcp_resource(session, uri="file:///data/config.json")

    # Load a prompt and convert to LangChain messages
    messages = await load_mcp_prompt(session, name="summarize", arguments={"text": "..."})

from langchain_mcp_adapters.tools import to_fastmcp
from langchain_community.tools import DuckDuckGoSearchRun

lc_tools = [DuckDuckGoSearchRun()]
fastmcp_server = to_fastmcp(lc_tools)   # expose as MCP server
fastmcp_server.run()

MultiServerMCPClient(
    connections: dict[str, Connection] | None = None,
    callbacks: Callbacks | None = None,
    tool_interceptors: list[ToolCallInterceptor] | None = None,
    tool_name_prefix: bool = False,
)

Method	Returns	Description
`get_tools()`	`list[BaseTool]`	All tools from all connected servers
`get_prompt(server, name, args)`	`list[BaseMessage]`	Load a prompt as LangChain messages
`get_resources(server)`	`list[Blob]`	Load resources from a server
`session(server)`	context manager	Access raw MCP session for a server

Type	Transport	Best For
`StdioConnection`	`"stdio"`	Local CLI tools, local MCP servers
`StreamableHttpConnection`	`"streamable_http"`	Remote servers (recommended)
`SSEConnection`	`"sse"`	Legacy remote servers
`WebsocketConnection`	`"websocket"`	Bidirectional real-time connections

Function	Description
`load_mcp_tools(session, ...)`	Convert all MCP tools to `BaseTool` list
`convert_mcp_tool_to_langchain_tool(tool, session)`	Convert one MCP tool
`to_fastmcp(tools)`	Expose LangChain tools as FastMCP server
`load_mcp_resources(session)`	Load all resources as Blobs
`get_mcp_resource(session, uri)`	Load specific resource by URI
`load_mcp_prompt(session, name, args)`	Load prompt as LangChain messages
`create_session(connection)`	Create a raw MCP session (context manager)