Workflow Automation Architecture For Human Sign-Off: Durable Approval Steps With LangGraph 1.2 In Python

"""A durable approval step for a client bank-detail change, built on LangGraph.

Python 3.11+, langgraph 1.2.14 (typed interrupts need 1.2.12+), langgraph-checkpoint-sqlite 3.1.1,
pydantic 2.14.0. Client records are sample data. Use PostgresSaver rather than SQLite in production.
"""
from __future__ import annotations

import hashlib
import json
import sqlite3
from datetime import datetime, timezone
from typing import Any, Literal, TypedDict

from langgraph.checkpoint.base import BaseCheckpointSaver
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, START, StateGraph
from langgraph.graph.state import CompiledStateGraph
from langgraph.types import Command, interrupt
from pydantic import BaseModel


class Decision(BaseModel):
    """The only shape a resume value may take. Anything else raises before the node continues."""
    approver_id: str
    decision: Literal["approve", "reject"]
    callback_verified: bool  # the approver rang the client on the number already on file
    note: str = ""


class State(TypedDict, total=False):
    requested_by: str
    proposal: dict[str, str]
    record_version: int
    decision: dict[str, Any]
    outcome: str


# Sample systems of record, standing in for your client master and an idempotency table.
CLIENTS: dict[str, dict[str, Any]] = {
    "CL-204518": {"version": 7, "sort_code": "20-00-00", "account_number": "55779911", "account_name": "J Smith"},
}
APPLIED: dict[str, str] = {}
AUDIT: list[dict[str, Any]] = []
CRASH_AFTER_WRITE = {"armed": False}  # lets the demo simulate a crash between the write and the checkpoint


def audit(event: str, **detail: Any) -> None:
    AUDIT.append({"ts": datetime.now(timezone.utc).isoformat(), "event": event, **detail})


def prepare(state: State) -> State:
    # In production an extraction step fills `proposal` from the client's email, using a
    # tool-less reader (see the prompt injection playbook). Here the proposal arrives ready.
    version = int(CLIENTS[state["proposal"]["client_id"]]["version"])
    audit("prepared", client_id=state["proposal"]["client_id"], record_version=version)
    return {"record_version": version}


def approve(state: State) -> Command[Literal["apply", "record"]]:
    current = CLIENTS[state["proposal"]["client_id"]]
    decision = interrupt(
        {"action": "change_bank_details", "proposal": state["proposal"],
         "current_account_ending": str(current["account_number"])[-4:], "record_version": state["record_version"]},
        response_schema=Decision,
    )
    audit("decided", approver=decision.approver_id, decision=decision.decision, callback=decision.callback_verified)
    if decision.approver_id == state["requested_by"]:
        return Command(goto="record", update={"decision": decision.model_dump(), "outcome": "rejected: four-eyes rule"})
    if decision.decision == "reject" or not decision.callback_verified:
        return Command(goto="record", update={"decision": decision.model_dump(), "outcome": "rejected"})
    return Command(goto="apply", update={"decision": decision.model_dump()})


def apply(state: State) -> Command[Literal["prepare", "record"]]:
    proposal = state["proposal"]
    key = hashlib.sha256(json.dumps({"p": proposal, "v": state["record_version"]}, sort_keys=True).encode()).hexdigest()
    if key in APPLIED:  # this node can run more than once: after a crash, a retry or a resume
        return Command(goto="record", update={"outcome": "applied"})
    record = CLIENTS[proposal["client_id"]]
    if record["version"] != state["record_version"]:  # the record changed while the approval waited
        audit("stale", client_id=proposal["client_id"], approved_version=state["record_version"], now=record["version"])
        return Command(goto="prepare", update={"outcome": "stale: sent back for a fresh approval"})
    record.update({k: proposal[k] for k in ("sort_code", "account_number", "account_name")})
    record["version"] += 1
    APPLIED[key] = datetime.now(timezone.utc).isoformat()
    if CRASH_AFTER_WRITE["armed"]:
        CRASH_AFTER_WRITE["armed"] = False
        raise RuntimeError("simulated crash after the write, before the checkpoint")
    return Command(goto="record", update={"outcome": "applied"})


def record(state: State) -> State:
    audit("closed", client_id=state["proposal"]["client_id"], outcome=state.get("outcome"),
          approver=state.get("decision", {}).get("approver_id"))
    return {}


def build(checkpointer: BaseCheckpointSaver[Any]) -> CompiledStateGraph[State, None, State, State]:
    g = StateGraph(State)
    g.add_node("prepare", prepare)
    g.add_node("approve", approve)
    g.add_node("apply", apply)
    g.add_node("record", record)
    g.add_edge(START, "prepare")
    g.add_edge("prepare", "approve")
    g.add_edge("record", END)
    return g.compile(checkpointer=checkpointer)


if __name__ == "__main__":
    from langchain_core.runnables import RunnableConfig
    from pydantic import ValidationError

    graph = build(SqliteSaver(sqlite3.connect("workflows.db", check_same_thread=False)))
    cfg: RunnableConfig = {"configurable": {"thread_id": "bank-change-CL-204518-0001"}}
    proposal = {"client_id": "CL-204518", "sort_code": "40-11-22", "account_number": "11223344", "account_name": "J Smith"}

    out = graph.invoke({"requested_by": "j.patel", "proposal": proposal}, cfg)
    print("paused:", out["__interrupt__"][0].value["action"])

    try:  # a resume that skips the call-back confirmation is refused by the schema
        graph.invoke(Command(resume={"approver_id": "s.okafor", "decision": "approve"}), cfg)
    except ValidationError as exc:
        print("refused resume:", exc.errors()[0]["loc"])

    CRASH_AFTER_WRITE["armed"] = True
    try:
        graph.invoke(Command(resume={"approver_id": "s.okafor", "decision": "approve", "callback_verified": True}), cfg)
    except RuntimeError as exc:
        print("crashed:", exc)
    out = graph.invoke(None, cfg)  # resume from the last checkpoint: apply runs again
    print("outcome:", out["outcome"], "| version:", CLIENTS["CL-204518"]["version"], "| writes:", len(APPLIED))

"""Approvals API: the only route by which a paused workflow resumes.

Python 3.11+, fastapi 0.143.0, langgraph 1.2.14. Runs behind your single sign-on proxy,
which authenticates the reviewer and sets X-Authenticated-User. Never expose it directly.
"""
from __future__ import annotations

import sqlite3
from typing import Any, Literal

from fastapi import Depends, FastAPI, Header, HTTPException
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import Command
from pydantic import BaseModel

from bank_details_workflow import Decision, build

graph = build(SqliteSaver(sqlite3.connect("workflows.db", check_same_thread=False)))
app = FastAPI(title="Approvals")


class DecisionIn(BaseModel):
    """What the reviewer submits. Their identity comes from the session, never from this body."""
    decision: Literal["approve", "reject"]
    callback_verified: bool
    note: str = ""


def reviewer(x_authenticated_user: str = Header()) -> str:
    return x_authenticated_user


def thread(thread_id: str) -> RunnableConfig:
    return {"configurable": {"thread_id": thread_id}}


@app.get("/approvals/{thread_id}")
def pending(thread_id: str, user: str = Depends(reviewer)) -> dict[str, Any]:
    snapshot = graph.get_state(thread(thread_id))
    if not snapshot.interrupts:
        raise HTTPException(status_code=404, detail="nothing is waiting for approval")
    return {"thread_id": thread_id, "request": snapshot.interrupts[0].value,
            "response_schema": Decision.model_json_schema()}


@app.post("/approvals/{thread_id}")
def decide(thread_id: str, body: DecisionIn, user: str = Depends(reviewer)) -> dict[str, Any]:
    if not graph.get_state(thread(thread_id)).interrupts:
        raise HTTPException(status_code=409, detail="nothing is waiting, or it was already decided")
    out = graph.invoke(Command(resume={"approver_id": user, **body.model_dump()}), thread(thread_id))
    if "__interrupt__" in out:  # for example the record changed and needs a fresh approval
        return {"status": "waiting", "request": out["__interrupt__"][0].value}
    return {"status": out.get("outcome", "unknown")}