LangGraph Human-in-the-Loop: interrupt(), Command and Phone Approvals

LangGraph human-in-the-loop works by calling interrupt(payload) inside a node. The graph saves its state through a checkpointer, stops, and hands the payload back under __interrupt__. You resume with graph.invoke(Command(resume=value), config) using the same thread_id, and interrupt() returns value inside the node.

Every tutorial shows that much with input() in a terminal. This one builds the production version: a durable checkpointer, approve/edit/reject decisions, the re-run trap that causes double sends, and a worker that sends the question to your phone and resumes the thread when you tap a button, even if the worker restarts in between.

All code here was run on Python 3.12 with langgraph 1.2.18, langgraph-checkpoint-sqlite 3.1.2, langchain 1.4.7 and fastmcp 4.1.4 (October 2026).

How interrupt() works

The LangGraph interrupts docs list three requirements:

1. A checkpointer to persist graph state. Use a durable one in production.
2. A thread ID in the config, so LangGraph knows which saved state to resume.
3. A call to interrupt() where you want to pause, with a JSON-serializable payload.

When a node calls interrupt(), LangGraph saves a checkpoint and invoke() returns early. The pending interrupts sit in result["__interrupt__"], a list of Interrupt objects, each with a value (your payload) and an id. The id maps resume values when several interrupts are pending, and later we use it as an idempotency key.

On resume, LangGraph runs the interrupted node again from its first line. This time interrupt() returns the resume value instead of pausing. Keep that in mind; it's the source of the most common bug.

You don't have to keep the invoke() result around to find a paused run. graph.get_state(config) reads the latest checkpoint for a thread, and its interrupts field lists whatever is still waiting for an answer. The worker later in this post relies on that: after a restart, it asks the checkpointer what is pending instead of remembering it in memory.

Two newer features are worth knowing about. interrupt() accepts response_schema= to validate resume input (langgraph 1.2.12 and later), and the docs now recommend driving interrupting graphs with graph.stream_events(..., version="v3"). Plain invoke() still works, and it keeps the worker below short, so that is what we use.

Setup

pip install -U langgraph langgraph-checkpoint-sqlite fastmcp

The SQLite checkpointer ships as its own package, langgraph-checkpoint-sqlite, imported as langgraph.checkpoint.sqlite. FastMCP is only needed for the phone section.

A minimal approval graph

The graph below plans an action, pauses for review, then either executes or cancels. The plan step is a stand-in for an LLM call, so you can run everything without a model key.

# graph.py
import sqlite3
from typing import Literal, TypedDict

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt


class State(TypedDict, total=False):
    sql: str
    status: str


def plan(state: State) -> State:
    # Stand-in for the LLM step that proposes an action.
    return {"sql": "DELETE FROM sessions WHERE last_seen < now() - interval '90 days'"}


def review(state: State) -> Command[Literal["execute", "cancel"]]:
    decision = interrupt({
        "question": "Run this cleanup query on prod?",
        "options": ["Reject", "Approve"],  # safest first
        "detail": state["sql"],
    })
    if decision["type"] == "approve":
        return Command(goto="execute")
    if decision["type"] == "edit":
        return Command(goto="execute", update={"sql": decision["sql"]})
    return Command(goto="cancel", update={"status": "rejected: " + decision.get("message", "")})


def execute(state: State) -> State:
    print("running:", state["sql"])  # the side effect lives after the interrupt
    return {"status": "done"}


def cancel(state: State) -> State:
    return {}


builder = StateGraph(State)
builder.add_node(plan)
builder.add_node(review)
builder.add_node(execute)
builder.add_node(cancel)
builder.add_edge(START, "plan")
builder.add_edge("plan", "review")
builder.add_edge("execute", END)
builder.add_edge("cancel", END)

conn = sqlite3.connect("checkpoints.db", check_same_thread=False)
graph = builder.compile(checkpointer=SqliteSaver(conn))

The review node returns a Command(goto=...), and the Command[Literal["execute", "cancel"]] annotation tells LangGraph which nodes it may route to, so no conditional edge is needed.

Drive it by hand first:

# minimal.py
from langgraph.types import Command

from graph import graph

config = {"configurable": {"thread_id": "cleanup-1"}}

result = graph.invoke({}, config)
print(result["__interrupt__"][0].value)

result = graph.invoke(Command(resume={"type": "approve"}), config)
print(result["status"])

{'question': 'Run this cleanup query on prod?', 'options': ['Reject', 'Approve'], 'detail': "DELETE FROM sessions WHERE last_seen < now() - interval '90 days'"}
running: DELETE FROM sessions WHERE last_seen < now() - interval '90 days'
done

Because the state lives in checkpoints.db, the two invoke() calls don't have to happen in the same process, or on the same day. That property is what makes out-of-band approval possible.

Approve, edit or reject

The resume value can be any JSON-serializable value. A small decision dict covers the three answers a reviewer usually needs, and it mirrors the vocabulary LangChain's own middleware uses:

• {"type": "approve"} runs the action as proposed.
• {"type": "edit", "sql": "..."} runs a corrected action.
• {"type": "reject", "message": "..."} cancels and records why, so a downstream LLM node can adapt instead of retrying.

from langgraph.types import Command

from graph import graph

config = {"configurable": {"thread_id": "cleanup-2"}}
graph.invoke({}, config)
result = graph.invoke(
    Command(resume={"type": "edit", "sql": "DELETE FROM sessions WHERE last_seen < now() - interval '180 days'"}),
    config,
)
print(result["status"], "|", result["sql"])

Treat an edited action like a fresh proposal. If the model's SQL has to pass a validator, the human's edit should pass the same one.

The re-run trap

This is the bug that bites most people once. Here is a node that announces the approval request before pausing:

# trap.py
from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import START, StateGraph
from langgraph.types import Command, interrupt


class State(TypedDict, total=False):
    approved: bool


def review(state: State) -> State:
    print("sending approval request")  # runs on the first pass AND again on resume
    answer = interrupt("Deploy to prod?")
    print("got answer:", answer)
    return {"approved": answer == "Approve"}


builder = StateGraph(State)
builder.add_node(review)
builder.add_edge(START, "review")
graph = builder.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "trap"}}
graph.invoke({}, config)
graph.invoke(Command(resume="Approve"), config)

sending approval request
sending approval request
got answer: Approve

The docs say it directly: the node restarts from the beginning when resumed, so any code before interrupt runs again. Swap that print for a Slack message or a push notification and the reviewer gets the request twice, once before and once after they answered it. Swap it for a database write and you write twice.

The rules that follow from it:

• Put side effects after the interrupt, or in a later node, as execute does above.
• Make anything before the interrupt idempotent, or move it out of the graph entirely. That's why the worker below sends the phone notification from outside the graph.
• Don't wrap interrupt() in a bare try/except. It pauses by raising, and a broad except swallows the pause.
• Keep multiple interrupts in a stable order. Resume values are matched to interrupts by position within a node, so don't skip one conditionally or loop over data that can change between runs. To re-prompt after invalid input, route back with a conditional edge instead of a while loop.

Prebuilt option: HumanInTheLoopMiddleware

If you build agents with LangChain's create_agent, you don't have to write the review node yourself. HumanInTheLoopMiddleware interrupts before the tools you name. This example needs pip install langchain langchain-openai and an OPENAI_API_KEY:

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command


@tool
def execute_sql(query: str) -> str:
    """Run a SQL statement against the production database."""
    return f"ran: {query}"


agent = create_agent(
    model="openai:gpt-5.5",
    tools=[execute_sql],
    middleware=[HumanInTheLoopMiddleware(interrupt_on={"execute_sql": {"allowed_decisions": ["approve", "edit", "reject"]}})],
    checkpointer=InMemorySaver(),
)
config = {"configurable": {"thread_id": "agent-1"}}
result = agent.invoke({"messages": [{"role": "user", "content": "Delete sessions older than 90 days"}]}, config)
print(result["__interrupt__"][0].value["action_requests"])
result = agent.invoke(Command(resume={"decisions": [{"type": "reject", "message": "Archive them instead"}]}), config)
print(result["messages"][-1].content)

The interrupt payload holds action_requests (tool name, args, description) and review_configs (which decisions each tool allows). You resume with one decision per action, in order:

• {"type": "approve"}
• {"type": "edit", "edited_action": {"name": ..., "args": ...}}
• {"type": "reject", "message": "optional reason"}
• {"type": "respond", "message": "required reply"}

respond returns your message to the model as a successful tool result without running the tool. It is meant for "ask the user" style tools. The docs warn against using it to deny a side-effecting action, because the model is told the tool succeeded; use reject for that. Since the middleware raises ordinary interrupts, the phone worker below works with it too: forward action_requests instead of our question payload.

Approving from your phone

In production nobody is watching the terminal. The pattern is a small driver loop:

1. Run the thread until it pauses.
2. Send the interrupt payload to a human through some channel.
3. Wait for the answer, however long the deadline allows.
4. Resume the thread with Command(resume=...).

Any channel works for step 2: Slack, Telegram, email or a push service. One-way push services such as ntfy and Pushover are great at delivery (we compare them in ntfy vs Pushover vs Slack webhooks), but here you also need a round trip: buttons, a reply you can wait for, and a defined outcome when nobody answers. We use AlertMCP here because its MCP tools do exactly that: ask sends a question with two to four buttons to your phone and desktop, and get_answer waits for the tap. If MCP is new to you, our MCP explainer covers the basics.

The ask is sent from the worker, not from inside the node. The node re-runs on resume, so a notification sent inside it would fire twice, and a node can't sensibly block for hours anyway.

# worker.py
import asyncio
import os
import sqlite3
import sys

from fastmcp import Client
from langgraph.types import Command

from graph import graph

ALERTMCP_URL = "https://api.alertmcp.com/mcp"

asks = sqlite3.connect("asks.db")
asks.execute("CREATE TABLE IF NOT EXISTS asks (interrupt_id TEXT PRIMARY KEY, ask_id TEXT)")


async def phone_answer(pending) -> str | None:
    """One ask per interrupt, even across restarts. Returns the tapped label or None."""
    async with Client(ALERTMCP_URL, auth=os.environ["ALERTMCP_API_KEY"]) as mcp:
        row = asks.execute("SELECT ask_id FROM asks WHERE interrupt_id = ?", (pending.id,)).fetchone()
        if row:
            ask_id = row[0]
        else:
            p = pending.value
            sent = await mcp.call_tool("ask", {
                "question": p["question"][:200],
                "options": p["options"],  # safest first: the timeout fallback
                "body": p["detail"][:2000],
                "priority": "high",
                "topic": "langgraph",
            })
            ask_id = sent.data["id"]
            asks.execute("INSERT INTO asks VALUES (?, ?)", (pending.id, ask_id))
            asks.commit()
        while True:  # each call waits up to 50 s, then reports "pending"
            r = await mcp.call_tool("get_answer", {"id": ask_id, "wait_seconds": 50}, timeout=70)
            if r.data["status"] != "pending":
                return r.data.get("answer")  # None if it timed out with no fallback


def run(thread_id: str) -> None:
    config = {"configurable": {"thread_id": thread_id}}
    if not graph.get_state(config).values:
        graph.invoke({}, config)  # new thread: plan, then pause at review
    while pending := graph.get_state(config).interrupts:
        answer = asyncio.run(phone_answer(pending[0]))
        if answer == "Approve":
            decision = {"type": "approve"}
        else:
            decision = {"type": "reject", "message": f"phone answer: {answer}"}
        graph.invoke(Command(resume=decision), config)
    print(graph.get_state(config).values)


if __name__ == "__main__":
    run(sys.argv[1])

Get an API key by signing in, then run it:

export ALERTMCP_API_KEY="amcp_sk_…"
python worker.py deploy-42

What each part is doing:

• Client(url, auth=key). FastMCP adds the Bearer prefix to a raw token. Because AlertMCP returns structured content, FastMCP's .data is a plain dict with id, status and answer.
• Limits. question is capped at 200 characters, each of the 2 to 4 options at 40, and body at 2,000, hence the slices.
• The wait loop. get_answer waits at most 50 seconds per call and returns pending if nobody has answered yet, so the worker keeps calling it. The status ends as answered or timed_out.
• The asks table. It maps each interrupt id to the AlertMCP ask id. If the worker crashes or redeploys mid-wait, the next run finds the pending interrupt in the checkpointer, finds the ask it already sent, and keeps waiting on that one. You get one notification per decision, not one per restart.
• Default to reject. Anything other than an explicit "Approve", including None after a timeout, resumes with a rejection.
• One interrupt at a time. This graph only ever has one pending interrupt. If parallel branches can pause together, collect an answer for each and resume once with a dict keyed by interrupt id: Command(resume={i.id: decision for i in pending}).

We tested the worker against a local server that uses the same MCP SDK and transport settings as the hosted endpoint, including a kill-and-restart in the middle of a wait.

Letting the agent call ask itself

The worker pattern gates actions from outside the model, which is the right default. If you would rather the agent decide when to ask, give it the MCP tools. LangChain's langchain[mcp] package (version 1.4.0 or later, currently beta) exposes MCP servers as tools through MCPAdapter, which wraps the same FastMCP Client shown above. It replaces the standalone langchain-mcp-adapters package and its MultiServerMCPClient; there's a migration guide if you're on the old one. Our remote MCP server guide covers connecting AlertMCP to Claude Code, Cursor and other clients.

One-way alerts over HTTP

Not everything needs an answer. For "the run finished" or "the cleanup was rejected", a one-way notification is enough, and it doesn't need MCP. AlertMCP accepts a plain POST /v1/alerts:

curl -X POST https://api.alertmcp.com/v1/alerts \
  -H "Authorization: Bearer $ALERTMCP_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"title": "Cleanup finished", "body": "deploy-42: done", "priority": "normal", "topic": "langgraph"}'

The same call from Python, with httpx, which LangGraph already installs:

import os

import httpx


def notify(title: str, body: str = "", priority: str = "normal") -> None:
    httpx.post(
        "https://api.alertmcp.com/v1/alerts",
        headers={"Authorization": f"Bearer {os.environ['ALERTMCP_API_KEY']}"},
        json={"title": title[:200], "body": body[:2000], "priority": priority, "topic": "langgraph"},
        timeout=10,
    ).raise_for_status()

Call it from the worker after the loop ends, not from a node that might re-run. Questions with buttons go through the MCP tools; the HTTP endpoint is for notifications.

Deadlines, fallbacks and durable state

Pick the checkpointer for your deployment. SQLite is fine for a single machine. If several workers can pick up the same thread, use Postgres:

pip install -U langgraph-checkpoint-postgres "psycopg[binary,pool]"

Then replace the last two lines of graph.py with:

import os

from langgraph.checkpoint.postgres import PostgresSaver
from psycopg import Connection
from psycopg.rows import dict_row

conn = Connection.connect(os.environ["DATABASE_URL"], autocommit=True, prepare_threshold=0, row_factory=dict_row)
checkpointer = PostgresSaver(conn)
checkpointer.setup()  # creates the checkpoint tables
graph = builder.compile(checkpointer=checkpointer)

Those connection settings are what PostgresSaver.from_conn_string() uses internally. Run setup() before the first use. The rest of the tutorial, including the worker, works unchanged. If more than one machine runs workers, move the asks table into the same database and make sure only one worker drives a given thread at a time.

Let the deadline live in one place. In AlertMCP, each workspace sets a maximum response time from 5 minutes to 24 hours, and a topic can override it, so a langgraph topic can have its own deadline. When it runs out, depending on the workspace setting, the agent is told nobody answered, the ask resolves to the first (safest) option, or it escalates to the team's admins. With options ordered ["Reject", "Approve"], the first two outcomes both reject in our worker, and nothing executes without an explicit tap.

Re-check before acting. An approval given six hours after the request may be stale. If execute depends on facts that can drift, such as row counts, the current deploy or a branch head, verify them again before running, and ask again if they changed.

Use meaningful thread IDs. A business key like deploy-42 makes it easy to find a paused run, inspect it with graph.get_state(config), and resume it by hand if you have to.

For deciding which actions deserve an interrupt in the first place, see our guide to human-in-the-loop AI agents.

Testing it

You don't need a phone or a model to test this. Resume values are just data, so drive the graph directly:

# test_graph.py
from langgraph.types import Command

from graph import graph


def test_reject_never_executes(capsys):
    config = {"configurable": {"thread_id": "test-reject"}}
    graph.invoke({}, config)
    out = graph.invoke(Command(resume={"type": "reject", "message": "no"}), config)
    assert out["status"] == "rejected: no"
    assert "running:" not in capsys.readouterr().out


def test_edit_runs_the_edited_query(capsys):
    config = {"configurable": {"thread_id": "test-edit"}}
    graph.invoke({}, config)
    out = graph.invoke(Command(resume={"type": "edit", "sql": "SELECT 1"}), config)
    assert out["status"] == "done"
    assert "running: SELECT 1" in capsys.readouterr().out

For the worker, replace the channel with a stub and check the timeout path:

# test_worker.py
import uuid

import worker


def test_worker_rejects_when_nobody_answers(monkeypatch):
    async def nobody_answered(pending):
        return None

    monkeypatch.setattr(worker, "phone_answer", nobody_answered)
    thread_id = f"test-{uuid.uuid4()}"
    worker.run(thread_id)
    state = worker.graph.get_state({"configurable": {"thread_id": thread_id}})
    assert state.values["status"].startswith("rejected")

Run both with pytest -q. Two more checks are worth doing by hand once: kill the worker while it waits and start it again (you should get no second notification), and count the side effects in any node that calls interrupt() to make sure nothing before the interrupt does real work.