TopolearnAgents that send emails, spend money or change data must let people approve, edit or reject actions.
LangGraph can pause execution and wait indefinitely for input. Inside a node or tool, call interrupt(payload):
decision = interrupt({"question": "Approve refund?", "amount": 120})Execution stops, the state is saved by the checkpointer (required), and invoke returns with an __interrupt__ entry containing the payload. Later, possibly from another process or days later, resume the same thread with:
graph.invoke(Command(resume={"approved": True}), config)The node re-runs from its start and this time interrupt returns the resume value. Therefore code before interrupt must be idempotent (no side effects such as charging a card before the pause), and multiple interrupts in a node are matched by order.
For agents, HumanInTheLoopMiddleware(interrupt_on={"send_email": True, "read_inbox": False}) pauses before selected tool calls. The reviewer can approve, edit the arguments, or reject with feedback, resuming with Command(resume={"decisions": [{"type": "approve"}]}).
Related tools:
interrupt_before=["node"]/interrupt_afterat compile time for static breakpoints while debugging.graph.get_state(config).nextshows which node is pending;update_statelets a human patch state before resuming.
Design the payload for the UI that will show it: what is proposed, why, and what choices exist.
Example
from typing import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
class State(TypedDict):
amount: int
status: str
def approval(state: State):
decision = interrupt({"question": "Approve refund?", "amount": state["amount"]})
return {"status": "refunded" if decision["approved"] else "rejected"}
builder = StateGraph(State)
builder.add_node("approval", approval)
builder.add_edge(START, "approval")
builder.add_edge("approval", END)
graph = builder.compile(checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "refund-1"}}
first = graph.invoke({"amount": 120, "status": "pending"}, cfg)
print(first["__interrupt__"][0].value) # {'question': 'Approve refund?', 'amount': 120}
print(graph.get_state(cfg).next) # ('approval',)
final = graph.invoke(Command(resume={"approved": True}), cfg)
print(final["status"]) # refundedRemember
interrupt(payload)pauses and saves state; a checkpointer is required- Resume with
Command(resume=value)on the same thread_id; the node re-runs from its start - HumanInTheLoopMiddleware adds approve/edit/reject for chosen agent tools
Watch out for
- Performing side effects before interrupt() in the same node, which then run twice when the node re-executes on resume.