LangGraph Human-in-the-Loop Approval with Actionbox

LangGraph's interrupt() can pause a graph before a consequential operation. Actionbox turns that surfaced interrupt into a durable request for a person, then the graph driver resumes the same checkpoint with Command(resume=...).

How do I connect a LangGraph interrupt to a human reviewer?

Call interrupt() inside the graph node with a JSON-serializable description of the proposed operation. When graph.ainvoke() returns __interrupt__, let the outer graph driver create an Actionbox request and wait for the reviewer's decision. Resume the saved checkpoint with the same thread_id and Command(resume={interrupt_id: decision}).

The handoff in the driver is short. The ask_actionbox() helper, shown in full below, sends the review request and returns {"approved": true} only for a valid human approval:

result = await graph.ainvoke(initial_input, config=config)
pending = result["__interrupt__"][0]

decision = await ask_actionbox(actionbox, pending, thread_id)
result = await graph.ainvoke(
    Command(resume={pending.id: decision}),
    config=config,
)

Keep the Actionbox request outside the interrupt node because LangGraph starts that node again when it resumes. This prevents a resume from creating a second review request.

The safe LangGraph human-in-the-loop boundary keeps state ownership explicit:

What LangGraph and Actionbox each own

Actionbox does not replace the LangGraph checkpointer. Use a durable checkpointer in production and resume with the same stable thread_id used when the interrupt was created.

Do you need an approval service for LangGraph?

LangGraph can pause and resume with your own input interface. Use that path when the person answering is already inside your application and your application handles their identity, decision record, and return to the graph.

Actionbox is useful when the reviewer needs a separate inbox shared with other workflows. The graph still needs its checkpointer and run driver. Adding an approval service does not make in-memory graph state survive a process restart.

If you are still choosing which operations need review, start with the human-in-the-loop automation design guide. This tutorial focuses on wiring that decision into LangGraph. The LangGraph integration guide is a shorter reference for state ownership and resume behavior.

Try the control flow without an account

Download the LangGraph demonstration, extract it into a new directory, and follow the README. It uses real LangGraph interrupts with simulated ActionBox responses. No Source key or LLM API key is required.

python3 -m venv .venv
. .venv/bin/activate
python -m pip install -r langgraph-evidence-requirements.txt
python langgraph_evidence.py

We ran these checks on September 10, 2026 with Python 3.12, LangGraph 1.2.11 and langgraph-checkpoint 4.2.0. All seven scenarios passed:

Each scenario created one simulated request. The bundle includes the assertions, dependency versions and JSON results so you can reproduce the checks.

These results validate this example's decision checks and graph resume behavior. They do not establish hosted API availability, human response times, restart recovery or exactly-once deployment. Timeout is a simulated empty response, not a measured waiting period. The execution node writes a state marker rather than deploying anything.

To try an actual review, create a free Source, keep its key in your environment, and use the hosted SDK integration below with a harmless synthetic proposal.

Install the SDKs

python -m pip install actionbox-sdk langgraph
export ACTIONBOX_API_KEY="axb_live_..."

Complete LangGraph interrupt approval example

Keep interrupt() inside the graph node. Create the Actionbox Action in the driver after LangGraph surfaces the interrupt; otherwise the node can repeat that network side effect when it restarts on resume.

from __future__ import annotations

import asyncio
import hashlib
import json
import os
from typing import Any, TypedDict

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


class ApprovalState(TypedDict, total=False):
    proposal: dict[str, Any]
    approved: bool
    outcome: str


def approval_node(state: ApprovalState) -> dict[str, bool]:
    proposal = state["proposal"]
    decision = interrupt({
        "kind": "actionbox_approval",
        "title": proposal["title"],
        "tool": proposal["tool"],
        "arguments": proposal["arguments"],
    })
    return {
        "approved": (
            isinstance(decision, dict)
            and decision.get("approved") is True
        )
    }


def execute_node(state: ApprovalState) -> dict[str, str]:
    if not state.get("approved"):
        return {"outcome": "rejected"}
    # Perform the exact proposed side effect here.
    return {"outcome": "executed"}


builder = StateGraph(ApprovalState)
builder.add_node("approval", approval_node)
builder.add_node("execute", execute_node)
builder.add_edge(START, "approval")
builder.add_edge("approval", "execute")
builder.add_edge("execute", END)
graph = builder.compile(checkpointer=InMemorySaver())


async def ask_actionbox(client, pending, thread_id: str) -> dict[str, bool]:
    value = pending.value
    rendered = json.dumps(value, sort_keys=True, ensure_ascii=False)
    identity = hashlib.sha256(
        f"{thread_id}\0{pending.id}".encode()
    ).hexdigest()

    action = await asyncio.to_thread(
        client.create,
        title=str(value.get("title") or "Approve LangGraph operation")[:200],
        description="LangGraph paused before this operation.",
        interaction={
            "type": "boolean",
            "label": "Allow this operation?",
            "true_label": "Approve",
            "false_label": "Reject",
        },
        context=[{
            "type": "code",
            "title": "Interrupt payload — remove secrets before sending",
            "language": "json",
            "content": rendered,
        }],
        metadata={
            "framework": "langgraph",
            "thread_id": thread_id,
            "interrupt_id": pending.id,
        },
        idempotency_key=f"langgraph:{identity}",
    )

    response = await asyncio.to_thread(action.wait, timeout=3600)
    if response is None and action.status == "open":
        await asyncio.to_thread(
            client.cancel,
            action.id,
            "LangGraph approval timed out; rejected fail-closed.",
        )

    return {
        "approved": (
            isinstance(response, dict)
            and response.get("type") == "boolean"
            and response.get("value") is True
        )
    }


async def main() -> None:
    config = {"configurable": {"thread_id": "release-2.18.0"}}
    thread_id = config["configurable"]["thread_id"]
    proposal = {
        "title": "Deploy version 2.18.0 to staging?",
        "tool": "deploy_service",
        "arguments": {"environment": "staging", "version": "2.18.0"},
    }

    with Actionbox(os.environ["ACTIONBOX_API_KEY"]) as actionbox:
        result = await graph.ainvoke({"proposal": proposal}, config=config)

        while result.get("__interrupt__"):
            pending = list(result["__interrupt__"])
            decisions = await asyncio.gather(
                *(ask_actionbox(actionbox, item, thread_id) for item in pending)
            )
            resume = {
                item.id: decision
                for item, decision in zip(pending, decisions, strict=True)
            }
            result = await graph.ainvoke(
                Command(resume=resume),
                config=config,
            )

    print(result["outcome"])


if __name__ == "__main__":
    asyncio.run(main())

The downloadable demonstration bundle includes the driver and reproducible checks. The example above uses InMemorySaver only to remain self-contained; replace it with a durable checkpointer before production use.

Why the Action belongs outside the interrupt node

LangGraph resumes by restarting the node from its beginning. Any code before interrupt() can run again. Creating an Actionbox Action there can therefore send duplicate requests unless the side effect is perfectly idempotent.

The cleaner pattern is:

1. Let the node return a JSON-serializable interrupt payload.
2. Let the driver observe __interrupt__.
3. Create or recover the idempotent Action outside the graph node.
4. Resume the same checkpoint with the typed decision.

Use a stable thread_id and interrupt ID

The thread_id reconnects a later invocation to the saved graph state. The interrupt ID identifies the exact paused operation within that thread. Hash both into the Actionbox idempotency key so a restarted driver recovers the same logical approval.

Do not use a random idempotency key on every retry. That turns one paused tool call into multiple human interruptions with potentially conflicting answers.

Resume multiple interrupts by ID

When LangGraph surfaces multiple interrupts, collect a decision for every item and pass an interrupt-ID map to Command(resume=resume). Do not assume the first interruption is the only one, and do not resume unrelated decisions by list position alone.

Fail closed and keep the payload complete

A missing, expired, malformed, or false typed response must resume as {"approved": false}. The execute node then returns without performing the side effect.

Redact secret-bearing arguments before creating the interrupt payload. Set a payload-size limit and reject an approval request that cannot show the complete reviewed operation; silent truncation breaks the connection between what the person saw and what the graph executes.

Troubleshoot a graph that does not resume as expected

Persist the Action ID after creation and re-read that Action during recovery. Create-request idempotency has a retention window; it is not a permanent replacement for the driver's saved mapping. Keep the proposal unchanged on a transport retry and use the idempotency documentation for the current contract.

Production checklist

• Replace InMemorySaver with a durable checkpointer.
• Use the same stable thread_id for start and resume.
• Create Actionbox Actions in the driver, not before interrupt().
• Hash the thread and interrupt IDs into a stable idempotency key.
• Handle every surfaced interrupt.
• Treat timeout, expiry, rejection, and malformed responses as false.
• Keep the side-effecting node after the approval node and make execution idempotent.
• Report the real execution outcome after the tool finishes.

For the broader architecture, read human-in-the-loop AI agents. If you use the OpenAI runtime instead, follow the OpenAI Agents SDK human-in-the-loop tutorial. For checkpoint recovery concepts, see durable execution explained. For production controls around the graph, use the AI agent governance and AI agent observability guides.

Create a free Source · Human approval API · Read the integration docs

Sources and further reading

• LangGraph interrupt documentation: checkpoint persistence, thread IDs, interrupt payloads, and Command(resume=...).
• Actionbox documentation: typed interactions, idempotency, callbacks, and outcomes.
• OpenAI Agents SDK human-in-the-loop with Actionbox: the equivalent tool-interruption integration.