Supercharge Your Gemini Flash Agent
with Financial Infrastructure

from google.genai import types

# ── Tool: Get Wallet Balance ───────────────────────────────
get_balance_tool = types.Tool(
    function_declarations=[types.FunctionDeclaration(
        name="get_wallet_balance",
        description="Get the current USDC balance in the agent's Purple Flea wallet.",
        parameters=types.Schema(
            type=types.Type.OBJECT,
            properties={},  # no params needed — uses API key from header
        )
    )]
)

# ── Tool: Place Trade Order ────────────────────────────────
place_order_tool = types.Tool(
    function_declarations=[types.FunctionDeclaration(
        name="place_trade_order",
        description="Place a perpetual futures order on Purple Flea Trading.",
        parameters=types.Schema(
            type=types.Type.OBJECT,
            properties={
                "market": types.Schema(
                    type=types.Type.STRING,
                    description="Market symbol, e.g. BTC-USDC-PERP, ETH-USDC-PERP"
                ),
                "side": types.Schema(
                    type=types.Type.STRING,
                    enum=["long", "short"],
                    description="Trade direction"
                ),
                "size_usdc": types.Schema(
                    type=types.Type.NUMBER,
                    description="Notional size in USDC"
                ),
                "leverage": types.Schema(
                    type=types.Type.INTEGER,
                    description="Leverage multiplier (1-20)"
                ),
                "order_type": types.Schema(
                    type=types.Type.STRING,
                    enum=["market", "limit"],
                    description="Order type — market fills immediately"
                ),
            },
            required=["market", "side", "size_usdc", "leverage"]
        )
    )]
)

# ── Tool: Create Escrow ────────────────────────────────────
create_escrow_tool = types.Tool(
    function_declarations=[types.FunctionDeclaration(
        name="create_escrow",
        description="Create a trustless escrow for agent-to-agent payments on Purple Flea.",
        parameters=types.Schema(
            type=types.Type.OBJECT,
            properties={
                "recipient_agent_id": types.Schema(
                    type=types.Type.STRING,
                    description="Agent ID of the payment recipient"
                ),
                "amount_usdc": types.Schema(
                    type=types.Type.NUMBER,
                    description="Amount in USDC to lock in escrow"
                ),
                "condition": types.Schema(
                    type=types.Type.STRING,
                    description="Task or condition for release"
                ),
            },
            required=["recipient_agent_id", "amount_usdc", "condition"]
        )
    )]
)

# ── Tool: Claim Faucet ─────────────────────────────────────
claim_faucet_tool = types.Tool(
    function_declarations=[types.FunctionDeclaration(
        name="claim_faucet",
        description="Claim free $1 USDC from the Purple Flea Faucet (one-time per agent).",
        parameters=types.Schema(
            type=types.Type.OBJECT,
            properties={}
        )
    )]
)

# Combine all tools
ALL_PF_TOOLS = [
    get_balance_tool, place_order_tool,
    create_escrow_tool, claim_faucet_tool
]

# gemini_flash_agent.py — Autonomous financial agent with Gemini Flash
import os, json, httpx, asyncio
from google import genai
from google.genai import types
from purple_flea_tools import ALL_PF_TOOLS

# ── Config ─────────────────────────────────────────────────
GEMINI_API_KEY = os.environ["GEMINI_API_KEY"]
PF_AGENT_KEY   = os.environ["PF_AGENT_KEY"]
PF_API         = "https://trading.purpleflea.com/api/v1"
PF_HEADERS     = {"X-Agent-Key": PF_AGENT_KEY}

client = genai.Client(api_key=GEMINI_API_KEY)

# ── Tool Executor ──────────────────────────────────────────
async def execute_tool(tool_name: str, args: dict) -> dict:
    async with httpx.AsyncClient(timeout=15) as c:

        if tool_name == "get_wallet_balance":
            r = await c.get(f"{PF_API}/wallet/balance", headers=PF_HEADERS)
            return r.json()

        elif tool_name == "place_trade_order":
            r = await c.post(
                f"{PF_API}/orders",
                headers=PF_HEADERS,
                json={
                    "market":     args["market"],
                    "side":       args["side"],
                    "size":       args["size_usdc"],
                    "leverage":   args.get("leverage", 5),
                    "type":       args.get("order_type", "market"),
                }
            )
            return r.json()

        elif tool_name == "create_escrow":
            r = await c.post(
                "https://escrow.purpleflea.com/api/v1/escrow",
                headers=PF_HEADERS,
                json=args
            )
            return r.json()

        elif tool_name == "claim_faucet":
            r = await c.post(
                "https://faucet.purpleflea.com/api/v1/claim",
                json={"agent_key": PF_AGENT_KEY}
            )
            return r.json()

        return {"error": f"Unknown tool: {tool_name}"}

# ── Agent Loop ─────────────────────────────────────────────
async def run_agent(task: str):
    print(f"\n{'='*60}")
    print(f"Task: {task}")
    print('='*60)

    messages = [types.Content(
        role="user",
        parts=[types.Part(text=task)]
    )]

    while True:
        response = client.models.generate_content(
            model="gemini-2.0-flash",
            contents=messages,
            config=types.GenerateContentConfig(
                tools=ALL_PF_TOOLS,
                system_instruction=(
                    "You are a financial agent with access to Purple Flea APIs. "
                    "Check balance before trading. Never risk more than 10% of balance per trade. "
                    "Always confirm order fills. Be concise in your reasoning."
                ),
            )
        )

        candidate = response.candidates[0]

        # Check for function calls
        fn_calls = [
            p.function_call
            for p in candidate.content.parts
            if p.function_call
        ]

        if not fn_calls:
            # No more function calls — model is done
            print("\nAgent response:", response.text)
            break

        # Execute all function calls (possibly parallel)
        messages.append(candidate.content)
        tool_results = []

        for fn in fn_calls:
            print(f"\n[Tool] {fn.name}({dict(fn.args)})")
            result = await execute_tool(fn.name, dict(fn.args))
            print(f"[Result] {result}")
            tool_results.append(types.Part.from_function_response(
                name=fn.name, response=result
            ))

        messages.append(types.Content(
            role="tool", parts=tool_results
        ))

# ── Example Usage ──────────────────────────────────────────
async def main():
    # Example 1: Balance check + trade decision
    await run_agent(
        "Check my Purple Flea wallet balance. If I have more than $5, "
        "open a $2 long BTC-USDC-PERP position at 3x leverage. "
        "Tell me the fill price."
    )

    # Example 2: Faucet claim on first run
    await run_agent(
        "Claim my free $1 USDC from the Purple Flea faucet, "
        "then check my updated balance."
    )

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

# Gemini Flash fires all tool calls in parallel when it decides
# the answers are independent. This cuts latency dramatically.

import asyncio, httpx
from google import genai
from google.genai import types

# Add market data tool alongside Purple Flea tools
get_market_data_tool = types.Tool(
    function_declarations=[types.FunctionDeclaration(
        name="get_market_data",
        description="Fetch current price and 24h stats for a market.",
        parameters=types.Schema(
            type=types.Type.OBJECT,
            properties={
                "symbol": types.Schema(
                    type=types.Type.STRING,
                    description="Market symbol, e.g. BTC-USDC-PERP"
                )
            },
            required=["symbol"]
        )
    )]
)

async def execute_parallel_tools(fn_calls: list) -> list:
    """Execute all function calls concurrently with asyncio.gather."""
    tasks = [execute_single_tool(fn.name, dict(fn.args)) for fn in fn_calls]
    results = await asyncio.gather(*tasks)
    return [
        types.Part.from_function_response(name=fn.name, response=res)
        for fn, res in zip(fn_calls, results)
    ]

# With parallel tools, Flash can simultaneously ask:
# 1. "What is my wallet balance?"  → get_wallet_balance()
# 2. "What is BTC price now?"      → get_market_data(symbol="BTC-USDC-PERP")
# 3. "What are my open positions?" → get_open_positions()
# All three fire at the same time, cutting round-trip from 450ms → 150ms.

async def run_parallel_agent():
    task = (
        "I want to make an optimal trade decision. "
        "Simultaneously check my wallet balance, the current BTC-USDC-PERP price, "
        "and my open positions. Then decide whether to add to my BTC long or stay flat."
    )
    # Gemini Flash will recognize these are independent and fire all 3 tools at once
    await run_agent(task)  # from main agent module