Human-in-the-Loop AI: Build a Simple Approval Queue

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:

1. 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.
2. Label the assistant's own explanation. It can be included, but marked as unverified.
3. Make the default "no". A request that nobody approves does nothing.
4. Let requests expire. A decision about yesterday's situation shouldn't authorize today's action.
5. Run it exactly once. An approval is one permission, and not a standing one.
6. Record who decided, and when.
7. Don't let the agent approve itself. The approving step must be out of the agent's reach.
8. 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:

1. The agent proposes an action. The request is stored, and a person is told. Nothing runs.
2. A person looks at it and approves or rejects it.
3. A runner executes approved requests, once each.
4. 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":
        show(args.id)
    else:
        approve = args.command == "approve"
        if approve:
            show(args.id)
            if not args.yes and input("\nType yes to approve: ").strip().lower() != "yes":
                sys.exit("Not approved.")
        try:
            decide(args.id, approve)
        except ValueError as error:
            sys.exit(str(error))
        print(f"Request {args.id} {'approved' if approve else 'rejected'}")


if __name__ == "__main__":
    main()

It's the longest piece of code in this guide, so here's what the main parts do.

• propose() is the only function the agent calls. It checks that the action is on the list and the arguments fit, stores the request, and sends the notification. Malformed requests are refused before a person ever sees them. If too many requests are waiting, it raises QueueFull, because a flood of requests is a warning sign in itself.
• notify() posts to Discord with allowed_mentions switched off, so text like @everyone in an assistant's note can't ping anyone. If the notification fails, the request is still safely stored.
• decide() is what a person triggers. It only works on a request that's still pending and hasn't expired, and each request can be decided once. Approving also records a fingerprint of exactly what was approved, and starts a one-hour window to run it.
• run_approved() is the runner. It claims each approved request before running it, so two runners at the same moment can't both execute it. Before running, it checks that the request still matches the fingerprint from approval. If something changed it in between, it refuses.
• expire_old() marks requests as expired when time runs out, whether they were waiting for a decision or for the runner.
• The events table is the audit trail: who did what, and when.
• The command line at the bottom lets a person list, show, approve and reject requests, and run the approved ones.

Notifications can be sent to a channel through a webhook. If webhooks are new to you, here's what a webhook is and how it works.

Step 3: Connect an agent

Your agent needs a version of its risky tools that proposes instead of acting. Create agent_demo.py:

from approvals import propose


def queued(tool):
    """Turn a risky action into a request that a person has to approve."""
    def call(reason="", **args):
        request_id = propose(tool, args, reason)
        return f"Queued for a person to approve (request {request_id}). It has NOT happened yet."
    return call


send_email = queued("send_email")
refund_order = queued("refund_order")

# What an agent's tool calls would look like:
print(send_email(reason="The customer asked where their order is",
                 to="[e-mail]", subject="Your order", body="Hi Maria, your order A-1042 has shipped."))
print(refund_order(reason="Customer says the lamp arrived broken", order_number="A-1043", amount=24.99))

The agent gets back a message that says the action has not happened yet. That line matters. Without it, a model might tell a customer their refund is done, when it's only waiting.

Run it with python3 agent_demo.py. Each request prints a notification like this, and the same text goes to Discord if you set a webhook:

Approval needed (request 8fa2a38f)
Send an email to [e-mail]
Subject: Your order
---
Hi Maria, your order A-1042 has shipped.

Assistant's note (not verified): The customer asked where their order is
Expires: 2026-10-17 07:08 UTC
Approve: python3 approvals.py approve 8fa2a38f
Reject:  python3 approvals.py reject 8fa2a38f

Step 4: Decide, then run

First, see what's waiting:

$ python3 approvals.py list
8fa2a38f  Send an email to [e-mail]
c3ad383c  Refund 24.99 on order A-1043
2 waiting

Approving shows the details first, then asks for a typed yes:

$ python3 approvals.py approve 8fa2a38f
Request 8fa2a38f (pending)
Send an email to [e-mail]
Subject: Your order
---
Hi Maria, your order A-1042 has shipped.

Assistant's note (not verified): The customer asked where their order is
  2026-10-16 07:08:26 UTC  proposed by agent

Type yes to approve: yes
Request 8fa2a38f approved

Approving doesn't run anything. The runner does that:

$ python3 approvals.py run
Ran 1 approved request(s)

You can schedule python3 approvals.py run to repeat every few minutes, or run it right after you approve. The full history is always available:

$ python3 approvals.py show 8fa2a38f
Request 8fa2a38f (done)
...
  2026-10-16 07:08:26 UTC  proposed by agent
  2026-10-16 07:08:27 UTC  approved by yourname
  2026-10-16 07:08:27 UTC  done by system

Keep the agent away from the approve button

This is the most important rule, and the easiest to break by accident.

If the agent can run commands on the same machine, it could run python3 approvals.py approve itself. The whole system would be useless. So:

• Don't give the agent a shell. It should only be able to call the tools you give it.
• Run it as a different user from the person who approves, and keep approval on a separate path, like your own laptop or a separate account.
• Remember that this version keeps everything in one database file. Anyone who can write to that file can approve requests. If the agent might be compromised, put the queue behind a small service that the agent can only call to add requests.

The checks in the code, like the fingerprint, protect against mistakes and casual tampering. They don't protect against an attacker who has full access to the database. Permissions and a separate service do that job.

Approving from a link

A link that approves with one click is convenient, and it's risky for one reason. Some email and chat tools open links automatically to check them or show a preview, and a link that changes something when it's opened can be triggered by a robot. If you build one, make the link open a page that asks for a confirmation click, rather than approving on the first visit. Use a long, random, single-use address, and make it expire.

For most small projects, a command or a button in a tool you already use is simpler and safer.

On GitHub Actions

GitHub has an approval feature of its own. A job that refers to an environment with required reviewers pauses until one of them approves. Secrets stored for that environment stay locked until then.

name: Agent with approval

on:
  workflow_dispatch:

jobs:
  propose:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4   # use the latest major version
      - run: python3 agent.py       # writes what it wants to do to the run summary

  act:
    needs: propose
    runs-on: ubuntu-latest
    environment: approvals          # waits here until a required reviewer approves
    steps:
      - uses: actions/checkout@v4
      - run: python3 do_the_approved_things.py

You set the reviewers in the repository settings, under Environments. Check your plan first. At the time of writing, GitHub's documentation says required reviewers are available for public repositories on all plans, and for private repositories only with GitHub Enterprise.

This suits a workflow with one clear gate. The queue above suits an agent that raises requests at any time.

Making approvals easy to get right

The queue is only as good as the questions it asks.

• Say what will happen, in one line. "Refund 24.99 on order A-1043" beats "Approve action?".
• Include what the person needs to decide. The recipient, the amount, the first lines of a message.
• Say why they're being asked. A line like "this run read an email from a stranger" is worth showing, because it tells the person to be careful.
• Group, don't repeat. If an agent proposes twenty similar actions, let one person approve them together, with a clear list.
• Be sparing. Auto-allow what's low-risk, within hard limits, and ask only about what matters.
• Make "no" easy. A rejected request should take one command, and the agent should be told it was declined.

What can go wrong

People stop reading. If requests are frequent and almost always fine, approving becomes a reflex. Fewer, better questions fix this. A rise in the number of requests is worth investigating too.

Requests sit forever. Expiry keeps them from becoming time bombs, but an unattended queue means things aren't getting done. Set up an alert when something has been waiting a long time. I explain how in how to monitor your automations when something breaks.

Notifications leak data. A message in a chat channel is visible to everyone in it. Keep summaries short, and avoid putting private details in places many people can read.

The runner has too much power. The runner executes whatever is approved, so give it the narrowest permissions it can work with.

A one-person queue. If the same person owns the agent and approves everything, you've built a speed bump, not a second pair of eyes. For high-stakes actions, get a second person to approve.

It's not a substitute for hard limits. Approval catches mistakes you can see. Rules that make some actions impossible, like a refund cap, catch the ones you can't.

Quick answers

What does "human in the loop" mean?
A person is part of the process, usually approving or reviewing what an AI does before it takes effect.

Do I need this for every agent?
No. A read-only assistant doesn't. Once an agent can send, spend or delete, it's worth adding.

Why not just ask in the chat?
That works when someone is watching. For scheduled agents, the question needs somewhere to wait, and that's what the queue is.

Can the AI write the summary people read?
It can add a note, but the main summary should come from your code, built from the real arguments.

How long should requests last?
Long enough for a person to see them, short enough to still be true. A day is a reasonable start for a decision, and an hour for running it.

Can I use Slack, email or Telegram instead of Discord?
Yes. Change the notify() function. The rest stays the same.

The short version

An approval is only worth something if the person understood what they approved. Show the real action, make the answer expire, run it once, and keep the approve button away from the agent.

Start with one risky action and one person. Once it runs smoothly, add the next.

You can find more practical automation guides on Procwire.