An email verification step has more outcomes than a straight line of nodes shows. The code arrives. It does not arrive, and the site offers a "resend" link. It does not arrive after three tries and a person has to look. The network drops in the middle of a wait. The process restarts an hour later with the inbox still holding the email. LangGraph is a good fit because each of those outcomes becomes an explicit edge in the graph instead of a branch buried in a loop, and its checkpointer gives you a place to resume from.
This post builds that graph in Python against the Lumbox REST API, using the LangGraph 1.x graph API: StateGraph, conditional edges, RetryPolicy on nodes, interrupt() for the human fallback, and a SQLite checkpointer.
State and nodes
import os, sqlite3
from datetime import datetime, timedelta, timezone
from typing import Literal, NotRequired, TypedDict
import httpx
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, RetryPolicy, interrupt
http = httpx.Client(
base_url="https://api.lumbox.co/v1",
headers={"X-API-Key": os.environ["LUMBOX_API_KEY"]},
timeout=130,
)
class SignupState(TypedDict):
service: str
inbox_id: NotRequired[str]
address: NotRequired[str]
requested_at: NotRequired[str]
attempts: NotRequired[int]
code: NotRequired[str | None]
status: NotRequired[str]
def create_inbox(state: SignupState) -> dict:
if state.get("inbox_id"):
return {}
r = http.post("/inboxes", json={})
r.raise_for_status()
body = r.json()
return {"inbox_id": body["id"], "address": body["address"], "attempts": 0}
def request_code(state: SignupState) -> dict:
requested_at = (datetime.now(timezone.utc) - timedelta(seconds=5)).isoformat()
if state["attempts"] == 0:
submit_signup_form(state["service"], state["address"]) # your browser step
else:
click_resend(state["service"]) # your browser step
return {"requested_at": requested_at, "attempts": state["attempts"] + 1}
def wait_for_code(state: SignupState) -> dict:
r = http.get(
f"/inboxes/{state['inbox_id']}/otp",
params={"timeout": 90, "since": state["requested_at"]},
)
if r.status_code == 408:
return {"code": None}
r.raise_for_status()
return {"code": r.json()["code"]}
def route_after_wait(state: SignupState) -> Literal["enter_code", "request_code", "ask_human"]:
if state.get("code"):
return "enter_code"
if state["attempts"] < 3:
return "request_code"
return "ask_human"
def ask_human(state: SignupState) -> dict:
code = interrupt({"address": state["address"], "reason": "no code after 3 requests"})
return {"code": code}
def enter_code(state: SignupState) -> dict:
submit_code(state["service"], state["code"]) # your browser step
http.delete(f"/inboxes/{state['inbox_id']}")
return {"status": "verified"}The inbox is the agent's identity for this one signup, created once and stored in state. create_inbox checks for an existing inbox_id before calling the API, because LangGraph re-runs a node from its first line when it is retried or resumed, and a second POST /v1/inboxes would create a second address. The empty body gives a generated address on trylumbox.com.
request_code records a timestamp just before it acts, minus five seconds for clock skew. That timestamp is the since for the next wait. The /otp endpoint returns the newest verification email received after since; without it, it looks back five minutes, and on the second attempt that window still contains the first code. With it, each wait only sees mail caused by the request that preceded it.
wait_for_code treats a 408 as a result. The endpoint holds the request for up to timeout seconds (the server caps it at 120) and answers 408 when nothing matching has arrived. Writing code: None into state and letting the router decide keeps "no email yet" on a visible edge instead of in an exception handler.
Edges, retries and the checkpointer
net_retry = RetryPolicy(max_attempts=3, retry_on=httpx.TransportError)
builder = StateGraph(SignupState)
builder.add_node("create_inbox", create_inbox, retry_policy=net_retry)
builder.add_node("request_code", request_code)
builder.add_node("wait_for_code", wait_for_code, retry_policy=net_retry)
builder.add_node("ask_human", ask_human)
builder.add_node("enter_code", enter_code)
builder.add_edge(START, "create_inbox")
builder.add_edge("create_inbox", "request_code")
builder.add_edge("request_code", "wait_for_code")
builder.add_conditional_edges("wait_for_code", route_after_wait)
builder.add_edge("ask_human", "enter_code")
builder.add_edge("enter_code", END)
checkpointer = SqliteSaver(sqlite3.connect("signups.db", check_same_thread=False))
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "signup-acme-42"}}
result = graph.invoke({"service": "acme"}, config)There are two kinds of retry here, and they are kept apart on purpose. RetryPolicy with retry_on=httpx.TransportError covers the network: a dropped connection or a read timeout re-runs the node. HTTP errors raised by raise_for_status() are not transport errors, so a 401 (bad key) or 402 (PLAN_LIMIT_EXCEEDED, the plan's inbox count is full) fails the run straight away instead of being retried into the same answer. The business retry, "no code, ask for another one", is the request_code edge, capped at three by state.
request_code has no retry policy because re-running it has a side effect on someone else's site. If it fails, you want the run to stop where you can see it.
The checkpointer saves state after each step, keyed by thread_id. The SQLite saver comes from the langgraph-checkpoint-sqlite package; InMemorySaver works for experiments but forgets everything when the process exits. If the process dies during wait_for_code, calling graph.invoke(None, config) with the same thread ID continues from the last checkpoint and re-runs that wait. Because the wait is anchored to the stored requested_at rather than to "now", an email that arrived while the process was down is still found. Lumbox inboxes do not expire, so the mail is there whenever the graph comes back. The code inside it may not be: the /otp response includes expires when the email stated a lifetime ("expires in 10 minutes"), and a graph resumed hours later should check it before typing the code.
The human fallback
After three requests with no code, the router sends the run to ask_human, and interrupt() pauses the graph there. The value passed to it comes back to the caller:
if "__interrupt__" in result:
print(result["__interrupt__"]) # address and reason, for whoever handles it
result = graph.invoke(Command(resume="482913"), config)Resuming with Command(resume=...) and the same thread ID makes interrupt() return that value, and the graph continues to enter_code. The node restarts from its first line on resume, which is why interrupt() is the first call in ask_human; anything above it would run twice. The person handling it can look at the inbox directly with GET /v1/inboxes/:id/emails to see whether mail arrived at all, or whether it landed in a category the OTP endpoint ignores.
Where this graph is thin
The parser that fills /otp finds 4 to 8 digit codes near words like "code" or "verification", in emails it categorises as verification mail. Services that send only a link, or an alphanumeric code, will never produce a code here. Add a branch that calls GET /v1/inboxes/:id/wait with the same since and reads parsed.verification_links or the text itself for those.
Parallel signups each need their own thread ID and their own inbox, and each inbox counts against your plan until enter_code deletes it (three on Free). A failed run that never reaches enter_code leaves its inbox behind, so a cleanup pass over old thread IDs is worth adding.
For the general idea without a framework, see email as a state machine. Long-polling for agents explains the wait endpoint, and the inbox docs list every field these nodes read.