agentagon.buyer

Self-custody buyer client. Picks the right wallet for an agent, signs the x402 EIP-3009 USDC TransferWithAuthorization, and runs the full probe -> 402 -> sign -> retry dance against any compliant x402 seller.

Overview

Use agentagon.buyer.Agent when you want to:

• Call paid x402 endpoints from a Python program with a CLI-managed local keystore
• Cap spending per-call (max_price_usd)
• Get the actual response body back, with metadata about what was paid
• Stay self-custody: the seed lives on your machine, encrypted by your passphrase

It’s not the right tool when you want to:

• Pay through a managed Treasury (Gate or KMS-managed seed): use agentagon.sdk.Client.pay or agentagon.wallet.Treasury
• Drive payments interactively via a terminal: use agentagon-cli buy
• Hook payment-on-tool-call into Claude / Cursor: use the buyer MCP server

Install

pip install agentagon
pip install agentagon-cli   # the buyer client reads keystores written by the CLI

agentagon-cli is required because the keystore decryption code lives there. The buyer module imports it lazily on the first sign attempt; you only need it if buy() actually has to sign (free endpoints work without it).

Quickstart

The CLI sets up the keystore once:

agentagon init                      # generates seed, derives Base + Solana addresses
agentagon fund --testnet            # opens a faucet to top up the new wallet

Then in code:

from agentagon.buyer import Agent
 
agent = Agent.from_local_active(passphrase="your-keystore-passphrase")
 
result = agent.buy(
    "https://demo-weather.agentagon.ai/v1/weather/sf",
    max_price_usd="0.10",
)
print(result.status_code, result.json())
print(f"spent ${result.spent_usd}")

That’s it. Probe, 402, sign, retry, return the response.

API reference

Agent

class Agent:
    @classmethod
    def from_local_active(
        cls,
        *,
        passphrase: str,
        api_url: str | None = None,
    ) -> "Agent": ...
 
    @classmethod
    def from_local(
        cls,
        *,
        agent_id: str,
        passphrase: str,
        api_url: str | None = None,
    ) -> "Agent": ...
 
    def buy(
        self,
        url: str,
        *,
        method: str = "GET",
        body: bytes | str | dict | None = None,
        headers: dict[str, str] | None = None,
        max_price_usd: str = "0.10",
        timeout: float = 30.0,
    ) -> BuyResult: ...
 
    async def abuy(self, ...) -> BuyResult: ...   # async variant; same args

Agent.from_local_active

agent = Agent.from_local_active(passphrase="...")

Reads $AGENTAGON_ACTIVE_AGENT_PATH (default ~/.agentagon/active), then loads the matching agent config from $AGENTAGON_AGENTS_DIR (default ~/.agentagon/agents/). Raises AgentNotFoundError if either file is missing, with a hint to run agentagon init.

Agent.from_local

agent = Agent.from_local(agent_id="agt_xxx", passphrase="...")

Loads a specific agent by id. Raises AgentNotFoundError with a hint to run agentagon agent list if the config doesn’t exist.

The seed decryption is lazy: the keystore is only touched if buy() has to sign (i.e. the seller returned a 402). Free endpoints get probed and returned without ever needing the passphrase to be correct.

buy(url, *, method="GET", body=None, headers=None, max_price_usd="0.10", timeout=30.0)

Runs the full x402 dance:

1. Probe: send the original request with no X-PAYMENT header
2. If the response is anything but 402, return it (free endpoint or unrelated error)
3. If it’s a 402, parse X-PAYMENT-REQUIRED, pick a compatible accept entry, and check the requested price against max_price_usd
4. Sign an EIP-3009 TransferWithAuthorization, base64 it into X-PAYMENT, and re-issue the original request with that header
5. Return the seller’s response

Returns BuyResult regardless of status code; HTTP errors from the retry come back as PaymentRejectedError. Network errors on the probe come back as SellerError.

BuyResult

@dataclass
class BuyResult:
    status_code: int
    headers: dict[str, str]
    content: bytes
    spent_usd: str            # "0.00" if no payment was made
    tx_hash: str              # "" if no payment
    scheme: str               # "" | "exact" | "upto"
 
    def json(self) -> Any: ...   # parses content as JSON
    def text(self) -> str: ...   # decodes content as UTF-8

spent_usd and tx_hash are populated only when a payment was actually made (i.e. the seller responded 402, the agent signed, and the retry came back 2xx). For free endpoints spent_usd == "0.00" and tx_hash == "".

Cookbook

Pay an x402 endpoint with a budget cap

from agentagon.buyer import Agent, InsufficientBudgetError
 
agent = Agent.from_local_active(passphrase="...")
 
try:
    result = agent.buy(
        "https://api.example.com/expensive-thing",
        max_price_usd="0.05",
    )
except InsufficientBudgetError as exc:
    print(f"seller wanted ${exc.requested_usd}, our cap was ${exc.cap_usd}")
    print("raise --max if you want to proceed")

The agent never signs above max_price_usd. Surface the cap to the user; don’t silently raise it.

POST with a JSON body

result = agent.buy(
    "https://api.example.com/v1/translate",
    method="POST",
    body={"text": "hello", "target": "ja"},
    headers={"Content-Type": "application/json"},
    max_price_usd="0.02",
)
print(result.json())

body accepts dict (JSON-encoded automatically), str (sent as-is), or bytes. Custom headers are merged onto both the probe and the retry.

Async usage

import asyncio
from agentagon.buyer import Agent
 
async def main():
    agent = Agent.from_local_active(passphrase="...")
    result = await agent.abuy(
        "https://api.example.com/data",
        max_price_usd="0.10",
    )
    print(result.json())
 
asyncio.run(main())

abuy() mirrors buy() exactly; use it inside async event loops.

Switch agents within a script

from agentagon.buyer import Agent
 
bot = Agent.from_local(agent_id="agt_research_bot", passphrase=os.environ["BOT_PASS"])
finance = Agent.from_local(agent_id="agt_finance_bot", passphrase=os.environ["FIN_PASS"])
 
bot.buy("https://research.example.com/...", max_price_usd="0.05")
finance.buy("https://markets.example.com/...", max_price_usd="2.00")

Each agent has its own seed, address, and PAT; the buy() calls don’t share state.

Handle a free endpoint cleanly

result = agent.buy("https://api.example.com/health")
if result.spent_usd == "0.00":
    print("free endpoint, response:", result.text())
else:
    print(f"paid ${result.spent_usd}, tx: {result.tx_hash}")

The same buy() call works whether the endpoint is paid or free; check spent_usd to know.

Cap budget by capability with the CLI

agentagon pat create \
  --label research-bot \
  --capability research.* \
  --daily-cap-usd 5.00

The CLI mints a PAT with capability and daily-cap policy applied at the platform level. The buyer agent sees the same cap as daily_cap_usd on whoami, and the platform refuses to mint signatures past it (returning MGMT_6014 with a raise_pat_cap recovery). See Authentication.

Diagnose a no-compatible-accept failure

from agentagon.buyer import Agent, NoCompatibleAcceptError
 
agent = Agent.from_local_active(passphrase="...")
try:
    agent.buy("https://solana-only-seller.example.com/...", max_price_usd="0.10")
except NoCompatibleAcceptError as exc:
    print("seller offered no accepts our agent can speak")
    print("raw 402 challenge:", exc.challenge)

exc.challenge is the parsed 402 body. Inspect accepts[] to see what the seller does support; if it’s only Solana and your agent is Base-only, you’ll need a Solana-derived agent (use agentagon agent wallets add --chain solana).

Inspect what was signed

result = agent.buy("https://...", max_price_usd="0.10")
if result.tx_hash:
    print("scheme:", result.scheme)         # "exact" or "upto"
    print("amount:", result.spent_usd)
    print("tx:    ", result.tx_hash)

For audit logs or local accounting, persist (url, scheme, spent_usd, tx_hash) per call.

Errors

The buyer raises a typed exception for every failure mode. They all inherit from BuyerError.

from agentagon.buyer import (
    BuyerError,
    AgentNotFoundError,
    NoCompatibleAcceptError,
    InsufficientBudgetError,
    PaymentRejectedError,
    SellerError,
)

Catalog

Catch the most specific class you care about and let the rest propagate. Catching BuyerError works as a fallback.

Troubleshooting

AgentNotFoundError: no active agent on disk

You haven’t run agentagon init yet, or the active pointer was deleted. Run:

agentagon init
# or, if you already have an agent:
agentagon agent switch <agent_id>

AgentNotFoundError: agent has no seed_id

The agent config exists but is a managed agent (server-side custody). agentagon-buyer v1 only supports self-custody. Either create a self-custody agent (agentagon agent wallets add --new-seed) or use agentagon.sdk.Client.pay for the managed path.

RuntimeError: agentagon-cli is required to decrypt the local keystore

The keystore decryption code lives in agentagon-cli, not agentagon-buyer. Install it:

pip install agentagon-cli

This is only required at sign time. Free endpoints work without it.

Wrong passphrase

The keystore decryption raises a typed exception from agentagon.cli.keystore; it propagates up through buy() as that exception class. Check agentagon agent show to confirm the active agent, and double-check the passphrase against the one used in agentagon init.

NoCompatibleAcceptError

Either the seller advertised only schemes / networks your agent doesn’t speak, or no accepts at all. Inspect exc.challenge.accepts to see what the seller offered. Common causes:

• Seller is Solana-only, agent is Base-only: add a Solana wallet (agentagon agent wallets add --chain solana)
• Seller advertises upto only, agent doesn’t have a Permit2 allowance: have the seller add exact to its schemes, or run scripts/permit2_approve.py once for your wallet

PaymentRejectedError after a successful sign

The signature was valid (the facilitator accepted it on the seller side), but the seller returned a non-2xx anyway. Check exc.status_code and exc.body for the seller’s reason. Common causes: the seller hit an internal error after settlement, or the request body changed between probe and retry.

SellerError on probe

DNS, TLS, or network error before any payment was attempted. The seller’s URL is wrong or unreachable. No payment was made; no tx_hash to worry about.

How does this differ from agentagon.sdk.Client.pay?

The buyer client is end-to-end (probe-to-response). Client.pay is just the signing step; you do the x402 dance yourself if you want raw control.