LangGraph Human in the Loop with FastAPI

A LangGraph and FastAPI example for business automation: pause for human approval, save the case and return the decision to the right workflow.

10 July 2026 · 5 min read · by the Yoonet team · Updated 13 September 2026

Educational example: the reviewer endpoint below is a non-working placeholder for a service your provider would need to implement. It is not a hitl.ph API. Use our setup kit to prepare a workflow and rehearse locally.

A business process may need to pause for a person without leaving a customer request unfinished or losing the case. This guide shows how LangGraph and FastAPI can be arranged around that checking step.

LangGraph manages the workflow. FastAPI provides the web endpoints that start work and receive the reviewer’s reply. Share the code with your technical provider; it is an architectural example, not a ready-to-deploy service. They need to implement the application functions, access controls and error handling, and agree the actual reviewer connection. The hitl.ph setup kit provides a local planning exercise.

Save the case before the decision

The example pauses before cancelling a customer’s subscription. A person sees the proposed action and its reason. The saved state, or checkpoint, allows the case to remain available while the reviewer responds.

# 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,
)

Your business sets who can approve a cancellation and what information they need. Your provider chooses and configures the storage and framework versions for the application.

Start the work and pass on the question

POST /runs starts the example workflow. When it finds a pending decision, it turns the request into a brief for the reviewer and returns a waiting status to the caller.

# 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()
REVIEWER = "https://reviewer.example.invalid/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['REVIEWER_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"{REVIEWER}/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}

A good brief identifies the customer, what is proposed, the reason and the relevant business rule. Keep the information understandable to the person doing the checking. The LangChain guide covers this handoff in more detail.

Match the reply to the waiting case

A webhook sends the reviewer’s reply back to your system. The receiving code finds the matching case and supplies the decision to LangGraph. A separate status endpoint lets the application show whether the work is still waiting.

# 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,
    }

Before using this with live customer accounts, your provider must verify who sent the reply, prevent duplicate actions and handle failures. Test that a reply to one case cannot approve another.

What this means for the business

Waiting work has a place. A case stays recorded while the person responds. Storage and operation still need to be managed; waiting is not a guarantee of zero cost.
Decisions can be traced. Keep the original request, reviewer, answer and reason together.
Staffing remains a real responsibility. The software can hold a queue, but people need time to work through it. Agree capacity and cover for the volume you expect.
The idea can support other tasks. The same handoff may support phone follow-ups and data checks, with the instructions and returned information adjusted to the work.

Agree what should happen to overdue requests, rejected decisions and cases needing more information. Those business rules are as necessary as the technical connection.

Make this work in your business.

A short conversation about your business is a good place to start. Tell us what you have automated, what still needs checking and how often it happens.

You can email us directly at ai@yoonet.io.

Save the case before the decision

Start the work and pass on the question

Match the reply to the waiting case

What this means for the business

Make this work in your business.

Read next