Wallet Architecture for Production AI Agents

HD Wallet Structure: BIP32, BIP39, BIP44

Hierarchical Deterministic (HD) wallets are the backbone of modern crypto key management. Defined by BIP32, they allow an entire tree of key pairs to be derived deterministically from a single 256-bit master seed. Add BIP39 mnemonic encoding and BIP44 path conventions, and you have a portable, auditable, multi-chain key management system in a single 12 or 24-word phrase.

The relationship between the standards:

• BIP39: Converts 128–256 bits of entropy into a human-readable mnemonic phrase (12–24 words). The mnemonic + optional passphrase derives a 512-bit seed via PBKDF2-HMAC-SHA512.
• BIP32: Takes the 512-bit seed, splits it with HMAC-SHA512 to create the master private key + chain code, then defines the math for child key derivation at each level of the tree.
• BIP44: Standardizes the path structure: m / purpose' / coin_type' / account' / change / index, where hardened levels (') are derivable only with the parent private key.

Child Key Derivation Math

At each derivation step, BIP32 computes:

The left 32 bytes become the child key tweak (added mod n to the parent private key), and the right 32 bytes become the child chain code. Hardened derivation (index ≥ 2^31) uses the parent private key as the HMAC data instead of the public key, preventing downward-exposure attacks where a leaked child private key could compromise the parent.

Multi-Chain Address Derivation

An AI agent operating across multiple blockchains needs deterministic address generation for each chain from a single master seed. The following Python code demonstrates derivation for the four most common chains an agent encounters on Purple Flea.

"""
Multi-chain address derivation from a single BIP39 mnemonic.
Requires: pip install mnemonic hdwallet eth-account base58
"""
from mnemonic import Mnemonic
from hdwallet import BIP44HDWallet
from hdwallet.cryptocurrencies import EthereumMainnet, BitcoinMainnet
from eth_account import Account
import hashlib, base58

mnemo = Mnemonic("english")
# Generate a fresh mnemonic — store this SECURELY (never log it)
MNEMONIC = mnemo.generate(strength=256)  # 24 words

def derive_ethereum(mnemonic: str, account: int = 0, index: int = 0) -> dict:
    Account.enable_unaudited_hdwallet_features()
    acct = Account.from_mnemonic(
        mnemonic,
        account_path=f"m/44'/60'/{account}'/0/{index}"
    )
    return {
        "address": acct.address,   # EIP-55 checksummed
        "private_key": acct.key.hex()
    }

def derive_tron(mnemonic: str, account: int = 0, index: int = 0) -> dict:
    """Tron uses the same derivation as Ethereum path 195', same secp256k1 key."""
    Account.enable_unaudited_hdwallet_features()
    acct = Account.from_mnemonic(
        mnemonic,
        account_path=f"m/44'/195'/{account}'/0/{index}"
    )
    # Convert EVM address to Tron base58check format
    evm_hex = acct.address[2:]  # strip '0x'
    raw = bytes.fromhex("41" + evm_hex)
    checksum = hashlib.sha256(hashlib.sha256(raw).digest()).digest()[:4]
    tron_addr = base58.b58encode(raw + checksum).decode()
    return {"address": tron_addr, "private_key": acct.key.hex()}

def derive_bitcoin(mnemonic: str, account: int = 0, index: int = 0) -> dict:
    wallet = BIP44HDWallet(cryptocurrency=BitcoinMainnet)
    wallet.from_mnemonic(mnemonic=mnemonic, language="english")
    wallet.clean_derivation()
    wallet.from_path(f"m/44'/0'/{account}'/0/{index}")
    return {
        "address": wallet.address(),  # P2PKH (1xxx)
        "private_key": wallet.private_key()
    }

# Example: derive first address on each chain
eth  = derive_ethereum(MNEMONIC)
tron = derive_tron(MNEMONIC)
btc  = derive_bitcoin(MNEMONIC)

print(f"ETH  : {eth['address']}")
print(f"TRON : {tron['address']}")
print(f"BTC  : {btc['address']}")

Hot, Warm, and Cold Wallet Tiers

A production agent needs a three-tier wallet architecture. Moving funds between tiers should be auditable and policy-driven, not ad-hoc.

Automatic Sweep Policy

Agents should implement automatic sweep policies to keep the hot wallet balance within safe bounds:

• Upward sweep: When hot wallet exceeds the high-water mark (e.g., 5% of total AUM), sweep excess to warm wallet automatically.
• Downward sweep: When hot wallet falls below the low-water mark (e.g., 1%), request a refill from warm wallet (with 2-of-2 multi-sig approval in high-security deployments).
• Emergency drain: If the process detects anomalous behavior (unexpected transactions, unusual gas spend), drain the hot wallet to cold immediately.

Key Management Patterns

Key management is the hardest part of wallet architecture. The following patterns are ordered from minimal to production-grade:

Pattern 1: Environment Variables (Dev Only)

Suitable for development and testing. Never use in production with real funds.

import os
from eth_account import Account

private_key = os.environ["AGENT_PRIVATE_KEY"]  # e.g., export AGENT_PRIVATE_KEY=0xabc...
account = Account.from_key(private_key)
print(f"Agent address: {account.address}")

Pattern 2: Encrypted Keystore File

The Ethereum keystore format (EIP-55) encrypts the private key with a passphrase using scrypt + AES-128-CTR. The passphrase itself should come from a secret manager at runtime.

import json, os
from eth_account import Account

def create_keystore(private_key: str, passphrase: str, path: str):
    acct = Account.from_key(private_key)
    encrypted = Account.encrypt(private_key, passphrase)
    with open(path, "w") as f:
        json.dump(encrypted, f, indent=2)
    print(f"Keystore written to {path} for {acct.address}")

def load_keystore(path: str, passphrase: str) -> Account:
    with open(path) as f:
        keystore = json.load(f)
    private_key = Account.decrypt(keystore, passphrase)
    return Account.from_key(private_key)

# Usage
PASSPHRASE = os.environ["KEYSTORE_PASSPHRASE"]  # from secret manager
acct = load_keystore("/etc/agent/keystore.json", PASSPHRASE)
print(f"Loaded wallet: {acct.address}")

Pattern 3: Cloud Secret Manager

For production agents running on cloud infrastructure, store the encrypted private key (or mnemonic) in a dedicated secret manager (HashiCorp Vault, AWS Secrets Manager, GCP Secret Manager). The agent retrieves it at startup using its service identity — no secrets ever touch the filesystem or environment.

Multi-Sig for High-Value Agents

When an agent controls significant funds, multi-signature security becomes critical. Multi-sig means a transaction requires M-of-N keys to sign before it is valid — no single key compromise can drain the wallet.

Threshold Signature Schemes (TSS)

Modern multi-sig avoids on-chain multi-sig contracts (which are visible and slightly more expensive) by using Threshold Signature Schemes. TSS distributes key generation and signing across multiple parties at the cryptographic level — the on-chain result looks like a regular single-key transaction. Protocols like FROST (for Ed25519) and GG20 (for secp256k1) are the state of the art.

Purple Flea Wallet API Deep-Dive

The Purple Flea wallet API provides a complete managed wallet infrastructure. Agents don't need to self-custody keys — they can delegate custody to Purple Flea and interact via API, retaining full programmatic control.

import requests

BASE = "https://purpleflea.com/api/v1"
KEY  = "pf_live_<your_key>"
HEADERS = {"X-API-Key": KEY}

# Get balance for all supported chains
resp = requests.get(f"{BASE}/wallet/balance", headers=HEADERS)
balances = resp.json()["balances"]
for asset in balances:
    print(f"{asset['symbol']:10s}  {asset['amount']:>14.8f}  (${asset['usd_value']:.2f})")

# Get a fresh receive address (rotated automatically after use)
resp = requests.get(
    f"{BASE}/wallet/receive",
    params={"chain": "ethereum", "asset": "USDC"},
    headers=HEADERS
)
data = resp.json()
print(f"Deposit address : {data['address']}")
print(f"Chain           : {data['chain']}")
print(f"Memo required   : {data['memo_required']}")
if data["memo_required"]:
    print(f"Memo            : {data['memo']}")

# Send USDC on Ethereum
send_body = {
    "chain":     "ethereum",
    "asset":     "USDC",
    "to":        "0xRecipientAddressHere",
    "amount":    "100.00",
    "memo":      "payment for service #42",
    "speed":     "standard"   # "slow" | "standard" | "fast"
}
resp = requests.post(
    f"{BASE}/wallet/send",
    json=send_body,
    headers={**HEADERS, "Content-Type": "application/json"}
)
tx = resp.json()
print(f"TX hash  : {tx['tx_hash']}")
print(f"Fee paid : {tx['fee']} {tx['fee_asset']}")
print(f"Status   : {tx['status']}")

# Paginated transaction history
def get_history(chain: str = "all", limit: int = 50, cursor: str = None) -> list:
    params = {"chain": chain, "limit": limit}
    if cursor:
        params["cursor"] = cursor
    resp = requests.get(f"{BASE}/wallet/history", params=params, headers=HEADERS)
    data = resp.json()
    return data["transactions"], data.get("next_cursor")

txns, cursor = get_history("ethereum", limit=20)
for tx in txns:
    direction = "IN " if tx["direction"] == "receive" else "OUT"
    print(f"{direction}  {tx['amount']:>12s} {tx['asset']:6s}  {tx['timestamp'][:10]}  {tx['tx_hash'][:12]}...")

Address Rotation Strategies

Using the same receive address indefinitely creates privacy risks and makes on-chain activity trivially linkable. Address rotation strategies mitigate this:

• Per-transaction rotation: Generate a new BIP44 address for every incoming payment (incrementing the index). Most privacy-preserving but requires tracking all used addresses.
• Time-based rotation: Rotate addresses on a schedule (daily, weekly). Easier to manage, moderate privacy improvement.
• Stealth addresses: Using ECDH, generate a one-time address for each payer that only you can link to your master key. Full sender-receiver unlinkability on-chain (supported on Monero natively, EIP-5564 for Ethereum).

Production AgentWallet Class

The following AgentWallet class provides a complete abstraction over the Purple Flea wallet API with retry logic, balance caching, transaction confirmation polling, and sweep policy enforcement.

"""
AgentWallet: Production-grade wallet abstraction for AI agents.
Wraps the Purple Flea wallet API with caching, retries, and sweep logic.
"""
import time, logging, requests
from dataclasses import dataclass, field
from typing import Optional
from functools import cached_property

log = logging.getLogger("agent-wallet")
BASE = "https://purpleflea.com/api/v1"

@dataclass
class WalletConfig:
    api_key: str
    hot_max_usd: float = 500.0     # Sweep above this
    hot_min_usd: float = 50.0      # Refill below this
    refill_amount_usd: float = 200.0
    cache_ttl_seconds: int = 300   # 5 min balance cache
    max_retries: int = 3
    retry_delay: float = 2.0

class AgentWallet:
    def __init__(self, config: WalletConfig):
        self.cfg = config
        self._session = requests.Session()
        self._session.headers.update({"X-API-Key": config.api_key})
        self._balance_cache: dict = {}
        self._cache_time: float = 0.0

    def _request(self, method: str, path: str, **kwargs) -> dict:
        url = f"{BASE}{path}"
        for attempt in range(self.cfg.max_retries):
            try:
                r = self._session.request(method, url, **kwargs)
                r.raise_for_status()
                return r.json()
            except requests.RequestException as e:
                if attempt == self.cfg.max_retries - 1:
                    raise
                log.warning(f"Request failed (attempt {attempt+1}): {e}. Retrying...")
                time.sleep(self.cfg.retry_delay * (attempt + 1))

    def get_balances(self, force: bool = False) -> dict:
        """Return dict of asset -> float balance, with caching."""
        now = time.time()
        if not force and (now - self._cache_time) < self.cfg.cache_ttl_seconds:
            return self._balance_cache
        data = self._request("GET", "/wallet/balance")
        self._balance_cache = {b["symbol"]: float(b["amount"]) for b in data["balances"]}
        self._cache_time = now
        return self._balance_cache

    def get_balance(self, asset: str, force: bool = False) -> float:
        return self.get_balances(force).get(asset, 0.0)

    def get_usd_value(self) -> float:
        data = self._request("GET", "/wallet/balance")
        return sum(float(b["usd_value"]) for b in data["balances"])

    def get_receive_address(self, chain: str = "ethereum",
                            asset: str = "USDC", force_new: bool = False) -> dict:
        params = {"chain": chain, "asset": asset}
        if force_new:
            params["force_new"] = "true"
        return self._request("GET", "/wallet/receive", params=params)

    def send(self, chain: str, asset: str, to: str,
             amount: str, memo: str = "", speed: str = "standard") -> dict:
        body = {"chain": chain, "asset": asset, "to": to,
                "amount": amount, "memo": memo, "speed": speed}
        result = self._request("POST", "/wallet/send", json=body)
        # Invalidate balance cache after send
        self._cache_time = 0.0
        log.info(f"Sent {amount} {asset} to {to}: tx={result.get('tx_hash', 'pending')}")
        return result

    def wait_for_confirmation(self, tx_hash: str, chain: str,
                               timeout: int = 300, poll: int = 10) -> dict:
        deadline = time.time() + timeout
        while time.time() < deadline:
            data = self._request("GET", f"/wallet/tx/{tx_hash}", params={"chain": chain})
            status = data.get("status")
            if status == "confirmed":
                log.info(f"TX {tx_hash} confirmed at block {data.get('block_number')}")
                return data
            elif status == "failed":
                raise RuntimeError(f"Transaction {tx_hash} failed on-chain")
            time.sleep(poll)
        raise TimeoutError(f"TX {tx_hash} not confirmed within {timeout}s")

    def enforce_sweep_policy(self, hot_asset: str = "USDC",
                              warm_address: str = None) -> Optional[dict]:
        """Sweep hot wallet if above high-water mark; alert if below low-water mark."""
        if warm_address is None:
            log.warning("No warm_address configured — skipping sweep policy")
            return None
        usd_value = self.get_usd_value()
        if usd_value > self.cfg.hot_max_usd:
            excess = usd_value - self.cfg.hot_max_usd
            log.info(f"Hot wallet ${usd_value:.2f} > max ${self.cfg.hot_max_usd:.2f}. Sweeping ${excess:.2f}.")
            # Get precise balance for sweep asset
            balance = self.get_balance(hot_asset, force=True)
            # Convert excess USD to asset units (approximate)
            price_resp = self._request("GET", "/market/price", params={"symbol": f"{hot_asset}/USD"})
            price = float(price_resp.get("price", 1.0))  # USDC/USD ≈ 1.0
            sweep_amount = str(round(excess / price, 6))
            return self.send("ethereum", hot_asset, warm_address, sweep_amount, memo="auto-sweep")
        elif usd_value < self.cfg.hot_min_usd:
            log.warning(f"Hot wallet ${usd_value:.2f} < min ${self.cfg.hot_min_usd:.2f}! Refill needed.")
        return None

    def get_history(self, chain: str = "all", limit: int = 50) -> list:
        data = self._request("GET", "/wallet/history", params={"chain": chain, "limit": limit})
        return data.get("transactions", [])


# ── Usage Example ────────────────────────────────────────────────────────────

if __name__ == "__main__":
    import os
    cfg = WalletConfig(
        api_key=os.environ.get("PURPLEFLEA_API_KEY", "pf_live_<your_key>"),
        hot_max_usd=500.0,
        hot_min_usd=50.0,
        cache_ttl_seconds=300,
    )
    wallet = AgentWallet(cfg)

    balances = wallet.get_balances()
    print("Current balances:", balances)

    addr = wallet.get_receive_address(chain="ethereum", asset="USDC")
    print(f"Receive address: {addr['address']}")

    # Enforce sweep policy (sweeps to warm wallet if hot wallet is too large)
    WARM_ADDR = os.environ.get("WARM_WALLET_ADDRESS")
    if WARM_ADDR:
        wallet.enforce_sweep_policy(hot_asset="USDC", warm_address=WARM_ADDR)

Security Checklist for Agent Wallets

Before deploying a production agent with real funds, validate each item:

• Never log private keys, mnemonics, or passphrases — even in debug mode
• Use hardened BIP44 paths for account/coin_type derivation levels
• Store encrypted keystores with scrypt/Argon2 KDF — never plaintext
• Implement a hot-wallet balance cap and automatic sweep to warm wallet
• Enable 2-of-3 multi-sig for warm and cold tiers
• Rotate hot wallet addresses on every inbound transaction
• Monitor for unexpected outbound transactions with real-time alerting
• Maintain a transaction log with full audit trail (hash, amount, fee, timestamp)
• Set withdrawal rate limits per time window at the API layer
• Test emergency drain procedure quarterly — it must work when you need it most

Related: Perpetual Futures Guide for AI Agents · Market Regime Detection for AI Agents · Agent Registration Tutorial