Attentium Developer Guide

Get verified human attention for your AI agent in minutes.

This guide shows you how to integrate Attentium into your AI agent using the x402 Payment Required protocol.

How It Works
Using the Price Oracle
Step 1: Get a Price Quote
Step 2: Make the Solana Payment
Step 3: Create Your Campaign
Step 4: Store Your Keys
Retrieving Results
Webhooks
Verifying Signatures

Base URL

Environment	Base URL
Production	`https://api.attentium.ai`
Local Development	`http://localhost:3000`

All endpoints in this guide are relative to the base URL. For example:

GET /v1/oracle/quote  →  https://api.attentium.ai/v1/oracle/quote
POST /v1/verify       →  https://api.attentium.ai/v1/verify

Python setup:

import requests

BASE_URL = "https://api.attentium.ai"  # or "http://localhost:3000" for local dev

# All requests use this base
response = requests.get(f"{BASE_URL}/v1/oracle/quote")

How It Works

Attentium uses the x402 (Payment Required) standard with non-custodial escrow:

┌─────────────────┐         ┌──────────────────┐         ┌─────────────────┐
│   Your Agent    │         │    Attentium     │         │  Payment Router │
└────────┬────────┘         └────────┬─────────┘         └────────┬────────┘
         │                           │                            │
         │  1. POST /verify (no tx)  │                            │
         │──────────────────────────>│                            │
         │                           │                            │
         │  402 Payment Required     │                            │
         │  (serialized escrow tx)   │                            │
         │<──────────────────────────│                            │
         │                           │                            │
         │  2. Sign & Submit Escrow  │                            │
         │───────────────────────────────────────────────────────>│
         │                           │       deposit_escrow()     │
         │                           │                            │
         │  3. POST /verify + tx_sig │                            │
         │──────────────────────────>│                            │
         │                           │  Verify Escrow Deposit     │
         │                           │<───────────────────────────│
         │  200 OK + keys            │                            │
         │<──────────────────────────│                            │
         │                           │                            │
         │  4. Humans verify         │                            │
         │       ...wait...          │                            │
         │                           │                            │
         │  5. Webhook/Poll results  │                            │
         │<──────────────────────────│                            │

Non-custodial: Funds go to an on-chain escrow PDA, not a wallet. No API keys. No registration.

Using the Price Oracle (Optional)

Don’t know what to bid? Use our Oracle endpoint to get a competitive price automatically:

import requests

# Get the market clearing price
response = requests.get(
    "https://api.attentium.ai/v1/oracle/quote",
    params={"duration": 30}  # 10, 30, or 60 seconds
)

quote = response.json()
print(quote)

Response:

{
  "duration": 30,
  "gross_bid_cents": 6,
  "market_depth": 3,
  "timestamp": "2024-12-20T10:00:00Z"
}

Field	Description
`gross_bid_cents`	Recommended bid in cents/second to beat current market
`market_depth`	Number of active orders at this duration

Using the Oracle price:

# Get competitive price from oracle
quote = requests.get("https://api.attentium.ai/v1/oracle/quote", params={"duration": 30}).json()

# Convert cents to USDC
bid_per_second = quote["gross_bid_cents"] / 100  # e.g., 6 cents = $0.06

# Use in your campaign
requests.post("/v1/verify", json={
    "duration": 30,
    "bid_per_second": bid_per_second,  # $0.06
    ...
})

Tip: The Oracle always returns a price that beats the highest current bid by $0.01. If the market is empty, it returns the floor price ($ 0.003/second).

Step 1: Get a Price Quote

First, call /verify without a payment signature to get an invoice:

import requests

# Request without payment - get invoice
response = requests.post(
    "https://api.attentium.ai/v1/verify",
    headers={"Content-Type": "application/json"},
    json={
        "duration": 30,           # 10, 30, or 60 seconds
        "quantity": 5,            # Number of human verifiers
        "bid_per_second": 0.05,   # Your bid in USDC per second
        "validation_question": "Does this image show a cat?",
        "content_url": "https://example.com/image.png"
    }
)

# Returns 402 Payment Required
invoice = response.json()
print(invoice)

Response (402 Payment Required):

{
  "error": "payment_required",
  "message": "Sign the attached transaction to fund escrow.",
  "transaction": "<base64_serialized_deposit_escrow_tx>",
  "amount": 7.50,
  "campaign_id": "ab1234..." // SAVE THIS: Required for Step 3
}

Escrow Calculation:

total = duration × quantity × bid_per_second
      = 30 × 5 × $0.05
      = $7.50 USDC

Step 2: Sign the Escrow Deposit Transaction

The 402 response includes a pre-built transaction that deposits USDC into your agent’s escrow PDA. Sign and submit it:

import base64
from solana.rpc.api import Client
from solana.transaction import Transaction
from solders.keypair import Keypair

# Your agent's wallet
agent_keypair = Keypair.from_bytes(YOUR_PRIVATE_KEY_BYTES)
client = Client("https://api.mainnet-beta.solana.com")

# 1. Decode the pre-built transaction from the 402 response
serialized_tx = base64.b64decode(invoice["transaction"])
tx = Transaction.from_bytes(serialized_tx)

# 2. Get a fresh blockhash and update the transaction
recent_blockhash = client.get_latest_blockhash().value.blockhash
tx.recent_blockhash = recent_blockhash

# 3. Sign with your agent keypair
tx.sign(agent_keypair)

# 4. Submit to Solana
result = client.send_transaction(tx)
tx_signature = str(result.value)
print(f"Escrow deposit sent: {tx_signature}")

Non-custodial: Your USDC goes to an escrow PDA derived from your wallet address. You retain control and can withdraw unspent funds after the campaign expires or completes.

Networks:

Network	RPC	Token
Mainnet	`api.mainnet-beta.solana.com`	USDC only
Devnet	`api.devnet.solana.com`	USDC (testing)

Step 3: Create Your Campaign

Now call /verify again with the transaction signature and the campaign ID from Step 1:

response = requests.post(
    "https://api.attentium.ai/v1/verify",
    headers={
        "Content-Type": "application/json",
        "X-Solana-Tx-Signature": tx_signature,    # Your escrow deposit proof
        "X-Agent-Key": str(agent_keypair.pubkey()),  # Your wallet address
        "X-Campaign-Id": invoice["campaign_id"],  # CRITICAL: Match the ID from Step 1
        "X-Builder-Code": "MY_AGENT_V1"           # Optional: Earn 3% revenue share
    },
    json={
        "duration": 30,
        "quantity": 5,
        "bid_per_second": 0.05,
        "validation_question": "Does this image show a cat?",
        "content_url": "https://example.com/image.png",
        "callback_url": "https://your-agent.com/webhook"  # Optional: Real-time webhooks
    }
)

data = response.json()
print(data)

Response (200 Success):

{
  "success": true,
  "message": "Verification slots reserved: 5x 30s @ $0.0425/s",
  "order": {
    "duration": 30,
    "quantity": 5,
    "bid_per_second": 0.0425,
    "tx_hash": "5abc123...",
    "referrer": null
  },
  "read_key": "a1b2c3d4e5f6...",
  "webhook_secret": "x9y8z7w6v5u4..."
}

Note: The bid_per_second in the response is the NET amount (85% of what you paid). The 15% spread covers protocol fees and gas costs.

Idempotency: If you accidentally submit the same tx_hash twice, the endpoint returns the existing order with its original keys. This is safe to retry.

Are you building an agent framework or platform? You can earn 3% of every bid routed through your code. simply add the X-Builder-Code header to your /verify requests:

headers={
    "X-Builder-Code": "MY_FRAMEWORK_V1", # Your unique identifier
    ...
}

No Registration Required: Just start sending the header.
Automatic Payouts: Fees accumulate in the Fee Vault.
Claiming: Use the smart contract to claim your balance (SDK coming soon).

Step 4: Store Your Keys

⚠️ CRITICAL: Save these immediately - they are only returned ONCE!

# Extract from response
campaign_id = data["order"]["tx_hash"]
read_key = data["read_key"]           # Needed to fetch results
webhook_secret = data["webhook_secret"]  # Needed to verify webhooks

# Store in your database
db.campaigns.insert({
    "campaign_id": campaign_id,
    "read_key": read_key,
    "webhook_secret": webhook_secret,
    "created_at": datetime.now()
})

Warning: If you lose these keys, you cannot retrieve your campaign results or verify webhooks. There is no recovery mechanism.

Retrieving Results (Polling)

Use the read_key to fetch human responses:

import requests

def get_campaign_results(campaign_id: str, read_key: str) -> dict:
    """Fetch all human responses for a campaign."""
    response = requests.get(
        f"https://api.attentium.ai/v1/campaigns/{campaign_id}/results",
        params={"key": read_key}
    )
    
    if response.status_code == 401:
        raise ValueError("Invalid read_key")
    
    return response.json()

# Usage
results = get_campaign_results(
    campaign_id="admin_123abc...",
    read_key="a1b2c3d4..."
)

print(f"Question: {results['validation_question']}")
print(f"Completed: {results['completed_quantity']}/{results['target_quantity']}")

for r in results["results"]:
    print(f"  - Answer: {r['answer']}")

Response Structure:

{
  "campaign_id": "admin_123abc...",
  "validation_question": "Does this image show a cat?",
  "status": "in_progress",
  "target_quantity": 5,
  "completed_quantity": 3,
  "results": [
    {
      "response_id": "resp_1",
      "answer": "Yes, it's definitely a cat",
      "duration_seconds": 28,
      "completed_at": "2024-12-20T09:30:00Z"
    }
  ],
  "aggregates": {
    "avg_duration_seconds": 26.4,
    "completion_rate": 0.6
  }
}

Webhooks (Real-Time)

Get notified instantly when a human completes your verification task.

Setup

Include callback_url when creating your campaign:

response = requests.post(
    "https://api.attentium.ai/v1/verify",
    json={
        "callback_url": "https://your-agent.com/attentium-webhook",
        # ... other fields
    }
)

Webhook Payload

When a human submits their answer, we POST to your callback URL:

{
  "event": "response_submitted",
  "campaign_id": "admin_123abc...",
  "validation_question": "Does this image show a cat?",
  "timestamp": "2024-12-20T09:30:00Z",
  "data": {
    "answer": "Yes, it's definitely a cat",
    "duration": 28,
    "exited_early": false,
    "completed_at": "2024-12-20T09:30:00Z"
  }
}

⚠️ Reliability Warning: Webhooks are currently “Fire and Forget.” We attempt delivery once with a 5-second timeout. If your server does not respond with 200 OK immediately, the event is not retried. Always implement the Polling endpoint as a backup to sweep for missed results.

Verifying Webhook Signatures

Every webhook includes an X-Attentium-Signature header. You must verify this signature to ensure the webhook is authentic.

Python (Copy-Paste Ready)

import hmac
import hashlib
import json
from flask import Flask, request, jsonify

app = Flask(__name__)

def verify_attentium_signature(
    payload: dict,
    signature_header: str,
    webhook_secret: str
) -> bool:
    """
    Verify that a webhook payload was signed by Attentium.
    
    Args:
        payload: The JSON body of the webhook request
        signature_header: Value of X-Attentium-Signature header
        webhook_secret: Your campaign's webhook_secret
    
    Returns:
        True if signature is valid, False otherwise
    """
    if not signature_header or not signature_header.startswith("sha256="):
        return False
    
    received_signature = signature_header[7:]  # Remove "sha256=" prefix
    
    # Compute expected signature
    payload_bytes = json.dumps(payload, separators=(',', ':')).encode('utf-8')
    expected_signature = hmac.new(
        webhook_secret.encode('utf-8'),
        payload_bytes,
        hashlib.sha256
    ).hexdigest()
    
    # Use constant-time comparison to prevent timing attacks
    return hmac.compare_digest(received_signature, expected_signature)


@app.route("/attentium-webhook", methods=["POST"])
def handle_webhook():
    # Get your webhook_secret from your database
    campaign_id = request.json.get("campaign_id")
    campaign = db.campaigns.find_one({"campaign_id": campaign_id})
    
    if not campaign:
        return jsonify({"error": "Unknown campaign"}), 404
    
    # Verify signature
    signature = request.headers.get("X-Attentium-Signature")
    if not verify_attentium_signature(request.json, signature, campaign["webhook_secret"]):
        return jsonify({"error": "Invalid signature"}), 401
    
    # Process the verified webhook
    answer = request.json["data"]["answer"]
    print(f"Human answered: {answer}")
    
    # Your business logic here...
    
    return jsonify({"status": "ok"}), 200

Node.js

const crypto = require('crypto');
const express = require('express');
const app = express();

app.use(express.json());

function verifyAttentiumSignature(payload, signatureHeader, webhookSecret) {
  if (!signatureHeader || !signatureHeader.startsWith('sha256=')) {
    return false;
  }
  
  const receivedSignature = signatureHeader.slice(7);
  const expectedSignature = crypto
    .createHmac('sha256', webhookSecret)
    .update(JSON.stringify(payload))
    .digest('hex');
  
  return crypto.timingSafeEqual(
    Buffer.from(receivedSignature),
    Buffer.from(expectedSignature)
  );
}

app.post('/attentium-webhook', (req, res) => {
  const signature = req.headers['x-attentium-signature'];
  const webhookSecret = getSecretForCampaign(req.body.campaign_id);
  
  if (!verifyAttentiumSignature(req.body, signature, webhookSecret)) {
    return res.status(401).json({ error: 'Invalid signature' });
  }
  
  // Process verified webhook
  console.log('Human answered:', req.body.data.answer);
  res.json({ status: 'ok' });
});

Common Mistakes

❌ Don’t: Forget to save your keys

# BAD - keys are lost forever
response = requests.post("/v1/verify", json={...})
campaign_id = response.json()["order"]["tx_hash"]
# read_key and webhook_secret are gone!

✅ Do: Save keys immediately

# GOOD - keys are persisted
data = response.json()
save_to_database(
    campaign_id=data["order"]["tx_hash"],
    read_key=data["read_key"],
    webhook_secret=data["webhook_secret"]
)

❌ Don’t: Skip signature verification

# BAD - accepts any request, including attackers
@app.route("/webhook", methods=["POST"])
def webhook():
    answer = request.json["data"]["answer"]  # Could be fake!
    return "ok"

✅ Do: Always verify signatures

# GOOD - rejects forged requests
@app.route("/webhook", methods=["POST"])
def webhook():
    if not verify_attentium_signature(request.json, ...):
        return "Unauthorized", 401
    answer = request.json["data"]["answer"]  # Verified!
    return "ok"

❌ Don’t: Use string comparison for signatures

# BAD - vulnerable to timing attacks
if received_signature == expected_signature:
    ...

✅ Do: Use constant-time comparison

# GOOD - secure against timing attacks
import hmac
if hmac.compare_digest(received_signature, expected_signature):
    ...

Troubleshooting

”Invalid signature” errors

JSON serialization mismatch: We use JSON.stringify() on our Node.js backend (compact format, no spaces). Your verification must use the exact same format.

# ✅ CORRECT - matches our backend
payload_bytes = json.dumps(payload, separators=(',', ':')).encode('utf-8')

# ❌ WRONG - default Python adds spaces after colons
payload_bytes = json.dumps(payload).encode('utf-8')  # Has ", " instead of ","

Re-serialization trap: If your framework parses JSON before verification, re-serializing may change the format. FastAPI (safer approach):

from fastapi import Request

@app.post("/webhook")
async def webhook(request: Request):
    raw_body = await request.body()  # Get raw bytes
    signature = request.headers.get("X-Attentium-Signature")
    
    # Verify against RAW body, not re-serialized JSON
    expected = "sha256=" + hmac.new(
        webhook_secret.encode(),
        raw_body,  # <-- Use raw bytes
        hashlib.sha256
    ).hexdigest()
    
    if not hmac.compare_digest(signature, expected):
        return {"error": "Invalid signature"}, 401
    
    # Now parse JSON
    payload = json.loads(raw_body)

Encoding issues: Both the secret and payload must be UTF-8 encoded.
Wrong secret: Each campaign has a unique webhook_secret. Make sure you’re using the correct one.

Debug helper:

# Print what we're comparing
payload_json = json.dumps(request.json, separators=(',', ':'))
print(f"Payload: {payload_json}")
print(f"Received sig: {signature_header}")
print(f"Expected sig: sha256={expected_signature}")

“401 Unauthorized” when fetching results

Verify you’re using the correct read_key for that specific campaign
Check that the read_key is passed as a query parameter: ?key=YOUR_READ_KEY

Error Codes

HTTP	Error	Description
400	`missing_fields`	Missing required fields (`duration`, `bid_per_second`, or `validation_question`)
400	`invalid_duration`	Duration must be 10, 30, or 60 seconds
400	`transaction_not_found`	Solana transaction not confirmed yet (wait and retry)
400	`transaction_failed`	Solana transaction execution failed on-chain
400	`missing_campaign_id`	`X-Campaign-Id` header not provided
400	`missing_memo`	Transaction has no memo instruction
400	`memo_mismatch`	Transaction memo doesn’t match `X-Campaign-Id` header
401	`unauthorized`	Invalid or missing `read_key` for results endpoint
402	`payment_required`	No payment header provided; returns invoice
402	`invalid_payment`	Transaction validation failed (wrong amount, recipient, or token)
403	`expired_transaction`	Transaction older than 2 minutes (replay protection)
403	`rejected_tos`	Content rejected by moderation (funds forfeited)
404	`campaign_not_found`	No campaign exists with that transaction hash
500	`server_error`	Internal server error

Best Practices

Store secrets securely: Use environment variables or a secrets manager, never hardcode.
Respond to webhooks fast: Our webhooks have a 5-second timeout. Return 200 OK immediately and process asynchronously.
Always poll as backup: Webhooks are not retried. Periodically poll /campaigns/:tx_hash/results to catch any missed events.
Implement idempotency: Use campaign_id + timestamp to deduplicate in case you receive duplicate events.

Support

Issues: Open a GitHub issue
Discord: Join our developer community

Built with ❤️ for AI Agent developers

Get Started

Reference

Introduction

Attentium Developer Guide

Table of Contents

Base URL

How It Works

Using the Price Oracle (Optional)

Step 1: Get a Price Quote

Step 2: Sign the Escrow Deposit Transaction

Step 3: Create Your Campaign

Step 4: Store Your Keys

Retrieving Results (Polling)

Webhooks (Real-Time)

Setup

Webhook Payload

Verifying Webhook Signatures

Python (Copy-Paste Ready)

Node.js

Common Mistakes

❌ Don’t: Forget to save your keys

✅ Do: Save keys immediately

❌ Don’t: Skip signature verification

✅ Do: Always verify signatures

❌ Don’t: Use string comparison for signatures

✅ Do: Use constant-time comparison

Troubleshooting

”Invalid signature” errors

“401 Unauthorized” when fetching results

Error Codes

Best Practices

Support

Get Started

Reference

Documentation Index

​Attentium Developer Guide

​Table of Contents

​Base URL

​How It Works

​Using the Price Oracle (Optional)

​Step 1: Get a Price Quote

​Step 2: Sign the Escrow Deposit Transaction

​Step 3: Create Your Campaign

​Builder Revenue Share

​Step 4: Store Your Keys

​Retrieving Results (Polling)

​Webhooks (Real-Time)

​Setup

​Webhook Payload

​Verifying Webhook Signatures

​Python (Copy-Paste Ready)

​Node.js

​Common Mistakes

​❌ Don’t: Forget to save your keys

​✅ Do: Save keys immediately

​❌ Don’t: Skip signature verification

​✅ Do: Always verify signatures

​❌ Don’t: Use string comparison for signatures

​✅ Do: Use constant-time comparison

​Troubleshooting

​”Invalid signature” errors

​“401 Unauthorized” when fetching results

​Error Codes

​Best Practices

​Support

Attentium Developer Guide

Table of Contents

Base URL

How It Works

Using the Price Oracle (Optional)

Step 1: Get a Price Quote

Step 2: Sign the Escrow Deposit Transaction

Step 3: Create Your Campaign

Builder Revenue Share

Step 4: Store Your Keys

Retrieving Results (Polling)

Webhooks (Real-Time)

Setup

Webhook Payload

Verifying Webhook Signatures

Python (Copy-Paste Ready)

Node.js

Common Mistakes

❌ Don’t: Forget to save your keys

✅ Do: Save keys immediately

❌ Don’t: Skip signature verification

✅ Do: Always verify signatures

❌ Don’t: Use string comparison for signatures

✅ Do: Use constant-time comparison

Troubleshooting

”Invalid signature” errors

“401 Unauthorized” when fetching results

Error Codes

Best Practices

Support