Audit and activity logs

Query the audit trail and subscribe to real-time events.

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

CLI

# 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.

Python SDK

from cobo_agentic_wallet import WalletAPIClient

owner_client = WalletAPIClient(base_url=API_URL, api_key=OWNER_API_KEY)

logs = await owner_client.list_audit_logs(
    wallet_id=WALLET_UUID,           # optional: filter by wallet
    action="transfer.initiate",      # optional: filter by action
    result="denied",                 # optional: "allowed", "denied", "pending_approval"
    principal_id=AGENT_ID,           # optional: filter by who acted
    limit=50,
)

for entry in logs.get("items", []):
    print(
        entry["created_at"],
        entry["action"],
        entry["result"],
        entry.get("authz_details", {}),
    )

TypeScript SDK

const logs = (await auditApi.listAuditLogs(
  walletId,          // wallet_id
  agentId,           // principal_id
  'transfer.initiate', // action
  'denied',          // result
  undefined, undefined, undefined, undefined, undefined,
  50,                // limit
)).data.result;

for (const entry of logs.items ?? []) {
  console.log(
    entry.created_at,
    entry.action,
    entry.result,
    entry.authz_details,
  );
}

Filter options

Parameter	Description
wallet_id	Filter to one wallet
action	Filter by action type (see table below)
result	allowed, denied, or pending_approval
principal_id	Filter by principal who initiated the action
limit	Maximum number of results to return

Action types

Action	When it fires
transfer.initiate	Agent submits a transfer
contract_call.initiate	Agent submits a contract call
approval.requested	Transfer/call enters pending state
approval.approved	Owner approves a pending operation
approval.rejected	Owner rejects a pending operation
approval.executed	Approved operation executes on-chain
approval.expired	Pending operation expires without action
delegation.freeze	Owner freezes a delegation
delegation.unfreeze	Owner 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 type	Fired when	Key payload fields
approval.requested	Agent operation enters pending state	resource_id (pending_operation_id), details.amount
transaction.status_changed	Transaction status updates	resource_id (cobo_transaction_id), details.status
policy.violated	Agent operation is denied	resource_id (wallet_id), details.denial_code
delegation.created	New delegation is created	resource_id (delegation_id)
delegation.revoked	Delegation is revoked	resource_id (delegation_id)
heartbeat	Keep-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:

Python SDK

logs = await owner_client.list_audit_logs(
    wallet_id=WALLET_UUID,
    action="transfer.initiate",
    result="denied",
    limit=50,
)

TypeScript SDK

const logs = (await auditApi.listAuditLogs(
  walletId, undefined, 'transfer.initiate', 'denied',
  undefined, undefined, undefined, undefined, undefined, 50,
)).data.result;

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")