How to Build an MCP Client using LangGraph

An architectural guide to modernizing Large Language Model (LLM) agents by replacing brittle, custom tool wrappers with standardized Model Context Protocol (MCP) clients. This note covers the transition from synchronous to asynchronous execution in LangGraph, the integration of local and remote MCP servers, and the construction of a robust hybrid agent setup.

1. The Context & Evolution of LangGraph Agents

Building functional LLM applications requires enabling agents to perform real-world actions via external tools. In standard LangGraph workflows, an agent receives a user prompt, determines whether an external action is required, invokes a tool node, and routes the tool's output back to the LLM to synthesize a final response.

While connecting simple functions (such as stock lookup or basic web search) works seamlessly in isolated prototypes, scaling custom tool integrations across enterprise applications introduces severe maintenance challenges. Modern agent architectures require a standard protocol to decouple tool execution logic from the agent framework.

2. The Inherent Flaws of Traditional Tool Integration

In traditional LangChain and LangGraph implementations, tools are built either via built-in library classes (e.g., DuckDuckGoSearchRun) or user-defined custom Python functions wrapped in @tool decorators.

Consider a scenario where an engineering team builds a custom LangGraph tool to fetch open Pull Requests (PRs) from a GitHub repository:

import requests
from langchain_core.tools import tool

@tool
def fetch_github_pull_requests(owner: str, repo: str, state: str = "open", limit: int = 5) -> str:
    """Fetches a list of pull requests from a specified GitHub repository."""
    url = f"https://api.github.com/repos/{owner}/{repo}/pulls"
    headers = {
        "Authorization": "Bearer YOUR_GITHUB_TOKEN",
        "Accept": "application/vnd.github.v3+json"
    }
    params = {"state": state, "per_page": limit}
    
    response = requests.get(url, headers=headers, params=params)
    if response.status_code != 200:
        return f"Failed to fetch PRs: {response.status_code} - {response.text}"
    
    prs = response.json()
    result = []
    for pr in prs:
        result.append(
            f"PR #{pr['number']}: {pr['title']} | Author: {pr['user']['login']} | State: {pr['state']} | URL: {pr['html_url']}"
        )
    return "\n".join(result)

Why This Approach Is Brittle:

1. Hardcoded API Schemas: The client tool explicitly depends on GitHub's REST API endpoint structure (/repos/{owner}/{repo}/pulls) and response JSON format (pr['number'], pr['user']['login']).
2. Upstream API Drift: If GitHub updates its API from v1.0 to v2.0 (e.g., changing /pulls to /pull_requests, or changing key names from title to title_name), the tool function immediately throws runtime key errors or HTTP failures.
3. Client-Side Coupling: Heavy lifting, authentication payload structures, endpoint paths, and data parsing logic are embedded directly within the client application code.

3. Real-World Analogy: The $N \times M$ Maintenance Nightmare

Real-Life Analogy

Imagine a multi-department enterprise that deploys $M$ separate specialized AI chatbots (e.g., HR Bot, Developer Bot, Executive Assistant Bot, Customer Support Bot). To function, each chatbot needs access to $N$ common corporate platforms (e.g., GitHub, Slack, Jira, Gmail, PostgreSQL).

If every chatbot writes its own native tool integration code for each of the $N$ services, any change in an external platform's API requires updating code in up to $N \times M$ separate places!

Structure & Concept Flow

Copy DiagramCopy

Traditional Tool Integration Complexity: O(N x M)

 

[Chatbot 1] ---> Custom GitHub Tool Code ---> GitHub API

---> Custom Slack Tool Code ---> Slack API

---> Custom Jira Tool Code ---> Jira API

 

[Chatbot 2] ---> Custom GitHub Tool Code ---> GitHub API

---> Custom Slack Tool Code ---> Slack API

---> Custom Jira Tool Code ---> Jira API

 

Developers end up spending a disproportionate amount of time repairing broken API wrappers across multiple code repositories rather than building core application logic.

4. Model Context Protocol (MCP) as the Architectural Remedy

The Model Context Protocol (MCP) solves the $N \times M$ maintenance crisis by introducing a standardized client-server architecture for LLM tools, resources, and prompts.

Key Concepts of MCP Integration:

• Separation of Concerns: Tool logic, API routing, authentication handling, and response formatting are completely pushed to the MCP Server.
• Declarative Client Configuration: The client (LangGraph Chatbot) retains zero domain-specific tool logic. It only contains configuration details required to connect to the MCP Server.
• Dynamic Tool Discovery: At runtime, the MCP Client queries the MCP Server using standardized methods (e.g., get_tools()). The server dynamically advertises available tools, parameters, descriptions, and JSON schemas.
• Resilience to Upstream Changes: If GitHub updates its backend API from v1.0 to v2.0, the server maintainer updates the MCP Server. The client application code requires zero changes and zero redeployments.

5. Architectural Comparison: Traditional Tools vs. MCP

6. Step 1: Refactoring LangGraph from Synchronous to Asynchronous Execution

The standard library for integrating MCP with LangChain/LangGraph—langchain-mcp-adapters—relies strictly on Python's asyncio event loop. Therefore, prior to instantiating an MCP client, existing synchronous LangGraph workflows must be refactored into asynchronous implementations.

Synchronous Baseline vs. Asynchronous Refactoring

1. Baseline Synchronous Workflow (chat_node & execution):

# Synchronous Chat Node
def chat_node(state: State):
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

# Synchronous Execution
graph = graph_builder.compile()
response = graph.invoke({"messages": [HumanMessage(content="Hello")]})

2. Asynchronous Refactored Workflow (chat_node & execution):

# Asynchronous Chat Node
async def chat_node(state: State):
    # Asynchronous call to the LLM using `ainvoke`
    response = await llm_with_tools.ainvoke(state["messages"])
    return {"messages": [response]}

# Asynchronous Graph Execution within main event loop
async def main():
    graph = graph_builder.compile()
    response = await graph.ainvoke({"messages": [HumanMessage(content="Hello")]})

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

Note: LangGraph native nodes such as ToolNode automatically support asynchronous execution internally. Only custom node functions (e.g., chat_node) require explicit async def declarations and await keywords.

7. Step 2: Constructing a LangGraph MCP Client

To establish communication with an MCP server, use the MultiServerMCPClient provided by langchain-mcp-adapters.

Local Execution via STDIO Transport

When the MCP Server resides locally on the same host system, the client spawns and communicates with the server process using Standard Input/Output (STDIO).

LangGraph Agent (Client Process)
   |
   | Spawns local process via STDIO
   v
Local MCP Math Server (`main.py` using FastMCP)

import asyncio
from typing import Annotated, TypedDict
from dotenv import load_dotenv

from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

# Import MCP MultiServer Client adapter
from langchain_mcp_adapters.client import MultiServerMCPClient

load_dotenv()

class State(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

async def build_graph():
    # 1. Initialize LLM
    llm = ChatOpenAI(model="gpt-4o-mini")

    # 2. Configure MCP Client connection details
    mcp_client = MultiServerMCPClient(
        {
            "math_server": {
                "command": "python",
                "args": ["/Users/username/Desktop/mcp_math_server/main.py"],
                "transport": "stdio",
            }
        }
    )

    # 3. Dynamically fetch available tools from the MCP Server
    mcp_tools = await mcp_client.get_tools()
    
    # 4. Bind fetched tools to the LLM
    llm_with_tools = llm.bind_tools(mcp_tools)

    # 5. Define graph nodes
    async def chat_node(state: State):
        response = await llm_with_tools.ainvoke(state["messages"])
        return {"messages": [response]}

    # 6. Assemble state graph
    builder = StateGraph(State)
    builder.add_node("chat_node", chat_node)
    builder.add_node("tools", ToolNode(mcp_tools))

    builder.add_edge(START, "chat_node")
    builder.add_conditional_edges("chat_node", tools_condition)
    builder.add_edge("tools", "chat_node")

    return builder.compile()

async def main():
    chatbot = await build_graph()
    
    query = "Perform a modulus operation on 125 and 12, and present the result like a cricket commentator."
    inputs = {"messages": [HumanMessage(content=query)]}
    
    response = await chatbot.ainvoke(inputs)
    print(response["messages"][-1].content)

if __name__ == "__main__":
    asyncio.run(main())

8. Step 3: Scaling to Multi-Server MCP Client Architecture

The MultiServerMCPClient class natively supports binding multiple independent MCP servers concurrently, utilizing different transport protocols (e.g., local stdio processes alongside remote streamable_http or Server-Sent Events sse endpoints).

Code Implementation: Multi-Server Configuration

# Multi-Server Configuration setup inside MultiServerMCPClient
mcp_client = MultiServerMCPClient(
    {
        # Server 1: Local Math Operations Server via STDIO
        "math_server": {
            "command": "python",
            "args": ["/Users/username/Desktop/mcp_math_server/main.py"],
            "transport": "stdio",
        },
        # Server 2: Remote Expense Tracking Server via Streamable HTTP
        "expense_tracker": {
            "url": "https://expense-tracker-mcp-server.onrender.com/mcp",
            "transport": "streamable_http",
        }
    }
)

# Dynamic Discovery aggregates tools across BOTH servers seamlessly
all_mcp_tools = await mcp_client.get_tools()

When await mcp_client.get_tools() is executed, the client automatically queries both endpoints and produces a unified list of tools containing signatures for add, subtract, modulus, add_expense, list_expenses, and summarize_expenses.

9. Full Production Architecture: Hybrid Tools, Async Persistence, and Streamlit

In production systems, agents often need to combine standard native Python tools (e.g., specialized web scraping or stock pricing APIs) with external MCP Servers, maintain long-term chat persistence via SQLite, and stream answers asynchronously to a web user interface.

Complete Implementation

Backend Pipeline (langgraph_mcp_backend.py):

import asyncio
import aiosqlite
from typing import Annotated, TypedDict
from dotenv import load_dotenv

from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_core.tools import tool
from langchain_core.messages import BaseMessage

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver

from langchain_mcp_adapters.client import MultiServerMCPClient

load_dotenv()

# --- 1. Define Standard Local Tools ---
search_tool = DuckDuckGoSearchRun()

@tool
def get_stock_price(ticker: str) -> str:
    """Mock function to fetch real-time stock pricing."""
    mock_prices = {"AAPL": "$230.50", "GOOGL": "$175.20", "MSFT": "$420.10"}
    return mock_prices.get(ticker.upper(), f"Ticker {ticker} not found.")

native_tools = [search_tool, get_stock_price]

# --- 2. Define State Schema ---
class State(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

# --- 3. Build Hybrid Async Graph ---
async def create_agent_graph(db_connection):
    llm = ChatOpenAI(model="gpt-4o-mini", streaming=True)

    # Configure Multi-Server MCP Client
    mcp_client = MultiServerMCPClient(
        {
            "math_server": {
                "command": "python",
                "args": ["/Users/username/Desktop/mcp_math_server/main.py"],
                "transport": "stdio",
            },
            "expense_tracker": {
                "url": "https://expense-tracker-mcp-server.onrender.com/mcp",
                "transport": "streamable_http",
            }
        }
    )

    # Fetch remote MCP tools dynamically
    mcp_tools = await mcp_client.get_tools()

    # Combine Native Tools + MCP Tools into a single unified list
    combined_tools = native_tools + mcp_tools

    # Bind aggregated toolset to the LLM
    llm_with_tools = llm.bind_tools(combined_tools)

    # Define Chat Node
    async def chat_node(state: State):
        response = await llm_with_tools.ainvoke(state["messages"])
        return {"messages": [response]}

    # Graph Setup
    builder = StateGraph(State)
    builder.add_node("chat_node", chat_node)
    builder.add_node("tools", ToolNode(combined_tools))

    builder.add_edge(START, "chat_node")
    builder.add_conditional_edges("chat_node", tools_condition)
    builder.add_edge("tools", "chat_node")

    # Use Async SQLite Saver for Checkpointing
    memory = AsyncSqliteSaver(db_connection)
    return builder.compile(checkpointer=memory)

Async Streamlit UI Stream (streamlit_frontend.py):

import streamlit as st
import asyncio
import aiosqlite
from langchain_core.messages import HumanMessage
from langgraph_mcp_backend import create_agent_graph

st.set_page_config(page_title="MCP Hybrid Agent", layout="wide")
st.title("LangGraph Hybrid Agent with MCP Support")

# Asynchronous runner helper for Streamlit
def run_async(coroutine):
    return asyncio.run(coroutine)

async def process_chat(user_input: str):
    async with aiosqlite.connect("agent_memory.db") as db:
        agent = await create_agent_graph(db)
        config = {"configurable": {"thread_id": "session_1"}}
        
        # Utilize asynchronous streaming (astream_events or astream)
        async for event in agent.astream(
            {"messages": [HumanMessage(content=user_input)]},
            config,
            stream_mode="values"
        ):
            latest_message = event["messages"][-1]
            st.write(f"**{latest_message.type.capitalize()}**: {latest_message.content}")

user_query = st.chat_input("Ask a question, query expenses, or execute math...")
if user_query:
    run_async(process_chat(user_query))

10. Real-World Debugging & Production Pitfalls

Integrating asynchronous MCP clients into production user interfaces introduces specific operational pitfalls:

1. The Streamlit Synchronous Framework Lock

• Problem: Streamlit is inherently a synchronous framework built on standard procedural execution. Calling await agent.ainvoke() directly inside a Streamlit script raises SyntaxError or runtime loop conflicts.
• Mitigation: Wrap asynchronous calls inside asyncio.run() wrappers or route request processing through an intermediate asynchronous web framework (such as FastAPI) serving a decoupled client framework (like Next.js or React).

2. Database Drivers for Persistence

• Problem: Standard sqlite3 drivers block the Python GIL during disk read/writes, stalling concurrent asynchronous tool execution.
• Mitigation: Replace standard checkpointers with AsyncSqliteSaver leveraging aiosqlite to ensure non-blocking I/O operations during state storage.

3. Transport Protocol Errors

• STDIO Failures: If a local MCP server script outputs non-JSON debug logs directly to standard output (sys.stdout), the JSON-RPC parser breaks.

  Correction: Configure servers to write logs exclusively to sys.stderr.

• Streamable HTTP Errors: Remote transport requires reliable network connectivity and persistent SSE/HTTP connections. Timeouts must be explicitly configured in production environments.

11. Summary and Key Takeaways

• Traditional Tool Flaws: Hardcoding API integration details within LLM tools leads to fragile, tightly coupled architectures and an $\mathcal{O}(N \times M)$ maintenance overhead.
• MCP Core Principle: Model Context Protocol establishes a clean separation of concerns, decoupling tool execution logic (MCP Server) from agent workflow orchestration (MCP Client).
• Dynamic Tool Discovery: LangGraph agents configured with MCP Clients dynamically fetch tool schemas via standard protocols (get_tools()), rendering client code resilient to upstream API schema changes.
• Asynchronous Prerequisite: Integrating langchain-mcp-adapters requires migrating LangGraph nodes from synchronous methods (invoke) to asynchronous execution paradigms (async def, ainvoke, asyncio.run).
• Multi-Server Aggregation: Using MultiServerMCPClient, a single agent can simultaneously orchestrate tools from local processes (STDIO) and remote endpoints (Streamable HTTP / SSE).
• Hybrid Support: LangGraph seamlessly supports hybrid configurations, enabling agents to execute traditional custom tools alongside dynamic MCP Client toolsets within a single graph.