Human-in-the-Loop Approval¶
Require human approval before executing sensitive tool calls. The agent pauses, sends an approval request to a human reviewer, and waits for a decision before proceeding or adapting.
from promptise import build_agent, ApprovalPolicy, CallbackApprovalHandler, ApprovalDecision
async def my_approver(request):
print(f"Agent wants to call: {request.tool_name}({request.arguments})")
answer = input("Approve? [y/n]: ").strip().lower()
return ApprovalDecision(approved=(answer == "y"))
agent = await build_agent(
servers=servers,
model="openai:gpt-5-mini",
approval=ApprovalPolicy(
tools=["send_email", "delete_*", "payment_*"],
handler=CallbackApprovalHandler(my_approver),
),
)
How It Works¶
User sends message
↓
Agent reasons and decides to call send_email(to="alice@acme.com", subject="Invoice")
↓
┌─ ApprovalPolicy checks: does "send_email" match any pattern in tools=["send_*"]?
│
├─ NO MATCH → tool executes immediately (zero overhead)
│
└─ MATCH → Approval flow begins:
↓
1. Build ApprovalRequest (unique ID, tool name, redacted arguments, caller identity)
↓
2. Send to ApprovalHandler (webhook POST / async callback / UI queue)
↓
3. Agent execution PAUSES (event loop is free — other agents/requests continue)
↓
4. Human reviews:
"Agent 'billing-bot' wants to send_email(to='[EMAIL]', subject='Invoice')"
[Approve] [Deny] [Approve with edits]
↓
5. Decision received (or timeout triggers on_timeout action)
↓
┌─ APPROVED → tool executes with original (or modified) arguments → agent continues
├─ DENIED → tool returns "DENIED: reason" as result → agent adapts
└─ TIMEOUT → on_timeout="deny" returns denial / on_timeout="allow" proceeds
Key principle: The LLM doesn't know approval exists. It calls tools normally. The approval wrapper intercepts matching tool calls transparently. On approval, the tool runs. On denial, the tool returns a "DENIED" message as its result — the LLM sees this and adapts (tries alternatives, asks for help, or reports it can't proceed).
Setup Guide¶
Step 1: Choose which tools need approval¶
Use glob patterns to match tool names:
tools=["send_email"] # Exact match — only send_email
tools=["send_*"] # Wildcard — send_email, send_notification, send_sms
tools=["delete_*", "payment_*"] # Multiple patterns
tools=["*"] # Every tool requires approval (for maximum control)
Patterns use Python's fnmatch — * matches any characters, ? matches one character.
Step 2: Choose a handler¶
Three built-in handlers + custom protocol:
| Handler | Best for | How it works |
|---|---|---|
CallbackApprovalHandler |
Quick integration, scripts, Slack bots | Call your async function, return decision |
WebhookApprovalHandler |
External approval systems, REST APIs | POST request + poll for decision |
QueueApprovalHandler |
In-process UIs (Gradio, Streamlit) | asyncio.Queue — UI reads, user clicks, decision flows back |
| Custom (protocol) | Anything else | Implement one async method |
Step 3: Configure the policy¶
from promptise import ApprovalPolicy
policy = ApprovalPolicy(
tools=["send_email", "delete_*"],
handler=my_handler,
timeout=300, # 5 minutes to decide
on_timeout="deny", # Deny if no response
include_arguments=True, # Show tool args to reviewer
redact_sensitive=True, # Redact PII/credentials in args
max_pending=10, # Max 10 concurrent pending approvals
max_retries_after_deny=3, # Permanent deny after 3 retries
)
Step 4: Pass to build_agent¶
That's it. Every tool call matching your patterns now requires human approval.
ApprovalPolicy — All Parameters¶
| Parameter | Type | Default | Description |
|---|---|---|---|
tools |
list[str] |
required | Glob patterns for tool names. "send_*" matches send_email, send_notification. |
handler |
ApprovalHandler \| Callable |
required | How to send and receive approval decisions. |
timeout |
float |
300 |
Seconds to wait for a decision before on_timeout triggers. Min: > 0. Max: 86,400 (24 hours). |
on_timeout |
"deny" \| "allow" |
"deny" |
Action when timeout expires. "deny" is the safe default. Use "allow" only for non-critical, low-risk tools. |
include_arguments |
bool |
True |
Include tool arguments in the approval request. Set to False to hide arguments from reviewers (e.g., when arguments contain data the reviewer shouldn't see). |
redact_sensitive |
bool |
True |
Run arguments through PII/credential detection before sending to the reviewer. Requires the guardrails module. Falls back to raw arguments if guardrails are not installed. |
max_pending |
int |
10 |
Maximum concurrent pending approvals per agent. When reached, additional tool calls are auto-denied with "Too many pending approval requests." |
max_retries_after_deny |
int |
3 |
After this many denials of the same tool name, return a permanent denial message. Prevents the LLM from retrying the same denied tool in an infinite loop. |
ApprovalRequest — What the Handler Receives¶
Every handler receives an ApprovalRequest with these fields:
| Field | Type | Description |
|---|---|---|
request_id |
str |
Cryptographically random unique ID (secrets.token_hex(16) — 32 hex chars). Use this to match requests with decisions. |
tool_name |
str |
The name of the tool the agent wants to call (e.g., "send_email"). |
arguments |
dict |
The tool arguments. Redacted if redact_sensitive=True — e.g., {"to": "[EMAIL]", "body": "..."}. Empty dict if include_arguments=False. |
agent_id |
str \| None |
Agent or process identifier (set automatically in runtime). |
caller_user_id |
str \| None |
The user_id from CallerContext — identifies which user triggered the action. |
context_summary |
str |
Brief context for the reviewer (can be customized). |
timestamp |
float |
When the request was created (time.time()). |
timeout |
float |
How long the handler has to respond before the default action triggers. |
metadata |
dict |
Developer-provided custom data (passed through unchanged). |
HMAC signature:
# Compute a signature for tamper-proof webhook delivery
signature = request.compute_hmac("your-secret-key")
# Returns a hex string — include in X-Promptise-Signature header
Serialization:
ApprovalDecision — What the Handler Returns¶
| Field | Type | Default | Description |
|---|---|---|---|
approved |
bool |
required | Whether the tool call is approved. |
modified_arguments |
dict \| None |
None |
If the reviewer edited the arguments. When set, the tool executes with these arguments instead of the originals. |
reviewer_id |
str \| None |
None |
Who made the decision (for audit trail). |
reason |
str \| None |
None |
Optional explanation shown to the agent on denial. |
timestamp |
float |
time.time() |
When the decision was made. |
Handlers — Detailed Setup¶
CallbackApprovalHandler¶
Wraps any Python async function. The simplest way to integrate.
from promptise import CallbackApprovalHandler, ApprovalDecision
# Full control — return an ApprovalDecision
async def slack_approver(request):
message = (
f"*Approval needed*\n"
f"Agent wants to call `{request.tool_name}`\n"
f"Arguments: `{request.arguments}`\n"
f"User: {request.caller_user_id or 'anonymous'}"
)
channel_id = await post_to_slack(message)
reaction = await wait_for_reaction(channel_id, timeout=request.timeout)
return ApprovalDecision(
approved=(reaction.emoji == "thumbsup"),
reviewer_id=reaction.user_id,
reason="Approved via Slack" if reaction.emoji == "thumbsup" else "Denied via Slack",
)
handler = CallbackApprovalHandler(slack_approver)
Shortcut — return a plain bool:
# Auto-approve everything (for testing)
handler = CallbackApprovalHandler(lambda req: True)
# Auto-deny everything (for dry runs)
handler = CallbackApprovalHandler(lambda req: False)
# Approve based on tool name
handler = CallbackApprovalHandler(
lambda req: req.tool_name != "delete_production_database"
)
WebhookApprovalHandler¶
For external approval systems. POSTs the request as JSON, then polls for the decision.
from promptise import WebhookApprovalHandler
handler = WebhookApprovalHandler(
url="https://your-approval-api.com/requests",
secret="your-hmac-secret",
poll_url="https://your-approval-api.com/decisions", # Optional — defaults to {url}/{request_id}
poll_interval=2.0,
headers={"Authorization": "Bearer your-api-token"},
)
All parameters:
| Parameter | Type | Default | Description |
|---|---|---|---|
url |
str |
required | Webhook URL to POST approval requests to. Validated against private IP blocklist (SSRF protection). |
secret |
str \| None |
Auto-generated | HMAC secret for signing requests. If not provided, a random secret is generated per process (not useful for cross-process verification — set your own for production). |
poll_url |
str \| None |
{url}/{request_id} |
URL to poll for the decision. Handler GETs this URL every poll_interval seconds. |
poll_interval |
float |
2.0 |
Seconds between poll attempts. Min: 0.5. |
headers |
dict[str, str] \| None |
None |
Custom HTTP headers (e.g., auth tokens, API keys). |
http_client |
httpx.AsyncClient \| None |
None |
Pre-configured HTTP client. Use for proxy, mTLS, custom CA certificates, or any httpx configuration. |
Webhook flow:
1. Handler POSTs to url:
POST https://your-api.com/requests
Headers:
Content-Type: application/json
X-Promptise-Signature: <hmac-sha256>
X-Promptise-Request-Id: <request_id>
Body: {"request_id": "...", "tool_name": "...", "arguments": {...}, ...}
2. Your API responds: 202 Accepted
3. Handler polls poll_url every poll_interval seconds:
GET https://your-api.com/requests/<request_id>
Headers:
X-Promptise-Request-Id: <request_id>
4. Your API responds with 200 + decision when ready:
{"approved": true, "reviewer_id": "alice", "reason": "Looks good"}
Or 202 if still pending (handler continues polling)
5. On timeout: handler raises TimeoutError → on_timeout action triggers
HMAC verification (server-side):
import hmac, hashlib, json
def verify_signature(request_id, tool_name, received_signature, secret):
payload = json.dumps({"request_id": request_id, "tool_name": tool_name}, sort_keys=True)
expected = hmac.new(secret.encode(), payload.encode(), hashlib.sha256).hexdigest()
return hmac.compare_digest(expected, received_signature)
Enterprise proxy / mTLS configuration:
import httpx
# Corporate proxy
client = httpx.AsyncClient(
proxies="http://corporate-proxy:8080",
verify="/path/to/corporate-ca-bundle.crt",
timeout=60,
)
handler = WebhookApprovalHandler(
url="https://internal-approval-api.corp.com/requests",
http_client=client,
)
QueueApprovalHandler¶
For in-process UIs where the human reviewer is in the same Python process (Gradio, Streamlit, terminal prompts).
from promptise import QueueApprovalHandler, ApprovalDecision
handler = QueueApprovalHandler(maxsize=100)
# Your UI reads pending requests from the queue
async def approval_ui_loop():
while True:
request = await handler.request_queue.get()
# Display to user
print(f"\n--- Approval Required ---")
print(f"Tool: {request.tool_name}")
print(f"Arguments: {request.arguments}")
print(f"User: {request.caller_user_id}")
# Collect decision
answer = input("Approve? [y/n/edit]: ").strip().lower()
if answer == "y":
handler.submit_decision(request.request_id, ApprovalDecision(approved=True))
elif answer == "edit":
new_to = input("New recipient: ")
handler.submit_decision(request.request_id, ApprovalDecision(
approved=True,
modified_arguments={"to": new_to},
))
else:
handler.submit_decision(request.request_id, ApprovalDecision(
approved=False, reason="Denied by operator"
))
| Parameter | Type | Default | Description |
|---|---|---|---|
maxsize |
int |
100 |
Maximum queue size. Prevents unbounded memory growth if no reviewer is consuming requests. |
| Method | Description |
|---|---|
handler.request_queue |
asyncio.Queue[ApprovalRequest] — read pending requests from here |
handler.submit_decision(request_id, decision) |
Submit a decision for a pending request. Raises KeyError if the request_id is unknown or already resolved. |
Custom Handler¶
Implement the ApprovalHandler protocol — one async method:
from promptise import ApprovalHandler, ApprovalRequest, ApprovalDecision
class JiraApprovalHandler:
"""Create a Jira ticket for each approval, wait for resolution."""
def __init__(self, jira_url: str, project: str, api_token: str):
self.jira_url = jira_url
self.project = project
self.api_token = api_token
async def request_approval(self, request: ApprovalRequest) -> ApprovalDecision:
# Create Jira ticket
ticket = await self._create_ticket(
summary=f"Agent approval: {request.tool_name}",
description=f"Arguments: {request.arguments}\nUser: {request.caller_user_id}",
)
# Poll ticket status
while True:
status = await self._get_ticket_status(ticket.id)
if status == "approved":
return ApprovalDecision(approved=True, reviewer_id=ticket.assignee)
if status == "rejected":
return ApprovalDecision(approved=False, reason=ticket.comment)
await asyncio.sleep(5)
The protocol is runtime-checkable:
What the Agent Sees¶
The approval system is transparent to the LLM. It doesn't know approval is happening — it just calls tools and receives results.
Approved¶
The tool executes normally. The agent sees the tool's actual result. It has no idea approval was involved.
Denied¶
The tool returns a denial message as its result:
The agent sees this as the tool output and must adapt. Typical agent behaviors: - Try an alternative approach that doesn't require the denied tool - Ask the user for clarification or permission - Report that it cannot complete the requested action
Denied with reason¶
The reviewer's reason is included, helping the agent understand why and adapt more intelligently.
Modified arguments¶
The reviewer approves but changes the arguments. The tool executes with the modified arguments. The agent sees the result of the modified call.
Example: Agent calls send_email(to="all-staff@acme.com"). Reviewer changes to to="marketing-team@acme.com". Agent sees the result of sending to marketing-team.
Permanent denial¶
After max_retries_after_deny denials of the same tool:
This stops the agent from wasting reviewer time by retrying the same denied action.
Security¶
HMAC Request Signing¶
WebhookApprovalHandler signs every request with HMAC-SHA256:
- Header:
X-Promptise-Signature - Payload: JSON of
{"request_id": "...", "tool_name": "..."}(sorted keys) - Secret: Your
secretparameter
Verify this in your approval API to reject spoofed requests.
Single-Use Request IDs¶
Every request_id is generated with secrets.token_hex(16) — 128 bits of cryptographic randomness. Request IDs are tracked per-agent to prevent replay attacks.
Argument Redaction¶
When redact_sensitive=True (default), tool arguments are processed through the guardrails system before being sent to the reviewer:
- PII patterns: email addresses, phone numbers, SSNs →
[EMAIL],[PHONE],[SSN] - Credential patterns: API keys, tokens →
[AWS_ACCESS_KEY],[OPENAI_KEY]
The reviewer sees enough to make a decision without seeing raw sensitive data.
SSRF Protection¶
WebhookApprovalHandler validates the webhook URL at construction time:
- Blocks private IP ranges (10.x, 172.16-31.x, 192.168.x)
- Blocks loopback (127.x)
- Blocks link-local (169.254.x — cloud metadata endpoints)
- Blocks
localhostand known internal hostnames
Timeout Enforcement¶
Timeouts are enforced server-side with asyncio.wait_for. A misbehaving handler cannot keep the agent suspended indefinitely. Maximum configurable timeout: 24 hours.
Multi-User¶
CallerContext.user_id is automatically extracted and included in every ApprovalRequest.caller_user_id. The reviewer knows which user's action they're approving.
# User identity flows through to the approval request
result = await agent.ainvoke(
{"messages": [{"role": "user", "content": "Send the invoice"}]},
caller=CallerContext(user_id="user-42"),
)
# ApprovalRequest will include caller_user_id="user-42"
Different users' pending approvals are completely independent. User A waiting for approval doesn't block User B's agent from executing.
YAML Configuration (.superagent)¶
approval:
tools:
- send_email
- "delete_*"
- "payment_*"
handler: webhook
webhook_url: https://your-api.com/approvals
timeout: 300
on_timeout: deny
max_pending: 10
redact_sensitive: true
max_retries_after_deny: 3
Supported handler types in YAML: webhook (requires webhook_url), queue.
Callback handler not available in YAML
CallbackApprovalHandler requires a Python function, so it can only be configured programmatically. Use webhook or queue in YAML files.
Integration with Agent Runtime¶
Approval works seamlessly with autonomous agents in the runtime:
from promptise.runtime import ProcessConfig, TriggerConfig
from promptise import ApprovalPolicy, WebhookApprovalHandler
config = ProcessConfig(
model="openai:gpt-5-mini",
instructions="Process customer refunds. Amounts over $500 need manager approval.",
triggers=[TriggerConfig(type="cron", cron_expression="*/5 * * * *")],
approval=ApprovalPolicy(
tools=["process_refund", "delete_account", "escalate_to_legal"],
handler=WebhookApprovalHandler(url="https://ops.internal/approvals"),
timeout=600,
),
)
When the autonomous agent hits a tool requiring approval at 3am, the webhook fires, the ops team gets notified, and the agent waits.
Integration with Guardrails¶
Approval and guardrails work together:
- Input guardrails run first — if the input is blocked, the agent never reaches tool calling
- Approval intercepts tool calls that match patterns — after the agent decides to call a tool but before execution
- Argument redaction uses guardrail detectors to sanitize arguments sent to the reviewer
- Output guardrails run on the tool result after execution (if approved)
Integration with Budget¶
- Pending approvals count as in-progress tool calls for budget tracking
- If the budget is exhausted while an approval is pending, the pending tool call is not affected (it was already counted when the request was created)
- The
max_irreversible_per_runlimit is checked before sending the approval request — if the limit is already reached, the tool is denied without asking the reviewer
Integration with Semantic Cache¶
Cached responses bypass approval entirely — tools aren't called on cache hits, so there's nothing to approve. This is correct behavior: the original response was already approved (or didn't need approval) when it was first generated and cached.
Edge Cases¶
| Scenario | Behavior |
|---|---|
| Tool doesn't match any pattern | Executes immediately, zero overhead. |
| Approval arrives after timeout | Discarded. Agent already received the timeout decision. Late approvals are logged. |
| Agent crash during pending approval | On restart, the pending approval is gone. The agent starts fresh. |
| Multiple tools need approval in one invocation | Each is handled independently, sequentially. The second approval starts after the first resolves. |
| Agent retries denied tool | Tracked per tool name. After max_retries_after_deny, returns permanent denial without asking reviewer again. |
| max_pending reached | Additional tool calls are auto-denied: "Too many pending approval requests." |
| Handler throws an exception | Treated as denial: "Approval handler error: {error type}." Agent continues. |
| Reviewer modifies arguments to invalid values | The tool executes with modified arguments. If the tool validates internally (Pydantic), it will fail with a tool error. |
| No CallerContext provided | Approval still works. caller_user_id is None in the request. |
API Summary¶
| Symbol | Import | Description |
|---|---|---|
ApprovalPolicy |
from promptise import ApprovalPolicy |
Main configuration — tools, handler, timeout, behavior |
ApprovalRequest |
from promptise import ApprovalRequest |
Data sent to the handler (tool name, args, caller, etc.) |
ApprovalDecision |
from promptise import ApprovalDecision |
Human's decision (approved, modified args, reason) |
ApprovalHandler |
from promptise import ApprovalHandler |
Runtime-checkable protocol for custom handlers |
CallbackApprovalHandler |
from promptise import CallbackApprovalHandler |
Wrap an async callable or plain function |
WebhookApprovalHandler |
from promptise import WebhookApprovalHandler |
POST to URL + poll for decision |
QueueApprovalHandler |
from promptise import QueueApprovalHandler |
asyncio.Queue for in-process UIs |
Manual Integration¶
If you're not using build_agent(), you can wrap tools manually:
from promptise import wrap_tools_with_approval, ApprovalPolicy, CallbackApprovalHandler
policy = ApprovalPolicy(
tools=["send_*", "delete_*"],
handler=CallbackApprovalHandler(my_handler),
)
# Wrap a list of LangChain BaseTool instances
wrapped_tools = wrap_tools_with_approval(tools, policy, event_notifier=notifier)
# Tools matching patterns are wrapped; others pass through unchanged
| Parameter | Type | Description |
|---|---|---|
tools |
list[BaseTool] |
Tools to potentially wrap |
policy |
ApprovalPolicy |
Which tools need approval and how |
event_notifier |
EventNotifier \| None |
Optional notifier for approval events |
| Returns | list[BaseTool] |
New list with matching tools wrapped |
What's Next?¶
- Guardrails -- input/output security scanning
- Semantic Cache -- reduce costs with response caching
- Conversations -- multi-user session persistence
- Building Agents -- full parameter reference