LangGraph interrupt(): Pause Agents for Human Approval

Why a Pause Button Beats a Smarter Prompt

Production agents fail in two ways: they confidently take an action you didn't sanction, or they refuse so often they're useless. The fix is not a better system prompt — it's a pause point. LangGraph 1.2 gives you that pause in one line: interrupt(). The graph freezes, you inspect the proposed action, you approve, edit, or reject, and execution resumes from the exact same state. No re-running the LLM, no lost context.

In this guide you'll build a flight-booking agent that calls a tool, hits an interrupt() right before the booking is committed, and resumes only after a human signs off. By the end you'll know how to wire MemorySaver, raise an interrupt with a payload, and resume with Command(resume=...) — the same pattern you'd ship to production behind an approval queue.

Prerequisites

• Python 3.10+ and a working virtualenv
• An Anthropic API key (or OpenAI — the pattern is identical)
• Comfort reading a small StateGraph; we'll keep it to two nodes

pip install -U langgraph langchain-anthropic langgraph-checkpoint

Step 1 — Define the State and a Risky Tool

The state is a typed dict that flows through every node. Our tool is a fake book_flight that we want a human to approve before it commits.

from typing import TypedDict, Optional
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool

class State(TypedDict):
    user_request: str
    proposed_action: Optional[dict]
    booking_confirmation: Optional[str]

@tool
def book_flight(origin: str, destination: str, date: str, price_usd: float) -> str:
    """Books a flight. Charges the user's saved card."""
    # imagine a real Stripe + airline API call here
    return f"BOOKED {origin}->{destination} on {date} for ${price_usd:.2f}"

Step 2 — The Planner Node

The planner asks the LLM what it wants to do and stuffs the structured tool call into proposed_action. We do not execute the tool here — that's the whole point.

llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0).bind_tools([book_flight])

def planner(state: State) -> dict:
    msg = llm.invoke(
        f"User wants: {state['user_request']}. "
        "Propose exactly one book_flight tool call."
    )
    call = msg.tool_calls[0]   # {"name": ..., "args": {...}, "id": ...}
    return {"proposed_action": call}

Step 3 — The Approval Node with interrupt()

interrupt() takes any JSON-serializable payload. Whatever you pass becomes the value the caller sees when the graph pauses. The return value of interrupt() is whatever the caller passes back via Command(resume=...).

from langgraph.types import interrupt, Command

def approval_gate(state: State) -> dict:
    action = state["proposed_action"]
    decision = interrupt({
        "kind": "approve_booking",
        "summary": f"Book {action['args']['origin']}->{action['args']['destination']} "
                   f"for ${action['args']['price_usd']}?",
        "action": action,
    })

    if decision["choice"] == "reject":
        return {"booking_confirmation": "Rejected by user: " + decision.get("reason", "")}
    if decision["choice"] == "edit":
        action = {**action, "args": {**action["args"], **decision["edits"]}}

    result = book_flight.invoke(action["args"])
    return {"booking_confirmation": result}

Three branches: reject records the reason and bails, edit merges the human's tweaks into the args, and the default approve falls through to the real tool call.

Step 4 — Wire the Graph with a Checkpointer

A checkpointer is non-optional: interrupt() serializes the state to it so the graph can resume in a fresh process if it needs to. MemorySaver is fine for development; swap to PostgresSaver in production.

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

g = StateGraph(State)
g.add_node("planner", planner)
g.add_node("approval_gate", approval_gate)
g.add_edge(START, "planner")
g.add_edge("planner", "approval_gate")
g.add_edge("approval_gate", END)

app = g.compile(checkpointer=MemorySaver())

Step 5 — Run It, Hit the Interrupt, Resume

Invoking the graph returns when it finishes or when an interrupt is raised. You inspect state.tasks[*].interrupts, get the human decision, then resume with Command(resume=...).

config = {"configurable": {"thread_id": "booking-42"}}

first = app.invoke(
    {"user_request": "Cheap flight LAX to JFK next Friday"},
    config=config,
)
print(first)

Example output:

{
  "user_request": "Cheap flight LAX to JFK next Friday",
  "proposed_action": {
     "name": "book_flight",
     "args": {"origin": "LAX", "destination": "JFK",
              "date": "2026-05-29", "price_usd": 187.0},
     "id": "toolu_01..."
  },
  "__interrupt__": [{
     "value": {"kind": "approve_booking",
               "summary": "Book LAX->JFK for $187.0?",
               "action": {...}},
     "id": "..."
  }]
}

Now resume. Pass back a structured decision — the same config with the same thread_id tells LangGraph which paused run to wake up.

final = app.invoke(
    Command(resume={"choice": "approve"}),
    config=config,
)
print(final["booking_confirmation"])
# -> BOOKED LAX->JFK on 2026-05-29 for $187.00

Reject path:

app.invoke(
    Command(resume={"choice": "reject", "reason": "Found cheaper on Kayak"}),
    config={"configurable": {"thread_id": "booking-43"}},
)
# booking_confirmation = "Rejected by user: Found cheaper on Kayak"

Edit path — bump the date one day later:

app.invoke(
    Command(resume={"choice": "edit", "edits": {"date": "2026-05-30"}}),
    config={"configurable": {"thread_id": "booking-44"}},
)

Common Pitfalls

• Forgetting the checkpointer. app.compile() with no checkpointer= will raise as soon as interrupt() fires. Always pass one.
• Reusing thread_id across runs. Each interrupted run is keyed on thread_id. Start a fresh thread per user task or your resume will land in the wrong conversation.
• Putting side effects before interrupt(). Anything before interrupt() in the same node runs every time the node resumes. Move charges, emails, or DB writes after the interrupt.
• Non-serializable payloads. The interrupt payload is pickled to the checkpointer. Stick to dicts, lists, strings, numbers, booleans — no live SQLAlchemy sessions or open file handles.
• Approving in a different process. That's fine, but the checkpointer must be shared. MemorySaver is per-process; use PostgresSaver or SqliteSaver for cross-process resumes.

Quick Reference

Where to Go Next

• Swap MemorySaver for PostgresSaver and run two workers against the same Postgres — you'll see one resume a thread the other paused.
• Add a timeout policy: a background job that calls app.update_state to auto-reject interrupts older than N minutes.
• Wire the interrupt payload into a Slack approval message with Approve / Edit / Reject buttons. Each click POSTs back the matching Command(resume=...).
• Read the LangGraph docs on dynamic interrupts — you can decide per-call whether to pause based on the tool args (e.g. only pause for bookings over $500).

Ship it: pick one tool in your agent that can spend money or send a message, and put an interrupt in front of it before Friday.