Integration

LangGraph Human in the Loop with FastAPI

A worked example of async human approvals: LangGraph interrupts checkpointed behind FastAPI endpoints, resumed by webhook when the human answers.

5 July 2026 · 11 min read · by the Yoonet team

"LangGraph human in the loop with FastAPI" is one of the most searched phrases around agent approvals, and nearly every answer stops at the same place: the graph pauses, and then a human is assumed — someone watching a terminal, a dashboard nobody has built, a Slack channel nobody owns.

This is the complete version. Three files' worth of code: a LangGraph agent whose risky tool interrupts, a FastAPI app that starts runs and reports their status, and a webhook that resumes the graph when a real reviewer — in this example, a managed specialist reached over the hitl.ph API — sends the decision back. No process ever blocks waiting for a person.

The graph: interrupt at the point of no return

The agent is a stock ReAct loop; the only unusual thing is the tool. The interrupt() sits inside cancel_subscription, so the model can reason, look things up and draft freely — the pause happens only at the moment of commitment. The Postgres checkpointer is what lets a paused run outlive the request that started it.

# graph.py — an agent whose risky tool pauses for approval
import os
from langchain.tools import tool
from langchain_anthropic import ChatAnthropic
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import create_react_agent
from langgraph.types import interrupt

@tool
def cancel_subscription(customer_id: str, reason: str) -> str:
    """Cancel a customer's subscription."""
    decision = interrupt({
        "action": "cancel_subscription",
        "customer_id": customer_id,
        "reason": reason,
        "question": "Approve this cancellation?",
    })
    if decision.get("approved"):
        do_cancel(customer_id)
        return f"Cancelled {customer_id}"
    return f"Not cancelled: {decision.get('reason', 'declined')}"

checkpointer = PostgresSaver.from_conn_string(os.environ["DATABASE_URL"])
agent = create_react_agent(
    ChatAnthropic(model="claude-sonnet-4-6"),
    tools=[cancel_subscription],
    checkpointer=checkpointer,
)

The API: start the run, detect the pause, brief the human

POST /runs kicks off the agent. If the graph parks at an interrupt, we read the interrupt payload from graph state and turn it into a task brief a stranger can act on — action, subject, the agent's reasoning, and a one line question. That brief goes to a reviewer as a decision.own task, and the endpoint returns awaiting_human to the caller immediately.

# app.py — start runs, surface pauses
import uuid
import requests
from fastapi import FastAPI
from langgraph.types import Command
from graph import agent

app = FastAPI()
HITL = "https://api.hitl.ph/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['HITL_API_KEY']}"}

@app.post("/runs")
def start_run(body: dict):
    thread_id = str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}
    agent.invoke({"messages": [("user", body["prompt"])]}, config)

    # Did the graph park itself at an interrupt?
    state = agent.get_state(config)
    if state.tasks and state.tasks[0].interrupts:
        payload = state.tasks[0].interrupts[0].value
        task = requests.post(f"{HITL}/tasks", headers=HEADERS, json={
            "type": "decision.own",
            "brief": (
                f"An automation wants to {payload['action']} for customer "
                f"{payload['customer_id']} because: {payload['reason']}. "
                "Approve or decline with one line of reasoning."
            ),
            "context": {"thread_id": thread_id},
            "return": {"mode": "webhook",
                       "url": "https://your-app.com/hooks/hitl"},
        }).json()
        return {"thread_id": thread_id, "status": "awaiting_human",
                "hitl_task": task["id"]}

    return {"thread_id": thread_id, "status": "complete",
            "result": state.values["messages"][-1].content}

The interrupt payload doubles as the task brief, which is why it pays to write interrupt payloads for humans rather than for logs — the argument made at greater length in the LangChain guide.

The webhook: resume on answer

When the reviewer decides, the structured result lands on /hooks/hitl. The handler finds the thread, wraps the verdict in a Command(resume=...), and the graph continues inside the tool as if interrupt() had returned normally — because it just did. A status endpoint lets your frontend poll the thread while it waits.

# app.py (continued) — the human answered; resume the graph
@app.post("/hooks/hitl")
def hitl_webhook(task: dict):
    thread_id = task["context"]["thread_id"]
    config = {"configurable": {"thread_id": thread_id}}

    agent.invoke(
        Command(resume={
            "approved": task["result"]["decision"] == "approve",
            "reason": task["result"].get("reasoning", ""),
        }),
        config,
    )
    return {"ok": True}

@app.get("/runs/{thread_id}")
def run_status(thread_id: str):
    state = agent.get_state({"configurable": {"thread_id": thread_id}})
    waiting = bool(state.tasks and state.tasks[0].interrupts)
    return {
        "status": "awaiting_human" if waiting else "complete",
        "last_message": state.values["messages"][-1].content if state.values else None,
    }

Why this shape holds up

Nothing blocks. The run is a checkpoint row while it waits; FastAPI workers serve other requests. Human latency — minutes to hours — costs nothing.
Every decision is owned. The webhook payload carries the reviewer's verdict and reasoning, which lands in the graph transcript. That is your audit trail, made automatically.
The reviewer scales independently. Ten pending approvals or a thousand, the code is identical — capacity is the endpoint's problem, not your rota's. When the volume is small you could point the same webhook at a Slack approver instead; the graph neither knows nor cares.
It generalises past approvals. Swap decision.own for voice.call or data.validate and the same park-and-resume shape gives your agent phone calls and judgement calls — the full task type list maps what a person can return.

Harden it with the usual production trims — verify the webhook source, timebox parked threads with an auto decline, idempotency keys on the resume — and this is not a demo; it is the whole feature.

The graph: interrupt at the point of no return

The API: start the run, detect the pause, brief the human

The webhook: resume on answer

Why this shape holds up

Read next