The safest AI agent is one that has to ask before it does anything important. The most annoying one asks about everything.
Both extremes fail in the same way. An agent that never asks can do real damage when it's wrong. An agent that asks constantly trains people to click "yes" without reading, which is no protection at all.
There's a second problem for automations that run on a schedule: nobody is sitting at the keyboard when the question comes up.
This guide covers the main ways to put a person in the loop, and helps you choose. Then it builds a small approval queue in Python that works for unattended agents. It's free to run and uses only what Python already includes.
Deciding which steps should stay with a person is the real design work, and I wrote a checklist for it in When Not to Automate: 6 Signs a Task Isn't Worth It.
Six ways to keep a person in charge
| Pattern | How it works | Good for | Watch out for |
|---|---|---|---|
| Ask first | The agent pauses and asks, and a person answers right away | Interactive tools, where someone is watching | Nobody is there when it runs on a schedule |
| Approval queue | The agent files a request and carries on. A person decides later, and only then does it happen | Unattended agents, actions that matter | Requests pile up if nobody checks |
| Draft and send | The agent prepares the thing, and a person does the final step | Emails, posts, anything public | Slower, but the safest |
| Batch review | The agent collects proposals, and a person reviews them together once a day | Many small, similar actions | Stale by the time they're approved |
| Veto window | The agent will do it in 30 minutes unless a person stops it | Low-risk, reversible actions | Silence means yes, so use it sparingly |
| Limits and after-the-fact review | The agent acts within hard limits, and a person reads the log later | Low-risk, high-volume work | Mistakes are found after the fact |
One of these shows up in my own products. My Repurpose Bot is a draft-and-send tool: it writes an X thread, a LinkedIn post and a newsletter blurb for each new article, then sends the drafts to Telegram, Discord or Slack. It doesn't publish anything itself, so a person always has the last word. You can see how it works on the Repurpose Bot page.
A bot that posts under your own name is a good test case for the veto window. If an AI writes posts on a schedule, as in my Bluesky automation bot guide, a short delay before posting, with an easy way to cancel, is a cheap safety net.
The rest of this guide builds the approval queue, because it's the pattern that fits most agents that run without anyone watching.
What makes an approval worth anything
A person clicking "approve" only protects you if they understood what they approved. A few rules make that likely:
- Show the real action. Write the summary with code, using the actual arguments, such as the real recipient and the real amount. Don't show a description the model wrote, because a fooled model can write a reassuring one.
- Label the assistant's own explanation. It can be included, but marked as unverified.
- Make the default "no". A request that nobody approves does nothing.
- Let requests expire. A decision about yesterday's situation shouldn't authorize today's action.
- Run it exactly once. An approval is one permission, and not a standing one.
- Record who decided, and when.
- Don't let the agent approve itself. The approving step must be out of the agent's reach.
- Ask rarely. The fewer the questions, the more carefully each one is read.
The queue below follows all eight.
How it works
The flow has four steps:
- The agent proposes an action. The request is stored, and a person is told. Nothing runs.
- A person looks at it and approves or rejects it.
- A runner executes approved requests, once each.
- Anything not approved in time expires.
The agent can only propose actions from a short list, so a request for anything else is refused before a person is bothered.
What you need
- Python 3 on a computer or server that stays on. The queue lives in a small database file, so this is for a machine you control, not for GitHub Actions, where files disappear between runs
- Optionally, a Discord webhook for notifications. Without one, notifications are printed to the screen
Step 1: List the actions
Create actions.py. It holds the only things the agent can ask for, and how each one is described to a person.
import json SENT, REFUNDED = [], [] # stand-ins: a real version would call your email service and your shop def send_email(to, subject, body): SENT.append((to, subject, body)) return "sent" def refund_order(order_number, amount): REFUNDED.append((order_number, amount)) return f"refunded {amount:.2f}" ACTIONS = {"send_email": send_email, "refund_order": refund_order} # The text a person sees is written by this code from the real arguments, never by the model. SUMMARIES = { "send_email": lambda a: f"Send an email to {a['to']}\nSubject: {a['subject']}\n---\n{a['body'][:600]}" + ("\n[message continues]" if len(a["body"]) > 600 else ""), "refund_order": lambda a: f"Refund {a['amount']:.2f} on order {a['order_number']}", } def summarize(tool, args): return SUMMARIES.get(tool, lambda a: json.dumps(a, indent=2))(args) A real version would call your email service and your shop system. These stand-ins just record what happened, so you can test safely.
Look at SUMMARIES. The text a person reads is built from the real arguments, such as a['to'] and a['amount']. A long email body is cut at 600 characters, and says so.
Step 2: Build the queue
Create approvals.py:
"""A small approval queue. The agent proposes, a person decides, and only then does anything happen.""" import argparse import getpass import hashlib import inspect import json import os import secrets import sqlite3 import sys import time import urllib.request from datetime import datetime, timezone from actions import ACTIONS, summarize DB = os.environ.get("APPROVALS_DB", "approvals.db") WEBHOOK = os.environ.get("DISCORD_WEBHOOK", "") MAX_PENDING = 20 # a flood of requests is a warning sign in itself ASK_FOR = 24 * 3600 # seconds a request waits for a decision RUN_WITHIN = 3600 # seconds an approved request stays valid class QueueFull(Exception): pass def connect(): conn = sqlite3.connect(DB, timeout=30) conn.execute("""CREATE TABLE IF NOT EXISTS requests (id TEXT PRIMARY KEY, tool TEXT, args TEXT, reason TEXT, status TEXT, created REAL, expires REAL, decided_by TEXT, decided_at REAL, approved_digest TEXT, result TEXT)""") conn.execute("CREATE TABLE IF NOT EXISTS events (at REAL, request_id TEXT, event TEXT, who TEXT)") return conn def log_event(conn, request_id, event, who): conn.execute("INSERT INTO events VALUES (?, ?, ?, ?)", (time.time(), request_id, event, who)) def digest_of(tool, args_json): return hashlib.sha256(f"{tool}\n{args_json}".encode()).hexdigest() def expire_old(conn): old = conn.execute("SELECT id FROM requests WHERE status IN ('pending', 'approved') AND expires <= ?", (time.time(),)).fetchall() for (request_id,) in old: conn.execute("UPDATE requests SET status = 'expired' WHERE id = ?", (request_id,)) log_event(conn, request_id, "expired", "system") def notify(text): if not WEBHOOK: print(text) return body = json.dumps({"content": text[:1900], "allowed_mentions": {"parse": []}}).encode() # never ping anyone request = urllib.request.Request(WEBHOOK, data=body, headers={"Content-Type": "application/json", "User-Agent": "approval-queue/1.0"}) try: urllib.request.urlopen(request, timeout=15).close() except Exception as error: # the request is safely stored, so a failed notification isn't fatal print(f"Warning: couldn't send the notification ({error})", file=sys.stderr) def propose(tool, args, reason="", ttl=ASK_FOR): """Called by the agent. Stores the request and tells a person. Nothing is executed.""" if tool not in ACTIONS: raise ValueError(f"unknown action {tool!r}") inspect.signature(ACTIONS[tool]).bind(**args) # reject malformed requests before bothering anyone conn = connect() with conn: expire_old(conn) with conn: if conn.execute("SELECT COUNT(*) FROM requests WHERE status = 'pending'").fetchone()[0] >= MAX_PENDING: raise QueueFull(f"{MAX_PENDING} requests are already waiting for a decision") request_id, now = secrets.token_hex(4), time.time() conn.execute("INSERT INTO requests (id, tool, args, reason, status, created, expires) VALUES (?, ?, ?, ?, 'pending', ?, ?)", (request_id, tool, json.dumps(args, sort_keys=True), reason, now, now + ttl)) log_event(conn, request_id, "proposed", "agent") expires = datetime.fromtimestamp(now + ttl, timezone.utc).strftime("%Y-%m-%d %H:%M UTC") notify(f"Approval needed (request {request_id})\n{summarize(tool, args)}\n\n" f"Assistant's note (not verified): {reason or 'none'}\nExpires: {expires}\n" f"Approve: python3 approvals.py approve {request_id}\nReject: python3 approvals.py reject {request_id}") return request_id def decide(request_id, approve): """Called by a person. Only a pending, unexpired request can be decided.""" who, conn = getpass.getuser(), connect() with conn: expire_old(conn) # saved first, so an error below can't undo it with conn: row = conn.execute("SELECT tool, args, status FROM requests WHERE id = ?", (request_id,)).fetchone() if not row: raise ValueError(f"no request called {request_id}") tool, args_json, status = row if status != "pending": raise ValueError(f"request {request_id} is {status}, so it can't be decided now") now = time.time() if approve: # remember exactly what was approved, and give it a short window to run conn.execute("UPDATE requests SET status='approved', decided_by=?, decided_at=?, expires=?, approved_digest=? WHERE id=?", (who, now, now + RUN_WITHIN, digest_of(tool, args_json), request_id)) else: conn.execute("UPDATE requests SET status='rejected', decided_by=?, decided_at=? WHERE id=?", (who, now, request_id)) log_event(conn, request_id, "approved" if approve else "rejected", who) def run_approved(): """Run each approved request exactly once. Returns the ids that were run.""" conn, ran = connect(), [] with conn: expire_old(conn) for (request_id,) in conn.execute("SELECT id FROM requests WHERE status = 'approved' ORDER BY decided_at").fetchall(): with conn: claimed = conn.execute("UPDATE requests SET status = 'running' WHERE id = ? AND status = 'approved'", (request_id,)).rowcount if claimed != 1: continue # another process got there first tool, args_json, approved = conn.execute("SELECT tool, args, approved_digest FROM requests WHERE id = ?", (request_id,)).fetchone() try: if digest_of(tool, args_json) != approved: raise RuntimeError("the request changed after it was approved") status, result = "done", json.dumps(ACTIONS[tool](**json.loads(args_json))) except Exception as error: # no automatic retry: a person should look at it status, result = "failed", str(error) with conn: conn.execute("UPDATE requests SET status = ?, result = ? WHERE id = ?", (status, result, request_id)) log_event(conn, request_id, status, "system") ran.append(request_id) return ran def show(request_id): conn = connect() row = conn.execute("SELECT tool, args, reason, status, expires FROM requests WHERE id = ?", (request_id,)).fetchone() if not row: sys.exit(f"No request called {request_id}") tool, args_json, reason, status, expires = row print(f"Request {request_id} ({status})\n{summarize(tool, json.loads(args_json))}\n\nAssistant's note (not verified): {reason or 'none'}") for at, event, who in conn.execute("SELECT at, event, who FROM events WHERE request_id = ? ORDER BY at", (request_id,)): print(f" {datetime.fromtimestamp(at, timezone.utc):%Y-%m-%d %H:%M:%S} UTC {event} by {who}") def main(): parser = argparse.ArgumentParser(description=__doc__) commands = parser.add_subparsers(dest="command", required=True) commands.add_parser("list", help="show requests waiting for a decision") commands.add_parser("run", help="run the requests that have been approved") for name in ("show", "approve", "reject"): command = commands.add_parser(name) command.add_argument("id") if name != "show": command.add_argument("--yes", action="store_true", help="don't ask for confirmation") args = parser.parse_args() if args.command == "list": conn = connect() with conn: expire_old(conn) rows = conn.execute("SELECT id, tool, args FROM requests WHERE status = 'pending' ORDER BY created").fetchall() for request_id, tool, args_json in rows: print(f"{request_id} {summarize(tool, json.loads(args_json)).splitlines()[0]}") print(f"{len(rows)} waiting") elif args.command == "run": print(f"Ran {len(run_approved())} approved request(s)") elif args.command == "show": sh...
