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

Building a self‑sustaining agent that can earn cryptocurrency without human intervention sounds like a sci‑fi dream. In practice it’s a mix of well‑known engineering pieces—task scheduling, reliable inference, payment handling, and careful observability—stitched together with enough safeguards to keep it from burning money or spamming endpoints.

Below is a walk‑through of the architecture I used, the concrete code that glues the pieces, and the trade‑offs I ran into along the way. The goal is to give you a reproducible baseline you can adapt, not a promise of effortless wealth.

---

1. High‑level flow

1. Scheduler – a cron‑like loop that wakes the agent every N minutes (default 5 min).
2. Task picker – pulls the highest‑priority job from a lightweight queue (Redis sorted set). Jobs are simple HTTP‑call specifications (method, URL, payload, expected reward in USDC).
3. Executor – runs the job, validates the response, and if successful, calls the payment micro‑service to claim the reward.
4. Ledger – records each attempt (success/fail, latency, payout) in an append‑only PostgreSQL table for audit and debugging.
5. Alerting – if error rate exceeds a threshold, the agent pauses and pushes a Slack/webhook notice.

The loop is intentionally stateless apart from the queue and DB; the agent can be killed and restarted without losing pending work (jobs stay in Redis, completed ones are logged).

---

2. Dependencies

Purpose	Library	Version (tested)
Async HTTP	httpx	0.27.0
Redis client	redis	5.0.1
PostgreSQL async	asyncpg	0.29.0
Structured logging	structlog	24.1.0
Env config	pydantic Settings	2.6.3
Cron‑like scheduler	apscheduler	3.10.4
Payment RPC (x402)	custom thin wrapper	—

Install with:

pip install httpx redis asyncpg structlog pydantic apscheduler

---

3. Configuration

All external touch‑points are injected via environment variables (or a .env file). Keeping config out of code makes the same binary runnable on a local dev machine, a Docker container, or a Workers‑style edge runtime.

# config.py
from pydantic import BaseSettings, PostgresDsn, RedisDsn

class Settings(BaseSettings):
    # Scheduler
    tick_interval_seconds: int = 300   # 5 min

    # Redis queue
    redis_url: RedisDsn = "redis://localhost:6379/0"
    queue_key: str = "agent:jobs"

    # PostgreSQL ledger
    pg_dsn: PostgresDsn = "postgresql://user:pwd@localhost:5432/agent"

    # Payment service (x402 wrapper)
    x402_endpoint: str = "https://payment.example.com/claim"

    # Alerting
    alert_webhook: str = ""   # Slack incoming webhook URL
    max_error_rate: float = 0.2   # 20% errors over last 20 jobs triggers pause

    class Config:
        env_file = ".env"

---

4. Job model

A job is a minimal JSON payload that the scheduler pushes onto the Redis sorted set, ordered by priority (lower number = higher priority). The payload also carries a max_retries field so the agent knows when to give up.

# job.py
from dataclasses import dataclass, asdict
from typing import Any, Dict
import json

@dataclass
class Job:
    id: str
    method: str          # "GET", "POST", etc.
    url: str
    payload: Dict[str, Any] | None = None
    headers: Dict[str, str] | None = None
    reward_usdc: float   # amount to claim if successful
    priority: int = 0
    max_retries: int = 3

    def to_redis(self) -> str:
        return json.dumps(asdict(self))

    @staticmethod
    def from_redis(raw: str) -> "Job":
        data = json.loads(raw)
        return Job(**data)

---

5. Core loop

The loop lives in agent.py. It uses APScheduler for the tick, but you could replace it with a simple while True: asyncio.sleep() if you prefer zero extra deps.

# agent.py
import asyncio
import logging
from typing import List

import httpx
import redis.asyncio as redis
import asyncpg
import structlog
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.interval import IntervalTrigger

from config import Settings
from job import Job
from ledger import Ledger
from payment import claim_reward
from alert import maybe_alert

settings = Settings()
logger = structlog.get_logger()

# ----------------------------------------------------------------------
# Helpers
# ----------------------------------------------------------------------
async def pop_job(r: redis.Redis) -> Job | None:
    """
    Atomic: get the highest‑priority job and remove it from the sorted set.
    Returns None if the queue is empty.
    """
    # ZRANGE with WITHSCORES gives us the element and its priority (score)
    result = await r.zrange(settings.queue_key, 0, 0, withscores=True)
    if not result:
        return None
    member_bytes, _score = result[0]
    # Remove it in the same call chain (ZREM)
    await r.zrem(settings.queue_key, member_bytes)
    raw = member_bytes.decode()
    return Job.from_redis(raw)


async def push_job(r: redis.Redis, job: Job):
    await r.zadd(settings.queue_key, {job.to_redis(): job.priority})


# ----------------------------------------------------------------------
# Main tick
# ----------------------------------------------------------------------
async def tick():
    r = redis.from_url(settings.redis_url)
    pg = await asyncpg.connect(dsn=str(settings.pg_dsn))
    ledger = Ledger(pg)
    async with httpx.AsyncClient(timeout=20.0) as client:
        job: Job | None = await pop_job(r)
        if job is None:
            logger.debug("tick: queue empty")
            return

        logger.info("tick: processing job", job_id=job.id, url=job.url)

        try:
            resp = await _execute_job(client, job)
            if not _validate_response(job, resp):
                raise ValueError("response validation failed")
            # Claim reward
            claimed = await claim_reward(settings.x402_endpoint, job.id, job.reward_usdc)
            await ledger.record_success(job.id, job.reward_usdc, resp.status_code, resp.elapsed.total_seconds())
            logger.info("job succeeded", job_id=job.id, usdc=claimed)
        except Exception as exc:  # noqa: BLE001
            await ledger.record_failure(job.id, str(exc))
            logger.warning("job failed", job_id=job.id, error=str(exc))
            # Retry logic – push back if we haven’t exhausted attempts
            if job.max_retries > 0:
                job.max_retries -= 1
                await push_job(r, job)
                logger.info("job requeued for retry", job_id=job.id, remaining=job.max_retries)
            else:
                logger.error("job exceeded max retries", job_id=job.id)
        finally:
            await r.close()
            await pg.close()

        # Alerting based on recent error rate
        await maybe_alert(settings, ledger)

# ----------------------------------------------------------------------
# Job execution helpers
# ----------------------------------------------------------------------
async def _execute_job(client: httpx.AsyncClient, job: Job) -> httpx.Response:
    if job.method.upper() == "GET":
        return await client.get(job.url, params=job.payload, headers=job.headers)
    if job.method.upper() == "POST":
        return await client.post(job.url, json=job.payload, headers=job.headers)
    raise ValueError(f"Unsupported method {job.method}")

def _validate_response(job: Job, resp: httpx.Response) -> bool:
    # Very basic validation – override per‑job if needed
    return 200 <= resp.status_code < 300

Why this shape?

• The loop is deliberately side‑effect free except for the external services (Redis, PG, HTTP). This makes unit‑testing the tick function straightforward: you can mock pop_job, _execute_job, and claim_reward.
• Error handling is explicit: we differentiate transient network errors (caught by the outer except) from logical validation errors. Retries go back onto the same queue, preserving order.
• The alert hook is decoupled; you can swap Slack for email, PagerDuty, or a simple log‑only mode.

---

6. Ledger implementation

A tiny wrapper around asyncpg that inserts into a table agent_runs. The schema is created once (see migrations.sql).

python
# ledger.py
import asyncpg
from typing import Tuple

class Ledger:
    def __init__(self, conn: asyncpg.Connection):
        self.conn = conn

    async def record_success(self, job_id: str, usdc: float, status: int, latency: float):
        await self.conn.execute(
            """
            INSERT INTO agent_runs (job_id, outcome, usdc, http_status, latency_sec, ts)
            VALUES ($1, 'success', $2, $3, $4, now())
            """,
            job_id, usdc, status, latency,
        )

    async def record_failure(self, job_id: str, error: str):
        await self.conn.execute(
            """
            INSERT INTO agent_runs (job_id, outcome, error_msg, ts)
            VALUES ($1, 'failure', $2, now())
            """,
            job_id, error,
        )

    async def recent_error_rate(self, window: int = 20) -> float:
        rows = await self.conn.fetch(
            """
            SELECT outcome FROM agent_runs
            ORDER BY ts DESC LIMIT $1
            """,
            window,