LangGraph

┌──────────────────────────────┐  HTTP POST /api/tasks  ┌──────────────────────┐
│ refund_agent.py              │ ──────────────────────►│ awaithumans server   │
│  - graph: triage → review    │                        │                      │
│         → process_refund     │                        │                      │
│  - drive_human_loop(graph,…) │   long-poll status     │ — human reviews ──►  │
│                              │ ──────────────────────►│ — completes task ──► │
│  ◄── interrupt / resume ──── │ ◄──────────────────────│                      │
└──────────────────────────────┘    response payload    └──────────────────────┘

pip install "awaithumans[langgraph]"

from typing import TypedDict
from pydantic import BaseModel
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

from awaithumans.adapters.langgraph import await_human


class State(TypedDict):
    customer_id: str
    amount_usd: int
    approved: bool


class RefundPayload(BaseModel):
    customer_id: str
    amount_usd: int


class RefundDecision(BaseModel):
    approved: bool
    notes: str | None = None


def review_node(state: State) -> dict:
    decision = await_human(
        task=f"Approve ${state['amount_usd']} refund for {state['customer_id']}?",
        payload_schema=RefundPayload,
        payload=RefundPayload(
            customer_id=state["customer_id"],
            amount_usd=state["amount_usd"],
        ),
        response_schema=RefundDecision,
        timeout_seconds=15 * 60,
    )
    return {"approved": decision.approved}


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

import asyncio
import os

from awaithumans.adapters.langgraph import drive_human_loop

# `graph` is the compiled StateGraph from the "Node side" snippet above.


async def main():
    config = {"configurable": {"thread_id": "wf-1"}}
    final_state = await drive_human_loop(
        graph,
        input_state={"customer_id": "cus_demo", "amount_usd": 250, "approved": False},
        config=config,
        server_url="http://localhost:3001",
        api_key=os.environ.get("AWAITHUMANS_ADMIN_API_TOKEN"),
    )
    print(final_state.values)


asyncio.run(main())

# Pretend this is your real money-moving call.
def process_refund(customer_id: str, amount_usd: int) -> str:
    ...  # returns a refund id


# ❌ DON'T do this — process_refund runs on every node re-execution.
def review_node_bad(state):
    decision = await_human(
        task="Approve refund?",
        payload_schema=RefundPayload,
        payload=RefundPayload(
            customer_id=state["customer_id"], amount_usd=state["amount_usd"]
        ),
        response_schema=RefundDecision,
        timeout_seconds=15 * 60,
    )
    refund_id = process_refund(state["customer_id"], state["amount_usd"])
    return {"refund_id": refund_id, "approved": decision.approved}


# ✅ DO this — split human review and the side-effect into two nodes.
def review_node(state):
    decision = await_human(
        task="Approve refund?",
        payload_schema=RefundPayload,
        payload=RefundPayload(
            customer_id=state["customer_id"], amount_usd=state["amount_usd"]
        ),
        response_schema=RefundDecision,
        timeout_seconds=15 * 60,
    )
    return {"approved": decision.approved}


def process_refund_node(state):
    if not state["approved"]:
        return {"refund_id": None}
    return {"refund_id": process_refund(state["customer_id"], state["amount_usd"])}