LangGraph

Use AllMCP tools inside LangGraph workflows — stateful, multi-step agent graphs.

All frameworks Browse providers

---

Install

install

pip install langchain-mcp-adapters langchain-anthropic langgraph

---

Minimal example (ReAct agent)

agent.py

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

async def main():
    client = MultiServerMCPClient({
        "allmcp": {
            "url": "https://go.allmcp.co/mcp/",
            "transport": "streamable_http",
            "headers": {"X-API-Key": "YOUR_API_KEY"},
        }
    })
    tools = await client.get_tools()

    agent = create_react_agent(
        ChatAnthropic(model="claude-sonnet-4-6"),
        tools,
    )

    result = await agent.ainvoke({
        "messages": [{"role": "user", "content": "List my top 5 deals in Bitrix24"}]
    })
    print(result["messages"][-1].content)

asyncio.run(main())

Note

LangGraph connects through the adapter. MultiServerMCPClient from langchain-mcp-adapters loads every AllMCP tool over streamable_http, so any provider you’ve connected becomes a tool the agent can call.

---

Custom graph with AllMCP tools

Build a graph that uses AllMCP tools as one node in a larger workflow:

crm_graph.py

import asyncio
from typing import TypedDict, Annotated
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, BaseMessage
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
import operator


class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]


async def build_crm_graph():
    client = MultiServerMCPClient({
        "allmcp": {
            "url": "https://go.allmcp.co/mcp/",
            "transport": "streamable_http",
            "headers": {"X-API-Key": "YOUR_API_KEY"},
        }
    })
    tools = await client.get_tools()
    model = ChatAnthropic(model="claude-sonnet-4-6").bind_tools(tools)

    def call_model(state: AgentState):
        response = model.invoke(state["messages"])
        return {"messages": [response]}

    def should_continue(state: AgentState):
        last = state["messages"][-1]
        if hasattr(last, "tool_calls") and last.tool_calls:
            return "tools"
        return END

    graph = StateGraph(AgentState)
    graph.add_node("agent", call_model)
    graph.add_node("tools", ToolNode(tools))
    graph.set_entry_point("agent")
    graph.add_conditional_edges("agent", should_continue)
    graph.add_edge("tools", "agent")

    app = graph.compile()

    result = await app.ainvoke({
        "messages": [HumanMessage(content="Summarize my open Bitrix24 deals")]
    })
    return result["messages"][-1].content

---

Persistent state across sessions

persistent_state.py

from langgraph.checkpoint.memory import MemorySaver

app = graph.compile(checkpointer=MemorySaver())

config = {"configurable": {"thread_id": "user-123-session"}}
await app.ainvoke({"messages": [HumanMessage(content="Connect Bitrix24")]}, config)
await app.ainvoke({"messages": [HumanMessage(content="List my contacts")]}, config)

Tip

Checkpointers keep conversation state. A MemorySaver plus a thread_id lets the graph resume the same session across invocations, so a “Connect Bitrix24” turn and a later “List my contacts” turn share context.

---

Multi-user setup

multi_user.py

def make_client_config(user_id: str) -> dict:
    return {
        "allmcp": {
            "url": f"https://go.allmcp.co/mcp/?user_id={user_id}",
            "transport": "streamable_http",
            "headers": {"X-API-Key": "YOUR_API_KEY"},
        }
    }

Note

One key, many users. Append ?user_id={user_id} to the endpoint to scope each request to a specific user’s connected credentials while reusing the same X-API-Key.

---

See also

Frameworks overview

Browse providers See every provider you can connect and the tools each one offers.