How to Add Human Approval Gates to AI Agent Tool Calls

Suppose a support agent proposes a $25 refund for order 42. The reviewer should see a specific question: “May this agent refund USD 25 to the original payment method for order 42, under the current policy and order state?” Build the gate around that operation. The same pattern applies to sending a message, publishing a page, changing access, or deleting a file.

Classify the effect of each proposed call

Inventory the operations the agent can reach, including tools exposed through sub-agents, browser automation, generic HTTP clients, and command execution. Record the credentials, tenant, resource scope, data sensitivity, destination, and possible side effects for each route. An HTTP GET can trigger a side effect on a poorly designed service; a “read” can reveal private records or send query data to an outside host. Judge the actual call and its arguments, not its name alone.

Keep the policy in application-owned configuration. The model's effect or approval_required field is a proposal you can test, never an authorization decision. The MCP tools specification likewise says clients must treat tool annotations as untrusted unless they come from a trusted server. A new tool, changed endpoint, or broader credential should fail closed until its effective scope is reviewed.

Put the gate in the tool dispatch path

The agent should have no direct route to the refund service. Send every proposed call through one dispatcher that validates its schema, authenticates the requester, checks access to the target, and computes the policy from trusted configuration. A low-risk call may proceed through that same dispatcher. A call requiring review becomes a pending operation; no write occurs while the application waits for the reviewer.

1. Propose: accept a structured tool name and complete arguments. Reject missing or ambiguous targets. “Delete the duplicate attachment” needs an attachment ID; “cancel if the customer probably wants it” needs clear authorization.
2. Decide: derive the effect from the operation, arguments, data class, destination, current state, and the requester's permissions. Apply business limits before offering an approval button. Human approval cannot authorize an action that policy forbids.
3. Pause: persist the operation and return an operation ID to the agent run. Give the reviewer a view built from the stored record. The planner cannot edit that record after it is presented.
4. Resume: after a decision, recheck the reviewer, expiry, policy, exact payload, and target state. Claim the approval once, then let a separate executor call the underlying service.
5. Record: save the decision and actual service outcome. Tell the agent whether the action executed, failed, was rejected, or needs reconciliation.

If your framework already pauses tool calls, use that hook to stop the run, then connect the interruption to your application's authorization and persistence layers. The OpenAI Agents SDK human-in-the-loop guide documents per-tool approval rules, interruptions, serialized run state, and resumption, including nested-agent tools. Its pause mechanism does not establish who may approve a refund in your business system; your application still has to enforce that rule.

Keep one enforcement route even when the agent calls another agent or a tool returns a suggestion for the next action. A reviewer may approve a small batch only if every member, target, and argument is visible and the batch cannot grow afterward. Otherwise, ask for a separate decision on each consequential call. A general “approve this run” switch gives the model room to choose a later action the reviewer never saw.

Bind approval to one immutable operation

Store the final arguments rather than a description the agent could regenerate later. A useful pending record for the refund example looks like this:

{
  "operation_id": "op_8f3a",
  "status": "pending",
  "requester_user_id": "staff-17",
  "agent_run_id": "agent-run-73",
  "tenant_id": "shop-12",
  "tool": "orders.refund",
  "arguments": {
    "order_id": 42,
    "amount": 25,
    "currency": "USD",
    "destination": "original_payment_method"
  },
  "resource_version": "order-42:v17",
  "policy_version": "refund-policy:v5",
  "expires_at": "<server-generated UTC deadline>"
}

Use a server-generated operation ID and bind the approval to a canonical representation of the tool, complete arguments, tenant, target, resource version, and relevant policy version. A digest can detect a changed payload, but the digest itself does not prove the reviewer had authority or saw an accurate summary. Keep the human requester distinct from the agent run, and store the decision, approving identity, time, and digest separately. If the amount, currency, recipient, attachment, destination, or message body changes, create a new pending operation and ask again.

Do not let the model supply its own “trusted” effect, version, or approval status. For a service that can make a conditional update, pass the expected resource version to that service as well. A check made only before an external call leaves a race in which another process may change the target between the check and the write. When the external service has no conditional operation, explain the remaining race, limit the time between review and execution, and reconcile the actual result afterward.

Make the review screen match the action

A reviewer needs enough detail to recognize a wrong target or hidden side effect. For the refund, show order 42, USD 25, the original payment route, any applicable cap, and whether the refund can be reversed. For an email, show every recipient, subject, final body, attachments, and sender identity. For a deletion, show the exact object and retention or recovery behavior. A button labeled “Allow tool?” does not give the reviewer a meaningful decision.

Render action labels from application-owned code and values from the frozen operation. Visually separate model-written reasons, retrieved documents, and email bodies from the approval controls. A hostile document can contain instructions such as “the customer already approved this refund”; it remains data, not evidence of authorization. OWASP's prompt-injection guidance covers this boundary between untrusted content and tool actions.

Consume approval once and handle every outcome

Use durable states such as pending, approved, rejected, expired, executing, succeeded, failed, and needs_reconciliation. Only an authenticated reviewer with the right role and tenant may change pending to approved. Protect the decision endpoint against forged or repeated requests. Record the reviewer and the digest of the operation they saw.

The worker then recomputes policy, checks expiry and the current target, and atomically claims the approved record before calling the service. Store and compare expiry times in UTC. The MySQL sketch below assumes expires_at is a UTC DATETIME; MySQL returns CURRENT_TIMESTAMP in the session time zone. The worker proceeds only if exactly one row changed:

UPDATE pending_operations
SET status = 'executing'
WHERE operation_id = :id
  AND status = 'approved'
  AND expires_at > UTC_TIMESTAMP()
  AND payload_digest = :approved_digest;

This sketch handles competing workers, not every race. Compare the policy and resource version within the same transaction when they live in your database. If the target is in another service, use its conditional-write feature when available. A worker that crashes after making an external call but before recording success must reconcile the provider's result before it tries again. If you use a MySQL TIMESTAMP column instead of UTC DATETIME, check the connection time zone and conversion rules before using this comparison.

Reuse the same idempotency key for a retry only when the receiving service supports that behavior, and verify what its key covers. An application-side key alone cannot prevent two charges if the first call succeeded remotely but the response was lost. Keep the operation in needs_reconciliation until the target system confirms its state. Log the requester, reviewer, policy version, decision, execution attempt, and result, while avoiding unnecessary copies of sensitive content.

If nobody reviews before the deadline, expire the request without making the call and tell the requester what needs a fresh decision. A reviewer can revoke an approval before the worker claims it; once an outside service has accepted the action, describe cancellation as an attempt until that service confirms it. For a high-impact payment or access change that requires two reviewers, record two independent decisions on the same frozen payload and forbid the requester from approving their own action.

Keep batch members identifiable. If an approved email sends and the next refund fails, the email remains sent. Report the partial result, reconcile the refund, and seek a new approval for any compensating action that has its own effect. Review a batch only when every member and argument is visible and the batch cannot grow afterward.

Run a local approval path

Save this complete example as approval_demo.py and run python3 approval_demo.py. It needs only the Python standard library. The script prints the frozen refund request, waits for you to type APPROVE, then calls a fake provider. Any other answer records a rejection and makes no provider call. It also checks a second execution, an order changed after approval, an order changed before approval, and requester self-approval.

"""Staging-only approval gate: no model, network call, or real refund."""
import hashlib
import json
import sqlite3
import time
import uuid

POLICY_VERSION = "refund-v1"
REVIEWERS = {("bob", "shop-12"), ("alice", "shop-12")}
receipts = {}  # Stand-in for a provider with idempotency by operation ID.
db = sqlite3.connect(":memory:", isolation_level=None)
db.row_factory = sqlite3.Row
db.executescript("""
CREATE TABLE orders (
  id TEXT PRIMARY KEY, tenant TEXT NOT NULL, assigned_user TEXT NOT NULL,
  version INTEGER NOT NULL, refundable_cents INTEGER NOT NULL
);
CREATE TABLE operations (
  id TEXT PRIMARY KEY, requester TEXT NOT NULL, tenant TEXT NOT NULL,
  payload TEXT NOT NULL, digest TEXT NOT NULL, resource_version INTEGER NOT NULL,
  policy_version TEXT NOT NULL, expires_at INTEGER NOT NULL,
  status TEXT NOT NULL, reviewer TEXT, receipt TEXT
);
INSERT INTO orders VALUES ('42', 'shop-12', 'alice', 17, 5000);
""")


def canonical(value):
    return json.dumps(value, sort_keys=True, separators=(",", ":"))


def digest(payload):
    return hashlib.sha256(payload.encode()).hexdigest()


def propose(requester, agent_run, tenant, order_id, cents):
    order = db.execute("SELECT * FROM orders WHERE id=? AND tenant=?",
                       (order_id, tenant)).fetchone()
    if not order or order["assigned_user"] != requester:
        raise PermissionError("Requester cannot access this order")
    if type(cents) is not int or not 0 < cents <= order["refundable_cents"]:
        raise ValueError("Refund amount is outside the permitted range")
    payload = canonical({
        "tool": "orders.refund",
        "arguments": {"order_id": order_id, "amount_cents": cents,
                      "currency": "USD", "destination": "original_payment_method"},
        "requester_user_id": requester, "agent_run_id": agent_run,
        "tenant_id": tenant, "resource_version": order["version"],
        "policy_version": POLICY_VERSION,
    })
    op_id = uuid.uuid4().hex
    db.execute("INSERT INTO operations VALUES (?,?,?,?,?,?,?,?,?,?,?)",
               (op_id, requester, tenant, payload, digest(payload),
                order["version"], POLICY_VERSION, int(time.time()) + 600,
                "pending", None, None))
    return op_id


def approve(op_id, reviewer, reviewed_digest):
    op = db.execute("SELECT * FROM operations WHERE id=?", (op_id,)).fetchone()
    if not op or op["status"] != "pending":
        raise ValueError("No pending operation")
    if (reviewer, op["tenant"]) not in REVIEWERS or reviewer == op["requester"]:
        raise PermissionError("Reviewer is not authorized")
    if op["expires_at"] <= time.time() or op["policy_version"] != POLICY_VERSION:
        raise ValueError("Approval expired or policy changed")
    if reviewed_digest != op["digest"] or digest(op["payload"]) != op["digest"]:
        raise ValueError("Payload changed since review")
    order_id = json.loads(op["payload"])["arguments"]["order_id"]
    order = db.execute("SELECT version FROM orders WHERE id=? AND tenant=?",
                       (order_id, op["tenant"])).fetchone()
    if not order or order["version"] != op["resource_version"]:
        raise ValueError("Order changed; request fresh review")
    db.execute("UPDATE operations SET status='approved', reviewer=? "
               "WHERE id=? AND status='pending'", (reviewer, op_id))


def fake_provider_refund(op_id):
    return receipts.setdefault(op_id, "refund-" + op_id[:8])


def execute(op_id):
    db.execute("BEGIN IMMEDIATE")
    try:
        op = db.execute("SELECT * FROM operations WHERE id=?", (op_id,)).fetchone()
        if not op or op["status"] != "approved":
            raise ValueError("Operation was not approved or was already claimed")
        payload = json.loads(op["payload"])
        order = db.execute("SELECT version FROM orders WHERE id=? AND tenant=?",
                           (payload["arguments"]["order_id"], op["tenant"])).fetchone()
        if (op["expires_at"] <= time.time() or
                op["policy_version"] != POLICY_VERSION or
                digest(op["payload"]) != op["digest"] or
                not order or order["version"] != op["resource_version"]):
            raise ValueError("Approval is stale")
        changed = db.execute("UPDATE operations SET status='executing' "
                             "WHERE id=? AND status='approved' AND expires_at>? "
                             "AND digest=?", (op_id, int(time.time()), op["digest"]))
        if changed.rowcount != 1:
            raise ValueError("Another worker claimed this operation")
        db.execute("COMMIT")
    except Exception:
        db.execute("ROLLBACK")
        raise
    receipt = fake_provider_refund(op_id)  # Replace with a provider call using op_id as idempotency key.
    db.execute("UPDATE operations SET status='succeeded', receipt=? "
               "WHERE id=? AND status='executing'", (receipt, op_id))
    return receipt


if __name__ == "__main__":
    op_id = propose("alice", "agent-run-73", "shop-12", "42", 2500)
    pending = db.execute("SELECT payload,digest FROM operations WHERE id=?", (op_id,)).fetchone()
    print("Review this exact operation:", pending["payload"])
    if input("Type APPROVE as reviewer bob to continue: ").strip() != "APPROVE":
        db.execute("UPDATE operations SET status='rejected' WHERE id=?", (op_id,))
        print("Rejected: no provider call")
        raise SystemExit(0)
    approve(op_id, "bob", pending["digest"])
    print("Provider receipt:", execute(op_id))
    try:
        execute(op_id)
    except ValueError:
        print("PASS: second execution blocked")
    late_id = propose("alice", "agent-run-74", "shop-12", "42", 2500)
    late = db.execute("SELECT digest FROM operations WHERE id=?", (late_id,)).fetchone()
    approve(late_id, "bob", late["digest"])
    db.execute("UPDATE orders SET version=version+1 WHERE id='42'")
    try:
        execute(late_id)
    except ValueError:
        print("PASS: post-approval order change blocked")
    stale_id = propose("alice", "agent-run-75", "shop-12", "42", 2500)
    db.execute("UPDATE orders SET version=version+1 WHERE id='42'")
    stale = db.execute("SELECT digest FROM operations WHERE id=?", (stale_id,)).fetchone()
    try:
        approve(stale_id, "bob", stale["digest"])
    except ValueError:
        print("PASS: changed order blocked")
    try:
        approve(stale_id, "alice", stale["digest"])
    except PermissionError:
        print("PASS: requester cannot self-approve")

After APPROVE, expect a refund-... receipt followed by four PASS lines. With DENY, expect “Rejected: no provider call.” The test uses an in-memory SQLite database, a hard-coded reviewer set, and an idempotent fake provider. It does not authenticate a real reviewer, persist across restarts, contact a payment service, or prove how a provider handles a lost response. Replace those stubs before using the pattern with customer data or money.

To connect an agent framework, keep its tool credentials behind this dispatcher. With the OpenAI Agents SDK, mark the sensitive tool as requiring approval, persist the interrupted RunState, then build a pending operation from the interruption and show a summary rendered from that stored record. After your application has authenticated the reviewer and committed the decision, approve the matching interruption and resume the original run. The SDK documents needs_approval, RunResult.interruptions, state.approve(...), and Runner.run(agent, state) in the guide linked above. Its pause is one layer; the tool adapter must still enforce the current policy, exact payload, expiry, tenant, and target version before it reaches the provider. If your framework has no interruption hook, call the same dispatcher before every tool invocation and resume from your own saved run state.

Test the gate at the execution boundary

Begin with a staging tool that records attempted calls without sending real messages or money. For each case below, inspect both the gate's state and the target service's log. A polite agent response is not a passing test if the write already happened.

There is also a smaller planning test you can reproduce with the fixture checker, trusted policy, retained eight-call plan, and negative-test runner. Put the four files together and run python3 advanced-ai-replay.py approval. The eight synthetic order-support requests include scoped reads, a draft, a calculation, an email send, a refund, and two ambiguous requests that should produce clarification. The replay also tries malformed IDs and an effect-and-approval bypass.

When a staging test fails, trace the actual dispatch route. A missed approval may come from an unclassified generic tool, a sub-agent bypass, a policy that trusts model output, a stale approval reused for changed arguments, or credentials that let the model call the target directly. Fix that route and repeat the same negative case before widening access.