AI agents paying
AI agents.

POST https://wallet.purpleflea.com/v1/wallet/send Headers: Authorization: Bearer {api_key} Content-Type: application/json Body: { "to": string, // recipient wallet address (required) "amount": string, // decimal string, e.g. "0.50" (required) "currency": string, // "USDC" | "USDT" | "ETH" | "MATIC" | ... (required) "chain": string, // "base" | "polygon" | "ethereum" | "arbitrum" | "solana" | ... "memo": string, // optional reference string — stored on-chain in tx data "speed": string // "normal" | "fast" | "instant" — affects gas bid } Response: { "tx_hash": string, // on-chain transaction hash "status": string, // "pending" | "confirmed" | "failed" "from": string, // your wallet address "to": string, // recipient address "amount": string, // amount transferred "currency": string, "chain": string, "fee_usd": string, // approximate gas cost in USD "block_number": number }

import requests import time # --- Configuration --- ORCHESTRATOR_API_KEY = "sk_live_orchestrator_abc..." SUBAGENT_WALLET_ADDRESS = "0xSubAgent1234..." # obtained from sub-agent's /v1/balance WALLET_API = "https://wallet.purpleflea.com" PAYMENT_AMOUNT = "0.50" # $0.50 USDC per completed research task CHAIN = "base" # cheap gas for micropayments def verify_subagent_exists(address: str) -> bool: """Confirm the sub-agent's wallet address is live before sending payment.""" resp = requests.get( f"{WALLET_API}/v1/balance/{CHAIN}/{address}", headers={"Authorization": f"Bearer {ORCHESTRATOR_API_KEY}"}, timeout=10 ) return resp.status_code == 200 def dispatch_task(subagent_url: str, task: dict) -> dict: """Send a task to the sub-agent and wait for completion.""" resp = requests.post( f"{subagent_url}/tasks", json=task, timeout=30 ) resp.raise_for_status() return resp.json() def pay_subagent(to_address: str, amount: str, task_id: str) -> dict: """Send USDC payment to sub-agent after task completion.""" resp = requests.post( f"{WALLET_API}/v1/wallet/send", headers={ "Authorization": f"Bearer {ORCHESTRATOR_API_KEY}", "Content-Type": "application/json" }, json={ "to": to_address, "amount": amount, "currency": "USDC", "chain": CHAIN, "memo": f"task-id:{task_id}" # on-chain reference }, timeout=15 ) resp.raise_for_status() return resp.json() def wait_for_confirmation(tx_hash: str, max_wait: int = 30) -> bool: """Poll until transaction is confirmed or timeout.""" for _ in range(max_wait): resp = requests.get( f"{WALLET_API}/v1/wallet/transaction/{tx_hash}", headers={"Authorization": f"Bearer {ORCHESTRATOR_API_KEY}"} ) data = resp.json() if data.get("status") == "confirmed": return True if data.get("status") == "failed": return False time.sleep(1) return False # --- Main orchestration flow --- def run_paid_task(task: dict) -> dict: # 1. Verify sub-agent is reachable if not verify_subagent_exists(SUBAGENT_WALLET_ADDRESS): raise ValueError("Sub-agent wallet address not found — verify address") # 2. Dispatch task to sub-agent print(f"Dispatching task: {task['type']}") result = dispatch_task("https://my-subagent.example.com", task) task_id = result["task_id"] print(f"Task completed: {task_id}") # 3. Pay sub-agent for completed work payment = pay_subagent(SUBAGENT_WALLET_ADDRESS, PAYMENT_AMOUNT, task_id) tx_hash = payment["tx_hash"] print(f"Payment sent: {tx_hash}") # 4. Wait for on-chain confirmation confirmed = wait_for_confirmation(tx_hash) if confirmed: print(f"Payment confirmed on {CHAIN}. Cost: ${payment['fee_usd']} gas") else: print("Warning: transaction not yet confirmed — check manually") return { "result": result, "payment_tx": tx_hash, "payment_confirmed": confirmed } # Example usage if __name__ == "__main__": output = run_paid_task({ "type": "market_research", "query": "Top 5 DeFi protocols by TVL this week", "format": "json" }) print(output) Copy

import requests def register_subagent_with_referral(my_referral_code: str) -> dict: """ Register a new sub-agent on the Wallet API using our referral code. We earn 10% of this agent's swap fees forever. """ resp = requests.post( "https://wallet.purpleflea.com/v1/auth/register", json={"referral_code": my_referral_code}, timeout=10 ) resp.raise_for_status() data = resp.json() # data["api_key"] — give this to the sub-agent # data["referral_code"] — the sub-agent's own code for further referrals return data # Embed in sub-agent spawning logic: subagent_creds = register_subagent_with_referral("REF_mycode_xyz") print("Sub-agent API key:", subagent_creds["api_key"]) # Now configure the sub-agent with this API key # You earn 10% of every swap it makes, indefinitely Copy

import requests import time WALLET_API = "https://wallet.purpleflea.com" SUBAGENT_API_KEY = "sk_live_subagent_..." MY_WALLET_ADDRESS = "0xSubAgentAddress..." def verify_payment_received(tx_hash: str, expected_amount: float, expected_sender: str, task_nonce: str) -> bool: """ Verify an incoming payment before executing irreversible work. Checks: amount, sender, memo nonce, and confirmation status. """ resp = requests.get( f"{WALLET_API}/v1/wallet/transaction/{tx_hash}", headers={"Authorization": f"Bearer {SUBAGENT_API_KEY}"}, timeout=10 ) if resp.status_code != 200: return False tx = resp.json() # Verify all conditions before proceeding checks = [ tx.get("status") == "confirmed", # on-chain confirmed float(tx.get("amount", 0)) >= expected_amount, # correct amount tx.get("from", "").lower() == expected_sender.lower(), # known sender tx.get("to", "").lower() == MY_WALLET_ADDRESS.lower(), # correct recipient task_nonce in tx.get("memo", ""), # nonce matches task ] return all(checks) # In the sub-agent's task handler: def handle_task_request(task_data: dict) -> dict: tx_hash = task_data.get("payment_tx_hash") task_id = task_data.get("task_id") # Wait up to 30 seconds for confirmation for attempt in range(30): if verify_payment_received( tx_hash=tx_hash, expected_amount=0.50, expected_sender=task_data["sender_address"], task_nonce=task_id ): # Payment verified — safe to execute irreversible work return execute_task(task_data) time.sleep(1) return {"error": "payment not confirmed within timeout"} def execute_task(task_data: dict) -> dict: # ... actual task logic return {"status": "completed", "result": "..."} Copy