Inserting Approval Gates Into Your Agents — A Production Design for Human-in-the-Loop With the Claude API

Designing Suspend and Resume — Persist the Message Loop

The core design of HITL is "pause the agent loop midway and resume later." Claude API is stateless by default, so to resume you need to persist an "awaiting-approval snapshot" yourself. This is not optional — if you try to hold state in memory across an approval request, you'll lose it on deploy, on pod restart, or simply when a request lands on a different instance behind the load balancer.

A minimal data model looks like this:

• session_id: Agent session identifier
• messages: The full messages array up to this point (JSON)
• pending_tool_use: The tool_use block awaiting approval (id, name, input)
• status: awaiting_approval / approved / rejected / expired
• created_at / expires_at: Expiration management

The following is the core loop that wires this state machine into the Claude API.

# agent_loop.py
# Purpose: Insert an approval gate just before tool calls, and resume from
# the same state once approval is granted.
# Requires: anthropic>=0.40, Python 3.11+; persistence works with either Redis or Postgres.
 
import json
import os
from dataclasses import dataclass
from typing import Any, Callable
 
from anthropic import Anthropic
 
client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
MODEL = "claude-sonnet-4-5"
 
# Red list: tool names that always require approval
RED_TOOLS = {"send_money", "delete_user", "send_email_to_customer"}
 
@dataclass
class ApprovalRequired(Exception):
    session_id: str
    tool_use_id: str
    tool_name: str
    tool_input: dict
 
def needs_approval(tool_name: str, tool_input: dict) -> bool:
    """Red/Yellow decision. Example: Yellow uses an amount threshold."""
    if tool_name in RED_TOOLS:
        return True
    if tool_name == "refund_payment":
        amount = tool_input.get("amount_jpy", 0)
        return amount >= 10_000  # approve at or above 10,000 JPY
    return False
 
def run_tool(tool_name: str, tool_input: dict) -> str:
    """Real tool execution is swappable by the caller."""
    # In production, call Stripe SDK, DB clients, etc.
    return json.dumps({"ok": True, "tool": tool_name, "input": tool_input})
 
def step_agent(
    session_id: str,
    messages: list,
    tools: list,
    store: "StateStore",
    on_approval_required: Callable[[ApprovalRequired], None],
) -> dict:
    """Advance one turn. Raises ApprovalRequired if approval is needed."""
    resp = client.messages.create(
        model=MODEL,
        max_tokens=1024,
        tools=tools,
        messages=messages,
    )
 
    # Append Claude's response to history
    messages.append({"role": "assistant", "content": resp.content})
 
    if resp.stop_reason == "end_turn":
        store.save(session_id, messages=messages, pending=None, status="completed")
        return {"status": "completed", "text": _final_text(resp)}
 
    tool_uses = [b for b in resp.content if b.type == "tool_use"]
 
    # Decide on every tool_use first; if any needs approval, run nothing and stop the whole turn
    gated = [b for b in tool_uses if needs_approval(b.name, b.input)]
    if gated:
        store.save(
            session_id,
            messages=messages,
            pending={
                "gated_ids": [b.id for b in gated],
                "tool_uses": [
                    {"id": b.id, "name": b.name, "input": b.input} for b in tool_uses
                ],
            },
            status="awaiting_approval",
        )
        exc = ApprovalRequired(
            session_id=session_id,
            tool_use_id=gated[0].id,
            tool_name=gated[0].name,
            tool_input=gated[0].input,
        )
        on_approval_required(exc)
        raise exc
 
    # No approval needed: return a tool_result for every tool_use in this turn
    tool_results = [
        {
            "type": "tool_result",
            "tool_use_id": b.id,
            "content": run_tool(b.name, b.input),
        }
        for b in tool_uses
    ]
    messages.append({"role": "user", "content": tool_results})
    store.save(session_id, messages=messages, pending=None, status="running")
    return {"status": "running"}
 
def _final_text(resp) -> str:
    return "".join(b.text for b in resp.content if b.type == "text")

The key point in this loop is that state is always persisted before raising ApprovalRequired. If you raise first and persist second, a downstream notification failure can leave the session orphaned. Also, when Claude returns multiple tool_use blocks in a single turn, the code classifies all of them before running anything, and stopping the entire turn even if only one needs approval is intentional. The API also expects a tool_result for every tool_use in the turn, so a half-executed turn can't be resumed cleanly. Partial execution risks leaving later tools operating on state the earlier tools assumed — and if approval ends up being rejected, you have partial results that never should have been produced. When in doubt, stop the whole turn and let Claude re-plan on resume; re-planning is cheap compared to cleanup.

Resuming After Approval — Rejection Handling Is Where Designs Differ

Once approval is granted, resume by appending the tool_result blocks for every tool_use in the stopped turn to the saved messages and calling step_agent again. On rejection, return is_error: true with the reason for the call that was waiting, and a short "skipped" error for the calls that were stopped alongside it.

# resume_agent.py
def resume_after_approval(
    session_id: str,
    decision: str,  # "approved" or "rejected"
    note: str,
    store: "StateStore",
    tools: list,
) -> dict:
    state = store.load(session_id)
    if state["status"] != "awaiting_approval":
        raise ValueError(f"session {session_id} is not waiting")
 
    pending = state["pending"]
    messages = state["messages"]
 
    # Return a tool_result for every tool_use in the turn (the API errors if any is missing)
    tool_results = []
    for tu in pending["tool_uses"]:
        if decision == "approved":
            content, is_error = run_tool(tu["name"], tu["input"]), False
        elif tu["id"] in pending["gated_ids"]:
            # Tell Claude why, so it can consider alternative approaches
            content = f"Human reviewer rejected this tool call. Reason: {note}"
            is_error = True
        else:
            content = "Skipped: another tool call in the same turn was rejected."
            is_error = True
        result = {"type": "tool_result", "tool_use_id": tu["id"], "content": content}
        if is_error:
            result["is_error"] = True
        tool_results.append(result)
 
    messages.append({"role": "user", "content": tool_results})
    store.save(session_id, messages=messages, pending=None, status="running")
 
    return step_agent(session_id, messages, tools, store, on_approval_required=lambda e: None)

Returning is_error: true nudges Claude toward "accomplish the goal another way" instead of just failing with an exception — this is the real difference from naive exception handling. That said, vague rejection notes cause Claude to retry the same tool. Keep the note field free-form but provide templates for reviewers ("threshold exceeded," "customer identity not verified," "refund requires manager approval," etc.). Repeat-request rates drop noticeably when reasons are specific. A useful advanced pattern: when the rejection reason indicates the approval was fundamentally impossible (e.g. "customer account is frozen"), include a hint like "do not retry this action" so Claude moves on to a different strategy rather than trying again with slightly different inputs.

See the Claude API LangGraph stateful agent production guide for a more declarative way to write this state machine — it helps when approval flows branch significantly across different escalation paths.

Pin Down the Multi-Call Turn With a Test

Approval-gate bugs tend to show up when one turn contains several calls. If a read-only lookup and a payment tool land in the same turn, a naive loop runs the lookup first and then stops at the payment — leaving a half-finished turn behind.

Here is the test I'd keep next to it. Claude's response is faked, so it runs instantly and never touches the API.

# test_multi_tool_turn.py
# Goal: lock in that a turn containing any gated call executes nothing
# Run: pytest with a dummy ANTHROPIC_API_KEY set (no real API call is made)
from types import SimpleNamespace as NS
 
import agent_loop
 
 
class FakeStore:
    def __init__(self):
        self.saved = []
 
    def save(self, session_id, **kw):
        self.saved.append(kw)
 
 
def fake_response():
    return NS(
        stop_reason="tool_use",
        content=[
            NS(type="tool_use", id="tu_1", name="lookup_order", input={"order_id": "A-100"}),
            NS(type="tool_use", id="tu_2", name="send_money", input={"amount_jpy": 50_000}),
        ],
    )
 
 
def test_nothing_runs_when_any_call_is_gated(monkeypatch):
    executed = []
    monkeypatch.setattr(agent_loop, "run_tool", lambda n, i: executed.append(n) or "{}")
    monkeypatch.setattr(agent_loop.client.messages, "create", lambda **kw: fake_response())
 
    store = FakeStore()
    try:
        agent_loop.step_agent("s1", [], [], store, lambda e: None)
    except agent_loop.ApprovalRequired:
        pass
 
    assert executed == []                       # even the read-only tool did not run
    last = store.saved[-1]
    assert last["status"] == "awaiting_approval"
    assert len(last["pending"]["tool_uses"]) == 2  # both calls are saved, so resume can answer both

It checks two things only: the read-only tool didn't run, and both tool_use blocks were saved for resume. Running it whenever you touch the gate logic is the cheapest insurance I know of.

Wiring the Approval UI and Async Notifications — Slack and Email

How you actually notify a human about a pending approval depends on the operations side. In my experience, Slack works for internal operators, while email plus a dedicated UI fits better for customer-facing flows. Mixing both is common; it's also fine for one organization to settle on a single channel as long as that channel is reliably monitored.

Slack Interactive Messages

Slack Block Kit plus Interactive Components lets you send a message with approve/reject buttons and receive the click as a webhook. Here is the sender side.

# notify_slack.py
# Purpose: Post an approval request to Slack, embedding the session ID into action_id.
import os
import requests
 
SLACK_WEBHOOK_URL = os.environ["SLACK_WEBHOOK_URL"]
 
def send_approval_request(session_id: str, tool_name: str, tool_input: dict) -> None:
    """Post an approval-request message to Slack."""
    text_input = "\n".join(f"• *{k}*: `{v}`" for k, v in tool_input.items())
    payload = {
        "blocks": [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Approval needed* — the agent wants to call `{tool_name}`.",
                },
            },
            {"type": "section", "text": {"type": "mrkdwn", "text": text_input}},
            {
                "type": "actions",
                "elements": [
                    {
                        "type": "button",
                        "text": {"type": "plain_text", "text": "Approve"},
                        "style": "primary",
                        "action_id": f"approve::{session_id}",
                    },
                    {
                        "type": "button",
                        "text": {"type": "plain_text", "text": "Reject"},
                        "style": "danger",
                        "action_id": f"reject::{session_id}",
                    },
                ],
            },
        ]
    }
    r = requests.post(SLACK_WEBHOOK_URL, json=payload, timeout=5)
    r.raise_for_status()

When the button webhook arrives, call resume_after_approval. One gotcha: Slack marks a button as failed if you don't return HTTP 200 within three seconds, so push heavy work to an async queue (Celery, BullMQ, SQS, etc.). Acknowledge immediately and let a background worker handle the actual agent resume — the system will be dramatically more stable. Another subtle thing to watch: if Slack retries the webhook because of a transient network blip, you'll receive the same interaction event twice. We'll talk about that shortly in the double-execution failure pattern, but the preventive design lives in your webhook handler, not in the queue.

Email Approval — Two Clicks With a Signed Token

For external approvers, email with an approval URL is the most straightforward option. Embed an HMAC-signed token in the URL so clicking the link is enough to approve.

# email_approval.py
# Purpose: Pack session ID and decision into a signed token and generate email URLs.
import hmac
import hashlib
import json
import base64
import os
import time
 
SIGNING_KEY = os.environ["APPROVAL_SIGNING_KEY"].encode()
 
def _sign(payload: dict) -> str:
    raw = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode()
    sig = hmac.new(SIGNING_KEY, raw, hashlib.sha256).digest()
    return base64.urlsafe_b64encode(raw + b"." + sig).decode()
 
def build_approval_url(session_id: str, decision: str, ttl_sec: int = 3600) -> str:
    token = _sign(
        {
            "session_id": session_id,
            "decision": decision,
            "exp": int(time.time()) + ttl_sec,
        }
    )
    return f"https://example.com/approve?token={token}"
 
def verify_token(token: str) -> dict:
    try:
        blob = base64.urlsafe_b64decode(token.encode())
        raw, sig = blob.rsplit(b".", 1)
        expected = hmac.new(SIGNING_KEY, raw, hashlib.sha256).digest()
        if not hmac.compare_digest(sig, expected):
            raise ValueError("bad signature")
        data = json.loads(raw)
        if data["exp"] < int(time.time()):
            raise ValueError("token expired")
        return data
    except Exception as e:
        raise ValueError(f"invalid token: {e}") from e

Two non-negotiables here: tokens must include an expiry, and HMAC comparison must use hmac.compare_digest to avoid timing attacks. One-click approval is convenient, but the URL can leak through email forwarding. For the riskiest operations, consider requiring the user to log into an internal system with 2FA before honoring the click. It's also worth building an endpoint that recognizes already-processed tokens and shows a friendly "this request was already handled" message — it prevents a lot of confused follow-up, especially when multiple approvers are on the same distribution list and two click at once. Treat the signed token as a bearer credential: short TTL, single-use, and revocable through session status.

Handling Expiration and Timeouts

Approval requests cannot wait forever. Operators go home at 6 PM, requests span weekends, and notifications get missed. You must plan for all three.

My default structure has three layers:

• Soft timeout (15 minutes): Re-send a reminder if nobody has responded after 15 minutes.
• Hard timeout (4 hours): Auto-reject and tell Claude "this tool call could not be executed due to a timeout."
• Escalation (1 hour): Notify the primary operator's manager.

A background job sweeps periodically:

# timeout_worker.py
# Purpose: Periodically scan pending approvals and handle reminder / escalation / timeout.
import time
 
def sweep(store, slack_sender, resumer):
    now = int(time.time())
    for session in store.list_awaiting_approval():
        age = now - session["created_at"]
        if age > 4 * 3600 and not session.get("expired"):
            store.mark_expired(session["id"])
            resumer.resume(
                session["id"],
                decision="rejected",
                note="Approval request expired after 4 hours without response.",
            )
        elif age > 3600 and not session.get("escalated"):
            slack_sender.escalate(session)
            store.mark_escalated(session["id"])
        elif age > 900 and not session.get("reminded"):
            slack_sender.remind(session)
            store.mark_reminded(session["id"])

Keep the history of timed-out sessions — don't delete them. Being able to explain afterward "why this transaction was never executed" is critical for compliance. Thresholds are tied to business hours, so extending the Hard timeout to "10 AM the next business day" outside working hours reduces failures substantially. If your operation spans time zones, consider attaching each gate to an on-call rotation rather than a fixed business calendar — the rotation already answers the question "who owns this request right now?"

Three Production Failure Patterns to Know

Below are three failure patterns I have either hit myself or seen during reviews. They are worth memorizing because each one produced real incidents in production, not hypothetical edge cases.

Pattern 1: Privilege Escalation — A Different Tool Fires After Approval

You approve payment A, and when the session resumes, Claude decides to "also" run payment B automatically in the next turn. Claude tends to expand context from the conversation flow, so you need a rule that forbids executing any operation other than the approved tool_use_id within that turn.

The fix is simple but must be enforced: every tool_use returned after resume_after_approval appends a tool_result is re-evaluated by needs_approval. Do not build an implicit "we just approved, so the next one is safe" assumption. Practically, keep needs_approval a pure, stateless function, and do not maintain a per-session set of "already approved tools" that gets carried forward. A related discipline: write a test that specifically tries to trick the function — "we just approved send_money at ¥5,000; does the next call to send_money at ¥6,000 still trigger the gate?" If it doesn't, your implementation has quietly become state-dependent.

Pattern 2: Double Execution — The Approval Button Gets Spammed

If a user double-clicks a Slack button, or a network retry delivers the same approval event twice, you can easily run the same tool twice. For payments or charging, that's fatal — a retried approval event is exactly how one transfer becomes two.

The mitigation combines "idempotency tokens" and "state-transition locks." resume_after_approval transitions from status = awaiting_approval to status = running using CAS (compare-and-swap); subsequent requests do nothing and return 409. In Redis use SET NX; in Postgres use UPDATE ... WHERE status = 'awaiting_approval' RETURNING id. On the Slack side, flipping the button UI to a "processing" state immediately prevents accidental re-clicks and improves the user experience noticeably. For email approval, render the result page server-side so a refresh or a second click on the same link naturally short-circuits on the CAS check rather than reaching the tool.

Pattern 3: State Divergence — The External System Times Out After Approval

Approval comes in, the tool executes, the external API times out on your side, but the server actually completed the operation. Classic example: a payment provider takes over 30 seconds, the client retries after a timeout, and you end up double-charging because the first call succeeded on the server.

This is a distributed-systems problem more than a HITL one, but in agent contexts the rule is absolute: every external API call must carry an idempotency key. Use Stripe's Idempotency-Key header if it's available; otherwise derive your own key from session_id + tool_use_id. The Claude API self-healing agent production patterns article goes deeper into retries and idempotency — it pairs well with this one. One more detail worth internalizing: after a timeout, do not auto-resume; surface "this tool call is in an unknown state" to a human and let them decide, because a blind retry can compound the problem rather than resolve it.

Designing Audit Logs That Survive Compliance Review

The biggest side benefit of HITL is that you naturally end up with a complete log of "who approved or rejected what, and when." Making that log usable as a compliance artifact requires a small amount of intentional design.

At minimum, record:

• The approval target's tool name, input parameters, and Claude's reasoning (the full tool_use block)
• The approver's ID and email address (Slack user_id alone is hard to trace later)
• The approver's justification (always include a free-form comment field)
• Decision timestamp (trust the server-side clock)
• Claude's model name and version at the time (so you can correlate behavior changes later)
• The trace ID tied to the request (your connection point to distributed tracing)

Ideally, you write these to an append-only store. In PostgreSQL, add rules that logically forbid deletes and updates; in regulated industries, you may need to persist to WORM (Write Once Read Many) storage.

Mask personally identifiable input data or keep encryption keys behind access control. If the approval log itself contains too much of what it reviews, a log leak becomes a secondary incident. Finally, build a weekly auto-summary (counts, approver distribution, per-tool ratios) from day one — it gives auditors and external reviewers a view they can actually use, and it saves you from scrambling when someone asks for a report months later. A small investment here has outsized value when a compliance audit lands with a two-week deadline.

Observability — Can You See That the Gates Are Working?

After adding approval gates, you need to measure whether they're actually doing their job. A common outcome is that the gate is shipped, nobody is watching the numbers, and six months later you discover it hasn't been firing correctly — perhaps a refactor removed a needs_approval check silently, or the threshold got changed for a load test and never reset.

These are the metrics I always track:

• Gate firing rate: Of all tool calls, what fraction requested approval?
• Approve / reject / timeout distribution: Where do decisions land?
• Time to decision (P50 and P95): How quickly do operators respond?
• Rejection reasons, categorized: Use frequent reasons to justify threshold adjustments or tool redesigns.
• Re-approval requests within the same session: If high, Claude is failing to understand the rejection reasons.

Review these weekly to decide when to revisit Yellow thresholds, what areas can be pushed further toward automation, and whether you need to raise a gate you previously relaxed. Hook an anomaly detector onto the firing rate as a lagging safety net: if that rate suddenly halves from one day to the next, something in your classification logic has probably regressed, even if no code change was intentional.

Start Small — A Three-Step Rollout for Existing Agents

Retrofitting HITL onto a running agent will stall if you try to be perfect from day one. The incremental path I recommend:

Step 1 — Shadow mode: Run the gate logic, but don't actually stop anything. Just log "would this have required approval?" Run for one week to measure real volumes. If you exceed 50 per week, revisit threshold design.

Step 2 — Red-only gate ON: Enable real approval for one or two of your most dangerous tools (payments, outbound emails). Hold sessions during the operator's ramp-up period — budget one to two weeks.

Step 3 — Yellow threshold tuning: Tune amount or count thresholds against real production data. Reviewing one "didn't need approval" and one "should have approved" case per week is enough to sharpen the thresholds quickly.

Going through these three steps surfaces facts like "Shadow mode showed three times the expected approval volume" before they become operational problems. Flipping the gate ON all at once in production tends to create approval fatigue within the first few hours, and the operation grinds to a halt. I've seen teams have to temporarily disable their newly added gates within a day of launch because operators couldn't keep up — a rollback that then took weeks to recover from organizationally, because confidence in the system had been shaken.

If you're on the Agent SDK, middleware fits naturally here. See the Claude Agent SDK production multi-agent system guide for a middleware mechanism that lets you plug the approval gate in cleanly as a before_tool hook. The SDK middleware pattern also makes it easier to share a single gate implementation across several agents rather than reimplementing it per use case.

Your Next Step

List every tool your agent can currently call, and classify each one along the three axes — irreversibility, blast radius, and amount threshold — into Red, Yellow, or Green. That exercise alone will usually surface two or three holes worth closing before you ship. Then start with Shadow mode, and you'll land HITL in operation without burning out your reviewers.