Human Interrupts and Approvals in LangGraph

Any agent that can send an email, run a database query, or spend money eventually needs a pause button. LangGraph interrupts are the mechanism for that pause: they stop a running graph at a specific node, hand control back to a human (or any external system), and let you resume execution later with new input. This is the backbone of human-in-the-loop approval flows in LangGraph, and it works because every graph run is checkpointed, so pausing does not mean losing state.

In this guide we build a real approval workflow: an agent drafts an action, a human reviews it, and only after explicit approval does the agent execute. Along the way we cover the interrupt() function, the Command primitive used to resume, checkpointers, editing state mid-pause, and the gotchas that trip people up the first time.

Why LangGraph needs interrupts

A LangGraph graph is a state machine. Nodes are functions, edges are transitions, and a checkpointer persists the state after every node runs. Because state is persisted at each step, LangGraph can stop execution at any point, serialize exactly where it stopped, and pick back up later, even in a different process, on a different machine, hours or days later.

That property is what makes LangGraph interrupts different from just raising an exception in your own code. A raised exception loses the call stack. A LangGraph interrupt suspends the graph at a node boundary, writes a checkpoint, and returns control to whatever called .invoke() or .stream(). Nothing is lost: not the messages so far, not the tool calls in flight, not any local state you had set on that turn.

This matters for a specific category of agent: ones that take consequential actions. Deleting a record, sending a customer email, approving a refund, merging a pull request, placing an order. You want the agent to reason and draft the action automatically, but you want a human gate before the side effect actually happens.

Setting up a graph with a checkpointer

Interrupts require a checkpointer. Without one, LangGraph has nowhere to persist the paused state, so the interrupt call will fail. For local development, MemorySaver is the fastest way to try this out. For anything running across requests in production, use a persistent checkpointer like PostgresSaver or SqliteSaver.

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict

class AgentState(TypedDict):
    task: str
    draft_action: str
    approved: bool
    result: str

builder = StateGraph(AgentState)

checkpointer = MemorySaver()

Every graph that uses interrupts also needs a thread_id. The thread_id is what ties a paused run to the checkpoint that will resume it. Think of it as a session id: one thread per conversation, per ticket, per approval request.

config = {"configurable": {"thread_id": "approval-run-42"}}

The interrupt() function

The core primitive is interrupt(), imported from langgraph.types. Call it inside a node, pass it a payload describing what you need from the human, and execution pauses right there. The payload can be a string, a dict, anything JSON-serializable, and it becomes visible to whatever is polling the graph for interrupts.

from langgraph.types import interrupt, Command

def draft_action_node(state: AgentState) -> AgentState:
    draft = f"Send a refund of $50 to customer for task: {state['task']}"
    return {"draft_action": draft}

def approval_node(state: AgentState) -> AgentState:
    decision = interrupt({
        "question": "Approve this action?",
        "action": state["draft_action"],
    })
    return {"approved": decision.get("approved", False)}

def execute_node(state: AgentState) -> AgentState:
    if not state["approved"]:
        return {"result": "Action was rejected by reviewer."}
    return {"result": f"Executed: {state['draft_action']}"}

builder.add_node("draft", draft_action_node)
builder.add_node("approval", approval_node)
builder.add_node("execute", execute_node)

builder.add_edge(START, "draft")
builder.add_edge("draft", "approval")
builder.add_edge("approval", "execute")
builder.add_edge("execute", END)

graph = builder.compile(checkpointer=checkpointer)

When approval_node calls interrupt(...), the graph run stops. graph.invoke(...) returns immediately with a special value describing the interrupt rather than the final state. Your application code, a web server handler, a CLI prompt, a Slack bot, then shows that payload to a human and waits for a decision.

Resuming with Command

To resume a paused graph, you call .invoke() or .stream() again on the same graph, with the same thread_id, but instead of passing new input you pass a Command object with a resume value. That resume value becomes the return value of the interrupt() call inside the node, and the node picks up right where it left off.

initial_input = {"task": "customer requested refund", "approved": False, "result": ""}

result = graph.invoke(initial_input, config=config)
print(result["__interrupt__"])
# shows the payload passed to interrupt(): the question and the draft action

human_decision = Command(resume={"approved": True})
final = graph.invoke(human_decision, config=config)
print(final["result"])
# Executed: Send a refund of $50 to customer for task: customer requested refund

Notice what happened here: graph.invoke was called twice on the same thread. The first call ran draft_action_node and approval_node, hit the interrupt, and stopped. The second call did not restart from START. It resumed inside approval_node, exactly at the interrupt() call, with decision bound to the dict we passed as resume. Execution then continued to execute_node normally.

This is the entire mental model for LangGraph interrupts: pause with interrupt(payload), inspect the payload outside the graph, resume with Command(resume=value).

A realistic tool-approval pattern

Most agents built with LangGraph use a ReAct-style loop with tool calling. The most common use case for interrupts is gating a specific tool, not the whole graph. Here is a pattern that inserts an approval step only before a sensitive tool runs, letting other tools execute freely.

from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver

SENSITIVE_TOOLS = {"delete_record", "send_payment"}

@tool
def delete_record(record_id: str) -> str:
    """Delete a record permanently."""
    return f"Deleted record {record_id}"

@tool
def lookup_record(record_id: str) -> str:
    """Look up a record, read-only, no approval needed."""
    return f"Record {record_id}: status=active"

def call_model(state: MessagesState) -> MessagesState:
    # your model call with tools bound goes here
    ...

def route_tools(state: MessagesState):
    last = state["messages"][-1]
    calls = getattr(last, "tool_calls", [])
    if not calls:
        return END
    if any(c["name"] in SENSITIVE_TOOLS for c in calls):
        return "human_gate"
    return "tools"

def human_gate(state: MessagesState) -> MessagesState:
    last = state["messages"][-1]
    decision = interrupt({
        "type": "tool_approval",
        "tool_calls": last.tool_calls,
    })
    if not decision.get("approved"):
        # short-circuit: skip execution, tell the model it was denied
        return {"messages": [{"role": "tool", "content": "Denied by reviewer."}]}
    return {}

builder = StateGraph(MessagesState)
builder.add_node("agent", call_model)
builder.add_node("human_gate", human_gate)
builder.add_node("tools", lambda state: state)  # your ToolNode here

builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", route_tools, ["tools", "human_gate", END])
builder.add_edge("human_gate", "tools")
builder.add_edge("tools", "agent")

graph = builder.compile(checkpointer=MemorySaver())

The route_tools function checks the tool calls the model just produced. Read-only tools like lookup_record go straight to execution. Anything in SENSITIVE_TOOLS gets routed through human_gate first, where the graph pauses until a human approves or denies. This is the shape of nearly every production human-in-the-loop agent: cheap and safe actions run automatically, expensive or irreversible ones stop for review.

Editing state during a pause, not just approving

interrupt() is not limited to a yes/no decision. Because the resume value becomes whatever the node receives, you can use it to let a human edit the draft before it runs, not just accept or reject it.

def approval_node(state: AgentState) -> AgentState:
    decision = interrupt({
        "question": "Approve or edit this action",
        "action": state["draft_action"],
    })
    if decision.get("edited_action"):
        return {"approved": True, "draft_action": decision["edited_action"]}
    return {"approved": decision.get("approved", False)}

Now the human reviewing the interrupt has three options: approve as-is, reject, or approve a modified version. This is the pattern behind editable drafts, think an agent that writes a support reply and a human tweaks the wording before it sends, or an agent that proposes a SQL query and an engineer edits the WHERE clause before it runs.

You can also update graph state directly, outside of the resume value, using graph.update_state(config, values) before resuming. This lets a reviewer change fields the node itself never asked about, useful for correcting upstream state rather than just answering the interrupt's question.

graph.update_state(config, {"draft_action": "Send a refund of $30 to customer"})
graph.invoke(Command(resume={"approved": True}), config=config)

Multiple interrupts in one run

A single graph run can hit interrupt() more than once, either because the same node calls it in a loop, or because the graph visits multiple approval nodes in sequence. Each call to interrupt() is matched to its resume value in call order within that node execution, so if a node calls interrupt() twice before returning, you must resume it with a matching sequence of resume values, most commonly by calling .invoke() again for each pause as it occurs, rather than batching resume values up front.

For graphs with several distinct approval points, check result["__interrupt__"] after every invoke call. If it is non-empty, the run paused and needs another resume. If it is empty, the run reached END and you have the final state. A simple driving loop looks like this:

def run_with_approvals(graph, initial_input, config, decide_fn):
    result = graph.invoke(initial_input, config=config)
    while result.get("__interrupt__"):
        payload = result["__interrupt__"][0].value
        decision = decide_fn(payload)  # your UI, CLI prompt, or Slack callback
        result = graph.invoke(Command(resume=decision), config=config)
    return result

decide_fn is where your actual human interface lives: a CLI input() call while prototyping, a webhook that writes to a database and waits for a UI click in production, or a Slack message with approve/deny buttons.

Static interrupts versus the interrupt() function

LangGraph also supports static interrupts, declared at compile time with interrupt_before or interrupt_after on the compile() call.

graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["execute_node"],
)

This pauses before execute_node runs, every single time, unconditionally. It is simpler but far less flexible than calling interrupt() inside a node: there is no payload describing what needs review, and you cannot decide dynamically, based on the tool being called or the size of the action, whether a pause is warranted. Static interrupts are good for a quick debugging breakpoint, stepping through a graph node by node during development. The dynamic interrupt() function is what you want for real approval logic, since it lets a single node decide, based on current state, whether to pause at all and what to show the reviewer.

Older LangGraph code sometimes uses NodeInterrupt, an exception-based approach that predates interrupt(). If you are starting a new project, use interrupt() and Command(resume=...); it is the current, more general mechanism and handles the multi-interrupt and state-editing cases cleanly.

Idempotency: the part everyone misses

When a graph resumes after an interrupt, LangGraph does not resume "in the middle of a Python function." It resumes by re-running the node from the top, and replaying up to the point of the interrupt() call using cached results, then continuing past it with the new resume value. In practice this means any code before interrupt() inside the same node will execute again on resume.

This matters if that code has side effects. If your node does this:

def bad_node(state):
    send_slack_notification("Reviewing action...")  # side effect before interrupt
    decision = interrupt({"action": state["draft_action"]})
    return {"approved": decision["approved"]}

the Slack notification can fire again when the node resumes, depending on how the checkpointer replays the node. The safe pattern is to keep everything before interrupt() pure (reading state, formatting a payload) and put side effects only after the resume value comes back, or in a separate node downstream of the approval node. Treat interrupt() as a boundary: pure logic before it, effects only after it.

Testing interrupts without a UI

While building the workflow, you do not need a real approval UI. Drive the graph directly in a script or a test, using Command(resume=...) with hardcoded decisions, to verify both the approve and reject paths.

def test_approval_flow():
    config = {"configurable": {"thread_id": "test-1"}}
    initial = {"task": "test task", "approved": False, "result": ""}

    paused = graph.invoke(initial, config=config)
    assert paused["__interrupt__"]

    approved = graph.invoke(Command(resume={"approved": True}), config=config)
    assert "Executed" in approved["result"]

    config2 = {"configurable": {"thread_id": "test-2"}}
    graph.invoke(initial, config=config2)
    rejected = graph.invoke(Command(resume={"approved": False}), config=config2)
    assert "rejected" in rejected["result"]

Each test uses its own thread_id so the two runs do not share checkpoint state. This is worth calling out because it is a common source of confusing test failures: reusing a thread_id across tests carries over the previous run's checkpoint, and a "fresh" invoke silently resumes old state instead of starting over.

Putting it behind an API

In a real service, the two graph.invoke() calls, the initial run and the resume, sit behind two endpoints. The first endpoint kicks off the agent and returns the interrupt payload plus the thread_id to the client. The second endpoint accepts a thread_id and a decision, and calls resume.

def start_task(task: str) -> dict:
    thread_id = generate_thread_id()
    config = {"configurable": {"thread_id": thread_id}}
    result = graph.invoke({"task": task, "approved": False, "result": ""}, config=config)
    return {"thread_id": thread_id, "interrupt": result.get("__interrupt__")}

def submit_decision(thread_id: str, decision: dict) -> dict:
    config = {"configurable": {"thread_id": thread_id}}
    result = graph.invoke(Command(resume=decision), config=config)
    return {"result": result.get("result"), "interrupt": result.get("__interrupt__")}

With a persistent checkpointer like PostgresSaver, this works correctly even if start_task and submit_decision run in different server processes, minutes or hours apart, which is exactly the durability guarantee approval workflows need: the reviewer might not click approve for a day, and the graph state has to survive that gap without holding a thread open.

FAQ

What is an interrupt in LangGraph? It is a checkpoint-backed pause inside a graph node, created by calling interrupt(payload). The graph run stops at that exact point, the state is persisted by the checkpointer, and execution resumes later when the caller passes Command(resume=value) with the same thread_id.

Do I need a checkpointer to use interrupt()? Yes. interrupt() relies on the checkpointer to save the paused state. Compile the graph with checkpointer=MemorySaver() for local testing or a persistent saver like PostgresSaver for production, otherwise the interrupt call will fail.

How is interrupt() different from interrupt_before? interrupt_before (and interrupt_after) are static, declared once at compile() time, and pause unconditionally before or after a named node. The interrupt() function is dynamic: it is called inside a node's own code, can decide conditionally whether to pause, and can carry a custom payload describing what needs review.

Can I resume with different data than what was requested? Yes. The resume value is whatever you pass to Command(resume=...). Nothing forces it to match the shape of the payload passed into interrupt(). This is how edit-before-approve flows work: the node asks for approval, but the human resume value carries an edited draft instead of a plain yes or no.

What happens if I resume the wrong thread_id? Nothing resumes. The thread_id in the config selects which checkpoint to load. Using a thread_id with no matching checkpoint, or one that already reached END, will not raise the interrupt again; it starts a fresh run or does nothing meaningful with the stale resume value, so track thread ids carefully in your application.

Is code before interrupt() inside a node safe to run twice? Only if it has no side effects. LangGraph resumes a node by replaying it up to the interrupt point, so any network calls, writes, or notifications placed before interrupt() inside the same node function can execute again on resume. Keep the code before interrupt() pure and put side effects after the resume value is available.

Can multiple interrupts happen in a single graph run? Yes, either from one node calling interrupt() in a loop or from several nodes each pausing in sequence. Drive this with a loop in your application code: invoke, check result["__interrupt__"], resume, and repeat until the interrupt list comes back empty.