DEV Community

Nikhil Ranka
Nikhil Ranka

Posted on

x402 Explained: HTTP-Native Micropayments for AI Agents (With Real Code)

x402 Explained: HTTP-Native Micropayments for AI Agents (With Real Code)

TL;DR – x402 turns the HTTP 402 Payment Required status code into a concrete, on‑chain micropayment flow. An AI agent can request a resource, receive a 402 response that contains a payment instruction, settle the tiny fee in USDC on Base, and then retry the request with a proof‑of‑payment header. The pattern works with any HTTP service; the only extra pieces are a wallet capable of signing ERC‑20 transfers and a tiny amount of Base gas.


Why a “402” for Agents?

Most agent‑to‑service interactions today rely on API keys, subscription tiers, or ad‑hoc invoicing. Those models assume a relatively stable relationship and incur operational overhead (key rotation, billing cycles, fraud checks). Micropayments flip the script: each call pays for itself, eliminating long‑term contracts and enabling truly pay‑as‑you‑go autonomy.

The HTTP specification already defined status code 402 for “Payment Required,” but it never gained traction because no standard payment method existed. x402 fills that gap by specifying:

  1. How a server signals a price (via a Payment-Required header containing a JSON payload).
  2. How the client settles (an ERC‑20 USDC transfer on Base, with the transaction hash returned in a X-Payment header on retry).
  3. How the server verifies (by checking the transaction on‑chain and matching the amount, token, and recipient).

Because the flow lives entirely in HTTP headers and a single on‑chain transaction, it adds only a few hundred milliseconds of latency and works with any language or framework that can make HTTP calls and sign Ethereum transactions.


The Wire Format

Server → Client (402 Response)

HTTP/1.1 402 Payment Required
Content-Type: application/json
Payment-Required: {"token":"0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913","amount":"1000000","recipient":"0xAbc...Def","nonce":"2024-09-26T14:32:10Z"}
{
  "error": "payment required",
  "payload": {...}
}
Enter fullscreen mode Exit fullscreen mode
  • token – ERC‑20 contract address (USDC on Base = 0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913).
  • amount – smallest unit (USDC has 6 decimals → 1000000 = $1.00).
  • recipient – the service’s wallet that should receive the funds.
  • nonce – a timestamp or random string to prevent replay attacks.

Client → Server (Retry with Proof)

GET /data HTTP/1.1
Host: agent.example.com
Authorization: Bearer <agent‑token>
X-Payment: {"tx":"0x123abc…","block":12345678}
Accept: application/json
Enter fullscreen mode Exit fullscreen mode

The server validates:

  • Transaction tx exists on Base.
  • It transfers amount USDC from the sender (derived from the transaction’s from field) to recipient.
  • The nonce matches the one originally sent (or is within an allowed window).
  • No replay: the server keeps a short‑term cache of seen nonces/tx hashes.

If validation passes, the server proceeds with the original request logic and returns a 200 response.


Minimal Working Example (Python)

Below is a self‑contained Flask service that implements the server side of x402, and a tiny agent that consumes it. The code deliberately avoids abstractions to show the exact steps.

1. Server (micropayment‑protected endpoint)

# server.py
from flask import Flask, request, jsonify, make_response
import json
from eth_account import Account
from web3 import Web3
import time

app = Flask(__name__)

# Configuration – replace with your own values
USDC_ADDRESS = Web3.to_checksum_address("0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913")
RECIPIENT    = Web3.to_checksum_address("0xAbcDef...YourWallet")
CHAIN_ID     = 8453   # Base
RPC_URL      = "https://base.mainnet.rpc.dev"
w3           = Web3(Web3.HTTPProvider(RPC_URL))

# Simple in‑memory replay cache (in prod use Redis with TTL)
_seen = set()
CACHE_TTL = 30  # seconds

def _clean_cache():
    now = time.time()
    global _seen
    _seen = {item for item in _seen if now - item[1] < CACHE_TTL}

@app.route("/price")
def price():
    """Return a 402 with payment instruction for a dummy data point."""
    amount_usdc = 1_000_000          # $0.01 (USDC has 6 decimals)
    nonce = f"{time.time()}"
    payload = {
        "token": USDC_ADDRESS.lower(),
        "amount": str(amount_usdc),
        "recipient": RECIPENT.lower(),
        "nonce": nonce,
    }
    resp = make_response(jsonify({"error": "payment required"}), 402)
    resp.headers["Payment-Required"] = json.dumps(payload)
    return resp

@app.route("/data")
def data():
    """Protected endpoint – expects X-Payment header."""
    _clean_cache()
    auth = request.headers.get("X-Payment")
    if not auth:
        return jsonify({"error": "missing X-Payment"}), 401
    try:
        info = json.loads(auth)
        tx_hash = info["tx"]
        block_number = int(info["block"])
    except (json.JSONDecodeError, KeyError, ValueError):
        return jsonify({"error": "malformed X-Payment"}), 400

    # Verify transaction exists and is correct
    try:
        tx = w3.eth.get_transaction(tx_hash)
        receipt = w3.eth.get_transaction_receipt(tx_hash)
    except Exception:
        return jsonify({"error": "tx not found"}), 400

    if receipt["status"] != 1:
        return jsonify({"error": "tx failed"}), 400

    # Check token, amount, recipient
    if tx["to"].lower() != USDC_ADDRESS:
        return jsonify({"error": "wrong token contract"}), 400
    # USDC transfer ABI: function transfer(address to, uint256 amount)
    if tx["input"][:10] != "0xa9059cbb":  # function selector for transfer
        return jsonify({"error": "not a USDC transfer"}), 400
    # Decode parameters (simple slice)
    to_addr = "0x" + tx["input"][34:74]
    amount = int.from_bytes(bytes.fromhex(tx["input"][74:]), "big")
    if to_addr.lower() != RECIPENT.lower():
        return jsonify({"error": "wrong recipient"}), 400
    # Amount must match what we advertised (allow 1 wei tolerance)
    expected = int(request.headers.get("Payment-Required", "{}"))  # not ideal – see note below
    if abs(amount - expected) > 1:
        return jsonify({"error": "incorrect amount"}), 400

    # Replay protection – use nonce from original 402 (we would have stored it)
    # For demo we just check tx hash not seen recently
    if tx_hash in {h for h, _ in _seen}:
        return jsonify({"error": "replay"}), 402
    _seen.add((tx_hash, time.time()))

    # If we reach here, payment is valid – serve the resource
    return jsonify({"value": 42, "timestamp": time.time()})

if __name__ == "__main__":
    app.run(port=5000, debug=True)
Enter fullscreen mode Exit fullscreen mode

Explanation of the server code

  • The /price endpoint returns a 402 with a JSON blob that tells the client exactly how much USDC to send, where, and a nonce.
  • /data reads the X-Payment header, checks the transaction on Base via web3.py, verifies the ERC‑20 transfer matches the advertised parameters, and ensures the nonce/tx hash hasn’t been seen recently.
  • The server returns 200 with dummy data once the checks pass.

Note: In a production service you would store the expected amount and nonce from the original 402 response (e.g., in a short‑lived Redis key) rather than trying to re‑parse the Payment-Required header on retry. The snippet omits that for brevity.

2. Agent (client side)


python
# agent.py
import json
import time
import requests
from eth_account import Account
from web3 import Web3

# ---- Configuration ----
BASE_RPC = "https://base.mainnet.rpc.dev"
USDC = Web3.to_checksum_address("0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913")
RECIPIENT = Web3.to_checksum_address("0xAbcDef...YourWallet")
PRIVATE_KEY = "0xYOUR_PRIVATE_KEY"   # never commit this!
ACCOUNT = Account.from_key
Enter fullscreen mode Exit fullscreen mode

Top comments (0)