The Complete Production SOP: How to Deploy and Run LangGraph Human-in-the-Loop Approval Checkpoints in Autonomous AI Agents

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

class AgentState(TypedDict):
    input: str
    action_payload: dict
    approved: bool
    result: str

def planner(state: AgentState):
    return {"action_payload": {"action": "delete_records", "target_id": 42}, "approved": False}

def sensitive_action(state: AgentState):
    if not state.get("approved", False):
        raise PermissionError("Action executed without explicit human approval!")
    return {"result": f"Successfully executed {state['action_payload']}"}

builder = StateGraph(AgentState)
builder.add_node("planner", planner)
builder.add_node("sensitive_action", sensitive_action)

builder.set_entry_point("planner")
builder.add_edge("planner", "sensitive_action")
builder.add_edge("sensitive_action", END)

checkpointer = MemorySaver()
# Compile graph with deterministic interruption before the sensitive execution node
app = builder.compile(checkpointer=checkpointer, interrupt_before=["sensitive_action"])

from graph_pipeline import app

config = {"configurable": {"thread_id": "prod_audit_session_101"}}

# 1. Start execution until breakpoint
for event in app.stream({"input": "Purge stale test tenants"}, config):
    print(f"Stream Event: {event}")

# 2. Inspect state at breakpoint
state_snapshot = app.get_state(config)
print("Next node pending execution:", state_snapshot.next)
print("Pending Action Payload:", state_snapshot.values.get("action_payload"))

# 3. Simulate human approval & state modification
app.update_state(config, {"approved": True}, as_node="planner")

# 4. Resume execution by passing None as input
for event in app.stream(None, config):
    print(f"Post-Approval Event: {event}")

import pytest
from graph_pipeline import app

def test_interruption_blocks_unauthorized_execution():
    config = {"configurable": {"thread_id": "test_thread_block"}}
    events = list(app.stream({"input": "Run action"}, config))
    
    snapshot = app.get_state(config)
    assert "sensitive_action" in snapshot.next
    assert snapshot.values.get("approved") is False
    assert "result" not in snapshot.values