how to build a langgraph stateful email agent workflow

LangGraph gives you something LangChain alone can't: a stateful email agent workflow where each step knows what happened before it. Your agent reads an email, classifies it, drafts a reply, waits for human approval, and sends. State flows through the entire graph.

That's the theory, at least. Most LangGraph email tutorials stop at the orchestration layer. They show you how to wire up nodes and edges but hand-wave the actual email infrastructure. You end up with a clean graph that can't touch a real inbox without bolting on IMAP and SMTP plumbing yourself.

This guide covers both sides. We'll walk through building a stateful email workflow in LangGraph, then address the infrastructure gap that most tutorials ignore. If you'd rather skip the email plumbing, LobsterMail lets your agent provision its own inbox with a single function call. Set up an inbox for your agent and focus on the graph logic instead.

What is a stateful email agent in LangGraph?#

LangGraph is a framework from the LangChain team for building AI agent workflows as directed graphs. Unlike a plain LangChain chain (which runs start to finish in a straight line), LangGraph lets you define nodes, edges, and conditional branching. The "stateful" part means each node can read and modify a shared state object that travels through the graph.

For email automation, this means your agent can receive a message, store it in state, classify the email type, route to different processing nodes based on that classification, draft a response using the full thread history, and pause for human review before sending. A StateGraph in LangGraph is the container that holds all of this logic together. You define the state shape, add nodes as functions, connect them with edges, compile the graph, and invoke it.

How to build a LangGraph stateful email agent#

Here's the process from zero to a working workflow:

1. Define your EmailState using a TypedDict with fields for messages, classification, and draft
2. Create processing nodes for classification, summarization, and response drafting
3. Add conditional edges that route emails by type (support, sales, spam)
4. Attach a human-review interrupt node for sensitive replies
5. Wire up email infrastructure so the agent can actually receive and send
6. Compile with workflow.compile() and invoke with your initial state
7. Add checkpointing to persist state across sessions

Let me walk through the important pieces.

Defining EmailState#

Your state object is the backbone of the whole workflow. Every node reads from it and writes to it.

from typing import TypedDict, List, Optional

class EmailState(TypedDict):
    sender: str
    subject: str
    body: str
    thread_history: List[dict]
    classification: Optional[str]
    draft_reply: Optional[str]
    human_approved: bool
    iteration: int

Building processing nodes#

Each node is a Python function that takes the current state and returns a partial state update. Here's a classification node and a reply-drafting node:

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

def classify_email(state: EmailState) -> dict:
    prompt = f"Classify this email as 'support', 'sales', or 'spam':\n\n{state['body']}"
    result = llm.invoke(prompt)
    return {"classification": result.content.strip().lower()}

def draft_reply(state: EmailState) -> dict:
    context = "\n".join(
        [f"From: {m['from']}\n{m['body']}" for m in state["thread_history"]]
    )
    prompt = (
        f"Draft a reply to this {state['classification']} email.\n\n"
        f"Thread:\n{context}\n\nLatest message:\n{state['body']}"
    )
    result = llm.invoke(prompt)
    return {"draft_reply": result.content, "iteration": state["iteration"] + 1}

The draft_reply function pulls thread_history from state, which means it has context from every previous node. No magic, no hidden memory store. Whatever you put in the TypedDict is what every downstream node sees.

Conditional routing#

This is where LangGraph diverges from a linear LangChain chain. Instead of running every step in order, you define routing functions that inspect state and decide which node executes next.

from langgraph.graph import StateGraph, END

def route_by_classification(state: EmailState) -> str:
    if state["classification"] == "spam":
        return "discard"
    if state["classification"] == "sales":
        return "sales_handler"
    return "support_handler"

workflow = StateGraph(EmailState)
workflow.add_node("classify", classify_email)
workflow.add_node("support_handler", draft_reply)
workflow.add_node("sales_handler", draft_reply)
workflow.add_node("discard", lambda s: {})

workflow.set_entry_point("classify")
workflow.add_conditional_edges("classify", route_by_classification)
workflow.add_edge("support_handler", END)
workflow.add_edge("sales_handler", END)
workflow.add_edge("discard", END)

graph = workflow.compile()

The add_conditional_edges call does the heavy lifting. After classification, the graph calls route_by_classification, inspects the return value, and sends execution down the matching branch. Spam gets discarded with no LLM call. Support and sales each follow their own handling path.

Human-in-the-loop review#

LangGraph supports interrupt nodes where execution pauses and waits for external input. For email, this is useful when you don't want the agent auto-sending replies to anything sensitive.

from langgraph.checkpoint.memory import MemorySaver

def human_review(state: EmailState) -> dict:
    return {"human_approved": True}

workflow.add_node("review", human_review)
workflow.add_edge("support_handler", "review")

checkpointer = MemorySaver()
graph = workflow.compile(checkpointer=checkpointer, interrupt_before=["review"])

With interrupt_before=["review"], the graph pauses before executing the review node. Your application displays the draft to a human, collects their decision, and resumes execution. The MemorySaver checkpointer persists state in memory during the pause. Swap it for a PostgreSQL-backed checkpointer when you go to production.

Where LangGraph stops and real email begins#

You now have a graph that classifies, routes, drafts, and reviews. But it can't actually touch a real inbox.

To connect this to Gmail, you'd need OAuth 2.0 credentials, a consent screen, token refresh logic, and IMAP polling. For Outlook, it's the Microsoft Graph API with similar OAuth complexity. Either way, you're writing hundreds of lines of infrastructure code that has nothing to do with your agent's actual logic. We covered this same friction in langchain email integration without the OAuth headache. The short version: building the graph is the fun part, and email plumbing is where the momentum dies.

LobsterMail fills this gap. Your agent pinches its own inbox with one SDK call, receives email through polling or webhooks, and sends replies through a clean API. No OAuth, no IMAP, no credential management. Connect LobsterMail as the entry and exit nodes in your graph, and the middle stays pure agent logic.

from lobstermail import LobsterMail

async def receive_email_node(state: EmailState) -> dict:
    lm = await LobsterMail.create()
    inbox = await lm.create_smart_inbox(name="support-agent")
    emails = await inbox.receive(limit=1)
    if emails:
        email = emails[0]
        return {
            "sender": email.sender,
            "subject": email.subject,
            "body": email.text,
            "thread_history": [],
            "iteration": 0,
            "human_approved": False,
        }
    return state

async def send_reply_node(state: EmailState) -> dict:
    if state["human_approved"] and state["draft_reply"]:
        lm = await LobsterMail.create()
        inbox = lm.get_inbox("support-agent")
        await inbox.send(
            to=state["sender"],
            subject=f"Re: {state['subject']}",
            body=state["draft_reply"],
        )
    return state

Plug these in as the first and last nodes. The classification, routing, drafting, and review nodes in the middle don't change at all.

Shipping to production#

MemorySaver works during development but won't survive a restart. LangGraph supports SQLite and PostgreSQL checkpointers out of the box, so switch to one of those before deploying. Your agent needs to resume long-running email threads across sessions rather than losing context every time the process recycles.

If you're processing hundreds of emails per hour, you'll hit LLM rate limits before you hit email rate limits. Batch your classification calls, cache common spam patterns with a regex pre-filter, and use a cheaper model (GPT-4o-mini works well) for triage. Save the bigger models for reply drafting where quality actually matters.

Turn on LangSmith tracing early. Stateful workflows have more failure modes than linear chains, and most bugs hide in unexpected state mutations between nodes. Being able to see the exact state at each step saves hours of print-statement debugging.

Watch your thread history size. If you're stuffing every message into thread_history, you'll blow past LLM context windows within a few exchanges. Summarize older messages and keep only the last three to five in full. Your agent's replies will be better for it because it's working from a clean signal rather than 40 pages of quoted replies.

Start with the infrastructure. Get your agent a working inbox first, then build the LangGraph workflow around it. Debugging classification logic is a lot more fun when you're not also debugging SMTP timeouts.