Skip to main content
Every decision made by the Policy Engine — allowed, denied, or pending approval — is recorded in the audit log with full structured context. The audit trail is append-only and immutable.

Query audit logs

# Audit logs are not available as a standalone CLI command.
# Use the Python SDK or query the API directly via caw fetch.
# See the Python SDK tab for the equivalent query.

Filter options

ParameterDescription
wallet_idFilter to one wallet
actionFilter by action type (see table below)
resultallowed, denied, or pending_approval
principal_idFilter by principal who initiated the action
limitMaximum number of results to return

Action types

ActionWhen it fires
transfer.initiateAgent submits a transfer
contract_call.initiateAgent submits a contract call
approval.requestedTransfer/call enters pending state
approval.approvedOwner approves a pending operation
approval.rejectedOwner rejects a pending operation
approval.executedApproved operation executes on-chain
approval.expiredPending operation expires without action
delegation.freezeOwner freezes a delegation
delegation.unfreezeOwner unfreezes a delegation

Audit entry structure

{
  "id": "...",
  "created_at": "2026-02-24T10:01:00Z",
  "principal_id": "agent-uuid",
  "principal_type": "agent",
  "wallet_id": "wallet-uuid",
  "action": "transfer.initiate",
  "result": "denied",
  "authz_details": {
    "denial_code": "TRANSFER_LIMIT_EXCEEDED",
    "policy_id": "policy-uuid",
    "limit_value": "100",
    "requested_amount": "500"
  },
  "request": {
    "chain_id": "SETH",
    "token_id": "SETH_USDC",
    "amount": "500",
    "dst_addr": "0x..."
  },
  "error": null
}

Real-time events (SSE)

Subscribe to the event stream for real-time push of all decisions affecting your wallets:
import httpx
import json

async def stream_events():
    headers = {
        "X-API-Key": OWNER_API_KEY,
        "Accept": "text/event-stream",
    }
    async with httpx.AsyncClient(timeout=None) as client:
        async with client.stream(
            "GET", f"{API_URL}/api/v1/events/stream", headers=headers
        ) as response:
            current_event_type = None
            async for line in response.aiter_lines():
                if line.startswith("event:"):
                    current_event_type = line[7:].strip()
                elif line.startswith("data:"):
                    data = json.loads(line[5:].strip())
                    await handle_event(current_event_type, data)
                elif line == "":
                    current_event_type = None  # reset between events

Event types

Event typeFired whenKey payload fields
approval.requestedAgent operation enters pending stateresource_id (pending_operation_id), details.amount
transaction.status_changedTransaction status updatesresource_id (cobo_transaction_id), details.status
policy.violatedAgent operation is deniedresource_id (wallet_id), details.denial_code
delegation.createdNew delegation is createdresource_id (delegation_id)
delegation.revokedDelegation is revokedresource_id (delegation_id)
heartbeatKeep-alive (every 30s)timestamp

Scope

The SSE stream is scoped to the authenticated principal:
  • Owners receive events for all wallets they own
  • Operators receive events for wallets they have active delegations on

Common audit queries

Show all denied transfers today:
logs = await owner_client.list_audit_logs(
    wallet_id=WALLET_UUID,
    action="transfer.initiate",
    result="denied",
    limit=50,
)
Show approval lifecycle for a specific operation:
# Query by action prefix to get the full lifecycle
for action in ["approval.requested", "approval.approved", "approval.executed"]:
    logs = await owner_client.list_audit_logs(action=action, limit=10)
    for entry in logs.get("items", []):
        if entry.get("authz_details", {}).get("pending_operation_id") == TARGET_ID:
            print(action, entry["created_at"], entry["result"])
Verify self-correction (deny → retry → allow):
logs = await owner_client.list_audit_logs(
    wallet_id=WALLET_UUID,
    action="transfer.initiate",
    limit=20,
)
allowed = sum(1 for e in logs["items"] if e["result"] == "allowed")
denied = sum(1 for e in logs["items"] if e["result"] == "denied")
print(f"Transfers: {allowed} allowed, {denied} denied")