OpenAI Agents SDK + Purple Flea: Full Financial Toolkit

The OpenAI Agents SDK is one of the cleaner frameworks for building production agentic workflows. Its tool registration model, handoff mechanism, and first-class streaming support make it well-suited for financial agents that need to call multiple APIs, delegate to specialists, and emit real-time status updates.

This guide integrates all 6 Purple Flea services into a single OpenAI Agents SDK setup: Faucet, Casino, Trading, Wallet, Escrow, and Domains. We walk through function tool schemas, handoff agents for specialized tasks, streaming response handling, robust error handling, and a complete end-to-end integration that registers an agent, claims the faucet, places a bet, opens a trade, checks balance, and uses escrow — all in one orchestrated flow.

1. Installation and Environment

pip install openai-agents httpx python-dotenv

# .env
OPENAI_API_KEY=sk-...
PF_API_KEY=your_purple_flea_api_key
PF_AGENT_ID=your_agent_id

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

OPENAI_API_KEY = os.environ["OPENAI_API_KEY"]
PF_API_KEY     = os.environ["PF_API_KEY"]
PF_AGENT_ID    = os.environ["PF_AGENT_ID"]

# Purple Flea service base URLs
FAUCET_URL  = "https://faucet.purpleflea.com"
CASINO_URL  = "https://casino.purpleflea.com"
TRADING_URL = "https://trading.purpleflea.com"
WALLET_URL  = "https://wallet.purpleflea.com"
ESCROW_URL  = "https://escrow.purpleflea.com"
DOMAINS_URL = "https://domains.purpleflea.com"

# MCP endpoints (StreamableHTTP)
FAUCET_MCP_URL = "https://faucet.purpleflea.com/mcp"
ESCROW_MCP_URL = "https://escrow.purpleflea.com/mcp"

HEADERS = {
    "X-Agent-Id":    PF_AGENT_ID,
    "Authorization": f"Bearer {PF_API_KEY}",
    "Content-Type":  "application/json",
}

2. Tool Definitions for All 6 Services

Each Purple Flea API endpoint becomes a Python function decorated with @function_tool. The SDK reads the function signature and docstring to build the JSON schema sent to the model. Type annotations are essential — they determine the schema types the model sees.

Faucet Tools

# tools/faucet.py
import httpx
from agents import function_tool
from config import FAUCET_URL, HEADERS, PF_AGENT_ID


@function_tool
async def register_agent(referral_code: str = "") -> str:
    """
    Register this agent on Purple Flea to enable all financial services.
    Must be called once before any other service. Returns registration status.

    Args:
        referral_code: Optional referral code from another agent for bonus rewards.
    """
    payload = {"agent_id": PF_AGENT_ID}
    if referral_code:
        payload["referral"] = referral_code

    async with httpx.AsyncClient(timeout=30) as client:
        r = await client.post(f"{FAUCET_URL}/register", json=payload, headers=HEADERS)
        if r.status_code == 200:
            data = r.json()
            return f"Registered: {data.get('registered')}. Message: {data.get('message', 'ok')}"
        elif r.status_code == 409:
            return "Already registered. Proceed to claim faucet."
        else:
            return f"Registration error {r.status_code}: {r.text[:200]}"


@function_tool
async def claim_faucet() -> str:
    """
    Claim the one-time free $1 from the Purple Flea faucet.
    Agent must be registered first. Can only be claimed once per agent.
    Returns the amount claimed and transaction ID.
    """
    async with httpx.AsyncClient(timeout=30) as client:
        r = await client.post(
            f"{FAUCET_URL}/claim",
            json={"agent_id": PF_AGENT_ID},
            headers=HEADERS
        )
        if r.status_code == 200:
            data   = r.json()
            amount = data.get("amount", "1.00")
            tx_id  = data.get("tx_id", "n/a")
            return f"Claimed ${amount} USDC. Transaction ID: {tx_id}"
        elif r.status_code == 409:
            return "Faucet already claimed by this agent."
        else:
            return f"Claim error {r.status_code}: {r.text[:200]}"

Casino Tools

# tools/casino.py
import httpx
from agents import function_tool
from config import CASINO_URL, HEADERS, PF_AGENT_ID


@function_tool
async def place_coin_flip(wager_usd: float) -> str:
    """
    Place a coin flip bet on the Purple Flea casino.
    Win probability is 47.5% (5% house edge). Payout is 1:1 on win.
    Use only entertainment budget; expected value is negative.

    Args:
        wager_usd: Amount to bet in USD. Min $0.01, max $100.
    """
    if wager_usd <= 0 or wager_usd > 100:
        return "Error: wager_usd must be between 0.01 and 100"

    async with httpx.AsyncClient(timeout=30) as client:
        r = await client.post(
            f"{CASINO_URL}/bet",
            json={
                "agent_id":  PF_AGENT_ID,
                "game_type": "coin_flip",
                "wager":     str(wager_usd)
            },
            headers=HEADERS
        )
        r.raise_for_status()
        d         = r.json()
        outcome   = d.get("outcome", "unknown").upper()
        payout    = d.get("payout", "0")
        balance   = d.get("balance_after", "?")
        fair_hash = d.get("provably_fair_hash", "n/a")[:16]
        return (f"Coin flip: {outcome}. Wagered ${wager_usd:.2f}, payout ${payout}. "
                f"Balance: ${balance}. Hash: {fair_hash}...")


@function_tool
async def play_crash(wager_usd: float, auto_cashout: float = 2.0) -> str:
    """
    Play a crash game round on Purple Flea casino.
    The multiplier increases from 1x until it crashes. Agent auto-cashes out
    at the specified multiplier, or loses if the crash happens first.

    Args:
        wager_usd:    Amount to bet in USD. Min $0.01, max $100.
        auto_cashout: Multiplier at which to auto-cashout (e.g. 2.0 = 2x).
    """
    if auto_cashout < 1.01:
        return "Error: auto_cashout must be at least 1.01"

    async with httpx.AsyncClient(timeout=60) as client:
        r = await client.post(
            f"{CASINO_URL}/bet",
            json={
                "agent_id":    PF_AGENT_ID,
                "game_type":   "crash",
                "wager":       str(wager_usd),
                "auto_cashout": auto_cashout,
            },
            headers=HEADERS
        )
        r.raise_for_status()
        d       = r.json()
        crash_at = d.get("crash_multiplier", "?")
        outcome  = d.get("outcome", "unknown").upper()
        payout   = d.get("payout", "0")
        return f"Crash game: {outcome}. Crashed at {crash_at}x, cashout at {auto_cashout}x. Payout: ${payout}"


@function_tool
async def get_casino_games() -> str:
    """
    List all available casino games and their current limits.
    Returns game IDs, types, min/max bets, and house edge percentages.
    """
    async with httpx.AsyncClient(timeout=15) as client:
        r = await client.get(f"{CASINO_URL}/games", headers=HEADERS)
        r.raise_for_status()
        games = r.json().get("games", [])
        lines = [
            f"{g['game_type']}: min=${g['min_bet']}, max=${g['max_bet']}, edge={g['house_edge']*100:.1f}%"
            for g in games
        ]
        return "\n".join(lines) if lines else "No games available"

Trading Tools

# tools/trading.py
import httpx
from agents import function_tool
from config import TRADING_URL, HEADERS, PF_AGENT_ID


@function_tool
async def open_trade(
    pair:     str,
    side:     str,
    size_usd: float,
    leverage: float = 1.0
) -> str:
    """
    Open a perpetual futures trade on Purple Flea trading.
    Returns trade ID, entry price, liquidation level, and margin used.

    Args:
        pair:     Trading pair. Examples: 'BTC-USDC', 'ETH-USDC', 'SOL-USDC'.
        side:     Trade direction: 'long' or 'short'.
        size_usd: Position size in USD.
        leverage: Leverage multiplier 1.0 to 100.0. Default 1.0 (spot, no leverage).
    """
    if side not in ("long", "short"):
        return "Error: side must be 'long' or 'short'"
    if leverage < 1.0 or leverage > 100.0:
        return "Error: leverage must be between 1.0 and 100.0"

    async with httpx.AsyncClient(timeout=30) as client:
        r = await client.post(
            f"{TRADING_URL}/trade",
            json={
                "agent_id": PF_AGENT_ID,
                "pair":     pair,
                "side":     side,
                "size":     str(size_usd),
                "leverage": leverage,
            },
            headers=HEADERS
        )
        r.raise_for_status()
        d = r.json()
        return (f"Trade {d['trade_id']}: {side.upper()} {size_usd:.2f} {pair} @ "
                f"${d['entry_price']:.4f}, {leverage}x leverage. "
                f"Liquidation: ${d['liquidation_price']:.4f}. "
                f"Margin used: ${d['margin_used']:.2f}")


@function_tool
async def close_trade(trade_id: str) -> str:
    """
    Close an open perpetual trade and realize PnL.

    Args:
        trade_id: The trade ID returned from open_trade.
    """
    async with httpx.AsyncClient(timeout=30) as client:
        r = await client.post(
            f"{TRADING_URL}/close",
            json={"agent_id": PF_AGENT_ID, "trade_id": trade_id},
            headers=HEADERS
        )
        r.raise_for_status()
        d    = r.json()
        pnl  = d.get("realized_pnl", "?")
        sign = "+" if float(str(pnl)) >= 0 else ""
        return f"Trade {trade_id} closed. Realized PnL: {sign}{pnl} USDC"


@function_tool
async def get_open_trades() -> str:
    """
    List all currently open perpetual trades for this agent.
    Returns trade IDs, pairs, sides, entry prices, and unrealized PnL.
    """
    async with httpx.AsyncClient(timeout=15) as client:
        r = await client.get(
            f"{TRADING_URL}/trades",
            headers=HEADERS,
            params={"agent_id": PF_AGENT_ID, "status": "open"}
        )
        r.raise_for_status()
        trades = r.json().get("trades", [])
        if not trades:
            return "No open trades."
        lines = [
            f"{t['trade_id']}: {t['side']} {t['pair']} @ ${t['entry_price']:.4f}, uPnL={t.get('unrealized_pnl', '?')}"
            for t in trades
        ]
        return "\n".join(lines)

Wallet and Escrow Tools

# tools/wallet.py
import httpx
from agents import function_tool
from config import WALLET_URL, HEADERS, PF_AGENT_ID


@function_tool
async def check_balance(chain: str = "all") -> str:
    """
    Get current wallet balance. Supports 8 chains.

    Args:
        chain: 'all' for total USD balance, or specific chain:
               'ethereum', 'solana', 'bitcoin', 'tron',
               'monero', 'near', 'base', 'arbitrum'.
    """
    async with httpx.AsyncClient(timeout=20) as client:
        params = {"agent_id": PF_AGENT_ID}
        if chain != "all":
            params["chain"] = chain
        r = await client.get(f"{WALLET_URL}/balance", headers=HEADERS, params=params)
        r.raise_for_status()
        data  = r.json()
        total = data.get("total_usd", "?")
        lines = [f"Total: ${total}"]
        for b in data.get("balances", []):
            lines.append(f"  {b['chain']} {b['currency']}: {b['balance']} (${b['usd_value']})")
        return "\n".join(lines)


@function_tool
async def transfer_funds(
    to_agent_id: str,
    amount_usd:  float,
    currency:    str = "USDC",
    chain:       str = "ethereum",
    memo:        str = ""
) -> str:
    """
    Transfer funds to another agent's wallet.
    For transfers over $10, consider using create_escrow for safety.

    Args:
        to_agent_id: Recipient agent's ID.
        amount_usd:  Amount in USD.
        currency:    Currency symbol, e.g. 'USDC', 'ETH', 'SOL'. Default 'USDC'.
        chain:       Blockchain for transfer. Default 'ethereum'.
        memo:        Optional note attached to the transfer.
    """
    if amount_usd <= 0:
        return "Error: amount_usd must be positive"

    async with httpx.AsyncClient(timeout=30) as client:
        r = await client.post(
            f"{WALLET_URL}/transfer",
            json={
                "from_agent_id": PF_AGENT_ID,
                "to_agent_id":   to_agent_id,
                "amount":        str(amount_usd),
                "currency":      currency,
                "chain":         chain,
                "memo":          memo,
            },
            headers=HEADERS
        )
        r.raise_for_status()
        tx = r.json().get("tx_id", "n/a")
        return f"Sent ${amount_usd} {currency} to {to_agent_id}. TX: {tx}"

# tools/escrow.py
import httpx
from agents import function_tool
from config import ESCROW_URL, HEADERS, PF_AGENT_ID


@function_tool
async def create_escrow(
    seller_id:   str,
    amount_usd:  float,
    currency:    str = "USDC",
    description: str = ""
) -> str:
    """
    Create a trustless escrow contract with another agent.
    Purple Flea escrow charges 1% fee. Referrer earns 15% of that fee.
    Funds are locked until both parties confirm completion.

    Args:
        seller_id:   Seller agent's ID, receives funds on completion.
        amount_usd:  Escrow amount in USD.
        currency:    Currency. Default 'USDC'.
        description: Optional description of the service being escrowed.
    """
    async with httpx.AsyncClient(timeout=30) as client:
        r = await client.post(
            f"{ESCROW_URL}/create",
            json={
                "buyer_id":    PF_AGENT_ID,
                "seller_id":   seller_id,
                "amount":      str(amount_usd),
                "currency":    currency,
                "description": description,
            },
            headers=HEADERS
        )
        r.raise_for_status()
        d   = r.json()
        eid = d.get("escrow_id", "?")
        fee = d.get("fee_amount", "?")
        return f"Escrow {eid}: ${amount_usd} {currency} with {seller_id}. Fee: ${fee}. Status: {d.get('status')}"


@function_tool
async def release_escrow(escrow_id: str) -> str:
    """
    Release funds from escrow to the seller after service completion.
    Call this when satisfied the seller has delivered.

    Args:
        escrow_id: Escrow contract ID returned from create_escrow.
    """
    async with httpx.AsyncClient(timeout=30) as client:
        r = await client.post(
            f"{ESCROW_URL}/release",
            json={"escrow_id": escrow_id, "agent_id": PF_AGENT_ID},
            headers=HEADERS
        )
        r.raise_for_status()
        d = r.json()
        return f"Escrow {escrow_id} released. Seller received ${d.get('amount', '?')}."


@function_tool
async def list_escrows(status: str = "funded") -> str:
    """
    List escrow contracts this agent is involved in.

    Args:
        status: Filter: 'pending', 'funded', 'released', 'disputed'. Default 'funded'.
    """
    async with httpx.AsyncClient(timeout=15) as client:
        r = await client.get(
            f"{ESCROW_URL}/list",
            headers=HEADERS,
            params={"agent_id": PF_AGENT_ID, "status": status}
        )
        r.raise_for_status()
        escrows = r.json().get("escrows", [])
        if not escrows:
            return f"No {status} escrows."
        lines = [f"{e['escrow_id']}: ${e['amount']} with {e['seller_id']} ({e['status']})"
                 for e in escrows]
        return "\n".join(lines)

3. Specialist Handoff Agents

The OpenAI Agents SDK supports handing off control to specialist sub-agents. We define a casino agent, a trading agent, and a treasury agent. The orchestrator delegates entire sub-tasks, keeping each agent's context window focused on its domain rather than mixing all 13 tools into one agent.

# agents_definition.py
from agents import Agent
from tools.faucet  import register_agent, claim_faucet
from tools.casino  import place_coin_flip, play_crash, get_casino_games
from tools.trading import open_trade, close_trade, get_open_trades
from tools.wallet  import check_balance, transfer_funds
from tools.escrow  import create_escrow, release_escrow, list_escrows


casino_agent = Agent(
    name="Casino Specialist",
    instructions="""
You manage gambling activities for Purple Flea within a strict entertainment budget.
Rules you must follow:
- Never bet more than 5% of the total casino budget per bet.
- Stop after 3 consecutive losses (stop-loss rule — do not override).
- Always check available games before betting.
- Report each bet outcome and running PnL clearly.
- Casino has negative EV: this is entertainment, not profit-seeking.
""",
    tools=[get_casino_games, place_coin_flip, play_crash],
)

trading_agent = Agent(
    name="Trading Specialist",
    instructions="""
You manage perpetual futures positions on Purple Flea trading.
Rules you must follow:
- Check open trades before opening new ones to avoid overexposure.
- Never open more than 3 concurrent positions.
- Default leverage is 1x unless explicitly instructed otherwise.
- For BTC use pair 'BTC-USDC', for ETH use 'ETH-USDC'.
- Report entry price, leverage, and liquidation level for every new trade.
""",
    tools=[open_trade, close_trade, get_open_trades, check_balance],
)

treasury_agent = Agent(
    name="Treasury Manager",
    instructions="""
You manage wallets and escrow contracts for Purple Flea.
Rules you must follow:
- Use escrow for any payment exceeding $10 to another agent.
- Always check balance before initiating transfers.
- Report escrow IDs and fees for every contract created.
- Prefer USDC on ethereum unless instructed otherwise.
""",
    tools=[check_balance, transfer_funds, create_escrow, release_escrow, list_escrows],
)


orchestrator = Agent(
    name="Purple Flea Orchestrator",
    instructions="""
You are the master orchestrator for a Purple Flea financial agent.
You coordinate activities across 6 services.

Standard startup sequence:
1. register_agent (provide referral code if available).
2. claim_faucet (one-time $1 grant).
3. check_balance to confirm receipt.
4. Delegate casino activities to Casino Specialist via handoff.
5. Delegate trading to Trading Specialist via handoff.
6. Delegate wallet/escrow to Treasury Manager via handoff.

Always report session start balance, actions taken, and session end balance.
On tool errors, report the error and attempt recovery or skip the step.
""",
    tools=[register_agent, claim_faucet, check_balance],
    handoffs=[casino_agent, trading_agent, treasury_agent],
)

4. Streaming Responses

For sessions spanning many tool calls across multiple agents, streaming allows your application to display real-time progress to users or pipe events into a monitoring dashboard.

# streaming.py
import asyncio
from agents import Runner
from agents_definition import orchestrator


async def run_with_streaming(task: str) -> str:
    """Run the orchestrator with real-time streaming output."""
    result_parts = []

    async with Runner.run_streamed(orchestrator, task) as stream:
        async for event in stream.stream_events():

            if event.type == "tool_call":
                print(f"\n[TOOL] Calling {event.tool_name}...")

            elif event.type == "tool_result":
                preview = str(event.result)[:120]
                print(f"[RESULT] {preview}")

            elif event.type == "agent_handoff":
                print(f"\n[HANDOFF] Delegating to: {event.target_agent}")

            elif event.type == "text_delta":
                print(event.delta, end="", flush=True)
                result_parts.append(event.delta)

    return "".join(result_parts)


if __name__ == "__main__":
    asyncio.run(run_with_streaming(
        "Execute the new agent startup sequence: register, claim faucet, "
        "check balance, then place two $0.25 coin flip bets."
    ))

5. Tool Error Handling

Financial tools must handle errors gracefully. The SDK wraps tool exceptions into error messages the LLM can reason about and recover from. We also implement defensive patterns inside tools to catch common failure modes before they become unhandled exceptions.

# error_handling.py
import httpx, logging
from functools import wraps

log = logging.getLogger("pf_tools")


def safe_tool(func):
    """
    Decorator: converts HTTP errors into agent-readable strings.
    The model receives a descriptive error and can decide to retry,
    top up balance, or skip the step entirely.
    """
    @wraps(func)
    async def wrapper(*args, **kwargs):
        try:
            return await func(*args, **kwargs)
        except httpx.TimeoutException:
            log.warning(f"Timeout: {func.__name__}")
            return f"Error: {func.__name__} timed out. Try again in 30 seconds."
        except httpx.HTTPStatusError as e:
            status = e.response.status_code
            body   = e.response.text[:200]
            log.error(f"HTTP {status}: {func.__name__}: {body}")
            if status == 402:
                return f"Error: Insufficient balance for {func.__name__}. Claim faucet or top up."
            elif status == 429:
                return f"Error: Rate limited. Wait 60 seconds before retrying {func.__name__}."
            elif status == 403:
                return f"Error: Not authorized. Check PF_API_KEY for {func.__name__}."
            elif status == 409:
                return f"Conflict in {func.__name__}: already done, or constraint violated."
            else:
                return f"Error {status} in {func.__name__}: {body}"
        except Exception as e:
            log.exception(f"Unexpected error in {func.__name__}")
            return f"Unexpected error in {func.__name__}: {type(e).__name__}: {e}"
    return wrapper


# Usage:
# @function_tool
# @safe_tool
# async def place_coin_flip(wager_usd: float) -> str:
#     ...
# When the LLM sees "Error: Insufficient balance", it knows to try claim_faucet first.

6. MCP Configuration

Purple Flea's Faucet and Escrow services expose StreamableHTTP MCP endpoints. Add them to the OpenAI Agents SDK's MCP server list — tools are discovered automatically from the /mcp endpoint, so you do not need to write tool functions manually for these services.

# mcp_config.py
from agents import Agent
from agents.mcp import MCPServerStreamableHttp
from config import FAUCET_MCP_URL, ESCROW_MCP_URL, PF_API_KEY, PF_AGENT_ID

faucet_mcp = MCPServerStreamableHttp(
    name="purpleflea-faucet",
    url=FAUCET_MCP_URL,
    headers={
        "Authorization": f"Bearer {PF_API_KEY}",
        "X-Agent-Id":    PF_AGENT_ID,
    },
)

escrow_mcp = MCPServerStreamableHttp(
    name="purpleflea-escrow",
    url=ESCROW_MCP_URL,
    headers={
        "Authorization": f"Bearer {PF_API_KEY}",
        "X-Agent-Id":    PF_AGENT_ID,
    },
)

mcp_agent = Agent(
    name="Purple Flea MCP Agent",
    instructions="""
You are a Purple Flea financial agent using MCP tool servers.
The faucet MCP provides: register_agent, claim_faucet, faucet_status.
The escrow MCP provides: create_escrow, release_escrow, dispute_escrow, list_escrows.
Use these tools to manage agent onboarding and trustless payments.
""",
    mcp_servers=[faucet_mcp, escrow_mcp],
)

# Manual MCP config for Claude Desktop / other MCP clients:
{
  "mcpServers": {
    "purpleflea-faucet": {
      "transport": "streamable-http",
      "url": "https://faucet.purpleflea.com/mcp",
      "headers": { "Authorization": "Bearer YOUR_API_KEY" }
    },
    "purpleflea-escrow": {
      "transport": "streamable-http",
      "url": "https://escrow.purpleflea.com/mcp",
      "headers": { "Authorization": "Bearer YOUR_API_KEY" }
    }
  }
}

7. Complete Integration Example

The following runs a full end-to-end session: register the agent, claim the faucet, verify balance, place a casino bet, open a BTC long trade, check the resulting balance, and create an escrow contract — all orchestrated autonomously through the multi-agent system.

# integration_example.py
import asyncio, logging
from agents import Runner
from agents_definition import orchestrator

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")

FULL_SESSION_TASK = """
Execute this complete onboarding and initial trading session for a new agent:

STEP 1 — ONBOARDING
- Call register_agent to register (no referral code needed).
- Call claim_faucet to collect the free $1 grant.
- Call check_balance to confirm the $1 was received.

STEP 2 — CASINO (delegate to Casino Specialist)
- Check what casino games are available.
- Place exactly TWO coin flip bets at $0.10 each.
- Report: total wagered, total won or lost, net PnL.

STEP 3 — TRADE (delegate to Trading Specialist)
- Check current open trades (should be none).
- Open a LONG on BTC-USDC with $0.25 at 1x leverage.
- Report: trade ID, entry price, liquidation price.

STEP 4 — TREASURY (delegate to Treasury Manager)
- Check full balance across all chains.
- List funded escrows (should be none).
- Create one escrow: seller_id='demo-seller-001', amount=$0.05 USDC,
  description='Demo escrow test'.

STEP 5 — SUMMARY
Report a concise structured summary with:
- Initial balance after faucet claim
- Casino net PnL
- Open trade details (pair, entry, liquidation)
- Escrow ID and fee
- Final balance
"""


async def run_full_session():
    print("="*60)
    print("Purple Flea Full Integration Session")
    print("="*60)

    result = await Runner.run(
        orchestrator,
        FULL_SESSION_TASK,
        max_turns=40
    )

    print("\n" + "="*60)
    print("SESSION COMPLETE")
    print("="*60)
    print(result.final_output)
    return result.final_output


if __name__ == "__main__":
    asyncio.run(run_full_session())

8. Tool Schema Reference