Human-in-the-loop approval AI agents need a hard stop between autonomous action and side effects. This guide shows how to build a durable approval gateway that pauses an agent, requests human sign-off, and resumes only after cryptographically verified confirmation. We’ll use Python and FastAPI, but the pattern ports to any stack. The core idea is simple: treat human approval as a state transition in a durable machine, not a synchronous blocking call.
Step 1: Define the approval boundary
Not every tool call needs a human. Start by enumerating high-risk operations: payments, data deletion, sending external emails, or any write to production systems. Model each as a typed action so your orchestrator can make decisions uniformly.
from enum import Enum
from pydantic import BaseModel
class RiskTier(str, Enum):
LOW = "low"
HIGH = "high"
class AgentAction(BaseModel):
action_id: str
tool: str
params: dict
risk: RiskTier
run_id: strAny action with risk == HIGH enters the approval queue. Keep the classifier simple: a static map or a small LLM prompt. If you call a model for classification, route through n4n.ai’s OpenAI-compatible endpoint to get automatic fallback when a provider is rate-limited, so the pre-approval step never becomes a single point of failure. The pattern for human-in-the-loop approval AI agents is to separate the risk assessment from the execution path entirely.
Step 2: Persist the pending action
A paused agent must survive process restarts. Write the pending action to a durable store before yielding control. Below is a minimal SQLite schema; in production use Postgres or a managed queue with replication.
import sqlite3
from contextlib import contextmanager
@contextmanager
def get_db():
conn = sqlite3.connect("approvals.db")
conn.execute("""CREATE TABLE IF NOT EXISTS pending (
action_id TEXT PRIMARY KEY,
run_id TEXT,
payload TEXT,
status TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)""")
yield conn
conn.close()
def save_pending(action: AgentAction):
with get_db() as conn:
conn.execute(
"INSERT INTO pending (action_id, run_id, payload, status) VALUES (?,?,?,?)",
(action.action_id, action.run_id, action.json(), "AWAITING")
)
conn.commit()Make the insert idempotent: if the same action_id already exists, skip. The agent coroutine should save_pending then return a sentinel. Do not block on the human; suspend the run and let the orchestrator poll. This decoupling is what keeps human-in-the-loop approval AI agents responsive under load.
Step 3: Notify the reviewer
Push a compact message to a human channel. Include a direct link to the approval UI and a short summary. Use a signed URL so the reviewer’s click carries provenance.
{
"action_id": "act_123",
"tool": "stripe.charge",
"params": {"amount": 5000, "currency": "usd"},
"approve_url": "https://app.example.com/approve?token=eyJhbGciOi...",
"reject_url": "https://app.example.com/reject?token=eyJhbGciOi..."
}Send via your existing alerting path. For a quick test, curl:
curl -X POST https://hooks.slack.com/services/T000/B000/XXXX \
-H 'Content-type: application/json' \
-d '{"text":"Approve agent action act_123: charge $50.00"}'If you have multiple reviewers, fan out to a rotation list. Record which reviewer acknowledged the message; that identity goes into the audit log later.
Step 4: Implement the signed callback
The approval endpoint must verify the reviewer’s intent and prevent replay. Generate an HMAC token per action; check it on callback. Never trust a bare GET from an email link; require a POST with the token in the body.
from fastapi import FastAPI, Request, HTTPException
import hmac, hashlib, os
app = FastAPI()
SECRET = os.environ["APPROVAL_SECRET"]
def sign(action_id: str) -> str:
return hmac.new(SECRET.encode(), action_id.encode(), hashlib.sha256).hexdigest()
@app.post("/approve")
async def approve(request: Request):
body = await request.json()
action_id = body["action_id"]
token = body["token"]
if not hmac.compare_digest(sign(action_id), token):
raise HTTPException(403, "Invalid signature")
with get_db() as conn:
conn.execute("UPDATE pending SET status='APPROVED' WHERE action_id=?", (action_id,))
conn.commit()
return {"status": "ok"}Reject follows the same shape with status REJECTED. Add a nonce column if you need to invalidate a token after first use. For human-in-the-loop approval AI agents, this signature is the only thing standing between an autonomous loop and an impersonated approval.
Step 5: Resume or abort the agent
A worker polls for status changes and reconstructs the action. If approved, re-inject into the agent’s execution context; if rejected, raise a guarded exception.
def poll_decision(action_id: str) -> str:
with get_db() as conn:
row = conn.execute("SELECT status FROM pending WHERE action_id=?", (action_id,)).fetchone()
return row[0] if row else "UNKNOWN"
def resume_agent(action_id: str, run_context: dict):
status = poll_decision(action_id)
if status == "APPROVED":
action = load_action(action_id)
return execute_tool(action, run_context)
elif status == "REJECTED":
raise PermissionError(f"Human rejected {action_id}")
else:
# still awaiting or unknown: keep suspended
return NoneThe orchestrator should call resume_agent on a 30-second tick. The agent run stays inert until then. For human-in-the-loop approval AI agents, the resume step is where guardrails either hold or fail—if you resume on a stale read, you bypass the human.
Step 6: Add timeout and escalation
An approval that sits for hours is worse than an auto-reject. Set a TTL (e.g., 15 minutes). A cron job flips stale AWAITING rows to EXPIRED and notifies the agent to abort.
from datetime import datetime, timedelta
def expire_stale():
cutoff = datetime.utcnow() - timedelta(minutes=15)
with get_db() as conn:
conn.execute(
"UPDATE pending SET status='EXPIRED' WHERE status='AWAITING' AND created_at < ?",
(cutoff.isoformat(),)
)
conn.commit()Escalate to a secondary reviewer if the primary hasn’t acted in 5 minutes. That’s just another notification with a higher severity tag. Track time-to-approval as a metric; if it climbs, your humans are the bottleneck, not the model.
Step 7: Verify success
You need proof the loop works end to end. Write a pytest that mocks the human and exercises the state machine.
def test_approval_flow(tmp_path):
# monkeypatch DB path, save pending, sign, approve, resume
action = AgentAction(action_id="test1", tool="echo", params={"msg":"hi"}, risk="high", run_id="r1")
save_pending(action)
token = sign("test1")
assert poll_decision("test1") == "AWAITING"
# simulate callback internally
with get_db() as conn:
conn.execute("UPDATE pending SET status='APPROVED' WHERE action_id=?", ("test1",))
conn.commit()
assert poll_decision("test1") == "APPROVED"
result = resume_agent("test1", {})
assert result is not NoneRun pytest -q and confirm green. For a live check, deploy the FastAPI app locally, trigger an agent action that hits a HIGH risk tool, and watch the Slack message arrive. Click approve, then confirm the agent proceeds and the DB row shows APPROVED. If you reject, the agent must raise and the run must halt without side effects. Testing human-in-the-loop approval AI agents requires simulating the reviewer as well as the agent.
Operational notes
- Log every state transition with the reviewer’s identity. Audit trails are the point.
- Keep the approval UI read-only on params; show exactly what the agent will do.
- For multi-tenant systems, scope the signing secret per tenant.
- Metric the queue depth; a growing backlog means your humans need help or the agent is too trigger-happy.
Building human-in-the-loop approval AI agents is mostly plumbing: durable storage, signed callbacks, and a poll loop. Get those right and the autonomy stays safe without grinding the system to a halt.