Query audit logs
- CLI
- Python SDK
- TypeScript SDK
# 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.
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", {}),
)
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
- TypeScript SDK
logs = await owner_client.list_audit_logs(
wallet_id=WALLET_UUID,
action="transfer.initiate",
result="denied",
limit=50,
)
const logs = (await auditApi.listAuditLogs(
walletId, undefined, 'transfer.initiate', 'denied',
undefined, undefined, undefined, undefined, undefined, 50,
)).data.result;
# 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"])
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")