Add a Human-Approval Checkpoint to an AI Pipeline

You will put one human in the path of one risky AI action. The pipeline pauses, writes a pending item to a queue, pings a person, and only continues once they say yes. This guide wires that up with a Postgres table as the queue, a Slack webhook for the ping, and a small FastAPI endpoint to catch the decision. Budget about forty-five minutes.

#Before you start

• A running AI pipeline with at least one action you would not want to fire blind (send email, refund, publish, delete).
• A Postgres database you can create a table in.
• A Slack incoming webhook URL, or any chat webhook, stored as an env var.
• Python 3.10+ with fastapi, uvicorn, psycopg2-binary, and requests installed.

#Find the one action that needs a human

Do not put a human in front of everything. Pick the single step where a wrong AI decision is expensive or hard to undo. Everything before it can run automatically. The moment you reach that step, the job stops and waits for a person instead of calling the risky function directly.

#Create the approval queue

The queue is just a table of pending decisions. Each row holds what the AI wants to do, its current status, and enough context for a human to judge it. payload stores the proposed action as JSON so you can replay it later.

CREATE TABLE approvals (
  id          BIGSERIAL PRIMARY KEY,
  action      TEXT NOT NULL,
  payload     JSONB NOT NULL,
  status      TEXT NOT NULL DEFAULT 'pending',
  created_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
  decided_at  TIMESTAMPTZ
);

CREATE INDEX approvals_pending_idx ON approvals (status) WHERE status = 'pending';

#Pause the pipeline and enqueue the request

At the risky step, insert a pending row instead of running the action. Then stop. The function returns the approval id and hands control off. Nothing irreversible has happened yet. You have only written a request to the queue.

import json
import os
import psycopg2

def request_approval(action: str, payload: dict) -> int:
    conn = psycopg2.connect(os.environ["DATABASE_URL"])
    with conn, conn.cursor() as cur:
        cur.execute(
            "INSERT INTO approvals (action, payload) VALUES (%s, %s) RETURNING id",
            (action, json.dumps(payload)),
        )
        approval_id = cur.fetchone()[0]
    conn.close()
    return approval_id

# at the risky step:
# approval_id = request_approval("send_refund", {"user": "u_123", "amount": 4200})
# then return and wait. do NOT call send_refund() here.

#Notify a real person

A row in a table nobody looks at is not an approval step. Send the reviewer a message with the id and the details so they can act. This uses a Slack incoming webhook, but any chat webhook works. Keep the URL in an env var, never in the code.

import os
import requests

def notify_reviewer(approval_id: int, action: str, payload: dict) -> None:
    text = (
        f":warning: Approval needed (#{approval_id})\n"
        f"Action: {action}\n"
        f"Details: {payload}\n"
        f"Approve: POST /approvals/{approval_id}/approve"
    )
    resp = requests.post(
        os.environ["SLACK_WEBHOOK_URL"],
        json={"text": text},
        timeout=10,
    )
    resp.raise_for_status()

#Catch the decision with a webhook

Stand up a small endpoint the reviewer hits to approve or reject. On approve, mark the row and run the real action using the stored payload. On reject, mark it and do nothing. Because the action is rebuilt from payload, the human decision is the only thing that lets it fire.

import os
import psycopg2
from fastapi import FastAPI, HTTPException

app = FastAPI()

def _decide(approval_id: int, status: str) -> dict:
    conn = psycopg2.connect(os.environ["DATABASE_URL"])
    with conn, conn.cursor() as cur:
        cur.execute(
            "UPDATE approvals SET status = %s, decided_at = now() "
            "WHERE id = %s AND status = 'pending' RETURNING action, payload",
            (status, approval_id),
        )
        row = cur.fetchone()
    conn.close()
    if row is None:
        raise HTTPException(404, "not pending or not found")
    return {"action": row[0], "payload": row[1]}

@app.post("/approvals/{approval_id}/approve")
def approve(approval_id: int):
    job = _decide(approval_id, "approved")
    # run the real action now, from the stored payload:
    # send_refund(**job["payload"])
    return {"ok": True, "ran": job["action"]}

@app.post("/approvals/{approval_id}/reject")
def reject(approval_id: int):
    _decide(approval_id, "rejected")
    return {"ok": True}

#Run it and test both paths

Start the server and walk one request all the way through. Enqueue a pending item, check that the Slack message arrives, then hit approve and confirm the action runs exactly once. Test the reject path too, and confirm nothing fires.

uvicorn app:app --port 8000

# approve request #1
curl -X POST http://localhost:8000/approvals/1/approve

# a second approve should return 404 and not run the action twice
curl -X POST http://localhost:8000/approvals/1/approve

#Watch out for

• The action must be idempotent or guarded, or a double-click approves twice. The WHERE status = 'pending' clause is doing that job here: the second approve updates zero rows and 404s instead of refunding again.
• A pending item with no owner and no deadline sits forever. Decide what happens to stale approvals, whether they expire, escalate, or default to reject, and enforce it with a job that scans old pending rows.
• The approve endpoint runs a real action, so it needs auth. An open URL means anyone who guesses an id can approve a refund. Put it behind your normal auth and verify the webhook signature if the sender supports it.

#What you built

You now have a pipeline that stops at its one risky step, parks the proposed action in a queue, pings a person, and only fires when they approve. The decision is auditable, since every request and outcome is a row with a timestamp. Next, add a small view over the approvals table so reviewers can see the queue instead of living in Slack, and start logging how often they reject. That number tells you how much the AI was about to get wrong.