langchain-ai/langchain langchain==1.4.0a2 on GitHub

Alpha preview of langchain.mcp — a first-party adapter that turns any MCP server into LangChain tools you can hand straight to create_agent.

Connection handling is FastMCP's, so its client features are available as-is rather than re-implemented behind a narrower interface.

pip install "langchain[mcp]==1.4.0a2"

Connect

MCPAdapter takes any target fastmcp.Client accepts — transport is inferred, so there is one entry point rather than one per protocol.

from langchain.agents import create_agent
from langchain.mcp import MCPAdapter

async with MCPAdapter("https://example.com/mcp") as adapter:
    agent = create_agent("anthropic:claude-sonnet-5", await adapter.get_tools())
    result = await agent.ainvoke({"messages": [{"role": "user", "content": "..."}]})

Valid targets: a URL, a local script path (launched over stdio), an in-process FastMCP server, a config naming several servers at once, or a fastmcp.Client you built yourself.

Tools returned by get_tools() hold the adapter's client, so they stay callable after the context exits — the async with block scopes discovery, not tool lifetime.

Auth, caching, timeouts — build the client

MCPAdapter takes two arguments: the target and elicitation. Everything else FastMCP supports is configured on a fastmcp.Client that you build and hand over as the target. This is the pattern to reach for whenever you need more than a bare connection:

from fastmcp.client import Client
from langchain.mcp import MCPAdapter

client = Client(
    "https://example.com/mcp",
    auth="oauth",       # or a bearer token string, or any httpx auth
    cache=True,         # opt-in response caching
    timeout=30,
)

async with MCPAdapter(client) as adapter:
    tools = await adapter.get_tools()

Auth accepts "oauth" to run the OAuth flow, a token string for bearer auth, or an httpx.Auth instance for anything custom — see FastMCP's auth docs. Per-server headers and auth can also be set in a multi-server config (below).

Caching is opt-in and off by default: cache=True enables it with defaults, honoring the server's own ttlMs and cacheScope hints; a CacheConfig customizes it. The cache is per-client and in-memory.

Everything else on fastmcp.Client — timeout, log_handler, progress_handler, message_handler, roots, sampling_handler — works the same way. The adapter passes your client through untouched, so FastMCP behavior is not re-implemented or restricted.

One caveat: with elicitation="interrupt", the adapter clones your client so it does not overwrite a callback you set. Configuration (auth, cache settings, handlers) carries over to the clone; cached entries do not, since the clone gets its own store.

adapter.client exposes the underlying client for prompts, resources, and anything else the adapter does not wrap.

Multiple servers

Point the adapter at a config and it fans out to every server through one connection, presenting a single tool list to your agent.

config = {
    "mcpServers": {
        "weather": {"url": "https://weather.example.com/mcp"},
        "calendar": {
            "url": "https://calendar.example.com/mcp",
            "headers": {"Authorization": "Bearer ..."},
        },
    }
}

async with MCPAdapter(config) as adapter:
    agent = create_agent("anthropic:claude-sonnet-5", await adapter.get_tools())

With more than one server, tools are namespaced by server name — weather_get_forecast, calendar_create_event — so collisions between servers are impossible. With exactly one server, the adapter connects directly and names are unprefixed. Each entry takes its own headers, auth, transport, and timeout, so servers with different credentials compose in one agent. A local server uses command/args instead of url and is launched over stdio. The config follows FastMCP's MCP JSON schema, so a config you already use elsewhere works here unchanged.

Old and new protocol servers, side by side

MCP has moved from the initialize handshake to server/discover, and servers in the wild sit on both sides of that line. FastMCP negotiates the era per connection, so the adapter reaches either without you selecting one:

# handshake-era server over SSE
legacy = MCPAdapter("https://legacy.example.com/sse")

# modern-era server over streamable HTTP
modern = MCPAdapter("https://modern.example.com/mcp")

Separate adapters negotiate independently and can run concurrently, each on its own era. This is covered by integration tests that stand up one server of each era and call both.

The one rule worth knowing: a multi-server config exposes a single era, so its oldest backend sets the era for the whole fleet. Mixing a handshake-era server into a config pulls the modern backends back to the handshake era — they still work, but era-gated features go with it. Keep a legacy server in its own adapter when you want the others on the modern protocol:

async with (
    MCPAdapter({"mcpServers": {...modern servers...}}) as modern,
    MCPAdapter("https://legacy.example.com/sse") as legacy,
):
    tools = await modern.get_tools() + await legacy.get_tools()

Results

Each tool is async. An MCP tool that runs and reports failure comes back as a ToolMessage with status="error" carrying the server's own error text, so the agent can correct itself and retry. Transport failures and unconvertible content raise instead — a model cannot act on those.

Structured output rides along on the tool message artifact:

from langchain.mcp import MCPToolArtifact

artifact: MCPToolArtifact | None = tool_message.artifact  # None when there is no structured content
artifact["structured_content"]

Elicitation — servers that ask questions mid-call

Some MCP tools need input before they can finish. Opt in, and the request surfaces as a LangGraph interrupt(), so the human already reviewing the agent's work answers the server too.

adapter = MCPAdapter(target, elicitation="interrupt")

The capability is opt-in rather than default because declaring it is a promise made on the wire: an agent with no path to a human cannot keep it. Left unset, nothing is declared, and a server whose tool requires an answer declines the call rather than running without one.

The run stops with a typed payload, and resumes with one answer per request key:

from langgraph.types import Command

result = await agent.ainvoke({"messages": [...]}, config)

[pause] = result["__interrupt__"]
pause.value["type"]       # "mcp_elicitation"
pause.value["tool_name"]  # the tool that is waiting
pause.value["requests"]   # each question, in the order to ask them

answer = {"responses": {key: {"action": "accept", "content": {"guests": 4}}}}
result = await agent.ainvoke(Command(resume=answer), config)

Requests narrow on mode: a "form" request carries requested_schema for the answer to satisfy, a "url" request carries an address for the human to visit. Answers narrow on action — "accept" (with content), "decline" (skip the question, let the call proceed), or "cancel" (abandon the tool call). Requires a checkpointer, as any interrupt does.

Sampling and roots are not answered through interrupts; leave those to your client's own handlers.

Types for handlers — MCPElicitationInterrupt, MCPElicitationRequest, MCPElicitationResponse, MCPElicitationResume, and the ELICITATION_INTERRUPT_TYPE discriminator — live in langchain.mcp.elicitation.