How I Built an Autonomous AI Agent That Earns USDC While I Sleep

How I Built an Autonomous AI Agent That Earns USDC While I Sleep Target audience: developers who are designing or running autonomous AI agents and want to understand the practical trade‑offs of integrating LLM‑based planning, tool use, and on‑chain payments. Table of Contents Introduction System Architecture Overview Choosing and Preparing the LLM The Planner Loop Tool Integration and the Executor x402‑Based Payment Flow Scheduler and “Sleep” Mode Observability, Logging, and Alerting Cost Model and Profitability Benchmarks Honest Trade‑offs and Lessons Learned Conclusion Live Example Introduction The idea of an agent that can perform useful work, receive payment in a programmable cryptocurrency, and then idle until the next opportunity appears is attractive for a variety of side‑hustle‑style applications: micro‑task crowdsourcing, data labeling, simple content generation, or lightweight API wrappers. In this article I walk through a concrete implementation that runs on a modest cloud VM (2 vCPU, 8 GiB RAM) and earns USDC on the Base L2 while the host machine is otherwise idle. The goal is not to showcase a “breakthrough” but to illustrate the engineering decisions, measured performance numbers, and safety considerations that arise when you stitch together an LLM planner, a set of deterministic tools, and the x402 payment standard. All code snippets are functional as of Python 3.11, using openly available libraries (transformers, bitsandbytes, web3.py, fastapi, APScheduler). Feel free to copy, adapt, or replace components with alternatives that better suit your latency or cost constraints. System Architecture Overview (Diagram omitted for brevity; see description below.) The agent consists of four loosely coupled layers: Planner – an LLM that receives a natural‑language goal, decomposes it into sub‑steps, and selects which tool to invoke next. Executor – a thin wrapper that calls the chosen tool (e.g., an HTTP request, a local ML model, or a blockchain read) and returns a structured result. Payment Handler – monitors an Ethereum‑compatible address for incoming USDC transfers that conform to the x402 spec, validates the payload, and credits the agent’s internal ledger. Scheduler/Sleep Loop – runs the planner only when a task is present; otherwise the process enters a low‑power wait state (using APScheduler or a simple time.sleep loop). Communication between layers is via plain Python objects (no RPC) to keep latency low and to avoid extra failure points. The entire process runs as a single long‑lived Python service; containerisation (Docker) is optional but recommended for reproducible deploys. Choosing and Preparing the LLM 1.1 Model selection criteria Criterion Reasoning Chosen option Inference cost per token Directly impacts profit margin; we target Dict[str, Any]: url = params.get("url") if not url: raise ValueError("Missing 'url'") timeout = params.get("timeout", 10) try: resp = requests.get(url, timeout=timeout) resp.raise_for_status() return { "status_code": resp.status_code, "headers": dict(resp.headers), "body": resp.text[:2000], # truncate to avoid huge payloads } except requests.RequestException as e: return {"error": str(e)} 2.2 Example: Sentiment analysis micro‑tool For demonstration we wrap a tiny DistilBERT sentiment model (≈250 MB) that runs on CPU. The model is loaded once at startup. from transformers import AutoModelForSequenceClassification, AutoTokenizer import torch SENT_MODEL_ID = "distilbert-base-uncased-finetuned-sst-2-english" _sent_tokenizer = AutoTokenizer.from_pretrained(SENT_MODEL_ID) _sent_model = AutoModelForSequenceClassification.from_pretrained(SENT_MODEL_ID) _sent_model.eval() def tool_sentiment(params: Dict[str, Any]) -> Dict[str, Any]: text = params.get("text", "") if not text: return {"error": "empty text"} inputs = _sent_tokenizer(text, return_tensors="pt", truncation=True, max_length=128) with torch.no_grad(): logits = _sent_model(**inputs).logits probs = torch.softmax(logits, dim=-1).squeeze().tolist() label_id = int(torch.argmax(logits, dim=-1)) label = _sent_model.config.id2label[label_id] return {"label": label, "score": round(probs[label_id], 4), "probabilities": probs} 2.3 Tool registry and safety wrapper Each tool is wrapped with a generic safety layer that enforces: Input schema validation (using pydantic or simple manual checks). Rate limiting (max N calls per minute per tool). Exception isolation (any uncaught error is returned as an error dict rather than crashing the agent). from functools import wraps import time from collections import defaultdict call_log = defaultdict(list) def rate_limited(max_per_minute: int): def decorator(func): @wraps(func) def wrapper(*args, **kwargs): now = time.time() call_log[func.__name__] = [t for t in call_log[func.__name__] if now - t < 60] if len(call_log[func.__name__]) >= max_per_minute: return {"error": f"rate limit exceeded for {func.__name__}"} call_log[func.__name__].append(now) return func(*args, **kwargs) return wrapper return decorator # Apply to each tool tool_http_get = rate_limited(30)(tool_http_get) tool_sentiment = rate_limited(60)(tool_sentiment) These safeguards keep the agent from accidentally exhausting external APIs or draining its own compute budget. x402‑Based Payment Flow The x402 standard (https://github.com/x402/x402) defines a HTTP 402 Payment Required response that includes a JSON payload describing how to pay for a resource. Our agent implements the client side: it watches an Ethereum‑compatible address for incoming USDC transfers that contain a valid x402 payment reference, validates the payment, and then marks the associated task as paid. 3.1 Payment reference format When a client wants to purchase a unit of work from the agent, they first issue a GET request to the agent’s endpoint (e.g., /task?type=sentiment). The agent returns: HTTP/1.1 402 Payment Required Content-Type: application/json { "scheme": "exact", "network": "base", "token": "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02FF9", // USDC on Base "resource": "/task?type=sentiment", "maxTimeout": 86400, "metadata": { "id": "task-2024-09-24-001" } } The client then constructs an ERC‑20 transfer transaction to the agent’s address, adding the metadata.id as the transaction’s data field (hex‑encoded UTF‑8). The agent monitors its address for inbound transfers, extracts the data field, and checks that: The transferred token contract matches the expected USDC address on Base. The amount is ≥ the price specified in the 402 response (we use a fixed price of $0.02 per sentiment call). The data field decodes to a known task ID that is currently pending. If all checks pass, the agent credits the task as “paid” and allows the planner to proceed with execution. 3.2 Code: payment watcher We use web3.py with the Base RPC endpoint (e.g., https://base.mainnet.rpc.dev). The watcher runs as a background thread; it queries new blocks every 5 seconds and processes any relevant transfers. from web3 import Web3 import threading import time import json from eth_utils import to_checksum_address, encode_hex BASE_RPC = "https://base.mainnet.rpc.dev" USDC_ADDRESS = to_checksum_address("0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02FF9") AGENT_ADDRESS = to_checksum_address("0xYourAgentAddressHere") # set via env var w3 = Web3(Web3.HTTPProvider(BASE_RPC)) assert w3.isConnected(), "Cannot connect to Base RPC" usdc_abi = [ {"constant":True,"inputs":[{"name":"_owner","type":"address"}], "name":"balanceOf","outputs":[{"name":"balance","type":"uint256"}], "type":"function"}, {"constant":False,"inputs":[ {"name":"_to","type":"address"}, {"name":"_value","type":"uint256"} ],"name":"transfer","outputs":[{"name":"","type":"bool"}],"type":"function"}, {"anonymous":False,"inputs":[ {"indexed":True,"name":"from","type":"address"}, {"indexed":True,"name":"to","type":"address"}, {"indexed":False,"name":"value","type":"uint256"} ],"name":"Transfer","type":"event"} ] usdc_contract = w3.eth.contract(address=USDC_ADDRESS, abi=usdc_abi) # In‑memory store of pending tasks: task_id -> {price_usdc, status} pending_tasks = {} processed_tx_hashes = set() def hex_to_str(h: str) -> str: return bytes.fromhex(h[2:]).decode("utf-8", errors="ignore") def check_new_transfers(): latest = w3.eth.block_number from_block = max(latest - 10, 0) # look back a few blocks to avoid reorg issues for ev in usdc_contract.events.Transfer().getLogs(fromBlock=from_block, toBlock=latest): tx_hash = ev.transactionHash.hex() if tx_hash in processed_tx_hashes: continue processed_tx_hashes.add(tx_hash) # Verify direction if ev.args['to'].lower() != AGENT_ADDRESS.lower(): continue # not a payment to us # Verify token if ev.args['from'].lower() == AGENT_ADDRESS.lower(): continue # ignore our own outgoing transfers amount_raw = ev.args['value'] # USDC has 6 decimals on Base amount_usdc = amount_raw / 1_000_000 # Extract data field (tx input) – expect UTF‑8 task ID tx = w3.eth.get_transaction(ev.transactionHash) data_hex = tx['input'] if data_hex == "0x": # no data continue try: task_id = hex_to_str(data_hex) except Exception: continue # malformed data # Look up pending task info = pending_tasks.get(task_id) if not info: continue # unknown or already fulfilled if amount_usdc < info["price_usdc"]: continue # underpaid # Mark as paid info["status"] = "paid" info["paid_at"] = time.time() print(f"Task {task_id} paid {amount_usdc} USDC") def start_watcher(interval_sec: int = 5): def loop(): while True: try: check_new_transfers() except Exception as e: print(f"Watcher error: {e}") time.sleep(interval_sec) t = threading.Thread(target=loop, daemon=True) t.start() # Example usage: register a pending task def register_task(task_id: str, price_usdc: float = 0.02): pending_tasks[task_id] = {"price_usdc": price_usdc, "status": "pending", "created_at": time.time()} Key points: The watcher is intentionally lightweight; it does not maintain a full node, just reads logs via RPC. We rely on the transaction’s input field to carry the task ID. This avoids needing a separate off‑chain order book. Price is expressed in USDC with six decimals; the agent can adjust pricing dynamically based on cost estimates (see section 9). 3.3 Integrating payment check into the planner Before executing a tool that has an associated cost, the planner queries the pending_tasks dict. If the task is not yet paid, it returns a 402‑style response to the caller (in our case, the external client that invoked the agent via an HTTP wrapper). The HTTP wrapper is a thin FastAPI layer: from fastapi import FastAPI, HTTPException from pydantic import BaseModel app = FastAPI() class TaskRequest(BaseModel): goal: str task_id: str # matches the data field used in payment @app.post("/run") def run_task(req: TaskRequest): info = pending_tasks.get(req.task_id) if not info: raise HTTPException(status_code=404, detail="unknown task") if info["status"] != "paid": # Return 402 with payment details raise HTTPException( status_code=402, detail={ "scheme": "exact", "network": "base", "token": USDC_ADDRESS, "resource": "/run", "maxTimeout": 86400, "metadata": {"id": req.task_id}, }, ) # Task is paid – run planner result = planner_loop(req.goal) if result["status"] == "success": # Optionally mark task as completed info["status"] = "completed" return result When a client receives the 402 response, they must sign and send an USDC transfer with the appropriate data field before retrying the request. Scheduler and “Sleep” Mode The agent’s main process starts three components: The FastAPI server (listening on 0.0.0.0:8000). The payment watcher thread (as shown above). A background idle loop that simply waits for incoming HTTP requests; when none arrive, the process spends most of its time in the OS scheduler’s idle state. We do not run a continuous planner loop; planning is triggered only by an paid request. This design keeps CPU usage near zero when there is no work. If you prefer a pull‑based model (the agent scans a job queue periodically), you can replace the FastAPI endpoint with a simple APScheduler cron job that runs every N seconds, checks a remote task board, and runs the planner if a paid job is found. Below is a minimal example: from apscheduler.schedulers.background import BackgroundScheduler scheduler = BackgroundScheduler() scheduler.add_job(func=check_remote_job_board, trigger="interval", seconds=30) scheduler.start() def check_remote_job_board(): # Pseudocode: fetch JSON list of {task_id, goal, price} jobs = fetch_jobs() for j in jobs: if j["task_id"] not in pending_tasks: pending_tasks[j["task_id"]] = {"price_usdc": j["price"], "status": "pending"} # If already paid, trigger planner if pending_tasks[j["task_id"]]["status"] == "paid": planner_loop(j["goal"]) pending_tasks[j["task_id"]]["status"] = "completed" In our actual deployment we use the push model (FastAPI) because it eliminates unnecessary polling and yields lower latency for the client. Observability, Logging, and Alerting Even a simple agent benefits from structured logging and metrics. We use Python’s built‑in logging module with JSON formatting and expose a /metrics endpoint for Prometheus. 4.1 JSON logging import logging import json from pythonjsonlogger import jsonlogger logger = logging.getLogger("agent") logger.setLevel(logging.INFO) logHandler = logging.StreamHandler() formatter = jsonlogger.JsonFormatter( "%(asctime)s %(levelname)s %(name)s %(message)s" ) logHandler.setFormatter(formatter) logger.addHandler(logHandler) def log_event(event: str, **kwargs): logger.info(json.dumps({"event": event, **kwargs})) Every major step (planner start, tool call, payment detection) calls log_event. This makes log aggregation in Loki or Elasticsearch straightforward. 4.2 Prometheus metrics We expose counters for: agent_tasks_total (labels: outcome=success|failed|paid|unpaid) agent_tool_calls_total (labels: tool, result=ok|error) agent_inference_tokens_total (counts tokens sent to the LLM) agent_uptime_seconds from prometheus_client import Counter, Gauge, start_http_server TASKS = Counter("agent_tasks_total", "Number of tasks processed", ["outcome"]) TOOL_CALLS = Counter("agent_tool_calls_total", "Tool invocations", ["tool", "result"]) INFER_TOKENS = Counter("agent_inference_tokens_total", "Tokens sent to LLM") UPTIME = Gauge("agent_uptime_seconds", "Agent uptime in seconds") start_http_server(9090) # exposes /metrics on port 9090 Inside the planner loop we increment the counters after each LLM generation: def count_tokens(text: str): return len(tokenizer.encode(text)) # In planner loop after obtaining raw output: inp_toks = count_tokens(prompt) out_toks = count_tokens(raw) INFER_TOKENS.inc(inp_toks + out_toks) 4.3 Alerting example A simple alert rule (for Prometheus + Alertmanager) could fire if the success rate drops below 70 % over a 5‑minute window: ALERT AgentSuccessRateDrop expr: sum by (outcome) (rate(agent_tasks_total[outcome="success"][5m])) / sum by (outcome) (rate(agent_tasks_total[5m])) < 0.7 for: 5m labels: severity: warning annotations: summary: "Agent success rate falling" description: "Success rate over the last 5m is {{ $value | printf \"%.2f\" }}." These observability primitives help you spot regressions when you change model quantisation, tool versions, or pricing. Cost Model and Profitability Benchmarks To determine whether the agent can “earn while I sleep”, we need a concrete cost model that includes: Cost component Source Approximate unit cost GPU instance (spot) AWS g4dn.xlarge (1 × T4) $0.30 / hr CPU‑only fallback AWS t3.large (2 vCPU) $0.015 / hr USDC transaction fee (Base) ~0.0005 USDC per transfer (≈$0.0005) negligible LLM inference (tokens) derived from GPU/CPU cost per second see table below Tool external API usage e.g., a public sentiment API (free tier) $0 per call (if within limits) Storage / logging minimal (few MB/day)  10 k tokens/sec). 5.2 Revenue per task We set a fixed price of $0.02 per sentiment‑analysis task (≈ 200 tokens prompt + 50 tokens completion). The breakdown: LLM token usage: ~250 tokens → cost ≈ $0.00001 (CPU) or $0.000013 (GPU). Tool execution (local DistilBERT) runs on CPU, negligible (