Real-World Apps — Chatbot, RAG Agent, MCP Client¶

1. Why this matters¶

You've learned state, nodes, conditional edges, loops, persistence, streaming, tools, agents. Real apps stitch all of those together. This chapter shows the wiring patterns the CampusX playlist actually ships:

A working chatbot you can deploy.
A RAG agent that decides when to retrieve — better than always-retrieve RAG.
An MCP client so your agent can use any MCP server (file system, web, custom tools).

2. Mental model¶

Every real LangGraph app is some combination of:

StateGraph + checkpointer + (tools or retriever) + streaming + (optional HITL)

flowchart LR
    UI[Streamlit / FastAPI / CLI] -->|stream| G[Compiled Graph]
    G -->|reads/writes| CP[Checkpointer]
    G -->|tool calls| EXT[Tools / Retriever / MCP servers]
    EXT --> G

3. Blueprint 1 — Streaming Streamlit chatbot¶

flowchart LR
    USR[User input] --> STR[Streamlit chat_input]
    STR --> SES[Session state holds thread_id, history]
    SES --> GR[LangGraph backend<br/>MessagesState + chat node + checkpointer]
    GR -->|stream_mode=messages| TOK[Token chunks]
    TOK --> WS[st.write_stream]

langgraph_backend.py — the graph definition:

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_core.messages import BaseMessage
from langchain_openai import ChatOpenAI
import sqlite3

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

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.4, streaming=True)

def chat_node(state: ChatState):
    return {"messages": [llm.invoke(state["messages"])]}

b = StateGraph(ChatState)
b.add_node("chat", chat_node)
b.add_edge(START, "chat")
b.add_edge("chat", END)

_conn = sqlite3.connect("chat.db", check_same_thread=False)
chatbot = b.compile(checkpointer=SqliteSaver(_conn))

app.py — the Streamlit frontend:

import streamlit as st
from langchain_core.messages import HumanMessage
from langgraph_backend import chatbot

st.title("LangGraph Chatbot")

if "thread_id" not in st.session_state:
    st.session_state["thread_id"] = "user-1"
if "history" not in st.session_state:
    st.session_state["history"] = []

CONFIG = {"configurable": {"thread_id": st.session_state["thread_id"]}}

# Replay prior turns
for msg in st.session_state["history"]:
    st.chat_message(msg["role"]).write(msg["content"])

if user_input := st.chat_input("Type a message"):
    st.session_state["history"].append({"role": "user", "content": user_input})
    st.chat_message("user").write(user_input)

    with st.chat_message("assistant"):
        ai_text = st.write_stream(
            chunk.content
            for chunk, _ in chatbot.stream(
                {"messages": [HumanMessage(user_input)]},
                config=CONFIG,
                stream_mode="messages",
            )
            if chunk.content
        )
    st.session_state["history"].append({"role": "assistant", "content": ai_text})

That's a full chatbot — memory, streaming, persistence — in ~50 lines. Repository: campusx-official/chatbot-in-langgraph.

4. Blueprint 2 — RAG Agent (retriever as a tool)¶

The classic LangChain pattern always-retrieves. A smarter pattern: make retrieval a tool the agent decides to call. The agent skips it for "hi" / "thanks" and uses it only for substantive questions.

flowchart TD
    Q[Question] --> A[Agent LLM]
    A -->|trivial chat| ANS[Direct answer]
    A -->|"needs PDF context"| R[rag_tool]
    R -->|retrieved chunks| A
    A --> ANS
    ANS --> END([END])

from langchain_core.tools import tool
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3

# 1. Pre-built vector store (built offline once)
vs = FAISS.load_local("./pdf_index", OpenAIEmbeddings(),
                      allow_dangerous_deserialization=True)

# 2. Wrap retrieval as a tool
@tool
def rag_tool(query: str) -> str:
    """Use this when the question asks about content in the PDF notes."""
    docs = vs.similarity_search(query, k=4)
    return "\n\n".join(f"[{i+1}] {d.page_content}" for i, d in enumerate(docs))

# 3. Build the agent — it decides if/when to call rag_tool
agent = create_react_agent(
    model=ChatOpenAI(model="gpt-4o-mini", temperature=0),
    tools=[rag_tool],
    prompt=(
        "You are an assistant with access to a PDF knowledge base via rag_tool. "
        "Call rag_tool ONLY when the question is about the PDF content. "
        "For greetings/small-talk, answer directly. "
        "Cite chunk numbers like [1] when you use retrieved content."
    ),
    checkpointer=SqliteSaver(sqlite3.connect("rag.db", check_same_thread=False)),
)

cfg = {"configurable": {"thread_id": "demo"}, "recursion_limit": 8}

# Direct answer — no tool call
print(agent.invoke({"messages": [("user", "Hi!")]}, config=cfg)["messages"][-1].content)
# Triggers the rag_tool
print(agent.invoke({"messages": [("user", "Using the PDF notes, summarize LCEL.")]}, config=cfg)["messages"][-1].content)

This is the modern "agentic RAG" pattern — strictly better than always-retrieve for chat apps.

5. Blueprint 3 — MCP Client¶

MCP (Model Context Protocol) is a standard for exposing tools and data from external servers (file system, web search, custom internal APIs) to LLMs. LangChain has an adapter (langchain-mcp-adapters) that turns MCP servers into LangChain tools — which you then drop into a LangGraph agent.

flowchart LR
    A[LangGraph Agent] -->|tool_call| CL[MultiServerMCPClient]
    CL --> S1[MCP Server: files]
    CL --> S2[MCP Server: web]
    CL --> S3[MCP Server: custom]

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

async def main():
    # 1. Spin up clients to one or more MCP servers
    async with MultiServerMCPClient({
        "filesystem": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
            "transport": "stdio",
        },
        "fetch": {
            "url": "http://localhost:3000/sse",
            "transport": "sse",
        },
    }) as client:
        # 2. Get tools from all servers, exposed as LangChain Tools
        tools = client.get_tools()

        # 3. Drop them into a LangGraph agent
        agent = create_react_agent(
            model=ChatOpenAI(model="gpt-4o-mini"),
            tools=tools,
            prompt="You have access to filesystem and web fetch tools via MCP.",
        )

        result = await agent.ainvoke(
            {"messages": [("user", "List files in /tmp and fetch example.com.")]}
        )
        print(result["messages"][-1].content)

asyncio.run(main())

Reference: campusx-official/mcp-client-langgraph.

6. Production checklist¶

Before shipping any of these:

7. Common pitfalls¶

❗ Mixing streaming and non-streaming patterns inconsistently. Pick one for each user-facing endpoint and stick with it.
❗ Forgetting to flush Streamlit state on new sessions. When thread_id changes, also clear st.session_state["history"] to avoid stale UI.
❗ MCP servers as long-lived background processes. Manage their lifecycle (start/stop) properly; use async with to clean up.
❗ RAG-as-tool with bad descriptions. The agent will call retrieval on every "hi". Be explicit: "ONLY call this for PDF-related questions."
❗ Skipping LangSmith tracing. You will regret it the first time the agent loops weirdly in production.

8. When to use each blueprint¶

Blueprint	When
Streamlit chatbot	Internal tools, prototypes, demos
FastAPI + SSE	Production chat APIs (multi-user, scalable)
Agentic RAG	Any chatbot over private documents
Multi-agent	Tasks spanning distinct domains
MCP client	You need to call external tools without writing custom integrations

9. Cheatsheet¶

# Streaming chatbot — the canonical pattern
for chunk, _ in graph.stream(
    {"messages": [HumanMessage(user_input)]},
    config={"configurable": {"thread_id": tid}},
    stream_mode="messages",
):
    if chunk.content:
        send_to_ui(chunk.content)

# RAG-as-tool — sharper than always-retrieve
@tool
def rag_tool(query: str) -> str:
    """Use ONLY when the question asks about <CORPUS>."""
    return retrieve_and_format(query)

agent = create_react_agent(model=llm, tools=[rag_tool], checkpointer=cp)

# MCP client
async with MultiServerMCPClient({...}) as client:
    tools = client.get_tools()
    agent = create_react_agent(model=llm, tools=tools)
    await agent.ainvoke({"messages": [...]})

# Always pass thread_id + recursion_limit
config = {
    "configurable": {"thread_id": user_session_id},
    "recursion_limit": 10,
}

10. Q&A — recall test¶

Q: Streaming chatbot — which stream_mode? A: "messages" — yields LLM token chunks for the typewriter effect.
Q: Why expose retrieval as a tool instead of always retrieving? A: Saves cost and improves quality on non-RAG queries (greetings, small talk, off-topic). The agent only retrieves when actually needed.
Q: What is MCP? A: Model Context Protocol — a standard for exposing tools/data from external servers to LLMs. langchain-mcp-adapters turns MCP servers into LangChain tools usable in a LangGraph agent.
Q: What checkpointer would you use for a deployed Streamlit chatbot? A: SqliteSaver if it's a single-server app; PostgresSaver if you scale to multiple workers.
Q: Top 3 things to verify before shipping any LangGraph app? A: (1) Persistent checkpointer + per-user thread_id. (2) recursion_limit set. (3) LangSmith tracing on.

Practice¶

What does this print?

Expected: True

# Production readiness checklist
checks = ["persistent_checkpointer", "per_user_thread_id", "recursion_limit", "tracing"]
print(len(checks) >= 3)

Use a persistent checkpointer (not InMemorySaver) in production

Expected: True

checkpointer = "InMemorySaver"        # bug: state lost on restart — use PostgresSaver
is_persistent = checkpointer in ["PostgresSaver", "SqliteSaver"]
print(not is_persistent)

Quiz — Quick check¶

What you remember

Q1. What are the three production essentials for a LangGraph app?

Persistent checkpointer, per-user thread_id, recursion_limit
More LLMs, more tools, more agents
Async, batch, stream
Tests, docs, monitoring

Why: These three prevent the most common production failures: data loss on restart, cross-user state leakage, and runaway recursion costs. Hard to recover from these in production; easy to prevent at design time.

Q2. Why use LangSmith with LangGraph?

Traces every node call, tool call, and LLM message — essential for debugging multi-step workflows
Required by LangGraph
For caching
For deployment

Why: Without tracing, debugging "why did the agent take this path?" is nearly impossible. LangSmith records every input/output of every node — you can replay, diff, and audit any historical run.

Q3. What's the MCP (Model Context Protocol)?

A standard for connecting LLM apps to external tools/data sources via a universal protocol
A LangGraph-specific feature
A LangChain replacement
An LLM model family

Why: MCP standardizes how LLM apps connect to tools (file system, databases, APIs). One server exposes capabilities; any MCP client can use them. Reduces the per-integration bespoke code that plagues agent tooling.

Common doubts¶

When is my LangGraph app ready to ship?

Checklist: (1) Persistent checkpointer wired up. (2) thread_id derived from authenticated user. (3) recursion_limit set explicitly. (4) Errors are caught and reported (not silent crashes). (5) LangSmith tracing on for debugging. (6) Costs monitored. (7) Load tested at expected traffic. Without all of these, you're shipping a prototype.

How do I deploy a LangGraph app?

Same as any FastAPI/Flask app: containerize it, deploy to your platform (Vercel/Render/AWS/etc.), point a load balancer at it. The state lives in your persistent checkpointer (Postgres/Redis/etc.); the app servers are stateless. Scale horizontally as needed.

What if the LLM provider has an outage during a long-running graph?

Checkpointing saves you. The graph pauses at the failing step; state is preserved. Once the provider is back, retry from the checkpoint. No need to restart from scratch. Add retries with exponential backoff for transient failures.