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=...).
The safe LangGraph human-in-the-loop boundary keeps state ownership explicit:
What LangGraph owns—and what Actionbox owns
| Concern | Owner |
|---|---|
| Graph state and checkpoint | LangGraph checkpointer |
Interrupt ID and thread_id | LangGraph |
| Reviewer notification and inbox | Actionbox |
| Typed approval or rejection | Actionbox |
| Side effect and its credentials | Your graph node or tool adapter |
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.
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 repository contains the maintained runnable version at integrations/agent_frameworks/langgraph_interrupts.py. 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:
- Let the node return a JSON-serializable interrupt payload.
- Let the driver observe
__interrupt__. - Create or recover the idempotent Action outside the graph node.
- 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.
Production checklist
- Replace
InMemorySaverwith a durable checkpointer. - Use the same stable
thread_idfor 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.
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.
