Skip to main content

Approval Workflows

When a policy blocks an agent, you can handle it with human approval instead of failing. The on_policy_block parameter lets you define what happens when a block is raised — prompt in the terminal, send a Slack message, call a webhook, or any custom logic.

Quick Start​

import waxell_observe as waxell

waxell.init()

@waxell.observe(
    agent_name="data-manager",
    workflow_name="delete",
    enforce_policy=True,
    on_policy_block=waxell.prompt_approval,  # terminal Y/N prompt
)
async def delete_records(table: str) -> dict:
    return {"deleted": 500, "table": table}

When a policy blocks execution, prompt_approval shows a terminal prompt:

  !! BLOCKED BY APPROVAL POLICY
  Reason:    Workflow 'delete' requires approval before execution
  Operation: delete
  Approvers: dba-team@company.com, compliance@company.com
  Timeout:   5 min

  Approve this operation? (y/n): y

If approved, the function executes. If denied or timed out, PolicyViolationError propagates.

Built-in Handlers​

prompt_approval — Terminal​

Interactive terminal prompt with timeout and approver display:

@waxell.observe(
    agent_name="my-agent",
    enforce_policy=True,
    on_policy_block=waxell.prompt_approval,
)
async def sensitive_operation():
    ...

auto_approve — Testing​

Always approves. Use in tests and dry-run scenarios:

@waxell.observe(
    agent_name="my-agent",
    enforce_policy=True,
    on_policy_block=waxell.auto_approve,
)
async def test_approval_flow():
    ...

auto_deny — Testing​

Always denies:

on_policy_block=waxell.auto_deny

Custom Handlers​

Write your own handler for any approval channel. The handler receives a PolicyViolationError and returns an ApprovalDecision:

Slack Approval​

from waxell_observe.types import ApprovalDecision
from waxell_observe.approval import extract_block_metadata

def slack_approval(error):
    meta = extract_block_metadata(error)

    # Send Slack message
    send_slack_message(
        channel="#approvals",
        text=f"🔒 *Approval Required*\n"
             f"Operation: {meta['action_type']}\n"
             f"Reason: {meta['reason']}",
        actions=["approve", "deny"],
    )

    # Wait for reaction
    response = wait_for_slack_reaction(timeout=300)

    return ApprovalDecision(
        approved=(response == "approve"),
        approver=response.user,
        elapsed_seconds=response.elapsed,
    )

@waxell.observe(
    agent_name="my-agent",
    enforce_policy=True,
    on_policy_block=slack_approval,
)
async def guarded_operation():
    ...

Webhook Approval​

import httpx
from waxell_observe.types import ApprovalDecision
from waxell_observe.approval import extract_block_metadata

async def webhook_approval(error):
    meta = extract_block_metadata(error)

    async with httpx.AsyncClient() as client:
        resp = await client.post(
            "https://approvals.internal/request",
            json={
                "operation": meta["action_type"],
                "reason": meta["reason"],
                "approvers": meta["approvers"],
            },
        )
        result = resp.json()

    return ApprovalDecision(
        approved=result["decision"] == "approved",
        approver=result.get("approved_by", ""),
    )
Async handlers supported

Both sync and async handlers work. Async handlers are awaited automatically.

ApprovalDecision​

The return type from all approval handlers:

FieldTypeDefaultDescription
approvedboolrequiredWhether to proceed with execution
approverstr""Who approved (email, username, system)
timed_outboolFalseWhether the approval window expired
elapsed_secondsfloatNoneTime from block to decision

extract_block_metadata​

Helper to extract structured data from a PolicyViolationError:

from waxell_observe.approval import extract_block_metadata

meta = extract_block_metadata(error)
meta["reason"]          # "Workflow 'delete' requires approval"
meta["action_type"]     # "delete"
meta["approvers"]       # ["dba-team@company.com"]
meta["timeout_minutes"] # 5.0 or None

With the Context Manager​

from waxell_observe import WaxellContext

with WaxellContext(
    agent_name="my-agent",
    workflow_name="delete",
    enforce_policy=True,
    on_policy_block=waxell.prompt_approval,
) as ctx:
    result = perform_deletion()
    ctx.set_result(result)

What Gets Traced​

When an approval handler runs, Waxell automatically records:

  1. approval_request governance event — logged before the handler is invoked
  2. human_turn:approval IO span — wraps the handler, captures prompt/response/duration
  3. approval_response governance event — logged after the handler returns

All three appear in the trace timeline without any extra instrumentation code.

Full Example​

import asyncio
import waxell_observe as waxell
from waxell_observe.errors import PolicyViolationError

waxell.init()


@waxell.tool(tool_type="database")
def delete_records(table: str, filter_criteria: str = "") -> dict:
    """Simulated deletion — triggers approval policy."""
    return {"table": table, "deleted": 500, "status": "completed"}


@waxell.observe(
    agent_name="data-manager",
    enforce_policy=True,
    on_policy_block=waxell.prompt_approval,
)
async def run_agent():
    command = waxell.input("> ")

    if "delete" in command:
        try:
            result = delete_records(table="users")
            print(f"Done: {result}")
        except PolicyViolationError as e:
            print(f"Blocked: {e}")


asyncio.run(run_agent())

Next Steps​