TopolearnApprovals, edits and missing information from people are essential for agents that take real actions.
interrupt(value) pauses the graph inside a node and hands value (any JSON-serializable data, like a question or a proposed action) to the caller. Requirements: the graph has a checkpointer and you invoke it with a thread_id, because the paused state must be saved.
What the caller sees: invoke returns early and the result contains "__interrupt__", a list of Interrupt objects with your value. graph.get_state(config).next shows the paused node. The run can wait minutes or weeks; nothing is held in memory.
To continue, invoke again on the same thread with Command(resume=answer). The node runs again from its beginning, and this time interrupt(...) returns answer instead of pausing.
That restart has consequences:
- Code before
interrupt()runs twice. Keep it free of side effects (don't send the email before asking), or move side effects into a separate node after approval. - If a node has several
interrupt()calls, keep their order fixed; resume values are matched by position. - Don't wrap
interrupt()in a baretry/exceptthat swallows its special exception.
Common patterns: approve/reject a tool call, let a reviewer edit a draft, ask for missing details, validate input in a loop. Tools themselves can call interrupt() too. LangGraph 1.2 adds an optional response_schema= so clients can render a typed form and invalid answers are rejected.
Command(resume=...) is the only Command you pass into invoke; for new user messages pass a normal input dict.
Example
from typing import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
class State(TypedDict):
action: str
status: str
def approve(state: State) -> dict:
ok = interrupt({"question": f"Run '{state['action']}'?"}) # pauses here
return {"status": "approved" if ok else "rejected"}
b = StateGraph(State)
b.add_node(approve)
b.add_edge(START, "approve")
b.add_edge("approve", END)
graph = b.compile(checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "deploy-7"}}
first = graph.invoke({"action": "deploy prod", "status": ""}, cfg)
print(first["__interrupt__"][0].value) # {'question': "Run 'deploy prod'?"}
final = graph.invoke(Command(resume=True), cfg)
print(final["status"]) # approvedRemember
- interrupt(value) pauses inside a node; needs a checkpointer + thread_id
- Paused runs return '__interrupt__' with your payload
- Resume with invoke(Command(resume=answer), same config)
- The node re-runs from the top: no side effects before interrupt()
Watch out for
- Calling an external API before interrupt() in the same node, so it runs again on resume.