Purple Flea for Google Gemini Agents

pip install google-generativeai purpleflea-python

import google.generativeai as genai
import google.ai.generativelanguage as glm

# ── Wallet balance tool declaration ─────────────────────
wallet_tool = glm.Tool(
    function_declarations=[
        glm.FunctionDeclaration(
            name="wallet_get_balance",
            description=(
                "Returns the native token and stablecoin balance of the agent's "
                "non-custodial HD wallet on the specified blockchain network. "
                "Supported chains: ethereum, bitcoin, solana, bnb, polygon, avalanche, tron."
            ),
            parameters=glm.Schema(
                type=glm.Type.OBJECT,
                properties={
                    "chain": glm.Schema(
                        type=glm.Type.STRING,
                        description="Blockchain to query (e.g. 'ethereum')",
                        enum=["ethereum", "bitcoin", "solana",
                              "bnb", "polygon", "avalanche", "tron"],
                    ),
                    "include_tokens": glm.Schema(
                        type=glm.Type.BOOLEAN,
                        description="Include ERC-20/SPL token balances alongside native",
                    ),
                },
                required=["chain"],
            ),
        ),
    ]
)

# ── Trading tool declaration ─────────────────────────────
trading_tool = glm.Tool(
    function_declarations=[
        glm.FunctionDeclaration(
            name="trading_open_position",
            description=(
                "Opens a leveraged perpetual futures position on the specified market. "
                "Returns position_id, entry_price, liquidation_price, and notional_size."
            ),
            parameters=glm.Schema(
                type=glm.Type.OBJECT,
                properties={
                    "market":     glm.Schema(type=glm.Type.STRING,  description="Perpetual market, e.g. BTC-PERP"),
                    "side":       glm.Schema(type=glm.Type.STRING,  enum=["long", "short"]),
                    "collateral": glm.Schema(type=glm.Type.NUMBER,  description="USDC collateral amount"),
                    "leverage":   glm.Schema(type=glm.Type.INTEGER, description="Leverage 1–100",
                                                             format="int32"),
                },
                required=["market", "side", "collateral", "leverage"],
            ),
        ),
        glm.FunctionDeclaration(
            name="trading_get_markets",
            description="Returns available perpetual markets with funding rates and 24h volume.",
            parameters=glm.Schema(type=glm.Type.OBJECT, properties={}, required=[]),
        ),
    ]
)

# ── Casino tool declaration ──────────────────────────────
casino_tool = glm.Tool(
    function_declarations=[
        glm.FunctionDeclaration(
            name="casino_coin_flip",
            description="Flips a provably fair coin. Returns outcome, multiplier, and fairness proof.",
            parameters=glm.Schema(
                type=glm.Type.OBJECT,
                properties={
                    "choice": glm.Schema(type=glm.Type.STRING, enum=["heads", "tails"]),
                    "wager":  glm.Schema(type=glm.Type.NUMBER,  description="USDC wager amount"),
                },
                required=["choice", "wager"],
            ),
        ),
    ]
)

import os, json
import google.generativeai as genai
from purpleflea import PurpleFlea

# ── Configure API keys ──────────────────────────────────
genai.configure(api_key=os.environ["GEMINI_API_KEY"])
pf = PurpleFlea(api_key=os.environ["PURPLEFLEA_API_KEY"])

# ── Initialise model with all six Purple Flea tools ─────
model = genai.GenerativeModel(
    model_name="gemini-2.0-flash",
    tools=[wallet_tool, trading_tool, casino_tool,
           domains_tool, faucet_tool, escrow_tool],
    generation_config=genai.GenerationConfig(temperature=0.1),
)

# ── Tool dispatcher ──────────────────────────────────────
def dispatch(fn_name: str, fn_args: dict) -> dict:
    match fn_name:
        case "wallet_get_balance":   return pf.wallet.get_balance(**fn_args)
        case "wallet_send":          return pf.wallet.send(**fn_args)
        case "wallet_swap":          return pf.wallet.swap(**fn_args)
        case "trading_open_position":  return pf.trading.open_position(**fn_args)
        case "trading_close_position": return pf.trading.close_position(**fn_args)
        case "trading_get_markets":    return pf.trading.get_markets(**fn_args)
        case "casino_coin_flip":      return pf.casino.coin_flip(**fn_args)
        case "casino_dice":           return pf.casino.dice(**fn_args)
        case "casino_roulette":       return pf.casino.roulette(**fn_args)
        case "domains_check":         return pf.domains.check(**fn_args)
        case "domains_register":      return pf.domains.register(**fn_args)
        case "faucet_claim":          return pf.faucet.claim(**fn_args)
        case "escrow_create":         return pf.escrow.create(**fn_args)
        case "escrow_release":        return pf.escrow.release(**fn_args)
        case _:                       return {"error": f"Unknown tool: {fn_name}"}

# ── Agentic loop ────────────────────────────────────────
chat    = model.start_chat(enable_automatic_function_calling=False)
user_msg = (
    "Claim the faucet for my new agent, then check my Ethereum "
    "balance, and open a 3x BTC long using 50 USDC collateral."
)
response = chat.send_message(user_msg)

while True:
    # Collect all function_call parts from the response
    fc_parts = [
        part for candidate in response.candidates
        for part in candidate.content.parts
        if part.function_call.name  # truthy when a function call is present
    ]

    if not fc_parts:
        # No more function calls — print the final text response
        print(response.text)
        break

    # Build function_response parts and dispatch each call
    response_parts = []
    for part in fc_parts:
        fc   = part.function_call
        result = dispatch(fc.name, dict(fc.args))
        response_parts.append(
            genai.protos.Part(
                function_response=genai.protos.FunctionResponse(
                    name=fc.name,
                    response={"result": result},
                )
            )
        )

    # Send all function responses back in a single turn
    response = chat.send_message(response_parts)

import asyncio, os
import google.generativeai as genai
from purpleflea import AsyncPurpleFlea

genai.configure(api_key=os.environ["GEMINI_API_KEY"])
pf = AsyncPurpleFlea(api_key=os.environ["PURPLEFLEA_API_KEY"])

async def dispatch_async(fn_name: str, fn_args: dict) -> dict:
    match fn_name:
        case "wallet_get_balance":   return await pf.wallet.get_balance(**fn_args)
        case "trading_open_position":  return await pf.trading.open_position(**fn_args)
        case "casino_coin_flip":      return await pf.casino.coin_flip(**fn_args)
        case "faucet_claim":          return await pf.faucet.claim(**fn_args)
        case _:                       return {"error": f"Unknown: {fn_name}"}

async def run_agent():
    model = genai.GenerativeModel("gemini-2.0-flash", tools=[wallet_tool, casino_tool])
    chat  = model.start_chat(enable_automatic_function_calling=False)
    resp  = chat.send_message("Check my ETH and SOL balance, then flip a coin for 1 USDC.")

    while True:
        fc_parts = [
            p for c in resp.candidates
            for p in c.content.parts
            if p.function_call.name
        ]
        if not fc_parts:
            print(resp.text)
            break

        # Dispatch all function calls concurrently
        results = await asyncio.gather(
            *[dispatch_async(p.function_call.name, dict(p.function_call.args))
              for p in fc_parts]
        )

        response_parts = [
            genai.protos.Part(
                function_response=genai.protos.FunctionResponse(
                    name=fc_parts[i].function_call.name,
                    response={"result": results[i]},
                )
            )
            for i in range(len(fc_parts))
        ]
        resp = chat.send_message(response_parts)

asyncio.run(run_agent())

# Instead of writing glm.Tool definitions by hand, import them:
from purpleflea.integrations.gemini import (
    wallet_tool,
    trading_tool,
    casino_tool,
    domains_tool,
    faucet_tool,
    escrow_tool,
    all_tools,     # convenience list of all six
)

model = genai.GenerativeModel(
    model_name="gemini-2.0-flash",
    tools=all_tools,  # all six Purple Flea services
)

pf = PurpleFlea(
    api_key     = os.environ["PURPLEFLEA_API_KEY"],
    referral_id = "your_referral_id",  # earn 20% of all agent-generated fees
)

# Escrow referral: escrow_create includes a referral_id param that gives
# the referrer 15% of the 1% escrow fee — separate from the 20% base referral.
# Both stack automatically when both are configured.