Build a Human Approval Step into a LangGraph Agent with FastAPI

What you'll build / learn

A customer-support agent that can issue refunds — except it can't, not on its own. Before executing the refund tool, the graph pauses mid-run, fires a webhook notification, and waits. A human approves or rejects the pending call through an HTTP endpoint, and the agent resumes exactly where it stopped. You'll learn LangGraph's interrupt() / Command(resume=...) mechanics and how to drive them from a stateless FastAPI app.

Prerequisites

• Python 3.10+ (tested on 3.13.5), macOS or Linux. On Windows, swap source .venv/bin/activate for .venv\Scripts\activate.
• Versions verified for this tutorial: langgraph 1.2.9, fastapi 0.139.2, langchain 1.3.14, langchain-anthropic 1.4.8, uvicorn 0.51.0.
• An Anthropic API key from the Anthropic Console. Any tool-calling model works; we use Claude Sonnet 5.
• curl for testing.

Step 1: Set up the project

mkdir approval-agent && cd approval-agent
python3 -m venv .venv && source .venv/bin/activate
pip install "langgraph==1.2.9" "fastapi[standard]==0.139.2" \
  "langchain==1.3.14" "langchain-anthropic==1.4.8" httpx
export ANTHROPIC_API_KEY="sk-ant-..."

fastapi[standard] pulls in uvicorn. httpx is for the outbound notification webhook.

Step 2: Build the interrupt-gated agent

Create agent.py:

from typing import Literal

from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langchain_core.messages import ToolMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode
from langgraph.types import Command, interrupt


@tool
def issue_refund(order_id: str, amount: float) -> str:
    """Refund a customer order. Irreversible once executed."""
    return f"Refunded ${amount:.2f} for order {order_id}."


TOOLS = [issue_refund]
model = init_chat_model("anthropic:claude-sonnet-5").bind_tools(TOOLS)


def agent(state: MessagesState):
    return {"messages": [model.invoke(state["messages"])]}


def route_after_agent(state: MessagesState) -> Literal["approval", "__end__"]:
    if state["messages"][-1].tool_calls:
        return "approval"
    return END


def approval(state: MessagesState) -> Command[Literal["tools", "agent"]]:
    call = state["messages"][-1].tool_calls[0]
    decision = interrupt(
        {
            "action": call["name"],
            "args": call["args"],
            "question": "Approve this irreversible action?",
        }
    )
    if decision.get("approved"):
        return Command(goto="tools")
    return Command(
        goto="agent",
        update={
            "messages": [
                ToolMessage(
                    content=f"Rejected by reviewer: {decision.get('reason') or 'no reason given'}. Do not retry.",
                    tool_call_id=call["id"],
                )
            ]
        },
    )


builder = StateGraph(MessagesState)
builder.add_node("agent", agent)
builder.add_node("approval", approval)
builder.add_node("tools", ToolNode(TOOLS))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", route_after_agent)
builder.add_edge("tools", "agent")

graph = builder.compile(checkpointer=InMemorySaver())

Three things here are load-bearing:

interrupt() needs a checkpointer. When the approval node calls interrupt(...), LangGraph raises internally, checkpoints the whole graph state under the run's thread_id, and returns control to whoever called invoke(). The payload you pass — the pending tool call — surfaces to the caller, and the value the human eventually supplies via Command(resume=...) becomes interrupt()'s return value. InMemorySaver is fine for one process; swap in a SQLite or Postgres checkpointer for anything real.

The gate is its own node. On resume, LangGraph re-runs the interrupted node from the top — it doesn't restart at the interrupt() line. Keeping the gate free of side effects means the re-run is harmless. Never put the destructive call before an interrupt in the same node.

Rejection still answers the tool call. Anthropic's API requires a tool_result for every tool_use block, so on rejection we route back to agent with a ToolMessage explaining the denial instead of silently dropping the call. The model then tells the user the refund was declined rather than retrying blind.

Step 3: Expose it through FastAPI

Create app.py:

import os
import uuid

import httpx
from fastapi import FastAPI, HTTPException
from langgraph.types import Command
from pydantic import BaseModel

from agent import graph

app = FastAPI(title="Approval-gated agent")

NOTIFY_URL = os.environ.get("APPROVAL_WEBHOOK_URL")


class RunRequest(BaseModel):
    message: str


class ApprovalRequest(BaseModel):
    approved: bool
    reason: str | None = None


def notify(thread_id: str, pending: dict) -> None:
    body = {"thread_id": thread_id, "pending_action": pending}
    if NOTIFY_URL:
        httpx.post(NOTIFY_URL, json=body, timeout=10)
    else:
        print(f"[APPROVAL NEEDED] {body}")


def run_graph(thread_id: str, graph_input) -> dict:
    config = {"configurable": {"thread_id": thread_id}}
    result = graph.invoke(graph_input, config)
    if "__interrupt__" in result:
        pending = result["__interrupt__"][0].value
        notify(thread_id, pending)
        return {
            "status": "pending_approval",
            "thread_id": thread_id,
            "pending_action": pending,
        }
    return {"status": "done", "reply": result["messages"][-1].content}


@app.post("/runs")
def start_run(req: RunRequest):
    thread_id = str(uuid.uuid4())
    return run_graph(
        thread_id, {"messages": [{"role": "user", "content": req.message}]}
    )


@app.post("/approvals/{thread_id}")
def resolve_approval(thread_id: str, req: ApprovalRequest):
    config = {"configurable": {"thread_id": thread_id}}
    if not graph.get_state(config).interrupts:
        raise HTTPException(404, "No paused run found for this thread_id")
    return run_graph(thread_id, Command(resume=req.model_dump()))

graph.invoke() returns as soon as the interrupt fires, with the pending payload under result["__interrupt__"]. The thread_id is the resume cursor — it's the only token the approver needs to send back. Both endpoints share run_graph because a resumed run can interrupt again if the model chains a second sensitive call; the approver just gets another pending_approval response.

The get_state().interrupts guard matters: resuming a thread that has no pending interrupt doesn't raise — LangGraph silently starts a fresh run on that thread, which is not what an approval click should ever do.

Set APPROVAL_WEBHOOK_URL to a Slack incoming webhook or a webhook.site URL to get real notifications; unset, it logs to the console.

sequenceDiagram
    participant C as Client
    participant A as FastAPI
    participant G as LangGraph
    participant R as Reviewer
    C->>A: POST /runs
    A->>G: invoke(messages)
    G-->>A: __interrupt__ (pending tool call)
    A->>R: webhook notification
    A-->>C: pending_approval + thread_id
    R->>A: POST /approvals/{thread_id}
    A->>G: invoke(Command(resume=decision))
    G-->>A: final state
    A-->>R: done + reply

Step 4: Run the server

uvicorn app:app --port 8000

You should see Uvicorn running on http://127.0.0.1:8000. Skip --reload here — every reload wipes InMemorySaver, killing pending approvals.

Verify it works

Start a run that triggers the refund tool:

curl -s -X POST http://127.0.0.1:8000/runs \
  -H 'Content-Type: application/json' \
  -d '{"message": "Refund order A-1001 for $42.50"}'

Expected response (thread_id and args come from your run — the shape is fixed):

{
  "status": "pending_approval",
  "thread_id": "dfc7025b-7acf-404b-8b19-18bf98c6a8a9",
  "pending_action": {
    "action": "issue_refund",
    "args": {"order_id": "A-1001", "amount": 42.5},
    "question": "Approve this irreversible action?"
  }
}

The server console (or your webhook receiver) shows the notification: [APPROVAL NEEDED] {'thread_id': '...', 'pending_action': {...}}. Now approve it, using the thread_id you got back:

curl -s -X POST http://127.0.0.1:8000/approvals/<thread_id> \
  -H 'Content-Type: application/json' \
  -d '{"approved": true}'

{"status": "done", "reply": "I've issued the refund of $42.50 for order A-1001."}

The exact wording is the model's, but the tool genuinely ran — Refunded $42.50 for order A-1001. is in the thread's message history. Run it again with {"approved": false, "reason": "wrong order"} on a fresh run: you get status: done with the model explaining the refund was declined, and no refund executes. An approval POST to a bogus thread returns 404 {"detail": "No paused run found for this thread_id"}.

Troubleshooting

TypeError: "Could not resolve authentication method. Expected one of api_key, auth_token, or credentials to be set..." — raised on the first model call, not at startup, so the server boots fine and then 500s on /runs. ANTHROPIC_API_KEY isn't set in the shell running uvicorn; export it in that same shell.

RuntimeError: Cannot use Command(resume=...) without checkpointer — you compiled with builder.compile() and dropped the checkpointer=InMemorySaver() argument. Without a checkpointer the interrupt still surfaces, but there's no saved state to resume from.

ValueError: Checkpointer requires one or more of the following 'configurable' keys: thread_id, checkpoint_ns, checkpoint_id — you invoked the graph without config={"configurable": {"thread_id": ...}}. Every run against a checkpointed graph needs a thread id.

Approvals 404 after a code change — with --reload, each file save restarts the worker and empties InMemorySaver, so every pending run vanishes. Use a durable checkpointer during development, or finish pending approvals before saving.

Next steps

Swap InMemorySaver for SqliteSaver (pip install langgraph-checkpoint-sqlite) or the Postgres checkpointer so approvals survive restarts and scale past one worker. Read the interrupts documentation for multi-interrupt resumes, where you map resume values to interrupt IDs. If you'd rather not hand-roll the gate, LangChain's create_agent ships a HumanInTheLoopMiddleware that wraps this same pattern, and the LangGraph Platform's Agent Inbox gives reviewers a UI instead of curl. Before production: authenticate the approval endpoint (anyone who can POST it can approve anything), and add an expiry/escalation policy for approvals nobody answers.