A Guide to MCP Clients with LangGraph

In the evolving landscape of Agentic AI, the way we connect Large Language Models (LLMs) to external data and tools is undergoing a massive shift. While traditional tool-calling has been the standard, it suffers from a significant “brittleness” that makes scaling difficult. The Model Context Protocol (MCP) has emerged as a standardized, robust solution to these challenges.

This blog explores why MCP is superior to traditional tools and provides a step-by-step guide on building an MCP client using LangGraph.

The Problem: Why Traditional Tools Fail at Scale #

Integrating tools directly into a chatbot involves writing specific code for each API (e.g., GitHub, Slack, or a calculator). However, this approach has a major flaw: Maintenance Overhead.

1. Brittleness: If an external service like GitHub updates its API from version 1.0 to 2.0, your hardcoded tool will likely break due to changes in URLs or attribute names (e.g., changing title to title_name).
2. The N x M Problem: If your company has 10 chatbots (M) and each uses 5 tools (N), a single API change might require updates in 50 different places. You end up spending more time maintaining code than building features.

---

The Solution: What is MCP? #

MCP (Model Context Protocol) is a standardized way to connect tools to LLM applications. It introduces a Separation of Concerns by dividing the system into two parts:

• MCP Server: This side contains the actual logic for the tools and handles communication with external APIs.
• MCP Client: This is your LangGraph application. Instead of complex tool logic, the client only needs a simple configuration code to connect to the server.

Even if the server’s underlying API changes, the client’s configuration remains the same, shifting the “heavy lifting” of maintenance to the server side and making the client future-proof.

For More Details, you can refer to the MCP Blog

---

Technical Requirements: The Need for Async #

Before implementing an MCP client, it is crucial to understand that the libraries used for MCP (like langchain-mcp-adapters) operate in asynchronous mode.

To use MCP with LangGraph, you must:

• Convert your LangGraph nodes into asynchronous functions using async def.
• Use ainvoke instead of invoke for model calls.
• Run the graph execution within an asyncio loop.

---

Step-by-Step Implementation with Example #

1. Install the Required Library #

You will need the MCP adapter for LangChain:

pip install langchain-mcp-adapters

2. Define the MCP Client Configuration #

import asyncio
import json
from langchain_mcp_adapters.client import MultiServerMCPClient

SERVERS = {
    # Local Math MCP Server
    "math_local": {
        "transport": "stdio",
        "command": "uv",
        "args": [
            "run",
            "fastmcp",
            "run",
            r"D:\Coding\MCP Remote\math.py"
        ]
    },

    # Local Manim MCP Server
    "manim-server": {
        "transport": "stdio",
        "command": r"C:\Users\sanjit\OneDrive\Desktop\mcp_project\venv\Scripts\python.exe",
        "args": [
            r"C:\Users\sanjit\OneDrive\Desktop\mcp_project\manim-mcp-server\src\manim_server.py"
        ],
        "env": {
            "MANIM_EXECUTABLE": r"C:\Users\sanjit\OneDrive\Desktop\mcp_project\venv\Scripts\manim.exe"
        }
    },

    # Remote MCP Server (Render / Railway / Cloud)
    "demo_remote": {
        "transport": "streamable_http",
        "url": "https://your-remote-server-url.com"
    }
}

• Local Server: Uses STDIO (Standard Input/Output) transport. You provide the command to run the server file on your machine.
• Remote Server: Uses Streamable HTTP (SSE) transport. You provide the URL where the server is hosted.

3. Fetch and Bind Tools to LangGraph #

Instead of manually defining tools, the client fetches them dynamically from the server.

# Inside your graph building function
async def build_graph():
    # Dynamically fetch tools from the configured MCP servers
    mcp_tools = await client.get_tools() 
    
    # Bind these tools to your LLM
    llm_with_tools = model.bind_tools(mcp_tools)
    
    # ... rest of your LangGraph setup

---

Example in Action: A Multi-Functional Chatbot #

In a live demonstration, an MCP-enabled chatbot can handle diverse tasks without any local tool logic for those tasks:

# backend Code

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_core.tools import tool, BaseTool
from langchain_mcp_adapters.client import MultiServerMCPClient
from dotenv import load_dotenv
import aiosqlite
import requests
import asyncio
import threading

load_dotenv()

# Dedicated async loop for backend tasks
_ASYNC_LOOP = asyncio.new_event_loop()
_ASYNC_THREAD = threading.Thread(target=_ASYNC_LOOP.run_forever, daemon=True)
_ASYNC_THREAD.start()


def _submit_async(coro):
    return asyncio.run_coroutine_threadsafe(coro, _ASYNC_LOOP)


def run_async(coro):
    return _submit_async(coro).result()


def submit_async_task(coro):
    """Schedule a coroutine on the backend event loop."""
    return _submit_async(coro)


# -------------------
# 1. LLM
# -------------------
llm = ChatOpenAI()

# -------------------
# 2. Tools
# -------------------
search_tool = DuckDuckGoSearchRun(region="us-en")


@tool
def get_stock_price(symbol: str) -> dict:
    """
    Fetch latest stock price for a given symbol (e.g. 'AAPL', 'TSLA') 
    using Alpha Vantage with API key in the URL.
    """
    url = f"https://www.alphavantage.co/query?function=GLOBAL_QUOTE&symbol={symbol}&apikey=C9PE94QUEW9VWGFM"
    r = requests.get(url)
    return r.json()


client = MultiServerMCPClient(
    {
        "arith": {
            "transport": "stdio",
            "command": "python3",
            "args": ["/Users/nitish/Desktop/mcp-math-server/main.py"],
        },
        "expense": {
            "transport": "streamable_http",  # if this fails, try "sse"
            "url": "https://splendid-gold-dingo.fastmcp.app/mcp"
        }
    }
)


def load_mcp_tools() -> list[BaseTool]:
    try:
        return run_async(client.get_tools())
    except Exception:
        return []


mcp_tools = load_mcp_tools()

tools = [search_tool, get_stock_price, *mcp_tools]
llm_with_tools = llm.bind_tools(tools) if tools else llm

# -------------------
# 3. State
# -------------------
class ChatState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

# -------------------
# 4. Nodes
# -------------------
async def chat_node(state: ChatState):
    """LLM node that may answer or request a tool call."""
    messages = state["messages"]
    response = await llm_with_tools.ainvoke(messages)
    return {"messages": [response]}


tool_node = ToolNode(tools) if tools else None

# -------------------
# 5. Checkpointer
# -------------------


async def _init_checkpointer():
    conn = await aiosqlite.connect(database="chatbot.db")
    return AsyncSqliteSaver(conn)


checkpointer = run_async(_init_checkpointer())

# -------------------
# 6. Graph
# -------------------
graph = StateGraph(ChatState)
graph.add_node("chat_node", chat_node)
graph.add_edge(START, "chat_node")

if tool_node:
    graph.add_node("tools", tool_node)
    graph.add_conditional_edges("chat_node", tools_condition)
    graph.add_edge("tools", "chat_node")
else:
    graph.add_edge("chat_node", END)

chatbot = graph.compile(checkpointer=checkpointer)

# -------------------
# 7. Helper
# -------------------
async def _alist_threads():
    all_threads = set()
    async for checkpoint in checkpointer.alist(None):
        all_threads.add(checkpoint.config["configurable"]["thread_id"])
    return list(all_threads)


def retrieve_all_threads():
    return run_async(_alist_threads())

---

#Fronted Code

import queue
import uuid

import streamlit as st
from langgraph_mcp_backend import chatbot, retrieve_all_threads, submit_async_task
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage

# =========================== Utilities ===========================
def generate_thread_id():
    return uuid.uuid4()


def reset_chat():
    thread_id = generate_thread_id()
    st.session_state["thread_id"] = thread_id
    add_thread(thread_id)
    st.session_state["message_history"] = []


def add_thread(thread_id):
    if thread_id not in st.session_state["chat_threads"]:
        st.session_state["chat_threads"].append(thread_id)


def load_conversation(thread_id):
    state = chatbot.get_state(config={"configurable": {"thread_id": thread_id}})
    # Check if messages key exists in state values, return empty list if not
    return state.values.get("messages", [])


# ======================= Session Initialization ===================
if "message_history" not in st.session_state:
    st.session_state["message_history"] = []

if "thread_id" not in st.session_state:
    st.session_state["thread_id"] = generate_thread_id()

if "chat_threads" not in st.session_state:
    st.session_state["chat_threads"] = retrieve_all_threads()

add_thread(st.session_state["thread_id"])

# ============================ Sidebar ============================
st.sidebar.title("LangGraph MCP Chatbot")

if st.sidebar.button("New Chat"):
    reset_chat()

st.sidebar.header("My Conversations")
for thread_id in st.session_state["chat_threads"][::-1]:
    if st.sidebar.button(str(thread_id)):
        st.session_state["thread_id"] = thread_id
        messages = load_conversation(thread_id)

        temp_messages = []
        for msg in messages:
            role = "user" if isinstance(msg, HumanMessage) else "assistant"
            temp_messages.append({"role": role, "content": msg.content})
        st.session_state["message_history"] = temp_messages

# ============================ Main UI ============================

# Render history
for message in st.session_state["message_history"]:
    with st.chat_message(message["role"]):
        st.text(message["content"])

user_input = st.chat_input("Type here")

if user_input:
    # Show user's message
    st.session_state["message_history"].append({"role": "user", "content": user_input})
    with st.chat_message("user"):
        st.text(user_input)

    CONFIG = {
        "configurable": {"thread_id": st.session_state["thread_id"]},
        "metadata": {"thread_id": st.session_state["thread_id"]},
        "run_name": "chat_turn",
    }

    # Assistant streaming block
    with st.chat_message("assistant"):
        # Use a mutable holder so the generator can set/modify it
        status_holder = {"box": None}

        def ai_only_stream():
            event_queue: queue.Queue = queue.Queue()

            async def run_stream():
                try:
                    async for message_chunk, metadata in chatbot.astream(
                        {"messages": [HumanMessage(content=user_input)]},
                        config=CONFIG,
                        stream_mode="messages",
                    ):
                        event_queue.put((message_chunk, metadata))
                except Exception as exc:
                    event_queue.put(("error", exc))
                finally:
                    event_queue.put(None)

            submit_async_task(run_stream())

            while True:
                item = event_queue.get()
                if item is None:
                    break
                message_chunk, metadata = item
                if message_chunk == "error":
                    raise metadata

                # Lazily create & update the SAME status container when any tool runs
                if isinstance(message_chunk, ToolMessage):
                    tool_name = getattr(message_chunk, "name", "tool")
                    if status_holder["box"] is None:
                        status_holder["box"] = st.status(
                            f"🔧 Using `{tool_name}` …", expanded=True
                        )
                    else:
                        status_holder["box"].update(
                            label=f"🔧 Using `{tool_name}` …",
                            state="running",
                            expanded=True,
                        )

                # Stream ONLY assistant tokens
                if isinstance(message_chunk, AIMessage):
                    yield message_chunk.content

        ai_message = st.write_stream(ai_only_stream())

        # Finalize only if a tool was actually used
        if status_holder["box"] is not None:
            status_holder["box"].update(
                label="✅ Tool finished", state="complete", expanded=False
            )

    # Save assistant message
    st.session_state["message_history"].append(
        {"role": "assistant", "content": ai_message}
    )

• Math Operations: By connecting to a local “Math MCP Server,” the bot can perform additions, multiplications, and modulus operations.
• Expense Tracking: By connecting to a remote “Expense Tracker Server,” the bot can add expenses (e.g., “Add 500 for a course”) and summarize monthly spending, despite having zero lines of expense-tracking code in the client application.

---

Conclusion: Why You Should Adopt MCP #

MCP is becoming an industry standard because it allows developers to “smartly use someone else’s hard work”. You can search for useful MCP servers online and plug them into your LangGraph agent instantly.

Whether you are building internal tools for developers or consumer-facing chatbots, moving toward MCP ensures your application is modular, robust, and easy to maintain. While you can still mix traditional tools with MCP, leaning toward MCP is the most future-proof path for Agentic AI

MCP Clients with LangGraph Quiz #