LangChain + Purple Flea: Building Financially Autonomous Agents

LangChain remains one of the most widely used frameworks for building AI agents in 2026. Its tool abstraction, LCEL chain composition, and agent executor model map naturally onto financial workflows where an LLM needs to reason across multiple APIs before taking action. Purple Flea's 6 financial APIs — casino, trading, wallet, escrow, faucet, and domains — each become a typed, tested LangChain Tool that an agent can invoke with full observability.

This guide walks through building a financially capable LangChain agent from scratch: wrapping each Purple Flea API as a BaseTool subclass, composing LCEL chains for multi-step financial flows, managing financial state in memory, handling errors with LangChain exception types, and adding callback handlers to log every API call for audit.

1. Understanding the Purple Flea API Surface

Before writing tool wrappers, map each service to the actions your agent will need. Purple Flea exposes 6 services with clean REST APIs:

2. Base Setup and Shared HTTP Client

Start with a shared session that handles auth headers, retries, and response validation. All Purple Flea tool wrappers will inherit from the same base class.

import os
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from typing import Optional, Type, Any
from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field
import logging

logger = logging.getLogger("purpleflea.tools")

# Shared HTTP session with retry logic
def build_session(agent_id: str, api_key: str) -> requests.Session:
    session = requests.Session()
    session.headers.update({
        "X-Agent-Id":     agent_id,
        "Authorization":  f"Bearer {api_key}",
        "Content-Type":   "application/json",
        "Accept":         "application/json",
    })
    retry = Retry(
        total=3,
        backoff_factor=1.0,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET", "POST"]
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount("https://", adapter)
    return session

# Load credentials from environment
AGENT_ID  = os.environ["PF_AGENT_ID"]
API_KEY   = os.environ["PF_API_KEY"]
SESSION   = build_session(AGENT_ID, API_KEY)

# Purple Flea base tool — all tools extend this
class PurpleFleasBaseTool(BaseTool):
    """Abstract base for all Purple Flea LangChain tools."""
    session: Any = Field(default=None, exclude=True)
    agent_id: str = Field(default="")

    def _call_api(self, method: str, url: str, **kwargs) -> dict:
        try:
            r = self.session.request(method, url, timeout=15, **kwargs)
            r.raise_for_status()
            return r.json()
        except requests.HTTPError as e:
            body = e.response.text[:200] if e.response else str(e)
            raise ToolException(f"Purple Flea API error {e.response.status_code}: {body}")
        except requests.Timeout:
            raise ToolException("Purple Flea API timeout after 15s")
        except requests.ConnectionError as e:
            raise ToolException(f"Purple Flea connection error: {e}")

3. PurpleFleasCasinoTool — Crash Game Interface

The casino tool exposes two actions: placing a bet on the crash game with a target multiplier, and reading the multiplier history for the last N rounds. The LLM uses history to reason about streak patterns before placing a bet.

from langchain_core.tools import ToolException

class CasinoBetInput(BaseModel):
    action:     str   = Field(description="'bet' or 'history'")
    amount:     float = Field(default=0.10, description="Bet amount in USDC (min 0.10)")
    target:     float = Field(default=2.0,  description="Auto-cashout multiplier (e.g. 2.0 = 2x)")
    rounds:     int   = Field(default=20,   description="Number of history rounds to fetch")

class PurpleFleasCasinoTool(PurpleFleasBaseTool):
    name:        str = "purple_flea_casino"
    description: str = (
        "Interact with Purple Flea crash casino. Use action='history' to read recent "
        "crash multipliers before betting. Use action='bet' to place a bet with amount "
        "(USDC) and target auto-cashout multiplier. Returns bet outcome or history list."
    )
    args_schema: Type[BaseModel] = CasinoBetInput
    handle_tool_error: bool = True

    BASE = "https://casino.purpleflea.com"

    def _run(self, action: str, amount: float = 0.10,
             target: float = 2.0, rounds: int = 20) -> str:
        if action == "history":
            data = self._call_api("GET", f"{self.BASE}/crash/history",
                                  params={"limit": rounds})
            history = data.get("rounds", [])
            multipliers = [r["multiplier"] for r in history]
            avg = sum(multipliers) / len(multipliers) if multipliers else 0
            return (f"Last {len(multipliers)} rounds: {multipliers}. "
                    f"Average multiplier: {avg:.2f}x. "
                    f"Min: {min(multipliers):.2f}x. Max: {max(multipliers):.2f}x.")

        elif action == "bet":
            if amount < 0.10:
                raise ToolException("Minimum bet is 0.10 USDC")
            if target < 1.01:
                raise ToolException("Target multiplier must be at least 1.01")
            data = self._call_api("POST", f"{self.BASE}/crash/bet", json={
                "amount": amount,
                "target": target,
            })
            outcome  = data.get("outcome")
            crashed  = data.get("crashed_at")
            pnl      = data.get("pnl", 0)
            return (f"Bet ${amount:.2f} at {target}x target. "
                    f"Crashed at {crashed}x. Outcome: {outcome}. PnL: ${pnl:+.4f} USDC.")
        else:
            raise ToolException(f"Unknown casino action: {action}. Use 'bet' or 'history'.")

4. PurpleFleasTradingTool — Perpetual Futures

The trading tool wraps open, close, and status actions for perpetual futures positions. It enforces a maximum position size at the tool level — the LLM cannot open a position larger than the configured limit regardless of what it reasons.

class TradingInput(BaseModel):
    action:    str   = Field(description="'open', 'close', 'status', or 'funding'")
    market:    str   = Field(default="BTC-PERP", description="Market symbol")
    side:      str   = Field(default="long",     description="'long' or 'short'")
    size_usdc: float = Field(default=10.0,       description="Position size in USDC")
    leverage:  int   = Field(default=5,          description="Leverage multiplier (1-20)")

class PurpleFleasTradingTool(PurpleFleasBaseTool):
    name:        str = "purple_flea_trading"
    description: str = (
        "Trade perpetual futures on Purple Flea. Actions: 'open' a leveraged long/short, "
        "'close' an open position, 'status' to read current positions and PnL, "
        "'funding' to check current funding rates before deciding to trade."
    )
    args_schema:     Type[BaseModel] = TradingInput
    handle_tool_error: bool = True
    max_position_usdc: float = 100.0  # Safety cap — LLM cannot exceed this

    BASE = "https://trading.purpleflea.com"

    def _run(self, action: str, market: str = "BTC-PERP",
             side: str = "long", size_usdc: float = 10.0, leverage: int = 5) -> str:

        if action == "funding":
            data  = self._call_api("GET", f"{self.BASE}/markets/{market}/funding")
            rate  = data.get("funding_rate_8h", 0)
            annl  = rate * 3 * 365
            return f"Funding rate for {market}: {rate:.6f} per 8h ({annl:.1f}% annualized)."

        elif action == "open":
            if size_usdc > self.max_position_usdc:
                raise ToolException(
                    f"Position size ${size_usdc} exceeds max ${self.max_position_usdc}.")
            if leverage > 20 or leverage < 1:
                raise ToolException("Leverage must be between 1 and 20.")
            data = self._call_api("POST", f"{self.BASE}/positions", json={
                "market":   market,
                "side":     side,
                "size":     size_usdc,
                "leverage": leverage,
            })
            pos_id = data.get("position_id")
            entry  = data.get("entry_price")
            return (f"Opened {side} {market} ${size_usdc}@{leverage}x. "
                    f"Entry: ${entry:,.2f}. Position ID: {pos_id}.")

        elif action == "status":
            data      = self._call_api("GET", f"{self.BASE}/positions")
            positions = data.get("positions", [])
            if not positions:
                return "No open positions."
            lines = []
            for p in positions:
                lines.append(
                    f"{p['market']} {p['side']} ${p['size']} | PnL: ${p['unrealized_pnl']:+.4f}")
            return "Open positions:\n" + "\n".join(lines)

        elif action == "close":
            data  = self._call_api("DELETE", f"{self.BASE}/positions/{market}")
            pnl   = data.get("realized_pnl", 0)
            return f"Closed {market} position. Realized PnL: ${pnl:+.4f} USDC."
        else:
            raise ToolException(f"Unknown trading action: {action}")

5. PurpleFleasWalletTool — Multi-Chain Wallet

The wallet tool handles balance queries, transfers, and token swaps. Transfers above $25 emit a structured log event at DEBUG level — useful for audit trails without blocking the agent.

class WalletInput(BaseModel):
    action:   str   = Field(description="'balance', 'transfer', or 'swap'")
    chain:    str   = Field(default="ethereum", description="Chain name")
    currency: str   = Field(default="USDC",     description="Token symbol")
    amount:   float = Field(default=0.0,         description="Amount to transfer or swap")
    to:       str   = Field(default="",          description="Destination agent ID for transfer")
    to_token: str   = Field(default="",          description="Target token symbol for swap")

class PurpleFleasWalletTool(PurpleFleasBaseTool):
    name:        str = "purple_flea_wallet"
    description: str = (
        "Manage multi-chain wallets on Purple Flea. Use 'balance' to check USDC/ETH/BTC "
        "balance on a specific chain. Use 'transfer' to send funds to another agent. "
        "Use 'swap' to exchange one token for another on the same chain."
    )
    args_schema:     Type[BaseModel] = WalletInput
    handle_tool_error: bool = True

    BASE = "https://wallet.purpleflea.com"
    SUPPORTED_CHAINS = {"ethereum", "solana", "bitcoin", "tron",
                        "monero", "near", "base", "arbitrum"}

    def _run(self, action: str, chain: str = "ethereum", currency: str = "USDC",
             amount: float = 0.0, to: str = "", to_token: str = "") -> str:

        if chain not in self.SUPPORTED_CHAINS:
            raise ToolException(f"Unsupported chain: {chain}. Options: {self.SUPPORTED_CHAINS}")

        if action == "balance":
            data    = self._call_api("GET", f"{self.BASE}/balance",
                                     params={"chain": chain, "currency": currency})
            balance = data.get("balance", 0)
            addr    = data.get("address", "unknown")
            return f"Balance: {balance:.6f} {currency} on {chain}. Address: {addr}"

        elif action == "transfer":
            if not to:
                raise ToolException("'to' field required for transfer")
            if amount <= 0:
                raise ToolException("Transfer amount must be positive")
            if amount > 25.0:
                logger.warning("Large transfer", extra={
                    "amount": amount, "currency": currency, "chain": chain, "to": to
                })
            data   = self._call_api("POST", f"{self.BASE}/transfer", json={
                "to": to, "amount": amount, "currency": currency, "chain": chain
            })
            tx     = data.get("tx_hash", "pending")
            return f"Sent {amount} {currency} to {to} on {chain}. Tx: {tx}"

        elif action == "swap":
            if not to_token:
                raise ToolException("'to_token' required for swap")
            data    = self._call_api("POST", f"{self.BASE}/swap", json={
                "from_token": currency, "to_token": to_token,
                "amount": amount, "chain": chain
            })
            received = data.get("received", 0)
            rate     = received / amount if amount > 0 else 0
            return f"Swapped {amount} {currency} → {received:.6f} {to_token} on {chain}. Rate: {rate:.6f}"
        else:
            raise ToolException(f"Unknown wallet action: {action}")

6. PurpleFleasEscrowTool — Trustless Agent Payments

The escrow tool is designed for multi-agent workflows where one agent hires another. The payer creates an escrow, the worker delivers, then the payer releases funds. If disputed, Purple Flea's arbitration resolves it on-chain.

class EscrowInput(BaseModel):
    action:     str   = Field(description="'create', 'fund', 'release', 'status', or 'dispute'")
    escrow_id:  str   = Field(default="",    description="Escrow ID for fund/release/status")
    payee:      str   = Field(default="",    description="Payee agent ID for create")
    amount:     float = Field(default=1.0,   description="USDC amount to escrow")
    description:str   = Field(default="",   description="Job description for escrow")
    referral:   str   = Field(default="",    description="Referral agent ID (earns 15% of fee)")

class PurpleFleasEscrowTool(PurpleFleasBaseTool):
    name:        str = "purple_flea_escrow"
    description: str = (
        "Create and manage trustless escrow between AI agents. Use 'create' to open an "
        "escrow with a payee agent and amount. Use 'fund' to deposit USDC. Use 'release' "
        "to pay out to payee when work is verified. Use 'status' to check escrow state. "
        "Fee is 1% of amount. Supports referral IDs for 15% of the fee."
    )
    args_schema:     Type[BaseModel] = EscrowInput
    handle_tool_error: bool = True

    BASE = "https://escrow.purpleflea.com"

    def _run(self, action: str, escrow_id: str = "", payee: str = "",
             amount: float = 1.0, description: str = "", referral: str = "") -> str:

        if action == "create":
            if not payee:
                raise ToolException("'payee' agent ID required to create escrow")
            payload = {"payee": payee, "amount": amount, "description": description}
            if referral:
                payload["referral"] = referral
            data  = self._call_api("POST", f"{self.BASE}/escrow", json=payload)
            eid   = data.get("escrow_id")
            fee   = amount * 0.01
            return (f"Escrow created: {eid}. Amount: ${amount} USDC. "
                    f"Fee: ${fee:.4f} (1%). Payee: {payee}. Status: unfunded.")

        elif action == "fund":
            if not escrow_id:
                raise ToolException("'escrow_id' required to fund")
            data   = self._call_api("POST", f"{self.BASE}/escrow/{escrow_id}/fund")
            status = data.get("status")
            return f"Escrow {escrow_id} funded. Status: {status}. Payee can now deliver work."

        elif action == "release":
            if not escrow_id:
                raise ToolException("'escrow_id' required to release")
            data = self._call_api("POST", f"{self.BASE}/escrow/{escrow_id}/release")
            paid = data.get("paid_out", 0)
            return f"Escrow {escrow_id} released. Payee received ${paid:.4f} USDC."

        elif action == "status":
            if not escrow_id:
                raise ToolException("'escrow_id' required for status")
            data = self._call_api("GET", f"{self.BASE}/escrow/{escrow_id}")
            return (f"Escrow {escrow_id}: status={data.get('status')}, "
                    f"amount=${data.get('amount')}, payee={data.get('payee')}")
        else:
            raise ToolException(f"Unknown escrow action: {action}")

7. Assembling the Agent Executor

With all four tools defined, wire them into a LangChain agent executor. The agent uses OpenAI functions calling or a ReAct prompt — both work well with typed tools that have structured schemas.

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# Initialize tools with shared session
casino_tool  = PurpleFleasCasinoTool(session=SESSION, agent_id=AGENT_ID)
trading_tool = PurpleFleasTradingTool(session=SESSION, agent_id=AGENT_ID)
wallet_tool  = PurpleFleasWalletTool(session=SESSION, agent_id=AGENT_ID)
escrow_tool  = PurpleFleasEscrowTool(session=SESSION, agent_id=AGENT_ID)

tools = [casino_tool, trading_tool, wallet_tool, escrow_tool]

# System prompt for financial agent
SYSTEM_PROMPT = """You are a financially autonomous AI agent operating on Purple Flea's
financial infrastructure. You have access to casino, trading, wallet, and escrow tools.

Rules:
- Always check balance before transferring funds
- Always check crash history before placing a casino bet
- Always check funding rates before opening a perpetual position
- Never place a bet larger than 5% of current USDC balance
- Log your reasoning before each financial action
- If any API call fails, report the error and stop — do not retry automatically"""

prompt = ChatPromptTemplate.from_messages([
    ("system", SYSTEM_PROMPT),
    MessagesPlaceholder("chat_history", optional=True),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),
])

llm   = ChatOpenAI(model="gpt-4o", temperature=0)
agent = create_tool_calling_agent(llm, tools, prompt)

executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=8,       # Prevents infinite reasoning loops
    handle_parsing_errors=True,
    return_intermediate_steps=True,  # For audit logging
)

8. LCEL Chains for Financial Workflows

For deterministic multi-step flows, LCEL (LangChain Expression Language) chains are more reliable than open-ended agents. Use chains when the sequence of API calls is known in advance.

from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# Chain: Read crash history → analyze with LLM → place bet if safe
fetch_history = RunnableLambda(lambda _: casino_tool._run(
    action="history", rounds=30
))

analyze_prompt = ChatPromptTemplate.from_template("""
Crash game history: {history}

Analyze the recent crash history. What is the percentage of rounds that crashed
below 2x? Below 1.5x? Given this data, is it safe to bet $0.50 at 1.5x target?
Answer with: SAFE_TO_BET or DO_NOT_BET, followed by one sentence rationale.
""")

decision_chain = (
    {"history": fetch_history}
    | analyze_prompt
    | llm
    | StrOutputParser()
)

bet_if_safe = RunnableLambda(lambda decision: (
    casino_tool._run(action="bet", amount=0.50, target=1.5)
    if "SAFE_TO_BET" in decision
    else f"Bet skipped. Reason: {decision}"
))

casino_chain = decision_chain | bet_if_safe

# Execute the chain
result = casino_chain.invoke({})
print(result)

9. Memory Management for Financial State

Agents that operate over long sessions need to track financial state without re-querying the API for every reasoning step. A custom memory class caches balances, open positions, and recent trades with configurable TTL.

import time
from langchain_core.memory import BaseMemory

class FinancialStateMemory(BaseMemory):
    """Caches financial state to reduce redundant API calls."""

    _cache: dict = {}
    _ttls:  dict = {}
    BALANCE_TTL:  int = 60   # seconds
    POSITION_TTL: int = 30
    HISTORY_TTL:  int = 120

    def _get_cached(self, key: str, ttl: int):
        if key in self._cache:
            if time.time() - self._ttls.get(key, 0) < ttl:
                return self._cache[key]
        return None

    def _set_cached(self, key: str, value):
        self._cache[key] = value
        self._ttls[key]  = time.time()

    def update_balance(self, chain: str, currency: str, amount: float):
        self._set_cached(f"balance_{chain}_{currency}", amount)

    def get_balance(self, chain: str, currency: str) -> Optional[float]:
        return self._get_cached(f"balance_{chain}_{currency}", self.BALANCE_TTL)

    def load_memory_variables(self, inputs: dict) -> dict:
        """Inject financial context into every LLM call."""
        context_parts = []
        for key, value in self._cache.items():
            if "balance" in key:
                context_parts.append(f"Cached {key}: {value}")
        context = "\n".join(context_parts)
        return {"financial_context": context if context else "No cached state."}

    def save_context(self, inputs: dict, outputs: dict):
        # Parse agent outputs to update cache
        output = outputs.get("output", "")
        if "Balance:" in output:
            # Parse and cache balance from tool response
            pass

    def clear(self):
        self._cache.clear()
        self._ttls.clear()

    @property
    def memory_variables(self) -> list:
        return ["financial_context"]

10. Callback Handlers for Audit Logging

Every tool call and LLM decision must be logged for financial audit purposes. LangChain's callback system makes this clean — attach a handler to the executor and it fires on every event.

import json
from langchain_core.callbacks import BaseCallbackHandler
from datetime import datetime

class FinancialAuditCallback(BaseCallbackHandler):
    """Logs every tool call and output to an append-only audit file."""

    def __init__(self, log_path: str = "/var/agent/financial_audit.jsonl"):
        self.log_path = log_path

    def _write(self, event: dict):
        event["ts"] = datetime.utcnow().isoformat()
        with open(self.log_path, "a") as f:
            f.write(json.dumps(event) + "\n")

    def on_tool_start(self, serialized, input_str, **kwargs):
        self._write({
            "event": "tool_call",
            "tool":  serialized.get("name"),
            "input": input_str,
        })

    def on_tool_end(self, output, **kwargs):
        self._write({"event": "tool_result", "output": str(output)[:500]})

    def on_tool_error(self, error, **kwargs):
        self._write({"event": "tool_error", "error": str(error)})

    def on_agent_finish(self, finish, **kwargs):
        self._write({"event": "agent_finish", "output": finish.return_values})

# Attach to executor
audit_cb = FinancialAuditCallback()
result   = executor.invoke(
    {"input": "Check my USDC balance, review crash history, then bet $0.25 at 2x if average is above 2x"},
    config={"callbacks": [audit_cb]}
)

Tool Summary