Approval Workflows

from agent_sentinel import guarded_action

@guarded_action(
    name="transfer_funds",
    cost_usd=0.10,
    requires_human_approval=True,
    approval_description="Transfer funds between customer accounts",
)
def transfer_funds(from_acct: str, to_acct: str, amount: float):
    # This will block until human approves/rejects
    return transfer_service.execute(from_acct, to_acct, amount)

from agent_sentinel import enable_remote_sync

enable_remote_sync(
    api_token="as_your_api_key_here",
    run_id="run-123",
)

from agent_sentinel import ApprovalClient, ApprovalPriority

client = ApprovalClient(
    api_token="as_your_api_key_here",
)

# Critical priority - notify immediately
response = client.request_approval_sync(
    action_name="delete_customer_data",
    description="Delete all data for customer #12345",
    agent_id="data-agent",
    run_id="run-456",
    priority=ApprovalPriority.CRITICAL,  # CRITICAL, HIGH, MEDIUM, LOW
    estimated_cost=0.0,
    timeout_seconds=300,
    action_inputs={"customer_id": 12345},
)

from agent_sentinel import ApprovalClient, RiskLevel

response = client.request_approval_sync(
    action_name="modify_pricing",
    description="Update product pricing in production",
    risk_level=RiskLevel.HIGH,  # CRITICAL, HIGH, MEDIUM, LOW, MINIMAL
    # ... other params
)

from agent_sentinel import ApprovalClient, ApprovalStatus

client = ApprovalClient(
    api_token="as_your_api_key_here",
)

response = client.request_approval_sync(
    action_name="execute_trade",
    description="Execute trade: Buy 100 shares AAPL",
    agent_id="trading-agent",
    run_id="run-789",
    estimated_cost=15000.0,
    timeout_seconds=600,  # 10 minute timeout
    action_inputs={"symbol": "AAPL", "quantity": 100, "side": "BUY"},
)

if response.status == ApprovalStatus.APPROVED:
    print(f"Approved by: {response.decided_by_email}")
    print(f"Notes: {response.decision_notes}")
    # Execute the action
    execute_trade(...)
elif response.status == ApprovalStatus.REJECTED:
    print(f"Rejected by: {response.decided_by_email}")
    print(f"Reason: {response.decision_notes}")
    # Handle rejection
elif response.status == ApprovalStatus.EXPIRED:
    print("Approval timed out - no decision within timeout window")
    # Handle timeout
elif response.status == ApprovalStatus.INFO_REQUESTED:
    print("Approver requested additional information")
    # Respond with more context
    client.respond_to_info_request(
        approval_id=response.approval_id,
        response_text="Additional context: This trade is part of rebalancing strategy..."
    )

from agent_sentinel import ApprovalClient

client = ApprovalClient(
    api_token="as_your_api_key_here",
)

async def execute_with_approval():
    response = await client.request_approval_async(
        action_name="send_customer_email",
        description="Send promotional email to 10,000 customers",
        agent_id="email-agent",
        run_id="run-999",
        estimated_cost=50.0,
        timeout_seconds=3600,  # 1 hour
    )

    if response.status == ApprovalStatus.APPROVED:
        await send_bulk_email(...)

client.cancel_approval(approval_id="approval-123")

from agent_sentinel import PolicyConfig, PolicyEngine

PolicyEngine._config = PolicyConfig(
    require_approval=True,
    approval_actions=["delete_user", "modify_prices", "transfer_funds"],
    approval_threshold_usd=100.0,  # Auto-require approval if cost > $100
    approval_timeout_seconds=600,
)
PolicyEngine._initialized = True

GET /api/v1/approvals/stats

from agent_sentinel import HumanApprovalHandler, ApprovalRequest, ApprovalResponse, ApprovalStatus

def my_approval_handler(request: ApprovalRequest) -> ApprovalResponse:
    # Send to Slack, custom UI, etc.
    slack_response = send_slack_approval_request(
        channel="#approvals",
        action=request.action_name,
        description=request.description,
        cost=request.estimated_cost,
    )

    # Wait for human response (blocking)
    decision = wait_for_slack_response(slack_response.ts)

    return ApprovalResponse(
        request_id=request.request_id,
        status=ApprovalStatus.APPROVED if decision.approved else ApprovalStatus.REJECTED,
        decided_by_email=decision.approver_email,
        decision_notes=decision.notes,
    )

# Register handler
HumanApprovalHandler.set_approval_handler(my_approval_handler)