MCP server LangChain integration

Why MCP and LangChain complement each other

LangChain excels at the orchestration layer: prompt construction, conversation memory, chain composition, and the tool-calling loop. MCP excels at the implementation layer: defining typed tool schemas, handling authentication, enforcing rate limits, and isolating external service logic behind a protocol boundary. Together they let you write clean agentic code in LangChain while keeping all external service complexity inside MCP servers that can be versioned, deployed, and monitored independently of the agent code.

Without MCP, LangChain tools are Python functions defined in the same codebase as the agent. Adding a new external service means adding a new dependency, new auth wiring, and new deployment coupling. With MCP, adding a new tool means pointing LangChain at a new MCP server endpoint — no agent code changes required.

Installing langchain-mcp-adapters

The adapter library is maintained by the LangChain team and published to PyPI:

pip install langchain-mcp-adapters langchain-anthropic
# or with uv:
uv add langchain-mcp-adapters langchain-anthropic

The library requires Python 3.10+ and langchain-core >= 0.3. It does not install a specific LLM provider — you bring your own (langchain-anthropic, langchain-openai, etc.).

MultiServerMCPClient — connection management

MultiServerMCPClient is the main entry point. It manages connections to one or more MCP servers and translates their tools into LangChain objects.

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_anthropic import ChatAnthropic
from langgraph.prebuilt import create_react_agent

async def main():
    async with MultiServerMCPClient(
        {
            "analytics": {
                "transport": "streamable_http",
                "url": "https://analytics.internal/mcp",
                "headers": {"Authorization": "Bearer " + os.environ["ANALYTICS_TOKEN"]},
            },
            "code_runner": {
                "transport": "stdio",
                "command": "npx",
                "args": ["-y", "@acme/code-runner-mcp"],
                "env": {"NODE_ENV": "production"},
            },
        }
    ) as client:
        tools = await client.get_tools()
        model = ChatAnthropic(model="claude-sonnet-4-6")
        agent = create_react_agent(model, tools)
        result = await agent.ainvoke({"messages": [("user", "Analyse last week's revenue")]})
        print(result["messages"][-1].content)

asyncio.run(main())

Three transport types are supported: streamable_http for HTTP/SSE MCP servers (the most common for production deployments), sse for legacy SSE-only servers, and stdio for locally-installed MCP packages. The headers field passes authentication to HTTP servers.

Tool schema translation

client.get_tools() calls tools/list on each MCP server and converts every tool's inputSchema into a LangChain StructuredTool. The tool's name and description are passed through verbatim to the LLM.

The tool description is the primary signal LLM models use to decide which tool to call. A description like "Get data" causes frequent wrong-tool selection. A description like "Returns daily revenue totals and per-SKU breakdown for a given date range. Use when the user asks about sales figures, revenue trends, or product performance." dramatically improves routing accuracy.

Persistent connections vs per-request connections

The most common production mistake with MultiServerMCPClient is reopening the connection for every request. Each async with MultiServerMCPClient(...): block opens a new transport connection to every configured MCP server. For HTTP servers this means a new TCP handshake and MCP initialize/initialized handshake on every agent invocation — adding 100–500 ms of latency per call and creating unnecessary load on the MCP servers.

The correct pattern for a long-running service (FastAPI app, Celery worker, background task loop) is to initialize the client once at startup and reuse it:

from contextlib import asynccontextmanager
from fastapi import FastAPI

mcp_client: MultiServerMCPClient | None = None
agent = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global mcp_client, agent
    mcp_client = MultiServerMCPClient({
        "analytics": {"transport": "streamable_http", "url": "https://analytics.internal/mcp",
                       "headers": {"Authorization": "Bearer " + os.environ["ANALYTICS_TOKEN"]}},
    })
    await mcp_client.__aenter__()
    tools = await mcp_client.get_tools()
    agent = create_react_agent(ChatAnthropic(model="claude-sonnet-4-6"), tools)
    yield
    await mcp_client.__aexit__(None, None, None)

app = FastAPI(lifespan=lifespan)

@app.post("/analyze")
async def analyze(prompt: str):
    result = await agent.ainvoke({"messages": [("user", prompt)]})
    return {"response": result["messages"][-1].content}

For batch pipelines and one-shot scripts, the context manager pattern is appropriate — you want the connection closed cleanly after the job completes.

Error propagation from MCP to LangChain

When an MCP tool returns { isError: true, content: [...] }, langchain-mcp-adapters raises a ToolException inside the LangChain tool execution. The agent executor's default behavior is to inject the error message back into the conversation as a tool result, allowing the LLM to decide whether to retry with different arguments, try a different tool, or give up.

from langchain_core.tools import ToolException

# The agent executor handles this automatically, but you can also
# intercept errors at the chain level:
from langchain_core.runnables import RunnableLambda

def handle_tool_error(error: ToolException) -> str:
    return f"Tool error: {error}. Try rephrasing your request."

agent_with_error_handling = agent.with_fallbacks(
    [RunnableLambda(handle_tool_error)]
)

Do not suppress ToolException silently — the LLM needs the error text to self-correct. MCP tool implementations should return descriptive error messages (not just "Internal server error") so the LLM can diagnose and retry effectively.

Streaming tool output and progress

LangChain's streaming interface (agent.astream_events()) surfaces MCP progress notifications as they arrive, enabling live progress UI in your application:

async for event in agent.astream_events(
    {"messages": [("user", "Export all records from Q1")]},
    version="v2",
):
    if event["event"] == "on_tool_start":
        print(f"Calling tool: {event['name']}")
    elif event["event"] == "on_tool_end":
        print(f"Tool complete: {event['name']}")
    elif event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="", flush=True)

Progress notifications from long-running MCP tools arrive between on_tool_start and on_tool_end events. The LangChain streaming interface does not expose them as distinct event types — they are delivered over the SSE connection and consumed internally. To surface them to your UI, implement a progress callback on the MCP server's SSE stream directly.

Multi-server tool namespacing

When connecting to multiple MCP servers, tool names may collide if two servers define a tool with the same name (e.g., both an analytics server and a database server define query). MultiServerMCPClient does not namespace tools automatically — you must handle this in the MCP server's tool definitions by using unique names (e.g., analytics_query vs db_query), or by using separate MultiServerMCPClient instances and merging the tools manually:

analytics_tools = await analytics_client.get_tools()
database_tools = await database_client.get_tools()

# Prefix to disambiguate:
for tool in database_tools:
    tool.name = f"db_{tool.name}"

all_tools = analytics_tools + database_tools
agent = create_react_agent(model, all_tools)

If you control the MCP servers, the cleaner approach is to give tools globally unique names at the server level. See MCP multi-server patterns for namespace prefixing conventions.

Health checks before agent invocation

A LangChain agent that invokes a dead MCP tool fails at tool execution time — the LLM has already consumed tokens to choose and call the tool before the failure surfaces. For high-cost or time-sensitive agents, verify MCP server health before invoking:

import httpx

async def check_mcp_health(url: str, token: str) -> bool:
    try:
        async with httpx.AsyncClient(timeout=5.0) as http:
            # Probe with a minimal initialize request
            resp = await http.post(
                url,
                json={"jsonrpc": "2.0", "method": "initialize", "params": {
                    "protocolVersion": "2025-03-26",
                    "capabilities": {},
                    "clientInfo": {"name": "health-check", "version": "1.0"},
                }, "id": 1},
                headers={"Authorization": f"Bearer {token}"},
            )
            return resp.status_code == 200
    except Exception:
        return False

async def run_agent_safely(prompt: str):
    if not await check_mcp_health(ANALYTICS_URL, os.environ["ANALYTICS_TOKEN"]):
        raise RuntimeError("Analytics MCP server is unavailable — aborting agent run")
    result = await agent.ainvoke({"messages": [("user", prompt)]})

Monitoring MCP servers that power LangChain pipelines

LangChain raises ToolException when an MCP tool call fails, but by that point the failure has already interrupted an agent run that may have spent seconds or minutes on prior steps. The problem is worse in batch pipelines where agent runs are triggered on a schedule — if the MCP server died at 3 AM, your 4 AM scheduled pipeline fails silently, and you discover the problem when the business team notices missing reports at 9 AM.

AliveMCP monitors your MCP server's protocol endpoint independently of LangChain — probing initialize → tools/list every minute and alerting on the first failure. When your LangChain pipeline starts at 4 AM, you already know whether the MCP server is healthy. Alert on MCP infrastructure separately from application-layer LangChain errors so that network and deployment failures never silently absorb tool-call retries.

Further reading

• MCP server LangGraph integration — stateful agents with MCP tools
• MCP server CrewAI integration — multi-agent teams with MCP tools
• MCP multi-server patterns — aggregating tools from multiple servers
• MCP server authentication — OAuth, JWT, and API key patterns
• MCP server tool design — naming, schemas, and LLM-friendly descriptions
• AliveMCP — continuous protocol monitoring for MCP servers