Python SDK

import asyncio
import json
import os

from cobo_agentic_wallet.client import WalletAPIClient

CHAIN_ID = "SETH"
TOKEN_ID = "SETH"
DESTINATION = "0x1111111111111111111111111111111111111111"
ALLOWED_AMOUNT = "0.001"
DENIED_AMOUNT = "0.005"
DENY_THRESHOLD = "0.002"


def parse_api_error(exc: Exception) -> dict:
    body = getattr(exc, "body", None)
    if not body:
        return {}
    try:
        payload = json.loads(body)
    except (TypeError, ValueError):
        return {}
    return payload if isinstance(payload, dict) else {}


async def main() -> None:
    api_url = os.environ["AGENT_WALLET_API_URL"]
    api_key = os.environ["AGENT_WALLET_API_KEY"]
    wallet_id = os.environ["AGENT_WALLET_WALLET_ID"]

    client = WalletAPIClient(base_url=api_url, api_key=api_key)

    try:
        # Step 1: Submit a pact requesting transfer permissions for 24 hours.
        pact_resp = await client.submit_pact(
            wallet_id=wallet_id,
            intent="Transfer tokens for integration testing",
            spec={
                "policies": [
                    {
                        "name": "max-tx-limit",
                        "type": "transfer",
                        "rules": {
                            "effect": "allow",
                            "when": {"chain_in": [CHAIN_ID], "token_in": [{"chain_id": CHAIN_ID, "token_id": TOKEN_ID}]},
                            "deny_if": {"amount_gt": DENY_THRESHOLD},
                        },
                    }
                ],
                "completion_conditions": [
                    {"type": "time_elapsed", "threshold": "86400"}
                ],
            },
        )
        pact_id = pact_resp["pact_id"]
        print(f"PACT SUBMITTED: {pact_id}")

        # Step 2: Poll until the owner approves the pact.
        print("Waiting for owner approval in Cobo Agentic Wallet app...")
        while True:
            pact = await client.get_pact(pact_id)
            status = pact.get("status", "")
            print(f"  pact status: {status}")
            if status == "active":
                break
            if status in ("rejected", "expired", "revoked", "completed"):
                raise RuntimeError(f"Pact reached terminal status before use: {status}")
            await asyncio.sleep(5)

        print("PACT ACTIVE")

        # Step 3: Use the pact-scoped API key for all subsequent calls.
        pact_api_key = pact["api_key"]
        pact_client = WalletAPIClient(base_url=api_url, api_key=pact_api_key)

        try:
            # Step 4: Execute an allowed transfer (within the deny threshold).
            allowed = await pact_client.transfer_tokens(
                wallet_id,
                chain_id=CHAIN_ID,
                dst_addr=DESTINATION,
                token_id=TOKEN_ID,
                amount=ALLOWED_AMOUNT,
            )
            print("ALLOWED:", allowed.get("status", "unknown"))

            # Step 5: Trigger a policy denial (amount exceeds the deny threshold).
            try:
                await pact_client.transfer_tokens(
                    wallet_id,
                    chain_id=CHAIN_ID,
                    dst_addr=DESTINATION,
                    token_id=TOKEN_ID,
                    amount=DENIED_AMOUNT,
                )
            except Exception as exc:
                payload = parse_api_error(exc)
                err = payload.get("error") or {}
                print(f"DENIED: code={err.get('code', '-')} reason={err.get('reason', '-')}")
                suggestion = payload.get("suggestion")
                if suggestion:
                    print(f"SUGGESTION: {suggestion}")

        finally:
            await pact_client.close()

        # Step 6: Verify allowed and denied events in audit logs.
        logs = await client.list_audit_logs(wallet_id=wallet_id, limit=20)
        items = logs.get("items", []) if isinstance(logs, dict) else []
        allowed_count = sum(1 for item in items if item.get("result") == "allowed")
        denied_count = sum(1 for item in items if item.get("result") == "denied")
        print(f"AUDIT: allowed={allowed_count}, denied={denied_count}")

    finally:
        await client.close()


if __name__ == "__main__":
    asyncio.run(main())