How I Built an Autonomous AI Agent That Earns USDC While I Sleep
Target audience: developers who are experimenting with self‑funding AI agents. The goal is to show a minimal, working prototype, not a product.
1. Why an “earning” agent?
An autonomous agent that can pay for its own compute or data needs removes a classic bottleneck: you have to fund a wallet manually before the agent can act. If the agent can receive micropayments for the services it provides, it can sustain itself as long as there is demand.
The prototype described here does three things repeatedly:
- Expose a paid HTTP endpoint (using the x402 “Payment Required” pattern).
- Perform a small unit of work when a client pays (e.g., run a lightweight inference model).
- Sweep the earned USDC to a reserve wallet so the agent can later pay for gas, storage, or external APIs.
The code is intentionally simple; it omits many production concerns (key rotation, audit logging, DoS protection) to keep the example readable.
2. High‑level architecture
+-------------------+ x402 (402) +-------------------+
| Client (curl, | <-------------------> | Agent Service |
| browser, etc.) | USDC payment header | (FastAPI + uvicorn)|
+-------------------+ +-------------------+
^ |
| v
| +-------------------+
| | Worker Process |
| | (model inference)|
| +-------------------+
| |
| v
| +-------------------+
+-------------------------------->| USDC Sweeper |
+-------------------+
(wallet → reserve)
-
Agent Service – a thin HTTP layer that checks for a valid
X-Paymentheader (the x402 spec). If the header is present and verifies, it enqueues a job. - Worker Process – pulls jobs from a Redis queue, runs the actual AI work, and writes the result to a temporary store (e.g., an S3‑compatible bucket).
- USDC Sweeper – a separate cron‑like task that reads the agent’s wallet balance, transfers any amount above a dust threshold to a reserve address, and logs the transaction.
All components run on the same cheap VPS (or a Docker Compose stack) for the prototype. In production you would split them across instances and add proper observability.
3. Wallet & key management
The agent needs an Ethereum-compatible wallet (Base is an EVM L2) to hold USDC and to sign the x402 payment verification message. For the demo we keep the private key in an environment variable; never do this in production. Use a hardware signer, AWS KMS, or a cloud‑based secret manager with rotation.
# wallet.py
import os
from eth_account import Account
from web3 import Web3
BASE_RPC = os.getenv("BASE_RPC", "https://base.mainnet.rpc.dev")
w3 = Web3(Web3.HTTPProvider(BASE_RPC))
PRIVATE_KEY = os.getenv("AGENT_PRIVATE_KEY")
if not PRIVATE_KEY:
raise RuntimeError("Set AGENT_PRIVATE_KEY in the environment")
acct = Account.from_key(PRIVATE_KEY)
ADDRESS = acct.address
print(f"Agent wallet: {ADDRESS} (balance will be checked later)")
Trade‑off: Storing the key in plain text is convenient for a local demo but exposes you to theft if the host is compromised. In a real deployment you would never export the key; instead you’d ask a signing service to sign the x402 challenge or the USDC transfer.
4. x402 payment verification
The x402 spec defines a 402 Payment Required response that includes a payment field with a signed message. The client must include a valid X-Payment header containing the signed receipt. The agent verifies the signature against the expected price and token.
# x402.py
from eth_account.messages import encode_defunct
from web3 import Web3
def verify_x402_payment(header: str, price_wei: int, token_address: str, spender: str) -> bool:
"""
header: value of the X-Payment header, expected format:
"<token_address>:<amount>:<signature>"
price_wei: expected amount in wei (USDC has 6 decimals, so 1 USDC = 1_000_000 wei)
token_address: USDC contract on Base (0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913)
spender: the agent's address that should receive the payment
"""
try:
token, amount_str, signature = header.split(":")
amount = int(amount_str)
if token.lower() != token_address.lower():
return False
if amount != price_wei:
return False
# Reconstruct the signed message: keccak256("\x19Ethereum Signed Message:\n32"<token><amount><spender>)
message = Web3.solidityKeccak(
["address", "uint256", "address"],
[token_address, amount, spender]
)
message_hex = Web3.toHex(message)
eth_message = encode_defunct(hexstr=message_hex)
recovered = Account.recover_message(eth_message, signature=signature)
return recovered.lower() == spender.lower()
except Exception:
return False
Trade‑off: The verification is cheap (a single ECDSA recover) but it assumes the client follows the exact header format. Malformed headers are rejected with a 400 error, which is fine for a low‑traffic prototype. In a high‑throughput service you would want to rate‑limit verification attempts to prevent DoS via bogus signatures.
5. The HTTP endpoint (FastAPI)
# main.py
import os
import json
from fastapi import FastAPI, Header, HTTPException, BackgroundTasks
from redis import Redis
from rq import Queue
from wallet import ADDRESS
from x402 import verify_x402_payment
app = FastAPI()
redis = Redis.from_url(os.getenv("REDIS_URL", "redis://localhost:6379"))
task_queue = Queue("agent-jobs", connection=redis)
USDC_ADDRESS = "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913"
PRICE_USDC = 0.01 # $0.01 per call
PRICE_WEI = int(PRICE_USDC * 1_000_000) # USDC has 6 decimals
@app.post("/infer")
async def infer(
background_tasks: BackgroundTasks,
x_payment: str = Header(None),
prompt: str = Body(..., embed=True)
):
if not x_payment:
raise HTTPException(status_code=402, detail="Payment required")
if not verify_x402_payment(x_payment, PRICE_WEI, USDC_ADDRESS, ADDRESS):
raise HTTPException(status_code=402, detail="Invalid payment")
# Enqueue the job; the actual work happens in a worker process.
job = task_queue.enqueue(
"worker.run_inference",
prompt,
job_timeout=120,
)
background_tasks.add_task(lambda: None) # placeholder for any post‑job cleanup
return {"job_id": job.get_id(), "status": "queued"}
Trade‑off: Using RQ (Redis Queue) adds a dependency on Redis and a separate worker process. For a toy service you could call the model directly in the request handler, but that would block the HTTP thread and make latency spikes visible to the client. Offloading to a worker keeps the API responsive, at the cost of extra operational complexity.
6. Worker – doing the real AI work
The worker runs a very small model (e.g., a distilled GPT‑2) hosted locally. In a real scenario you might call an external API (OpenAI, Hugging Face Inference Endpoints) and pay for that with the USDC you just earned.
python
# worker.py
import os
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from redis import Redis
from rq import Worker, Queue
from wallet import w3, acct, ADDRESS
from web3 import Web3
# Load a tiny model once at worker startup
MODEL_NAME = os.getenv("HF_MODEL", "sshleifer/tiny-gpt2")
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
model = AutoModelForCausalLM.from_pretrained(MODEL_NAME)
redis = Redis.from_url(os.getenv("REDIS_URL", "redis://localhost:6379"))
queue = Queue("agent-jobs", connection=redis)
def run_inference(prompt: str) -> str:
inputs = tokenizer(prompt, return_tensors="pt")
output = model.generate(**inputs, max_new_tokens=50)
text = tokenizer.decode(output[0], skip_special_tokens=True)
return text
if __name__ == "__main__":
# Start the worker – this process blocks waiting for jobs.
w = Worker([queue],
Top comments (0)