How to Add Human-in-the-Loop Approval to a LangGraph Agent (2026)

This guide is for Python engineers who have a working tool-calling agent and need a person to approve anything with side effects before it runs: restarting a service, sending an email, merging a pull request. By the end you will have a LangGraph agent that pauses with interrupt() before risky tools, stores its state in Postgres so the approval can arrive minutes or days later from a different process, and resumes with approve, edit or reject. I tested every snippet on 20 October 2026 with LangGraph 1.2.16, langgraph-checkpoint-postgres 3.2.0, langchain 1.4.6, psycopg 3.3.6, Python 3.14 and Postgres 16. You need Python 3.10 or newer, Docker (or any Postgres you can reach), and an API key for a model that supports tool calling. If you are still choosing a framework, my AI agent frameworks comparison explains why LangGraph is the one I use for approval gates.

How human-in-the-loop works in LangGraph

Three pieces do all the work. A checkpointer saves graph state after every step, keyed by a thread_id you supply in the config. The interrupt(value) function, called inside a node, raises a special exception that stops the run and hands value back to the caller under __interrupt__. Resuming with Command(resume=...) on the same thread_id loads the checkpoint, re-runs the interrupted node from its first line, and this time interrupt() returns your resume value instead of raising.

That "re-runs from its first line" detail drives most of the design decisions below. The official interrupts documentation lists the rules: do not wrap interrupt() in a bare try/except, keep interrupt calls in a deterministic order, make anything before the interrupt idempotent, and only pass JSON-serialisable payloads.

Step 1: Install LangGraph and the Postgres checkpointer

Create a virtual environment and pin everything. LangGraph ships frequently (1.2.14, 1.2.15 and 1.2.16 all came out in October), so an unpinned install will drift under you.

python3 -m venv .venv && source .venv/bin/activate
pip install \
  "langgraph==1.2.16" \
  "langgraph-checkpoint-postgres==3.2.0" \
  "psycopg[binary,pool]==3.3.6" \
  "langchain==1.4.6" \
  "langchain-openai==1.7.2"
pip list | grep -E "langgraph|psycopg"

Expected output (trimmed):

langgraph                     1.2.16
langgraph-checkpoint          4.3.0
langgraph-checkpoint-postgres 3.2.0
langgraph-prebuilt            1.1.1
psycopg                       3.3.6

langchain-openai is only there for the model; swap in langchain-anthropic or any other provider package that supports tool calling.

Step 2: Start Postgres for checkpoints

Any Postgres works. For local development, a container is enough:

docker run -d --name lg-postgres \
  -e POSTGRES_PASSWORD=devpass -e POSTGRES_DB=agents \
  -p 5432:5432 postgres:16

export DB_URI="postgresql://postgres:devpass@localhost:5432/agents?sslmode=disable"
export OPENAI_API_KEY="..."   # or the key for whichever provider you installed

The checkpointer creates four tables the first time you call setup(): checkpoints, checkpoint_blobs, checkpoint_writes and checkpoint_migrations. I confirmed that with \dt after the first run.

Step 3: Define the tools, state and approval node

Save this as agent.py. It uses MessagesState (a state with one messages list and the add_messages reducer), two tools, and a human_review node that interrupts only when a risky tool is requested.

# agent.py: LangGraph 1.2 agent with a human approval gate before risky tools
from typing import Literal

from langchain_core.messages import AIMessage, ToolMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode
from langgraph.types import Command, interrupt


@tool
def get_service_status(service: str) -> str:
    """Read-only: return the deployment status of a service."""
    return f"{service}: 3/3 replicas ready, image v1.4.2"


@tool
def restart_service(service: str, environment: str) -> str:
    """Restart a service. Side effect, needs human approval."""
    # Real code would call your deploy API here.
    return f"restarted {service} in {environment}"


TOOLS = [get_service_status, restart_service]
RISKY = {"restart_service"}  # tools that must pass the human gate


def build_graph(model, checkpointer):
    llm = model.bind_tools(TOOLS)

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

    def route_after_agent(state: MessagesState) -> Literal["human_review", "tools", "__end__"]:
        last = state["messages"][-1]
        if not getattr(last, "tool_calls", None):
            return END
        if any(tc["name"] in RISKY for tc in last.tool_calls):
            return "human_review"
        return "tools"

    def human_review(state: MessagesState) -> Command[Literal["tools", "agent"]]:
        last: AIMessage = state["messages"][-1]
        # Nothing with side effects above this line: the node re-runs on resume.
        decision = interrupt({
            "question": "Approve these tool calls?",
            "tool_calls": [
                {"id": tc["id"], "name": tc["name"], "args": tc["args"]}
                for tc in last.tool_calls
            ],
        })
        action = decision.get("action")
        if action == "approve":
            return Command(goto="tools")
        if action == "edit":
            # Same message id, so add_messages replaces the AIMessage instead of appending.
            new_calls = [
                {**tc, "args": decision["args"]} if tc["name"] in RISKY else tc
                for tc in last.tool_calls
            ]
            edited = AIMessage(content=last.content, tool_calls=new_calls, id=last.id)
            return Command(goto="tools", update={"messages": [edited]})
        # Reject: answer every tool call so the history stays valid, then let the model react.
        reason = decision.get("reason", "rejected by reviewer")
        return Command(
            goto="agent",
            update={"messages": [
                ToolMessage(content=f"Not executed: {reason}", tool_call_id=tc["id"])
                for tc in last.tool_calls
            ]},
        )

    builder = StateGraph(MessagesState)
    builder.add_node("agent", agent)
    builder.add_node("human_review", human_review)
    builder.add_node("tools", ToolNode(TOOLS))
    builder.add_edge(START, "agent")
    builder.add_conditional_edges("agent", route_after_agent)
    builder.add_edge("tools", "agent")
    return builder.compile(checkpointer=checkpointer)

Three details matter here. The Command[Literal["tools", "agent"]] return annotation tells LangGraph which edges human_review can take, so you do not add static edges from it. The edit branch reuses the original message id, which makes the add_messages reducer replace the tool call rather than append a second one. The reject branch adds a ToolMessage for every pending tool call, because most providers refuse a history where an assistant tool call has no matching tool result.

Step 4: Persist the graph with PostgresSaver

Save this as run.py. It is a small CLI: start sends a request and stops at the interrupt; approve, edit or reject resumes the same thread, potentially from a different process hours later.

# run.py: start or resume a thread. Usage: python run.py start|approve|edit|reject
import os
import sys

from langchain.chat_models import init_chat_model
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.types import Command

from agent import build_graph

DB_URI = os.environ["DB_URI"]
MODEL = os.environ.get("MODEL", "openai:gpt-5-mini")  # any tool-calling model
config = {"configurable": {"thread_id": "ticket-4812"}}  # one thread per request

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # idempotent: creates tables and runs migrations
    graph = build_graph(init_chat_model(MODEL), checkpointer)

    mode = sys.argv[1]
    if mode == "start":
        result = graph.invoke(
            {"messages": [{"role": "user", "content": "Restart checkout in prod"}]},
            config,
        )
        print(result.get("__interrupt__"))
        print("next:", graph.get_state(config).next)
    else:
        decision = {
            "approve": {"action": "approve"},
            "edit": {"action": "edit", "args": {"service": "checkout", "environment": "staging"}},
            "reject": {"action": "reject", "reason": "change freeze until Monday"},
        }[mode]
        result = graph.invoke(Command(resume=decision), config)
        print(result["messages"][-1].content)

from_conn_string opens a psycopg connection with autocommit=True, prepare_threshold=0 and row_factory=dict_row, which I checked in the 3.2.0 source. Those settings are not optional: setup() creates indexes concurrently, and Postgres will not do that inside a transaction. For a long-running service, use a pool with the same settings:

from psycopg.rows import dict_row
from psycopg_pool import ConnectionPool
from langgraph.checkpoint.postgres import PostgresSaver

pool = ConnectionPool(
    DB_URI,
    max_size=10,
    kwargs={"autocommit": True, "prepare_threshold": 0, "row_factory": dict_row},
)
checkpointer = PostgresSaver(pool)
checkpointer.setup()

For async apps, from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver has the same interface with await checkpointer.setup() and graph.ainvoke(...).

Step 5: Run, interrupt and resume

Start the thread:

python run.py start

This is the output from my test run, where I replaced the model with a stub that always requests restart_service, so the shape is exact:

[Interrupt(value={'question': 'Approve these tool calls?', 'tool_calls': [{'id': 'call_1', 'name': 'restart_service', 'args': {'service': 'checkout', 'environment': 'prod'}}]}, id='a5f76ef21b23e2e8042e10921ebe6946', response_schema=None)]
next: ('human_review',)

The process can exit now. The pending interrupt lives in Postgres, and get_state(config).next tells any other process that the thread is waiting on human_review. Resume it from a fresh process:

python run.py edit

Done: restarted checkout in staging

Running start then reject made the stub echo Done: Not executed: followed by the reviewer's reason, and approve produced Done: restarted checkout in prod. With a real model the final message is the model's own summary of the tool result. To see what happened on a thread, walk its history:

for snap in graph.get_state_history(config):
    print(snap.config["configurable"]["checkpoint_id"], snap.next)

Step 6: Stream updates to a UI

For a chat UI you want per-node updates rather than one final dict. With stream_mode="updates", the interrupt arrives as its own chunk under the __interrupt__ key:

for chunk in graph.stream(
    {"messages": [{"role": "user", "content": "Restart checkout in prod"}]},
    config,
    stream_mode="updates",
):
    if "__interrupt__" in chunk:
        payload = chunk["__interrupt__"][0].value
        print("WAITING FOR APPROVAL:", payload["tool_calls"])
    else:
        print("node finished:", list(chunk))

# later, after the reviewer clicks a button:
for chunk in graph.stream(Command(resume={"action": "approve"}), config, stream_mode="updates"):
    print("node finished:", list(chunk))

In my run the first loop printed agent then the interrupt; the resume printed human_review, tools, agent. If several parallel branches interrupt at once, each Interrupt has an id, and you resume with a dict mapping those ids to values, as shown in the interrupts docs.

Step 7: Deploy the agent

You have two realistic options.

Self-host it. Wrap run.py logic in your own FastAPI (or similar) service: one endpoint starts a thread and returns the interrupt payload, another accepts the decision and calls graph.invoke(Command(resume=...), config). You own Postgres, auth, and the queue of pending approvals. Use the connection pool above, and generate a thread_id per request (a ticket or job id works well).

Use the LangGraph server. The CLI (langgraph-cli 0.4.37) has dev, up, build and a beta deploy command targeting LangSmith Deployment. Describe the graph in langgraph.json:

{
  "dependencies": ["."],
  "graphs": {
    "ops_agent": "./server_graph.py:graph"
  },
  "env": ".env"
}

pip install "langgraph-cli[inmem]==0.4.37"
langgraph validate   # prints: Configuration file ... is valid. (1 graph found)
langgraph dev        # local API server with hot reload

The persistence docs say that with the Agent Server you do not configure checkpointers yourself, so server_graph.py should compile the graph without your PostgresSaver (for example graph = build_graph(init_chat_model(MODEL), None)). Clients then resume through the SDK:

from langgraph_sdk import get_client

client = get_client(url="http://localhost:2024")
thread = await client.threads.create()
await client.runs.wait(thread["thread_id"], "ops_agent",
                       input={"messages": [{"role": "user", "content": "Restart checkout in prod"}]})
await client.runs.wait(thread["thread_id"], "ops_agent", command={"resume": {"action": "approve"}})

If you only need approve, edit or reject on tool calls and are happy with a prebuilt agent loop, LangChain 1.x's create_agent with HumanInTheLoopMiddleware implements the same pattern with decision types approve, edit, reject and respond. I prefer the explicit node above because the approval logic sits in code I can read and test.

Troubleshooting

Symptom	Likely cause	Fix
ValueError: Checkpointer requires one or more of the following 'configurable' keys: thread_id, checkpoint_ns, checkpoint_id	Invoked a graph with a checkpointer but no config	Pass {"configurable": {"thread_id": "..."}} on every invoke and stream call
ActiveSqlTransaction: CREATE INDEX CONCURRENTLY cannot run inside a transaction block on setup()	Passed your own psycopg connection without autocommit=True	Use from_conn_string, or set autocommit=True, prepare_threshold=0, row_factory=dict_row on the connection or pool
Interrupt returned, but resume starts from scratch	Compiled without a checkpointer (interrupts still surface, state is not saved), or InMemorySaver in a process that restarted	Compile with PostgresSaver and keep the same thread_id
Resume crashes with IndexError or runs the agent on empty state	Wrong thread_id on resume; LangGraph treats it as a new thread	Store the thread_id alongside the pending approval and check get_state(config).next before resuming
Side effect (email, API call) happens twice	Code before interrupt() re-runs on resume	Move side effects after the interrupt or into a separate node, or make them idempotent
Graph never pauses	A broad try/except Exception around interrupt() swallowed it	Catch specific exceptions only, or re-raise
Provider error about a tool call without a tool result after reject	Rejected tool calls left unanswered in the history	Add a ToolMessage with the matching tool_call_id for each pending call
Edited arguments ignored, or two tool calls in history	New AIMessage created without the original id	Reuse id=last.id so add_messages replaces it
Wrong value delivered to the second of two interrupts	Interrupts matched by order and the order changed between runs	Keep interrupt calls deterministic, or resume with a map keyed by Interrupt.id

FAQ

Does LangGraph interrupt() work without a checkpointer?

The interrupt still surfaces: in my test, a graph compiled with no checkpointer returned the __interrupt__ payload. But nothing was saved, so there is nothing to resume. Always compile with a checkpointer and pass a thread_id.

Should I use InMemorySaver or PostgresSaver?

InMemorySaver is for tests and notebooks; it loses every pending approval when the process restarts. Use PostgresSaver (or AsyncPostgresSaver) for anything a human will approve later, because approvals routinely outlive a deploy.

How long can a LangGraph thread wait for approval?

As long as the checkpoint exists in your database. There is no timeout in the interrupt itself, so add your own expiry: store a created-at timestamp with each pending approval and reject stale ones automatically.

What is the difference between interrupt() and interrupt_before?

interrupt() is a function you call inside a node with a payload, and the resume value comes back as its return value, which makes it the documented way to build approval and editing flows. interrupt_before and interrupt_after are static breakpoints set at compile time, mainly useful for debugging.

In production, the first thing I check on any approval flow is what happens on a deploy in the middle of a pending approval: kill the process after start, bring it back, and resume. If that does not work, nothing else matters. Next I check that the approver sees the exact arguments that will run, that every rejection leaves a tool result in the history, and that the tools themselves run with least privilege, because a human click is a control, not a sandbox. The agent policy-as-code post covers the layer beneath it. Official references: the LangGraph interrupts docs, the LangGraph releases, and the deployment overview.

Related: It's 2 AM. Do You Know What Your AI Agent Is Doing?, Policy as Code for AI Agents in 2026: Cedar at the Gateway, Rego in the Hook, and the 11.2% the Model Still Lets Through and AI Agent Frameworks Compared (2026): LangGraph, CrewAI, ADK and 7 More.