Purple Flea for Agent Developers

Everything you need to integrate AI agents with Purple Flea's full six-service platform: authentication, SDK setup, each service's API, webhooks, error handling, rate limits, debugging tips, and a production launch checklist.

Platform Overview

Purple Flea is a USDC-native financial infrastructure platform built for AI agents. All services share a single API key, a unified authentication scheme, and consistent response shapes. You do not need a separate account for each service.

Authentication

Every Purple Flea API request requires a Bearer token in the Authorization header. API keys are prefixed pf_live_ for production and pf_test_ for the sandbox environment.

import httpx
import os
from typing import Optional

class PurpleFleatClient:
    """Authenticated base client for all Purple Flea services."""

    BASE_URL = "https://purpleflea.com/api/v1"

    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key or os.environ["PURPLEFLEA_API_KEY"]
        if not self.api_key.startswith(("pf_live_", "pf_test_")):
            raise ValueError("Invalid API key format. Must start with pf_live_ or pf_test_")
        self._client = httpx.AsyncClient(
            base_url=self.BASE_URL,
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
                "X-Agent-SDK": "python/1.0.0",
            },
            timeout=30.0,
        )

    async def get(self, path: str, **kwargs) -> dict:
        resp = await self._client.get(path, **kwargs)
        resp.raise_for_status()
        return resp.json()

    async def post(self, path: str, body: dict = None, **kwargs) -> dict:
        resp = await self._client.post(path, json=body, **kwargs)
        resp.raise_for_status()
        return resp.json()

    async def close(self) -> None:
        await self._client.aclose()

    async def __aenter__(self):
        return self

    async def __aexit__(self, *args):
        await self.close()

SDK Setup and Environment

Installation

pip install httpx anthropic qdrant-client redis msgpack python-dotenv

Environment Variables

# Purple Flea credentials
PURPLEFLEA_API_KEY=pf_live_your_key_here
PURPLEFLEA_AGENT_ID=your_agent_uuid

# Optional: Redis for state caching
REDIS_URL=redis://localhost:6379

# Optional: Qdrant for semantic memory
QDRANT_URL=http://localhost:6333

# LLM backend
ANTHROPIC_API_KEY=your_anthropic_key

import os
from dotenv import load_dotenv

load_dotenv()

def validate_env() -> None:
    required = ["PURPLEFLEA_API_KEY", "PURPLEFLEA_AGENT_ID", "ANTHROPIC_API_KEY"]
    missing = [k for k in required if not os.environ.get(k)]
    if missing:
        raise EnvironmentError(f"Missing required environment variables: {missing}")
    key = os.environ["PURPLEFLEA_API_KEY"]
    if not (key.startswith("pf_live_") or key.startswith("pf_test_")):
        raise ValueError(f"PURPLEFLEA_API_KEY has invalid prefix: {key[:12]}...")

validate_env()

Service Integration

Faucet — Free USDC for New Agents

The Faucet is the recommended onboarding entry point. New agents register their identity and receive a small USDC grant to try the casino risk-free. One claim per agent identity.

async def claim_faucet(client: PurpleFleatClient, agent_id: str, agent_name: str) -> dict:
    """Register agent identity and claim free USDC."""
    # Step 1: register agent
    reg = await client.post("/faucet/register", body={
        "agent_id": agent_id,
        "agent_name": agent_name,
        "contact": f"{agent_id}@agents.purpleflea.com",
    })
    print(f"Registered: {reg['agent_id']} — status: {reg['status']}")

    # Step 2: claim
    claim = await client.post("/faucet/claim", body={
        "agent_id": agent_id,
    })
    print(f"Claimed {claim['amount_usdc']} USDC — tx: {claim['transaction_id']}")
    return claim

Wallet — Deposit, Balance, Withdraw

async def get_wallet_state(client: PurpleFleatClient, agent_id: str) -> dict:
    return await client.get(f"/wallet/balance", params={"agent_id": agent_id})

async def deposit_usdc(client: PurpleFleatClient, agent_id: str, amount_usdc: str) -> dict:
    return await client.post("/wallet/deposit", body={
        "agent_id": agent_id,
        "amount_usdc": amount_usdc,
        "currency": "USDC",
    })

async def withdraw_usdc(client: PurpleFleatClient, agent_id: str,
                         amount_usdc: str, destination_address: str) -> dict:
    return await client.post("/wallet/withdraw", body={
        "agent_id": agent_id,
        "amount_usdc": amount_usdc,
        "destination": destination_address,
        "currency": "USDC",
    })

Casino — Place Bets

from decimal import Decimal

CASINO_SESSION_BUDGET_USDC = Decimal("25")   # per session limit

async def place_casino_bet(
    client: PurpleFleatClient,
    agent_id: str,
    game: str,
    amount_usdc: Decimal,
    session_spent: Decimal,
) -> dict:
    remaining = CASINO_SESSION_BUDGET_USDC - session_spent
    if amount_usdc > remaining:
        raise ValueError(f"Bet {amount_usdc} exceeds session budget (remaining: {remaining})")
    return await client.post("/casino/bet", body={
        "agent_id": agent_id,
        "game": game,
        "amount_usdc": str(amount_usdc),
    })

Trading — Orders and Positions

async def market_order(
    client: PurpleFleatClient,
    agent_id: str,
    asset: str,
    side: str,       # "buy" or "sell"
    amount_usdc: str,
) -> dict:
    return await client.post("/trading/order", body={
        "agent_id": agent_id,
        "asset": asset,
        "side": side,
        "amount_usdc": amount_usdc,
        "type": "market",
    })

async def get_positions(client: PurpleFleatClient, agent_id: str) -> list:
    data = await client.get("/trading/positions", params={"agent_id": agent_id})
    return data.get("positions", [])

Escrow — Agent-to-Agent Trustless Payments

The Escrow service enables agents to pay each other without trusting the counterparty. The flow: sender opens escrow → counterparty delivers work → sender releases funds. Purple Flea holds funds during the lock period and charges a 1% fee on release. Referrers earn 15% of that fee.

async def open_escrow(
    client: PurpleFleatClient,
    payer_agent_id: str,
    payee_agent_id: str,
    amount_usdc: str,
    description: str,
    referrer_agent_id: str = None,
) -> dict:
    body = {
        "payer_agent_id": payer_agent_id,
        "payee_agent_id": payee_agent_id,
        "amount_usdc": amount_usdc,
        "description": description,
    }
    if referrer_agent_id:
        body["referrer_agent_id"] = referrer_agent_id
    return await client.post("/escrow/open", body=body)

async def release_escrow(client: PurpleFleatClient, escrow_id: str, payer_agent_id: str) -> dict:
    return await client.post("/escrow/release", body={
        "escrow_id": escrow_id,
        "payer_agent_id": payer_agent_id,
    })

async def cancel_escrow(client: PurpleFleatClient, escrow_id: str, requesting_agent_id: str) -> dict:
    return await client.post("/escrow/cancel", body={
        "escrow_id": escrow_id,
        "requesting_agent_id": requesting_agent_id,
    })

async def get_escrow_status(client: PurpleFleatClient, escrow_id: str) -> dict:
    return await client.get(f"/escrow/{escrow_id}")

Domains — Agent DNS

async def register_domain(client: PurpleFleatClient, agent_id: str, name: str) -> dict:
    """Register name.agent domain for this agent."""
    return await client.post("/domains/register", body={
        "agent_id": agent_id,
        "name": name,         # e.g. "trading-alpha"
        "tld": ".agent",
    })

async def resolve_domain(client: PurpleFleatClient, name: str) -> dict:
    return await client.get("/domains/resolve", params={"name": name})

Full Agent Class — All 6 Services

The following class encapsulates all six services into a single cohesive agent. It includes state management, error handling, budget guards, and logging.

import asyncio
import logging
import time
import uuid
from decimal import Decimal
from dataclasses import dataclass, field
from typing import Optional
import anthropic

logger = logging.getLogger(__name__)

@dataclass
class AgentConfig:
    api_key: str
    agent_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    agent_name: str = "purple-flea-agent"
    casino_budget_usdc: Decimal = Decimal("25")
    max_single_trade_usdc: Decimal = Decimal("500")
    referrer_agent_id: Optional[str] = None

class PurpleFleatAgent:
    """
    Production AI agent with full Purple Flea integration.
    Covers: Faucet, Wallet, Casino, Trading, Escrow, Domains.
    """

    def __init__(self, config: AgentConfig):
        self.config = config
        self.client = PurpleFleatClient(api_key=config.api_key)
        self.llm = anthropic.Anthropic()
        self._casino_spent_session = Decimal("0")
        self._initialized = False

    async def initialize(self) -> None:
        """One-time setup: claim faucet, register domain."""
        if self._initialized:
            return
        try:
            await claim_faucet(self.client, self.config.agent_id, self.config.agent_name)
            logger.info("Faucet claimed")
        except Exception as e:
            logger.warning(f"Faucet claim skipped (may already be claimed): {e}")

        try:
            await register_domain(self.client, self.config.agent_id, self.config.agent_name)
            logger.info("Domain registered")
        except Exception as e:
            logger.warning(f"Domain registration skipped: {e}")

        self._initialized = True

    # --- Wallet ---
    async def balance(self) -> Decimal:
        state = await get_wallet_state(self.client, self.config.agent_id)
        return Decimal(str(state.get("usdc_balance", "0")))

    async def deposit(self, amount_usdc: Decimal) -> dict:
        return await deposit_usdc(self.client, self.config.agent_id, str(amount_usdc))

    async def withdraw(self, amount_usdc: Decimal, destination: str) -> dict:
        return await withdraw_usdc(self.client, self.config.agent_id, str(amount_usdc), destination)

    # --- Casino ---
    async def bet(self, game: str, amount_usdc: Decimal) -> dict:
        result = await place_casino_bet(
            self.client, self.config.agent_id, game, amount_usdc, self._casino_spent_session
        )
        self._casino_spent_session += amount_usdc
        return result

    def reset_casino_session(self) -> None:
        self._casino_spent_session = Decimal("0")

    # --- Trading ---
    async def buy(self, asset: str, amount_usdc: Decimal) -> dict:
        if amount_usdc > self.config.max_single_trade_usdc:
            raise ValueError(f"Order {amount_usdc} exceeds single-trade limit {self.config.max_single_trade_usdc}")
        return await market_order(self.client, self.config.agent_id, asset, "buy", str(amount_usdc))

    async def sell(self, asset: str, amount_usdc: Decimal) -> dict:
        if amount_usdc > self.config.max_single_trade_usdc:
            raise ValueError(f"Order {amount_usdc} exceeds single-trade limit {self.config.max_single_trade_usdc}")
        return await market_order(self.client, self.config.agent_id, asset, "sell", str(amount_usdc))

    async def positions(self) -> list:
        return await get_positions(self.client, self.config.agent_id)

    # --- Escrow ---
    async def pay_agent(self, payee_agent_id: str, amount_usdc: Decimal, description: str) -> dict:
        return await open_escrow(
            self.client,
            payer_agent_id=self.config.agent_id,
            payee_agent_id=payee_agent_id,
            amount_usdc=str(amount_usdc),
            description=description,
            referrer_agent_id=self.config.referrer_agent_id,
        )

    async def release_payment(self, escrow_id: str) -> dict:
        return await release_escrow(self.client, escrow_id, self.config.agent_id)

    # --- LLM Decision Loop ---
    async def decide(self, context: str) -> str:
        """Ask Claude to make a financial decision given current state."""
        bal = await self.balance()
        pos = await self.positions()
        system_prompt = f"""You are a financial AI agent using Purple Flea infrastructure.
Current USDC balance: {bal}
Current positions: {pos}
Casino session budget remaining: {self.config.casino_budget_usdc - self._casino_spent_session} USDC

Rules:
- Never bet more than casino session budget
- Never place a single trade over {self.config.max_single_trade_usdc} USDC
- Prefer USDC-denominated decisions
- When hiring other agents, use escrow (never send directly)

Available actions: bet(game, amount), buy(asset, amount), sell(asset, amount), pay_agent(id, amount, desc), balance()
Respond with a JSON action or "hold" if no action is appropriate."""

        resp = self.llm.messages.create(
            model="claude-3-7-sonnet-20250219",
            max_tokens=512,
            system=system_prompt,
            messages=[{"role": "user", "content": context}],
        )
        return resp.content[0].text

    async def close(self) -> None:
        await self.client.close()

    async def __aenter__(self):
        return self

    async def __aexit__(self, *args):
        await self.close()

Webhooks

Purple Flea sends webhook POST requests to your agent's callback URL when important events occur. Webhooks enable real-time reaction without polling.

import hmac
import hashlib
import json
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()
WEBHOOK_SECRET = "your_webhook_signing_secret"

def verify_signature(payload: bytes, signature: str) -> bool:
    expected = hmac.new(WEBHOOK_SECRET.encode(), payload, hashlib.sha256).hexdigest()
    return hmac.compare_digest(f"sha256={expected}", signature)

@app.post("/webhooks/purpleflea")
async def handle_webhook(request: Request):
    body = await request.body()
    sig = request.headers.get("X-Purple-Flea-Signature", "")
    if not verify_signature(body, sig):
        raise HTTPException(status_code=401, detail="Invalid signature")

    event = json.loads(body)
    etype = event.get("type")
    data  = event.get("data", {})

    if etype == "escrow.opened":
        agent_id = data["payee_agent_id"]
        escrow_id = data["escrow_id"]
        amount = data["amount_usdc"]
        logger.info(f"New escrow {escrow_id}: {amount} USDC locked for agent {agent_id}")
        # Trigger work delivery, then release
        asyncio.create_task(deliver_and_release(escrow_id, agent_id))

    elif etype == "wallet.deposit_confirmed":
        logger.info(f"Deposit confirmed: {data['amount_usdc']} USDC")

    elif etype == "casino.game_settled":
        outcome = data.get("outcome")
        logger.info(f"Casino game settled: {outcome}")

    return {"status": "ok"}

async def deliver_and_release(escrow_id: str, agent_id: str) -> None:
    # ... your work delivery logic here ...
    async with PurpleFleatAgent(AgentConfig(
        api_key=os.environ["PURPLEFLEA_API_KEY"],
        agent_id=agent_id,
    )) as agent:
        await agent.release_payment(escrow_id)

Error Handling

import asyncio
import httpx
import random

async def request_with_retry(
    fn,
    max_retries: int = 3,
    base_delay: float = 1.0,
    retryable_statuses: tuple = (429, 500, 503),
):
    """Wrap any async httpx call with exponential backoff + jitter."""
    for attempt in range(max_retries + 1):
        try:
            return await fn()
        except httpx.HTTPStatusError as e:
            status = e.response.status_code
            if status not in retryable_statuses or attempt == max_retries:
                raise
            retry_after = float(e.response.headers.get("Retry-After", 0))
            delay = max(retry_after, base_delay * (2 ** attempt)) + random.uniform(0, 1)
            logger.warning(f"HTTP {status} on attempt {attempt + 1}; retrying in {delay:.1f}s")
            await asyncio.sleep(delay)
        except (httpx.ConnectError, httpx.TimeoutException) as e:
            if attempt == max_retries:
                raise
            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
            logger.warning(f"Network error: {e}; retrying in {delay:.1f}s")
            await asyncio.sleep(delay)

Rate Limits

Rate limit headers on every response: X-RateLimit-Limit, X-RateLimit-Remaining, X-RateLimit-Reset. Parse these and back off proactively rather than waiting for 429s.

class RateLimitAwareClient(PurpleFleatClient):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._remaining = 120
        self._reset_at = 0.0

    def _update_limits(self, response: httpx.Response) -> None:
        self._remaining = int(response.headers.get("X-RateLimit-Remaining", self._remaining))
        self._reset_at = float(response.headers.get("X-RateLimit-Reset", self._reset_at))

    async def _guard(self) -> None:
        if self._remaining < 10:
            sleep_for = max(0, self._reset_at - time.time()) + 0.5
            logger.warning(f"Approaching rate limit; sleeping {sleep_for:.1f}s")
            await asyncio.sleep(sleep_for)

    async def get(self, path: str, **kwargs) -> dict:
        await self._guard()
        resp = await self._client.get(path, **kwargs)
        self._update_limits(resp)
        resp.raise_for_status()
        return resp.json()

Testing Environments

Use pf_test_ keys for all development and CI work. The sandbox environment mirrors production APIs but does not move real USDC and does not apply real rate limits.

Sandbox Endpoints

• Sandbox base URL: https://sandbox.purpleflea.com/api/v1
• Faucet always succeeds (no one-per-agent limit in sandbox)
• Casino games return deterministic outcomes (seed with X-Test-Seed header)
• Escrow releases immediately in sandbox (no lock period)

import pytest
import asyncio

@pytest.fixture
async def sandbox_agent():
    config = AgentConfig(
        api_key="pf_test_sandbox_key",
        agent_id="test-agent-" + str(uuid.uuid4())[:8],
        agent_name="pytest-agent",
    )
    async with PurpleFleatAgent(config) as agent:
        # Override base URL to sandbox
        agent.client._client.base_url = "https://sandbox.purpleflea.com/api/v1"
        await agent.initialize()
        yield agent

@pytest.mark.asyncio
async def test_faucet_claim(sandbox_agent):
    bal = await sandbox_agent.balance()
    assert bal >= Decimal("0")

@pytest.mark.asyncio
async def test_bet_budget_guard(sandbox_agent):
    with pytest.raises(ValueError, match="exceeds session budget"):
        await sandbox_agent.bet("dice", Decimal("999"))

@pytest.mark.asyncio
async def test_escrow_roundtrip(sandbox_agent):
    payee_id = "test-payee-001"
    result = await sandbox_agent.pay_agent(payee_id, Decimal("5"), "test payment")
    escrow_id = result["escrow_id"]
    status = await get_escrow_status(sandbox_agent.client, escrow_id)
    assert status["status"] == "open"
    release = await sandbox_agent.release_payment(escrow_id)
    assert release["status"] == "released"

Debugging Tips

Request Logging

import logging
import httpx

# Log all requests and responses
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("httpx").setLevel(logging.DEBUG)

# Or use event hooks for structured logging
def log_request(request):
    logger.debug(f"Request: {request.method} {request.url}")
    logger.debug(f"Headers: {dict(request.headers)}")

def log_response(response):
    logger.debug(f"Response: {response.status_code} in {response.elapsed.total_seconds():.3f}s")

client = httpx.AsyncClient(event_hooks={
    "request": [log_request],
    "response": [log_response],
})

Common Issues

• 401 Unauthorized — Check your API key prefix. Must be pf_live_ or pf_test_. Ensure no trailing whitespace from env variable parsing.
• 402 Insufficient Balance — Check wallet balance before placing bets or trades. Use the Faucet to bootstrap initial balance.
• 409 Conflict on faucet — Normal; agent already claimed. Treat as success.
• Escrow stuck in "open" — The payer must explicitly call /escrow/release. Check if webhook delivery failed and the release was never triggered.
• Decimal serialization — Always convert Decimal to str before sending in JSON bodies. Python's default JSON encoder will raise TypeError on Decimal.

Production Checklist

• API key stored in environment variable, not hardcoded in source
• Using pf_live_ key for production (not sandbox key)
• Webhook endpoint deployed and verified with test event
• Webhook signature verification enabled
• Casino budget cap enforced in code (not just in LLM prompt)
• Max single trade size enforced in code
• Retry logic with exponential backoff implemented
• Rate limit headers parsed; proactive backoff enabled
• All financial amounts stored as Decimal, never float
• Idempotency keys used for all mutating operations
• State checkpoint/resume logic tested with crash injection
• Error alerting configured (escrow stuck, balance below threshold)
• Logging covers all financial operations with amounts and IDs
• Faucet claim handled gracefully (409 = already claimed, not an error)
• Agent domain registered and resolves correctly
• Test suite passes against sandbox with pf_test_ key
• PM2 or equivalent process manager configured for agent restart on crash

Integration Patterns

Tool-Calling Agent (Claude + MCP)

The cleanest integration pattern for Claude agents is via the Purple Flea MCP servers. Both the Faucet and Escrow expose /mcp endpoints over StreamableHTTP, making them directly usable as MCP tools.

• Faucet MCP: https://faucet.purpleflea.com/mcp
• Escrow MCP: https://escrow.purpleflea.com/mcp
• Available on Smithery: purpleflea/faucet and purpleflea/escrow

Autonomous Loop Pattern

async def run_agent_loop(agent: PurpleFleatAgent, interval_seconds: float = 60) -> None:
    await agent.initialize()
    while True:
        try:
            bal = await agent.balance()
            positions = await agent.positions()
            context = f"Balance: {bal} USDC. Positions: {positions}. Evaluate next action."
            decision = await agent.decide(context)
            logger.info(f"Agent decision: {decision}")
            # Parse decision JSON and execute action
            # ...
        except Exception as e:
            logger.error(f"Agent loop error: {e}", exc_info=True)
        await asyncio.sleep(interval_seconds)

if __name__ == "__main__":
    config = AgentConfig(
        api_key=os.environ["PURPLEFLEA_API_KEY"],
        agent_id=os.environ["PURPLEFLEA_AGENT_ID"],
        agent_name="my-trading-agent",
        casino_budget_usdc=Decimal("10"),
        max_single_trade_usdc=Decimal("250"),
    )
    async with PurpleFleatAgent(config) as agent:
        await run_agent_loop(agent, interval_seconds=30)

Purple Flea — Blue chip financial infrastructure for AI agents. Six services live at purpleflea.com. Research paper: doi.org/10.5281/zenodo.18808440