Unified MCP Server with LangChain and LangGraph

pip install langchain langchain-mcp-adapters langchain-anthropic

npm install langchain @langchain/mcp-adapters @langchain/anthropic

import asyncio
import os

from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient

UNIFIED_API_KEY = os.environ["UNIFIED_API_KEY"]
CONNECTION_ID = os.environ["UNIFIED_CONNECTION_ID"]  # end-customer connection


async def main():
    client = MultiServerMCPClient(
        {
            "unified": {
                "transport": "streamable_http",
                "url": f"https://mcp-api.unified.to/mcp?connection={CONNECTION_ID}",
                "headers": {"Authorization": f"bearer {UNIFIED_API_KEY}"},
            }
        }
    )

    tools = await client.get_tools()
    print(f"Loaded {len(tools)} Unified tools")

    agent = create_agent("anthropic:claude-sonnet-4-5", tools)

    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "List the 10 most recently updated deals and summarize their stages."}]}
    )
    print(result["messages"][-1].content)


asyncio.run(main())

import { MultiServerMCPClient } from "@langchain/mcp-adapters";
import { createAgent } from "langchain";

const client = new MultiServerMCPClient({
  mcpServers: {
    unified: {
      transport: "http",
      url: `https://mcp-api.unified.to/mcp?connection=${process.env.UNIFIED_CONNECTION_ID}`,
      headers: { Authorization: `bearer ${process.env.UNIFIED_API_KEY}` },
    },
  },
});

const tools = await client.getTools();

const agent = createAgent({ model: "anthropic:claude-sonnet-4-5", tools });

const result = await agent.invoke({
  messages: [{ role: "user", content: "Find open jobs and count applications per job." }],
});
console.log(result.messages.at(-1)?.content);

await client.close();

url = (
    "https://mcp-api.unified.to/mcp"
    f"?connection={CONNECTION_ID}"
    "&permissions=crm_contact_read,crm_deal_read"
    "&hide_sensitive=true"
)

from langchain.chat_models import init_chat_model
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition

client = MultiServerMCPClient(
    {
        "unified": {
            "transport": "streamable_http",
            "url": f"https://mcp-api.unified.to/mcp?connection={CONNECTION_ID}",
            "headers": {"Authorization": f"bearer {UNIFIED_API_KEY}"},
        }
    }
)
tools = await client.get_tools()

model = init_chat_model("anthropic:claude-sonnet-4-5").bind_tools(tools)


async def call_model(state: MessagesState):
    return {"messages": [await model.ainvoke(state["messages"])]}


builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "call_model")
builder.add_conditional_edges("call_model", tools_condition)
builder.add_edge("tools", "call_model")
graph = builder.compile()

result = await graph.ainvoke(
    {"messages": [{"role": "user", "content": "Which candidates applied this week, and for which roles?"}]}
)

def unified_client_for(connection_id: str) -> MultiServerMCPClient:
    return MultiServerMCPClient(
        {
            "unified": {
                "transport": "streamable_http",
                "url": f"https://mcp-api.unified.to/mcp?connection={connection_id}",
                "headers": {"Authorization": f"bearer {UNIFIED_API_KEY}"},
            }
        }
    )


async def handle_request(customer, prompt: str):
    tools = await unified_client_for(customer.unified_connection_id).get_tools()
    agent = create_agent("anthropic:claude-sonnet-4-5", tools)
    return await agent.ainvoke({"messages": [{"role": "user", "content": prompt}]})

tools_by_name = {t.name: t for t in await client.get_tools()}
contacts = await tools_by_name["list_crm_contacts"].ainvoke({"limit": 50})