LangGraph + MCP Integration Guide: Building Governed Multi-Agent Systems

What you'll build

By the end of this guide you will have a working pattern for a LangGraph multi-agent system in which every agent reaches enterprise systems exclusively through MCP servers rather than direct API calls baked into the agent's code. Concretely, that means:

• One or more MCP servers, each exposing a scoped set of tools for one system of record (CRM, payments, documents), with authentication and audit logging built in at the server layer.
• LangGraph tool wrappers that call those MCP tools under the hood and translate results into graph state, so from LangGraph's point of view an MCP tool looks like any other bound tool.
• An agent graph with distinct reasoning, tool execution, and validation nodes, looping until a task is complete or handed off to a human.
• Workflow governance layered on top: checkpoints, timeouts, and error boundaries that MCP itself does not provide.
• A supervisor pattern for multi-agent coordination, where a routing agent delegates to specialists, and each specialist is scoped to only the MCP servers its job requires.

The result is a system where governance is not a policy document — it is enforced twice, once by the MCP server per tool call and once by the LangGraph runtime per workflow step.

Prerequisites

• Python 3.10 or later.
• The langgraph package (and langchain-core for message and tool types).
• An MCP SDK for your language of choice (the official Python and TypeScript SDKs both work; examples below use Python-style pseudocode).
• At least one running MCP server to connect to — either one you build in Step 1, or an existing internal server.
• Working familiarity with LangGraph's state machine model (nodes, edges, and a shared state object) and with the MCP tool protocol (tool discovery, typed inputs, and JSON results).

If either of those two foundations is new to you, it is worth reading What Is LangGraph? and What Is an MCP Server? before continuing — this guide assumes both concepts and moves straight to integration.

Step 1: Set up your MCP server

Before LangGraph enters the picture, you need at least one MCP server that exposes real tools with real schemas. A production MCP server does three things for every tool it registers: it publishes a name, description, and typed input schema so a model can decide when and how to call it; it authenticates and authorizes the caller before touching your system; and it logs the call, whether it succeeds or fails.

Conceptually, a tool definition looks like this — the schema is what the model reasons over, and the handler is where your authorization and logging actually live:

from mcp.server import Server
from mcp.types import Tool, TextContent

server = Server("crm-tools")

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="get_account",
            description="Fetch a CRM account by ID. Read-only.",
            inputSchema={
                "type": "object",
                "properties": {
                    "account_id": {"type": "string"}
                },
                "required": ["account_id"],
            },
        ),
        Tool(
            name="update_account_status",
            description="Update a CRM account's status. Requires write scope.",
            inputSchema={
                "type": "object",
                "properties": {
                    "account_id": {"type": "string"},
                    "status": {"type": "string", "enum": ["active", "paused", "churned"]},
                },
                "required": ["account_id", "status"],
            },
        ),
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict, context) -> list[TextContent]:
    identity = authenticate(context)          # who is calling
    authorize(identity, tool=name)            # are they allowed to call this tool
    audit_log.record(identity, name, arguments)

    if name == "get_account":
        result = crm_client.get_account(arguments["account_id"])
    elif name == "update_account_status":
        result = crm_client.update_status(
            arguments["account_id"], arguments["status"]
        )
    else:
        raise ValueError(f"Unknown tool: {name}")

    return [TextContent(type="text", text=json.dumps(result))]

Note what does not live in the agent's prompt or code: the CRM credentials, the authorization rule, and the audit trail all live inside the server. This is the entire point of the protocol — the agent only ever sees get_account and update_account_status as named, typed operations. It cannot compose a raw query against the CRM even if a prompt injection tries to talk it into one. If you're building this server for production rather than a prototype, our MCP Server Development service covers the auth mapping, PII handling, and audit export most enterprise deployments need beyond this skeleton.

Step 2: Create MCP tool wrappers for LangGraph

LangGraph does not speak MCP natively — you write a thin wrapper per tool (or a generic wrapper that discovers tools dynamically) that connects to the MCP server, calls the tool, and returns a result LangGraph can put into its state. The wrapper is responsible for exactly three things: opening or reusing an MCP client session, invoking the named tool with validated arguments, and parsing the MCP response back into a plain value the graph can reason over.

from langchain_core.tools import tool
from mcp import ClientSession
from mcp.client.sse import sse_client

MCP_SERVER_URL = "https://internal.example.com/mcp/crm"

async def call_mcp_tool(tool_name: str, arguments: dict) -> dict:
    async with sse_client(MCP_SERVER_URL) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            result = await session.call_tool(tool_name, arguments)
            # MCP results are a list of content blocks; unwrap the text block
            text = result.content[0].text
            return json.loads(text)

@tool
async def get_account(account_id: str) -> dict:
    """Fetch a CRM account by ID through the CRM MCP server."""
    return await call_mcp_tool("get_account", {"account_id": account_id})

@tool
async def update_account_status(account_id: str, status: str) -> dict:
    """Update a CRM account's status through the CRM MCP server."""
    return await call_mcp_tool(
        "update_account_status", {"account_id": account_id, "status": status}
    )

crm_tools = [get_account, update_account_status]

From here, crm_tools is a normal list of LangChain-style tools you can bind to a model or hand to a LangGraph tool node — LangGraph has no idea, and does not need to know, that MCP is underneath. In production, replace the per-call sse_client(...) with a long-lived, pooled connection (see the production checklist below) so you are not paying a handshake cost on every tool call.

Step 3: Build the agent graph

With MCP tools wrapped as LangGraph tools, the graph itself follows a pattern that generalizes well beyond this example: a reasoning node where the model decides what to do next, a tool execution node that actually calls the MCP tools the reasoning node requested, and a validation node that checks the tool results before the graph is allowed to either loop back or finish.

from langgraph.graph import StateGraph, END, MessagesState

def reasoning_node(state: MessagesState):
    # The LLM sees prior messages + tool results, decides the next tool call
    # (or decides it's done, in which case it emits a plain response).
    response = model.bind_tools(crm_tools).invoke(state["messages"])
    return {"messages": [response]}

async def tool_execution_node(state: MessagesState):
    last = state["messages"][-1]
    results = []
    for call in last.tool_calls:
        tool_fn = next(t for t in crm_tools if t.name == call["name"])
        output = await tool_fn.ainvoke(call["args"])
        results.append(ToolMessage(content=json.dumps(output), tool_call_id=call["id"]))
    return {"messages": results}

def validation_node(state: MessagesState):
    last_result = state["messages"][-1]
    if is_malformed(last_result) or violates_policy(last_result):
        # Route back to reasoning with an explicit correction, don't silently continue
        return {"messages": [SystemMessage(content="Tool result failed validation, retry.")]}
    return {}

def route_after_reasoning(state: MessagesState):
    last = state["messages"][-1]
    return "tool_execution" if getattr(last, "tool_calls", None) else END

graph = StateGraph(MessagesState)
graph.add_node("reasoning", reasoning_node)
graph.add_node("tool_execution", tool_execution_node)
graph.add_node("validation", validation_node)

graph.set_entry_point("reasoning")
graph.add_conditional_edges("reasoning", route_after_reasoning, {
    "tool_execution": "tool_execution",
    END: END,
})
graph.add_edge("tool_execution", "validation")
graph.add_edge("validation", "reasoning")

agent = graph.compile()

The loop is deliberate: reasoning → tool execution → validation → back to reasoning, until the reasoning node decides no further tool calls are needed and the conditional edge routes to END. The validation node is what keeps MCP's per-call governance from being the only check in the system — it is where you catch a tool result that is technically valid MCP output but wrong for this workflow (an empty account record, a status change that contradicts business rules) before the reasoning node acts on it as ground truth.

Step 4: Add governance

MCP gives you tool-level governance: which identity can call which tool, with what inputs, logged how. That is necessary but not sufficient. A perfectly scoped tool called in an infinite loop, or called without a human ever reviewing a high-stakes action, is still a runaway agent. LangGraph adds the workflow-level governance MCP does not attempt to provide:

• Human-in-the-loop checkpoints — use LangGraph's interrupt mechanism to pause the graph before any node that calls a state-changing MCP tool (for example, update_account_status), persist the paused state, and resume only after explicit approval.
• State validation between steps — the validation node from Step 3 is the enforcement point; make it strict rather than advisory, since it is the only place that sees both what the tool returned and what the workflow expected.
• Timeout guards — cap both the wall-clock time per node and the total number of reasoning → tool → validation loops, so a model stuck re-requesting the same tool cannot run indefinitely or burn an unbounded token budget.
• Error boundaries — wrap tool execution nodes so an MCP server error (timeout, auth failure, malformed response) routes to a defined recovery path instead of throwing an unhandled exception that kills the run.

This two-layer model — MCP enforcing what a single call may do, LangGraph enforcing what the workflow as a whole may do — is the same governance philosophy we describe more generally in our intelligent automation guide: control belongs at every layer that can fail independently, not just the outermost one.

Step 5: Multi-agent coordination

A single graph with one set of tools works for a narrow workflow. Most enterprise use cases span systems — a support ticket might need CRM lookups, a refund needs payment tools, a contract question needs document search. LangGraph's supervisor pattern handles this by adding a routing agent above the specialists: the supervisor reads the incoming task, decides which specialist (or sequence of specialists) should handle it, and each specialist is its own compiled graph connected to only the MCP server it needs.

def supervisor_node(state: MessagesState):
    decision = supervisor_model.invoke([
        SystemMessage(content=(
            "Route this task to exactly one specialist: "
            "'crm_agent', 'payments_agent', or 'docs_agent'."
        )),
        *state["messages"],
    ])
    return {"next": decision.content.strip()}

# Each specialist is a fully separate compiled LangGraph agent,
# connected only to its own MCP server.
crm_agent = build_agent(mcp_server="crm-tools")
payments_agent = build_agent(mcp_server="payments-tools")
docs_agent = build_agent(mcp_server="docs-tools")

graph = StateGraph(MessagesState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("crm_agent", crm_agent)
graph.add_node("payments_agent", payments_agent)
graph.add_node("docs_agent", docs_agent)

graph.set_entry_point("supervisor")
graph.add_conditional_edges("supervisor", lambda s: s["next"], {
    "crm_agent": "crm_agent",
    "payments_agent": "payments_agent",
    "docs_agent": "docs_agent",
})
graph.add_edge("crm_agent", "supervisor")
graph.add_edge("payments_agent", "supervisor")
graph.add_edge("docs_agent", "supervisor")

top_level = graph.compile()

The security property this buys you is the one worth underlining: the payments agent has no path to the CRM server and the CRM agent has no path to the payments server, enforced by which MCP client each specialist even has a connection to — not by a prompt instruction the model could be talked out of. The supervisor coordinates by delegating, but it never gains the union of every specialist's permissions itself. This is a stricter version of the coordination pattern we cover in MCP with n8n for enterprise AI agent systems, here implemented as code rather than a visual workflow. For systems that need this pattern built and hardened end-to-end, see our Custom AI Software Engineering practice. To see this multi-server, scoped-specialist approach working in production, see how we shipped 10 production MCP servers for Optevo.

Production checklist

The examples above are correct but minimal. Before running this in production, work through:

• Connection management — pool and reuse MCP client sessions rather than opening a new connection per tool call; a stateless connection-per-call pattern does not survive real load.
• Retry strategy — wrap MCP tool calls with exponential backoff for transient failures, and make retries safe by keeping tool operations idempotent on the server side.
• Checkpoint persistence — use a durable LangGraph checkpointer (Postgres or Redis backed, not in-memory) so a crashed process resumes mid-workflow instead of restarting from scratch and re-triggering completed tool calls.
• Monitoring and alerting — trace every reasoning → tool → validation cycle, alert on loop counts approaching your timeout guard, and alert separately on MCP call error rates per server.
• MCP server health checks — a lightweight heartbeat per server plus a circuit breaker in the tool wrapper, so a degraded server routes to a fallback or a human queue instead of stalling every agent that depends on it.

FAQ