agentagon.wallet

pip install agentagon

import os, asyncio
from agentagon.wallet import Treasury
 
async def main():
    async with Treasury.connect(api_key=os.environ["AGENTAGON_TREASURY_KEY"]) as t:
        sw = await t.create_sub_wallet(label="research-agent-1", chain="eip155:8453")
        b = await sw.balance()
        print(sw.address, b.balance_raw, "stale" if b.stale else "live")
 
asyncio.run(main())

from agentagon.wallet import create_treasury, Treasury
 
async def main():
    result = await create_treasury(
        base_url="https://treasury.agentagon.ai",
        owner_type="agent",
        owner_id="my-agent",
    )
    print("save this once:", result.api_key)
    async with Treasury.connect(api_key=result.api_key) as t:
        sw = await t.create_sub_wallet(label="default")
        print(sw.address)

class Treasury:
    @classmethod
    def connect(
        cls,
        *,
        api_key: str,
        base_url: str = "https://treasury.agentagon.ai",
        client: httpx.AsyncClient | None = None,
    ) -> "Treasury": ...
 
    async def create_sub_wallet(
        self,
        *,
        label: str,
        chain: str | None = None,
    ) -> SubWallet: ...
 
    async def get_sub_wallet(self, sub_wallet_id: str) -> SubWallet: ...
 
    async def aclose(self) -> None: ...

async with Treasury.connect(api_key="ag_pat_...") as t:
    ...
# underlying httpx.AsyncClient is closed on exit

class SubWallet:
    id: str
    chain: str
    address: str
    label: str
    status: str   # "active" | "revoked"
 
    async def balance(self) -> Balance: ...
    async def refresh(self) -> Balance: ...   # alias for balance()
    async def pay(
        self,
        *,
        to: str,
        amount_usd: str,
        capability: str | None = None,
        trace_id: str = "",
        valid_before: int = 0,
    ) -> SignedPayment: ...
    async def transfer(
        self,
        *,
        to: str,
        amount_usd: str,
        trace_id: str = "",
        valid_before: int = 0,
    ) -> SignedTransfer: ...
    async def rotate(self) -> "SubWallet": ...
    async def revoke(self) -> "SubWallet": ...

b = await sw.balance()
print(b.balance_raw, b.balance_pending_raw, b.stale, b.updated_at)

signed = await sw.pay(
    to="0xSellerAddress",
    amount_usd="0.05",
    capability="research.company",
)

signed = await sw.transfer(to="0xPayee", amount_usd="10.00")

await sw.rotate()
print("new address:", sw.address)

await sw.revoke()
print(sw.status)   # "revoked"

@dataclass(frozen=True)
class Balance:
    sub_wallet_id: str
    chain: str
    balance_raw: str           # token base units (USDC at 6 decimals: "10000" = $0.01)
    balance_pending_raw: str
    stale: bool                # True when Treasury fell back to its cache
    updated_at: str | None
 
@dataclass(frozen=True)
class SignedPayment:
    sub_wallet_id: str
    from_address: str
    to_address: str
    chain: str
    amount_raw: str
    authorization: dict[str, Any]
    signature: str
    trace_id: str
    decision_id: str
    capability: str | None
    scheme: str | None         # "exact" | "upto"
 
@dataclass(frozen=True)
class SignedTransfer:
    sub_wallet_id: str
    from_address: str
    to_address: str
    chain: str
    amount_raw: str
    authorization: dict[str, Any]
    signature: str
    trace_id: str
    decision_id: str
 
@dataclass(frozen=True)
class CreateTreasuryResult:
    treasury_id: str
    owner_type: str
    owner_id: str
    home_chain: str
    plan_tier: str
    status: str
    api_key: str               # shown once at bootstrap; save it
    api_key_id: str
    scopes: list[str]
 
class TreasuryAPIError(RuntimeError):
    status_code: int
    body: dict

async def create_treasury(
    *,
    base_url: str = "https://treasury.agentagon.ai",
    owner_type: str,                # "agent" | "user" | "service"
    owner_id: str,
    client: httpx.AsyncClient | None = None,
) -> CreateTreasuryResult: ...

import asyncio
from agentagon.wallet import Treasury, create_treasury
 
async def main():
    bootstrap = await create_treasury(
        owner_type="agent",
        owner_id="research-agent-1",
        base_url="https://treasury.agentagon.ai",
    )
    print("save this:", bootstrap.api_key)
 
    async with Treasury.connect(api_key=bootstrap.api_key) as t:
        sw = await t.create_sub_wallet(label="default", chain="eip155:8453")
        # ...top up sw.address with USDC out-of-band...
        signed = await sw.pay(
            to="0xSellerAddress",
            amount_usd="0.05",
            capability="research.company",
        )
        print("decision:", signed.decision_id)
 
asyncio.run(main())

async def rotate_with_sweep(sw):
    # 1. Drain to a custodian or your own EOA before rotating
    cold = "0xColdStorage"
    bal = await sw.balance()
    if int(bal.balance_raw) > 0:
        # Treasury policy may cap this; check the result
        await sw.transfer(to=cold, amount_usd=raw_to_usd(bal.balance_raw))
    # 2. Now rotate; the old address is orphaned with zero balance
    await sw.rotate()
    return sw
 
def raw_to_usd(raw: str, decimals: int = 6) -> str:
    units = int(raw)
    return f"{units / 10**decimals:.{decimals}f}".rstrip("0").rstrip(".")

async def fresh_balance(sw, max_age_seconds=30):
    from datetime import datetime, timezone
    b = await sw.balance()
    if not b.stale:
        return b
    if not b.updated_at:
        # No timestamp; surface the staleness without trying to age-check
        return b
    age = datetime.now(timezone.utc) - datetime.fromisoformat(b.updated_at)
    if age.total_seconds() <= max_age_seconds:
        return b
    raise RuntimeError(f"balance stale for {age.total_seconds():.0f}s, RPC outage?")

import httpx
from agentagon.wallet import Treasury
 
def handler(request: httpx.Request) -> httpx.Response:
    if request.url.path == "/v1/wallets/sw_1/balance":
        return httpx.Response(200, json={
            "sub_wallet_id": "sw_1",
            "chain": "eip155:8453",
            "balance_raw": "1000000",
            "balance_pending_raw": "0",
            "stale": False,
            "updated_at": "2026-04-29T00:00:00Z",
        })
    return httpx.Response(404)
 
transport = httpx.MockTransport(handler)
client = httpx.AsyncClient(transport=transport, base_url="http://t.test")
treasury = Treasury(api_key="ag_pat_test", base_url="http://t.test", client=client)
# tests...
await client.aclose()

async def rotate_and_resume(sw, *, to, amount_usd, capability):
    await sw.rotate()                     # mutate sw in place; new address
    # Top up sw.address with USDC out-of-band, then:
    return await sw.pay(to=to, amount_usd=amount_usd, capability=capability)

import asyncio
 
async def fan_out(sw, batch):
    return await asyncio.gather(*[
        sw.pay(to=item["to"], amount_usd=item["amount_usd"], capability=item["capability"])
        for item in batch
    ])

class TreasuryAPIError(RuntimeError):
    status_code: int
    body: dict

from agentagon.wallet import TreasuryAPIError
 
try:
    await sw.pay(to="0x...", amount_usd="100.00")
except TreasuryAPIError as exc:
    if exc.status_code == 403 and exc.body.get("reason") == "daily_cap_exceeded":
        # backoff, raise the cap, or wait for the next day
        ...
    elif exc.status_code == 402 and exc.body.get("reason") == "insufficient_balance":
        # body carries available_raw and required_raw
        topup = int(exc.body["required_raw"]) - int(exc.body["available_raw"])
        ...
    else:
        raise

sw = await treasury.get_sub_wallet("sw_xxx")
b = await sw.balance()

from agentagon.wallet import SubWallet
 
sw = SubWallet(
    treasury,
    sub_wallet_id="sw_xxx",
    chain="eip155:8453",
    address="0xabc",
    label="research-agent-1",
    status="active",
)