Purple Flea for AutoGen Studio

# autogen_purpleflea_tools.py — Purple Flea tools for AutoGen agents import requests from autogen_core.tools import FunctionTool from typing import Optional PURPLE_FLEA_API = "https://api.purpleflea.com/v1" HEADERS = {"X-API-Key": "pf_your_key_here"} def get_wallet_balance(wallet_id: str) -> dict: """Return wallet balances across all supported chains.""" r = requests.get(f"{PURPLE_FLEA_API}/wallet/balance?wallet_id={wallet_id}", headers=HEADERS) r.raise_for_status() return r.json()["balances"] def send_crypto_payment(from_wallet: str, to_address: str, amount: float, asset: str) -> dict: """Send a crypto payment. Returns tx hash and status.""" payload = {"from_wallet": from_wallet, "to_address": to_address, "amount": amount, "asset": asset} r = requests.post(f"{PURPLE_FLEA_API}/wallet/send", json=payload, headers=HEADERS) r.raise_for_status() return r.json() def open_perpetual_trade(wallet_id: str, market: str, side: str, size_usd: float, leverage: int = 5) -> dict: """Open a perpetual futures position. side: 'long' or 'short'.""" payload = {"wallet_id": wallet_id, "market": market, "side": side, "size_usd": size_usd, "leverage": leverage} r = requests.post(f"{PURPLE_FLEA_API}/trading/open", json=payload, headers=HEADERS) r.raise_for_status() return r.json() def create_escrow(sender_wallet: str, recipient_address: str, amount: float, asset: str, release_condition: str) -> dict: """Create a trustless escrow between two agents.""" payload = {"sender": sender_wallet, "recipient": recipient_address, "amount": amount, "asset": asset, "condition": release_condition} r = requests.post(f"https://escrow.purpleflea.com/api/create", json=payload, headers=HEADERS) r.raise_for_status() return r.json() # Wrap as AutoGen FunctionTool objects balance_tool = FunctionTool(get_wallet_balance, description="Check crypto wallet balance by wallet ID") payment_tool = FunctionTool(send_crypto_payment, description="Send crypto payment from agent wallet to address") trade_tool = FunctionTool(open_perpetual_trade, description="Open perpetual futures position on any of 275 markets") escrow_tool = FunctionTool(create_escrow, description="Create trustless escrow for agent-to-agent payments") PURPLE_FLEA_TOOLS = [balance_tool, payment_tool, trade_tool, escrow_tool]

# autogen_trading_team.py — coordinated trading with Purple Flea import asyncio from autogen_agentchat.agents import AssistantAgent from autogen_agentchat.teams import RoundRobinGroupChat from autogen_ext.models.openai import OpenAIChatCompletionClient from autogen_purpleflea_tools import PURPLE_FLEA_TOOLS model_client = OpenAIChatCompletionClient(model="gpt-4o") # Signal generator: finds trading opportunities signal_agent = AssistantAgent( name="SignalAgent", model_client=model_client, tools=PURPLE_FLEA_TOOLS, system_message="""You analyze crypto markets and generate trade signals. Use get_wallet_balance to check available capital. Output structured signals: {market, direction, confidence, reasoning}""" ) # Risk agent: validates signals against risk limits risk_agent = AssistantAgent( name="RiskAgent", model_client=model_client, tools=PURPLE_FLEA_TOOLS, system_message="""You validate trade signals. Reject if: - Size > 5% of portfolio - Leverage > 10x - More than 3 open positions Reply APPROVED or REJECTED with reasoning.""" ) # Executor: opens approved trades via Purple Flea executor_agent = AssistantAgent( name="ExecutorAgent", model_client=model_client, tools=PURPLE_FLEA_TOOLS, system_message="""You execute approved trade signals using open_perpetual_trade. Only execute trades with APPROVED status from RiskAgent. Confirm execution with tx ID.""" ) # Assemble the team trading_team = RoundRobinGroupChat( participants=[signal_agent, risk_agent, executor_agent], max_turns=6 ) async def run_trading_cycle(): task = "Analyze BTC, ETH, and SOL for trading opportunities. Check balances, generate signals, validate risk, and execute the best trade." result = await trading_team.run(task=task) return result asyncio.run(run_trading_cycle())

// autogen_studio_agent_config.json — import into AutoGen Studio UI { "name": "PurpleFleatrader", "agent_type": "AssistantAgent", "model": "gpt-4o", "system_message": "You are a crypto trading agent with access to Purple Flea's financial infrastructure. You can check balances, open trades on 275 perp markets, send payments, and create escrow agreements. Always check wallet balance before trading and never risk more than 5% per trade.", "tools": [ { "type": "mcp_server", "url": "https://api.purpleflea.com/mcp", "auth": { "type": "api_key", "key": "pf_your_key_here" } } ], "metadata": { "wallet_id": "wallet_abc123", "max_position_usd": 500, "max_leverage": 5 } }