Donner à un agent un portefeuille et un budget

pip install "x402[evm]" eth-account requests

import base64
import json
import os
import secrets
from datetime import datetime, timedelta, timezone

import requests
from eth_account import Account
from eth_account.messages import encode_defunct

BASE_URL = "https://api.venice.ai/api/v1"
DOMAIN = "api.venice.ai"
CHAIN_ID = 8453          # Base mainnet
MODEL = "qwen3-5-9b"
BUDGET_USD = 5.00

key = os.environ.get("WALLET_KEY")
account = Account.from_key(key) if key else Account.create()

print(f"wallet {account.address}")
print("funded" if key else "disposable, cannot pay yet")

def siwx_header():
    now = datetime.now(timezone.utc)
    stamp = lambda t: t.isoformat(timespec="milliseconds").replace("+00:00", "Z")
    issued_at, expires_at = stamp(now), stamp(now + timedelta(minutes=4))
    nonce = secrets.token_hex(8)

    message = (
        f"{DOMAIN} wants you to sign in with your Ethereum account:\n"
        f"{account.address}\n\nSign in to Venice AI\n\n"
        f"URI: https://{DOMAIN}\nVersion: 1\nChain ID: {CHAIN_ID}\n"
        f"Nonce: {nonce}\nIssued At: {issued_at}\nExpiration Time: {expires_at}"
    )
    signature = account.sign_message(encode_defunct(text=message)).signature.hex()

    payload = {
        "address": account.address,
        "message": message,
        "signature": signature if signature.startswith("0x") else "0x" + signature,
        "chainId": CHAIN_ID,
    }
    return base64.b64encode(json.dumps(payload).encode()).decode()

def wallet_get(path, **params):
    response = requests.get(
        f"{BASE_URL}{path}",
        headers={"SIGN-IN-WITH-X": siwx_header()},
        params=params,
        timeout=30,
    )
    response.raise_for_status()
    return response.json()["data"]


def balance():
    return wallet_get(f"/x402/balance/{account.address}")


print(balance())

{
  "walletAddress": "0xc5048ea84939bb7eb4c611b88ea17d5ee4f11a0c",
  "balanceUsd": 0,
  "canConsume": false,
  "minimumTopUpUsd": 5,
  "suggestedTopUpUsd": 10
}

from x402.client import SpendControls, x402ClientSync
from x402.http import PAYMENT_SIGNATURE_HEADER, encode_payment_signature_header
from x402.mechanisms.evm import EthAccountSigner
from x402.mechanisms.evm.exact.client import ExactEvmScheme
from x402.schemas.payments import PaymentRequired


def top_up():
    discovery = requests.post(f"{BASE_URL}/x402/top-up", timeout=30)
    required = PaymentRequired.model_validate(discovery.json())

    rail = next(a for a in required.accepts if a.network.startswith("eip155"))
    print(f"{rail.network}: {int(rail.amount) / 1e6:.2f} USDC to {rail.payTo}")

    client = x402ClientSync()
    client.register(rail.network, ExactEvmScheme(EthAccountSigner(account)))
    # The SDK caps one payment at $1 by default, which is below the $5 minimum
    # top-up, so every rail gets rejected until this is raised.
    client.set_spend_controls(SpendControls(max_amount_per_payment="$5", allowed_assets=True))

    payload = client.create_payment_payload(required)
    settlement = requests.post(
        f"{BASE_URL}/x402/top-up",
        headers={PAYMENT_SIGNATURE_HEADER: encode_payment_signature_header(payload)},
        timeout=90,
    )
    return settlement.json()

{
  "x402Version": 2,
  "accepts": [
    {
      "scheme": "exact",
      "network": "eip155:8453",
      "amount": "5000000",
      "asset": "0x833589fcd6edb6e08f4c7c32d4f71b54bda02913",
      "payTo": "0x2670b922ef37c7df47158725c0cc407b5382293f",
      "maxTimeoutSeconds": 300,
      "extra": { "name": "USD Coin", "version": "2" }
    },
    {
      "scheme": "exact",
      "network": "solana:5eykt4UsFv8P8NJdTREpY1vzqKqZKvdp",
      "amount": "5000000",
      "asset": "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v",
      "payTo": "8qUL23aSj7mDWdoLMXGHFvnVCT9wd7jXcysiekroADEL",
      "maxTimeoutSeconds": 300,
      "extra": { "name": "USD Coin", "version": "2", "feePayer": "BFK9TLC3edb13K6v4YyH3DwPb5DSUpkWvb7XnqCL9b4F" }
    }
  ]
}

def ask(question):
    response = requests.post(
        f"{BASE_URL}/chat/completions",
        headers={"SIGN-IN-WITH-X": siwx_header(), "Content-Type": "application/json"},
        json={
            "model": MODEL,
            "messages": [{"role": "user", "content": question}],
            "max_completion_tokens": 150,
            "venice_parameters": {
                "include_venice_system_prompt": False,
                "disable_thinking": True,
            },
        },
        timeout=90,
    )
    if response.status_code == 402:
        body = response.json()
        raise RuntimeError(
            f"balance ${body.get('currentBalanceUsd', 0)} is under the "
            f"${body.get('minimumBalanceUsd')} floor. Minimum top-up is "
            f"${body['topUpInstructions']['minimumAmountUsd']}."
        )
    response.raise_for_status()
    return response.json()["choices"][0]["message"]["content"].strip()

def charges():
    """Every debit against this wallet, newest first."""
    ledger = wallet_get(f"/x402/transactions/{account.address}", limit=100)
    return [t for t in ledger["transactions"] if t["type"] == "CHARGE"]

{
  "id": "ledger_01H...",
  "amount": -0.0000066,
  "balanceAfter": 4.9999934,
  "type": "CHARGE",
  "createdAt": "2026-08-21T18:22:10.000Z",
  "requestId": "chatcmpl-...",
  "modelId": "qwen3-5-9b"
}

TASKS = [
    "Name one concrete tradeoff of vector search versus keyword search. One sentence.",
    "In one sentence, when is a bloom filter the wrong choice?",
    "Give one reason CRDTs are hard to debug in production. One sentence.",
    "What is one failure mode of exponential backoff without jitter? One sentence.",
    "Name one thing consistent hashing does not solve. One sentence.",
    "Why is p99 latency more useful than the mean? One sentence.",
]


def run(budget=BUDGET_USD):
    opening = balance()
    print(f"balance ${opening['balanceUsd']:.4f}, budget ${budget:.4f}")

    if not opening["canConsume"]:
        print(f"cannot transact yet, minimum top-up is ${opening['minimumTopUpUsd']}")
        return

    baseline = sum(abs(c["amount"]) for c in charges())
    spent = 0.0

    for number, task in enumerate(TASKS, 1):
        if spent >= budget:
            print(f"\nstopped before task {number}: ${spent:.6f} of ${budget:.4f} spent")
            return

        answer = ask(task)
        spent = sum(abs(c["amount"]) for c in charges()) - baseline
        print(f"\n{number}. {task}")
        print(f"   {answer}")
        print(f"   ${spent:.6f} spent, ${budget - spent:.6f} left")

    print(f"\nfinished all {len(TASKS)} tasks for ${spent:.6f}")
    if spent:
        print(f"at this rate ${budget:.2f} covers about {int(budget / (spent / len(TASKS))):,} calls")

run()              # $5.00, finishes every task
run(budget=1e-5)   # stops partway, having spent about $0.000007 per call