[Home](/) / [Blog](/blog/) / Integration 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. 10 July 2026 · 11 min read · by the Yoonet team ![Marbled ink flowing in from both edges of the paper and locking together at the centre in a clean prismatic seam](/media/blog-integration.webp) "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](/docs/) — 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](/blog/human-in-the-loop-langchain/). ## 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](/pricing/), 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](/use-cases/) 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. ## Read next - [The Philippines' BPO Industry Is Growing Despite AI](/blog/philippines-bpo-growing-despite-ai/) - [Cloudflare's Agent Stack: Email, Traces, and Now Wallets](/blog/cloudflare-agent-infrastructure/) - [Human in the Loop Patterns for AI Agents](/blog/human-in-the-loop-patterns/) Or start with the product: [the hitl.ph API docs](/docs/), [the five task types](/use-cases/) and [FTE staff](/pricing/). *** Source: https://www.hitl.ph/blog/langgraph-human-in-the-loop-fastapi/ This is the markdown representation of that page, served to clients that send `Accept: text/markdown`. Machine-readable index: https://www.hitl.ph/llms.txt · Knowledge bundle: https://www.hitl.ph/okf/index.md