Why FastAPI + Purple Flea

The fastest stack for
agent payment backends

FastAPI's async-native design pairs perfectly with Purple Flea's REST APIs. You get type safety, automatic OpenAPI docs, and non-blocking HTTP calls to escrow and trading endpoints — all in one cohesive Python stack.

⚡

Fully Async

Purple Flea API calls via httpx.AsyncClient never block your event loop. Handle hundreds of concurrent agent payments without threads.

🔒

Pydantic Models

Every Purple Flea request and response is modelled with Pydantic v2. Automatic validation, serialization, and OpenAPI schema generation included.

🔧

Dependency Injection

The Purple Flea API client is a FastAPI dependency — injected once per request, shared across all endpoints, with lifecycle management built in.

📋

Background Tasks

Long-running escrow polls and trade status checks run in FastAPI BackgroundTasks — freeing the HTTP response immediately.

📊

OpenAPI Auto-Docs

FastAPI generates Swagger UI and ReDoc automatically from your Pydantic models. Your agent clients always have accurate API documentation.

🚀

Production Ready

Deploy with Uvicorn + Gunicorn, containerize with Docker, run on any cloud. Purple Flea APIs are globally accessible with 99.9% uptime SLA.

Installation

Get started in 60 seconds

Install dependencies, set your API key, and you have a running escrow endpoint.

terminal

# Create and activate virtual environment
python -m venv .venv && source .venv/bin/activate

# Install FastAPI, Uvicorn, and async HTTP client
pip install fastapi uvicorn httpx pydantic python-dotenv

# Create environment file
echo "PURPLE_FLEA_API_KEY=pf_live_your_key_here" > .env
echo "PURPLE_FLEA_BASE_URL=https://escrow.purpleflea.com" >> .env

# Start development server
uvicorn main:app --reload --host 0.0.0.0 --port 8000

Complete Application

Full FastAPI app with
Purple Flea integration

A production-ready FastAPI application with dependency injection, Pydantic models, background tasks, and all Purple Flea service endpoints.

main.py

"""
Purple Flea + FastAPI — Complete Integration
Handles escrow creation, funding, release, and trading.
"""

from __future__ import annotations

import asyncio
import logging
import os
from contextlib import asynccontextmanager
from datetime import datetime
from typing import Annotated, Any

import httpx
from dotenv import load_dotenv
from fastapi import (
    BackgroundTasks,
    Depends,
    FastAPI,
    HTTPException,
    Query,
    Request,
    status,
)
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field, field_validator

load_dotenv()

# ─────────────────────────────────────────────
#  Configuration
# ─────────────────────────────────────────────

API_KEY: str = os.getenv("PURPLE_FLEA_API_KEY", "")
ESCROW_BASE: str = os.getenv(
    "PURPLE_FLEA_BASE_URL", "https://escrow.purpleflea.com"
)
TRADING_BASE: str = "https://purpleflea.com/api/trading"
WALLET_BASE: str = "https://purpleflea.com/api/wallet"
FAUCET_BASE: str = "https://faucet.purpleflea.com"

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("purple_flea_api")

# ─────────────────────────────────────────────
#  Purple Flea API Client
# ─────────────────────────────────────────────

class PurpleFlealClient:
    """Async HTTP client for all Purple Flea services."""

    def __init__(self, api_key: str) -> None:
        self.api_key = api_key
        self._client: httpx.AsyncClient | None = None

    async def __aenter__(self) -> PurpleFlealClient:
        self._client = httpx.AsyncClient(
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
                "User-Agent": "PurpleFlea-FastAPI/1.0",
            },
            timeout=httpx.Timeout(30.0, connect=5.0),
        )
        return self

    async def __aexit__(self, *exc) -> None:
        if self._client:
            await self._client.aclose()

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

    async def post(self, url: str, **kwargs) -> dict[str, Any]:
        resp = await self._client.post(url, **kwargs)
        resp.raise_for_status()
        return resp.json()

    # ── Escrow methods ───────────────────────────

    async def create_escrow(
        self,
        payer: str,
        payee: str,
        amount_usdc: float,
        description: str,
        referral_code: str | None = None,
    ) -> dict[str, Any]:
        payload = {
            "payer": payer,
            "payee": payee,
            "amount": amount_usdc,
            "currency": "USDC",
            "description": description,
        }
        if referral_code:
            payload["referral_code"] = referral_code
        return await self.post(f"{ESCROW_BASE}/api/escrow/create", json=payload)

    async def release_escrow(self, escrow_id: str) -> dict[str, Any]:
        return await self.post(
            f"{ESCROW_BASE}/api/escrow/{escrow_id}/release"
        )

    async def cancel_escrow(self, escrow_id: str) -> dict[str, Any]:
        return await self.post(
            f"{ESCROW_BASE}/api/escrow/{escrow_id}/cancel"
        )

    async def get_escrow(self, escrow_id: str) -> dict[str, Any]:
        return await self.get(f"{ESCROW_BASE}/api/escrow/{escrow_id}")

    async def list_escrows(self, agent_id: str) -> dict[str, Any]:
        return await self.get(
            f"{ESCROW_BASE}/api/escrow",
            params={"agent_id": agent_id},
        )

    # ── Trading methods ──────────────────────────

    async def place_trade(
        self,
        agent_id: str,
        market: str,
        side: str,
        amount_usdc: float,
        leverage: int = 1,
    ) -> dict[str, Any]:
        return await self.post(
            f"{TRADING_BASE}/orders",
            json={
                "agent_id": agent_id,
                "market": market,
                "side": side,
                "amount": amount_usdc,
                "leverage": leverage,
            },
        )

    async def get_positions(self, agent_id: str) -> dict[str, Any]:
        return await self.get(
            f"{TRADING_BASE}/positions",
            params={"agent_id": agent_id},
        )

    # ── Wallet methods ───────────────────────────

    async def get_balance(self, agent_id: str) -> dict[str, Any]:
        return await self.get(
            f"{WALLET_BASE}/balance",
            params={"agent_id": agent_id},
        )

    async def claim_faucet(self, agent_id: str) -> dict[str, Any]:
        return await self.post(
            f"{FAUCET_BASE}/api/claim",
            json={"agent_id": agent_id},
        )


# ─────────────────────────────────────────────
#  Dependency: shared httpx client per request
# ─────────────────────────────────────────────

async def get_pf_client() -> PurpleFlealClient:
    """FastAPI dependency — injects the Purple Flea API client."""
    async with PurpleFlealClient(API_KEY) as client:
        yield client

PFClient = Annotated[PurpleFlealClient, Depends(get_pf_client)]

# ─────────────────────────────────────────────
#  Pydantic request / response models
# ─────────────────────────────────────────────

class CreateEscrowRequest(BaseModel):
    payer_agent_id: str = Field(..., min_length=3, description="Agent ID of payer")
    payee_agent_id: str = Field(..., min_length=3, description="Agent ID of payee")
    amount_usdc: float = Field(..., gt=0, le=100_000, description="USDC amount")
    description: str = Field(..., max_length=280)
    referral_code: str | None = Field(None)

    @field_validator("amount_usdc")
    @classmethod
    def round_usdc(cls, v: float) -> float:
        return round(v, 2)


class EscrowResponse(BaseModel):
    escrow_id: str
    status: str
    payer: str
    payee: str
    amount_usdc: float
    fee_usdc: float
    created_at: str
    released_at: str | None = None


class PlaceTradeRequest(BaseModel):
    agent_id: str = Field(..., min_length=3)
    market: str = Field(..., description="e.g. BTC-PERP, ETH-PERP")
    side: str = Field(..., pattern=r"^(long|short)$")
    amount_usdc: float = Field(..., gt=1, le=50_000)
    leverage: int = Field(1, ge=1, le=20)


class TradeResponse(BaseModel):
    order_id: str
    status: str
    market: str
    side: str
    amount_usdc: float
    leverage: int
    entry_price: float | None = None
    created_at: str


class BalanceResponse(BaseModel):
    agent_id: str
    balance_usdc: float
    locked_usdc: float
    available_usdc: float


class FaucetClaimResponse(BaseModel):
    agent_id: str
    claimed_usdc: float
    message: str


# ─────────────────────────────────────────────
#  Background task: poll escrow status
# ─────────────────────────────────────────────

async def poll_escrow_until_complete(
    escrow_id: str,
    api_key: str,
    max_retries: int = 30,
    interval: float = 2.0,
) -> None:
    """Background task: polls escrow until released/cancelled/timeout."""
    async with PurpleFlealClient(api_key) as client:
        for attempt in range(max_retries):
            try:
                data = await client.get_escrow(escrow_id)
                escrow_status = data.get("status", "unknown")
                log.info(f"Escrow {escrow_id} status: {escrow_status} (attempt {attempt+1})")
                if escrow_status in ("released", "cancelled", "completed"):
                    log.info(f"Escrow {escrow_id} finalized: {escrow_status}")
                    return
            except Exception as exc:
                log.warning(f"Poll error for {escrow_id}: {exc}")
            await asyncio.sleep(interval)
    log.warning(f"Escrow {escrow_id} polling timed out after {max_retries} attempts")


# ─────────────────────────────────────────────
#  App lifespan (startup / shutdown)
# ─────────────────────────────────────────────

@asynccontextmanager
async def lifespan(app: FastAPI):
    if not API_KEY:
        log.warning("PURPLE_FLEA_API_KEY not set — API calls will fail")
    log.info("Purple Flea FastAPI server starting")
    yield
    log.info("Purple Flea FastAPI server shutdown")


# ─────────────────────────────────────────────
#  FastAPI application
# ─────────────────────────────────────────────

app = FastAPI(
    title="Purple Flea Agent API",
    description="AI agent financial infrastructure — escrow, trading, wallet",
    version="1.0.0",
    lifespan=lifespan,
    docs_url="/docs",
    redoc_url="/redoc",
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)


async def handle_httpx_error(exc: httpx.HTTPStatusError) -> JSONResponse:
    return JSONResponse(
        status_code=exc.response.status_code,
        content={"error": "Purple Flea API error", "detail": str(exc)},
    )


# ─────────────────────────────────────────────
#  Escrow endpoints
# ─────────────────────────────────────────────

@app.post(
    "/api/escrow",
    response_model=EscrowResponse,
    status_code=status.HTTP_201_CREATED,
    tags=["escrow"],
    summary="Create a new escrow between two agents",
)
async def create_escrow(
    body: CreateEscrowRequest,
    background_tasks: BackgroundTasks,
    pf: PFClient,
) -> EscrowResponse:
    """
    Create a trustless USDC escrow between two AI agents.

    - **payer_agent_id**: the agent locking funds (1% fee applies)
    - **payee_agent_id**: the agent receiving funds on release
    - **amount_usdc**: gross amount; net = amount × 0.99
    - **referral_code**: optional referral code for 15% fee rebate to referrer
    """
    try:
        data = await pf.create_escrow(
            payer=body.payer_agent_id,
            payee=body.payee_agent_id,
            amount_usdc=body.amount_usdc,
            description=body.description,
            referral_code=body.referral_code,
        )
    except httpx.HTTPStatusError as exc:
        raise HTTPException(status_code=exc.response.status_code, detail=str(exc))

    escrow_id = data["escrow_id"]

    # Start background poll — returns HTTP 201 immediately
    background_tasks.add_task(
        poll_escrow_until_complete, escrow_id=escrow_id, api_key=API_KEY
    )

    return EscrowResponse(
        escrow_id=escrow_id,
        status=data.get("status", "pending"),
        payer=body.payer_agent_id,
        payee=body.payee_agent_id,
        amount_usdc=body.amount_usdc,
        fee_usdc=round(body.amount_usdc * 0.01, 2),
        created_at=datetime.utcnow().isoformat() + "Z",
    )


@app.get(
    "/api/escrow/{escrow_id}",
    response_model=EscrowResponse,
    tags=["escrow"],
    summary="Fetch escrow status by ID",
)
async def get_escrow(escrow_id: str, pf: PFClient) -> EscrowResponse:
    try:
        data = await pf.get_escrow(escrow_id)
    except httpx.HTTPStatusError as exc:
        raise HTTPException(status_code=exc.response.status_code, detail=str(exc))
    return EscrowResponse(**data)


@app.post(
    "/api/escrow/{escrow_id}/release",
    response_model=EscrowResponse,
    tags=["escrow"],
    summary="Release escrow funds to payee",
)
async def release_escrow(escrow_id: str, pf: PFClient) -> EscrowResponse:
    """
    Release funds from escrow to payee. Only the payer or an
    authorized agent may call this endpoint.
    Settlement occurs in USDC in under 2 seconds.
    """
    try:
        data = await pf.release_escrow(escrow_id)
    except httpx.HTTPStatusError as exc:
        raise HTTPException(status_code=exc.response.status_code, detail=str(exc))
    return EscrowResponse(**data)


@app.post(
    "/api/escrow/{escrow_id}/cancel",
    response_model=EscrowResponse,
    tags=["escrow"],
    summary="Cancel an open escrow and refund payer",
)
async def cancel_escrow(escrow_id: str, pf: PFClient) -> EscrowResponse:
    try:
        data = await pf.cancel_escrow(escrow_id)
    except httpx.HTTPStatusError as exc:
        raise HTTPException(status_code=exc.response.status_code, detail=str(exc))
    return EscrowResponse(**data)


@app.get(
    "/api/escrow",
    tags=["escrow"],
    summary="List all escrows for an agent",
)
async def list_escrows(
    agent_id: str = Query(..., description="Agent ID to list escrows for"),
    pf: PFClient = Depends(get_pf_client),
) -> dict[str, Any]:
    try:
        return await pf.list_escrows(agent_id)
    except httpx.HTTPStatusError as exc:
        raise HTTPException(status_code=exc.response.status_code, detail=str(exc))


# ─────────────────────────────────────────────
#  Trading endpoints
# ─────────────────────────────────────────────

@app.post(
    "/api/trading/orders",
    response_model=TradeResponse,
    status_code=status.HTTP_201_CREATED,
    tags=["trading"],
    summary="Place a perpetual futures order",
)
async def place_trade(body: PlaceTradeRequest, pf: PFClient) -> TradeResponse:
    """
    Place a long or short perpetual futures order on behalf of an agent.
    Markets: BTC-PERP, ETH-PERP, SOL-PERP. Leverage 1x–20x.
    """
    try:
        data = await pf.place_trade(
            agent_id=body.agent_id,
            market=body.market,
            side=body.side,
            amount_usdc=body.amount_usdc,
            leverage=body.leverage,
        )
    except httpx.HTTPStatusError as exc:
        raise HTTPException(status_code=exc.response.status_code, detail=str(exc))
    return TradeResponse(**data)


@app.get(
    "/api/trading/positions",
    tags=["trading"],
    summary="Get open positions for an agent",
)
async def get_positions(
    agent_id: str = Query(...),
    pf: PFClient = Depends(get_pf_client),
) -> dict[str, Any]:
    try:
        return await pf.get_positions(agent_id)
    except httpx.HTTPStatusError as exc:
        raise HTTPException(status_code=exc.response.status_code, detail=str(exc))


# ─────────────────────────────────────────────
#  Wallet endpoints
# ─────────────────────────────────────────────

@app.get(
    "/api/wallet/balance",
    response_model=BalanceResponse,
    tags=["wallet"],
    summary="Get USDC balance for an agent",
)
async def get_balance(
    agent_id: str = Query(...),
    pf: PFClient = Depends(get_pf_client),
) -> BalanceResponse:
    try:
        data = await pf.get_balance(agent_id)
    except httpx.HTTPStatusError as exc:
        raise HTTPException(status_code=exc.response.status_code, detail=str(exc))
    return BalanceResponse(**data)


@app.post(
    "/api/faucet/claim",
    response_model=FaucetClaimResponse,
    tags=["faucet"],
    summary="Claim free USDC from the Purple Flea faucet",
)
async def claim_faucet(
    agent_id: str = Query(..., description="Agent ID to receive free USDC"),
    pf: PFClient = Depends(get_pf_client),
) -> FaucetClaimResponse:
    """
    New agents can claim free USDC from faucet.purpleflea.com.
    One claim per agent. Use the funds to try the casino or fund escrow.
    """
    try:
        data = await pf.claim_faucet(agent_id)
    except httpx.HTTPStatusError as exc:
        raise HTTPException(status_code=exc.response.status_code, detail=str(exc))
    return FaucetClaimResponse(**data)


# ─────────────────────────────────────────────
#  Health check
# ─────────────────────────────────────────────

@app.get("/health", tags=["meta"])
async def health() -> dict[str, str]:
    return {"status": "ok", "service": "purple-flea-fastapi"}

Request / Response Models

Pydantic-validated
API contracts

Every Purple Flea API call is wrapped in Pydantic v2 models — giving you automatic validation, serialization, and free OpenAPI schema documentation.

models/escrow.py

from pydantic import BaseModel, Field
from typing import Literal

class CreateEscrowRequest(BaseModel):
    payer: str = Field(
        ..., min_length=3,
        description="Paying agent ID"
    )
    payee: str = Field(
        ..., min_length=3,
        description="Receiving agent ID"
    )
    amount_usdc: float = Field(
        ..., gt=0, le=100_000
    )
    description: str = Field(..., max_length=280)
    referral_code: str | None = None

class EscrowStatus(BaseModel):
    escrow_id: str
    status: Literal[
        "pending",
        "funded",
        "released",
        "cancelled",
    ]
    amount_usdc: float
    fee_usdc: float  # always 1%
    payer: str
    payee: str
    created_at: str
    released_at: str | None = None

models/trading.py

from pydantic import BaseModel, Field
from typing import Literal

Market = Literal[
    "BTC-PERP",
    "ETH-PERP",
    "SOL-PERP",
    "DOGE-PERP",
]

class PlaceOrderRequest(BaseModel):
    agent_id: str
    market: Market
    side: Literal["long", "short"]
    amount_usdc: float = Field(
        ..., gt=1, le=50_000
    )
    leverage: int = Field(
        1, ge=1, le=20
    )

class OrderResponse(BaseModel):
    order_id: str
    status: Literal[
        "open",
        "filled",
        "cancelled",
    ]
    market: Market
    side: Literal["long", "short"]
    amount_usdc: float
    leverage: int
    entry_price: float | None = None
    pnl_usdc: float | None = None

Background Tasks

Non-blocking escrow
status polling

FastAPI's BackgroundTasks lets you return an HTTP 201 immediately after creating an escrow, while the server polls Purple Flea in the background. The client gets instant feedback — no hanging requests.

Combine with WebSockets or Server-Sent Events to push escrow status updates to agent clients in real time.

Return 201 immediately, poll in background
Configurable retry count and interval
Structured logging at every poll attempt
Auto-stops on released/cancelled status
Graceful timeout with warning log
Zero blocking — event loop stays free

tasks/poll_escrow.py

import asyncio, logging

log = logging.getLogger("escrow.poll")

async def poll_escrow_until_complete(
    escrow_id: str,
    api_key: str,
    on_complete=None,
    max_retries: int = 30,
    interval: float = 2.0,
) -> None:
    async with PurpleFlealClient(api_key) as pf:
        for n in range(max_retries):
            data = await pf.get_escrow(escrow_id)
            s = data["status"]
            log.info(f"{escrow_id}: {s} ({n+1}/{max_retries})")
            if s in {"released", "cancelled"}:
                if on_complete:
                    await on_complete(escrow_id, s, data)
                return
            await asyncio.sleep(interval)
        log.warning(f"{escrow_id} timed out")


# Usage in endpoint:
background_tasks.add_task(
    poll_escrow_until_complete,
    escrow_id=escrow_id,
    api_key=API_KEY,
    on_complete=notify_agent,
    max_retries=30,
    interval=2.0,
)
# HTTP 201 returned immediately above ^^^

API Reference

Endpoints exposed
by this FastAPI app

All routes map 1:1 to Purple Flea service APIs. FastAPI generates interactive Swagger UI at /docs automatically.

Method	Path	Description	Purple Flea Service
POST	/api/escrow	Create escrow between two agents	escrow.purpleflea.com
GET	/api/escrow/{id}	Fetch escrow status and details	escrow.purpleflea.com
POST	/api/escrow/{id}/release	Release funds to payee (<2s)	escrow.purpleflea.com
POST	/api/escrow/{id}/cancel	Cancel escrow and refund payer	escrow.purpleflea.com
GET	/api/escrow?agent_id=	List all escrows for an agent	escrow.purpleflea.com
POST	/api/trading/orders	Place perp futures order	Trading API
GET	/api/trading/positions	Get open positions	Trading API
GET	/api/wallet/balance	Get agent USDC balance	Wallet API
POST	/api/faucet/claim	Claim free USDC (new agents)	faucet.purpleflea.com

Setup Guide

From zero to live
escrow endpoint

01

Install dependencies

Run pip install fastapi uvicorn httpx pydantic python-dotenv. All packages are stable, PyPI-hosted, and actively maintained. No proprietary SDKs required.

02

Set your API key

Add PURPLE_FLEA_API_KEY=pf_live_... to your .env file. Get your key from your Purple Flea dashboard. Never commit API keys to version control.

03

Copy the integration code

Use the full main.py example above as your starting point. The PurpleFlealClient class handles authentication, retries, and JSON parsing for all Purple Flea services.

04

Run the dev server

Run uvicorn main:app --reload. Visit http://localhost:8000/docs for the auto-generated Swagger UI. Test escrow creation directly from the browser.

05

Deploy to production

Use uvicorn main:app --workers 4 or wrap with Gunicorn. Set PURPLE_FLEA_API_KEY in your environment. Purple Flea APIs are globally accessible — no VPN or IP whitelisting needed.

Deployment

Containerize with Docker

Ship your FastAPI + Purple Flea backend as a Docker container for consistent deployments on any cloud or bare-metal server.

Dockerfile

# Multi-stage build for minimal image size
FROM python:3.12-slim AS base

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

FROM base AS runtime
COPY . .

# Non-root user for security
RUN adduser --disabled-password --gecos "" agent
USER agent

EXPOSE 8000
CMD ["uvicorn", "main:app",
     "--host", "0.0.0.0",
     "--port", "8000",
     "--workers", "2"]

docker-compose.yml

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      PURPLE_FLEA_API_KEY: "pf_live_your_key"
      PURPLE_FLEA_BASE_URL: "https://escrow.purpleflea.com"
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f",
             "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

FastAPI meets
Agent Financial Infrastructure

The fastest stack for
agent payment backends

Fully Async

Pydantic Models

Dependency Injection

Background Tasks

OpenAPI Auto-Docs

Production Ready

Get started in 60 seconds

Full FastAPI app with
Purple Flea integration

Pydantic-validated
API contracts

Non-blocking escrow
status polling

Endpoints exposed
by this FastAPI app

From zero to live
escrow endpoint

Install dependencies

Set your API key

Copy the integration code

Run the dev server

Deploy to production

Containerize with Docker

Ready to build your
agent payment backend?

FastAPI meets Agent Financial Infrastructure

The fastest stack foragent payment backends

Fully Async

Pydantic Models

Dependency Injection

Background Tasks

OpenAPI Auto-Docs

Production Ready

Get started in 60 seconds

Full FastAPI app withPurple Flea integration

Pydantic-validatedAPI contracts

Non-blocking escrowstatus polling

Endpoints exposedby this FastAPI app

From zero to liveescrow endpoint

Install dependencies

Set your API key

Copy the integration code

Run the dev server

Deploy to production

Containerize with Docker

Ready to build youragent payment backend?

FastAPI meets
Agent Financial Infrastructure

The fastest stack for
agent payment backends

Full FastAPI app with
Purple Flea integration

Pydantic-validated
API contracts

Non-blocking escrow
status polling

Endpoints exposed
by this FastAPI app

From zero to live
escrow endpoint

Ready to build your
agent payment backend?