LLM Function Calling
for Crypto Operations

PURPLE_FLEA_TOOLS = [

    # ── Casino ──────────────────────────────────────────────────────────────
    {
        "type": "function",
        "function": {
            "name": "casino_place_bet",
            "description": (
                "Place a provably-fair bet on Purple Flea Casino. "
                "Supported games: coin_flip, dice, crash."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "game": {
                        "type": "string",
                        "enum": ["coin_flip", "dice", "crash"],
                        "description": "Which casino game to play."
                    },
                    "amount": {
                        "type": "number",
                        "description": "Bet amount in USD (min 0.01, max 10000)."
                    },
                    "side": {
                        "type": "string",
                        "description": (
                            "For coin_flip: 'heads' or 'tails'. "
                            "For dice: 'high' (4-6) or 'low' (1-3). "
                            "For crash: target multiplier as string, e.g. '2.5'."
                        )
                    }
                },
                "required": ["game", "amount", "side"]
            }
        }
    },

    # ── Trading ─────────────────────────────────────────────────────────────
    {
        "type": "function",
        "function": {
            "name": "trading_open_position",
            "description": (
                "Open a perpetual futures position on 275+ markets via "
                "Purple Flea Trading (powered by Hyperliquid)."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "coin": {
                        "type": "string",
                        "description": "Coin ticker e.g. BTC, ETH, SOL, TSLA."
                    },
                    "side": {
                        "type": "string",
                        "enum": ["long", "short"],
                        "description": "Direction of the position."
                    },
                    "size_usd": {
                        "type": "number",
                        "description": "Notional position size in USD."
                    },
                    "leverage": {
                        "type": "integer",
                        "minimum": 1,
                        "maximum": 50,
                        "description": "Leverage multiplier (default 1 = no leverage)."
                    }
                },
                "required": ["coin", "side", "size_usd", "leverage"]
            }
        }
    },

    # ── Wallet — balance ────────────────────────────────────────────────────
    {
        "type": "function",
        "function": {
            "name": "wallet_get_balance",
            "description": "Fetch the agent's wallet balance on a given chain.",
            "parameters": {
                "type": "object",
                "properties": {
                    "chain": {
                        "type": "string",
                        "enum": [
                            "ethereum", "base", "polygon",
                            "arbitrum", "solana", "tron", "bnb"
                        ],
                        "description": "Blockchain to query."
                    }
                },
                "required": ["chain"]
            }
        }
    },

    # ── Wallet — swap ───────────────────────────────────────────────────────
    {
        "type": "function",
        "function": {
            "name": "wallet_swap",
            "description": (
                "Execute a best-rate cross-chain token swap via "
                "Purple Flea Wallet (Wagyu aggregator)."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "from_chain": {
                        "type": "string",
                        "enum": ["ethereum", "base", "polygon", "arbitrum", "solana", "tron", "bnb"]
                    },
                    "to_chain": {
                        "type": "string",
                        "enum": ["ethereum", "base", "polygon", "arbitrum", "solana", "tron", "bnb"]
                    },
                    "from_token": {"type": "string", "description": "Token symbol e.g. ETH, USDC, SOL."},
                    "to_token":   {"type": "string", "description": "Token symbol to receive."},
                    "amount":     {"type": "string", "description": "Amount as decimal string e.g. '0.5'."}
                },
                "required": ["from_chain", "to_chain", "from_token", "to_token", "amount"]
            }
        }
    },

    # ── Domains — search ────────────────────────────────────────────────────
    {
        "type": "function",
        "function": {
            "name": "domains_search",
            "description": "Search for available crypto/web3 domains on Purple Flea Domains.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Domain name or keyword to search for."
                    }
                },
                "required": ["query"]
            }
        }
    },

    # ── Domains — register ──────────────────────────────────────────────────
    {
        "type": "function",
        "function": {
            "name": "domains_register",
            "description": "Register a domain via Purple Flea Domains. Call domains_search first.",
            "parameters": {
                "type": "object",
                "properties": {
                    "domain": {
                        "type": "string",
                        "description": "Full domain name to register e.g. 'myagent.crypto'."
                    }
                },
                "required": ["domain"]
            }
        }
    },

]

import json, os, requests
from openai import OpenAI

client    = OpenAI()
PF_KEY    = os.environ["PURPLEFLEA_API_KEY"]
PF_BASE   = "https://api.purpleflea.com/v1"
PF_WALLET = "https://wallet.purpleflea.com/v1"
PF_DOM    = "https://domains.purpleflea.com/v1"

# Paste the PURPLE_FLEA_TOOLS list from the section above here.
# (Abbreviated for readability.)
tools = PURPLE_FLEA_TOOLS

def pf(method, url, **kwargs):
    """Thin wrapper: attach auth header, raise on HTTP error."""
    headers = {"Authorization": f"Bearer {PF_KEY}", **kwargs.pop("headers", {})}
    r = getattr(requests, method)(url, headers=headers, timeout=15, **kwargs)
    r.raise_for_status()
    return r.json()

def dispatch(name: str, args: dict) -> dict:
    """Execute a Purple Flea function call and return its result."""
    if name == "casino_place_bet":
        return pf("post", f"{PF_BASE}/casino/bet", json=args)

    elif name == "trading_open_position":
        return pf("post", f"{PF_BASE}/trading/positions", json=args)

    elif name == "wallet_get_balance":
        chain = args["chain"]
        return pf("get", f"{PF_WALLET}/wallet/balance/{chain}")

    elif name == "wallet_swap":
        return pf("post", f"{PF_WALLET}/wallet/swap", json=args)

    elif name == "domains_search":
        return pf("get", f"{PF_DOM}/domains/search", params=args)

    elif name == "domains_register":
        return pf("post", f"{PF_DOM}/domains/register", json=args)

    else:
        raise ValueError(f"Unknown tool: {name}")

def chat(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]

    while True:
        resp = client.chat.completions.create(
            model="gpt-4o",
            tools=tools,
            messages=messages,
        )
        msg = resp.choices[0].message

        if resp.choices[0].finish_reason != "tool_calls":
            return msg.content  # done — return final text reply

        messages.append(msg)  # append assistant message with tool_calls

        for tc in msg.tool_calls:
            name   = tc.function.name
            args   = json.loads(tc.function.arguments)
            result = dispatch(name, args)
            messages.append({
                "role":         "tool",
                "tool_call_id": tc.id,
                "content":      json.dumps(result),
            })
        # loop back — model will read tool results and continue

if __name__ == "__main__":
    reply = chat("Check my ETH balance, then flip a $10 coin on heads.")
    print(reply)

import json, os, requests
import anthropic

client  = anthropic.Anthropic()
PF_KEY  = os.environ["PURPLEFLEA_API_KEY"]
MODEL   = "claude-opus-4-6"

# Anthropic format: no outer type wrapper, input_schema instead of parameters
tools = [
    {
        "name": "casino_place_bet",
        "description": "Place a provably-fair bet. Games: coin_flip, dice, crash.",
        "input_schema": {
            "type": "object",
            "properties": {
                "game":   {"type": "string", "enum": ["coin_flip", "dice", "crash"]},
                "amount": {"type": "number", "description": "Bet amount in USD."},
                "side":   {"type": "string", "description": "Game-specific side selection."}
            },
            "required": ["game", "amount", "side"]
        }
    },
    {
        "name": "trading_open_position",
        "description": "Open a perp futures position on 275+ markets.",
        "input_schema": {
            "type": "object",
            "properties": {
                "coin":     {"type": "string"},
                "side":     {"type": "string", "enum": ["long", "short"]},
                "size_usd": {"type": "number"},
                "leverage": {"type": "integer", "minimum": 1, "maximum": 50}
            },
            "required": ["coin", "side", "size_usd", "leverage"]
        }
    },
    {
        "name": "wallet_get_balance",
        "description": "Get wallet balance on a given chain.",
        "input_schema": {
            "type": "object",
            "properties": {
                "chain": {"type": "string",
                          "enum": ["ethereum","base","polygon","arbitrum","solana","tron","bnb"]}
            },
            "required": ["chain"]
        }
    },
    {
        "name": "wallet_swap",
        "description": "Cross-chain token swap via Wagyu aggregator.",
        "input_schema": {
            "type": "object",
            "properties": {
                "from_chain":  {"type": "string"},
                "to_chain":    {"type": "string"},
                "from_token":  {"type": "string"},
                "to_token":    {"type": "string"},
                "amount":      {"type": "string"}
            },
            "required": ["from_chain", "to_chain", "from_token", "to_token", "amount"]
        }
    },
    {
        "name": "domains_search",
        "description": "Search available crypto domains.",
        "input_schema": {
            "type": "object",
            "properties": {"query": {"type": "string"}},
            "required": ["query"]
        }
    },
    {
        "name": "domains_register",
        "description": "Register a domain. Call domains_search first.",
        "input_schema": {
            "type": "object",
            "properties": {"domain": {"type": "string"}},
            "required": ["domain"]
        }
    },
]

def dispatch(name, args):  # same logic as OpenAI example above
    base = "https://api.purpleflea.com/v1"
    h = {"Authorization": f"Bearer {PF_KEY}"}
    if   name == "casino_place_bet":       return requests.post(f"{base}/casino/bet", headers=h, json=args, timeout=15).json()
    elif name == "trading_open_position":   return requests.post(f"{base}/trading/positions", headers=h, json=args, timeout=15).json()
    elif name == "wallet_get_balance":     return requests.get(f"https://wallet.purpleflea.com/v1/wallet/balance/{args['chain']}", headers=h, timeout=15).json()
    elif name == "wallet_swap":           return requests.post("https://wallet.purpleflea.com/v1/wallet/swap", headers=h, json=args, timeout=15).json()
    elif name == "domains_search":        return requests.get("https://domains.purpleflea.com/v1/domains/search", headers=h, params=args, timeout=15).json()
    elif name == "domains_register":      return requests.post("https://domains.purpleflea.com/v1/domains/register", headers=h, json=args, timeout=15).json()

def chat(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]

    while True:
        resp = client.messages.create(
            model=MODEL, max_tokens=1024, tools=tools, messages=messages
        )

        if resp.stop_reason != "tool_use":
            # Extract the text block from the response
            return next(b.text for b in resp.content if b.type == "text")

        # Append the full assistant response (may contain text + tool_use blocks)
        messages.append({"role": "assistant", "content": resp.content})

        # Build tool_result blocks for every tool_use block
        tool_results = []
        for block in resp.content:
            if block.type != "tool_use":
                continue
            result = dispatch(block.name, block.input)
            tool_results.append({
                "type":        "tool_result",
                "tool_use_id": block.id,
                "content":     json.dumps(result),
            })

        messages.append({"role": "user", "content": tool_results})

if __name__ == "__main__":
    print(chat("Search for 'defibot.crypto', then register it if available."))

import json

MAX_STEPS = 10  # safety cap — prevent runaway spending

def run_agent(system_prompt: str, user_message: str, tools: list) -> str:
    """
    Generic agent loop.
    - Calls the LLM in a while True loop.
    - Dispatches any tool_calls / tool_use blocks.
    - Returns the final text reply or raises after MAX_STEPS.
    """
    messages = [{"role": "user", "content": user_message}]
    steps    = 0

    while steps < MAX_STEPS:
        steps += 1

        resp = client.chat.completions.create(  # or anthropic client
            model="gpt-4o",
            tools=tools,
            messages=messages,
        )
        msg    = resp.choices[0].message
        reason = resp.choices[0].finish_reason

        if reason != "tool_calls":
            return msg.content  # agent is done

        messages.append(msg)

        for tc in msg.tool_calls:
            name   = tc.function.name
            args   = json.loads(tc.function.arguments)
            print(f"[step {steps}] calling {name}({args})")
            result = dispatch(name, args)          # your dispatcher
            print(f"[step {steps}] result: {result}")
            messages.append({
                "role":         "tool",
                "tool_call_id": tc.id,
                "content":      json.dumps(result),
            })

    raise RuntimeError(f"Agent exceeded {MAX_STEPS} steps without completing.")

# Example: multi-step agent
reply = run_agent(
    system_prompt="You are a crypto trading agent. Be concise. Always confirm amounts before acting.",
    user_message="Check my ETH and SOL balances, then open a $200 BTC long at 3x leverage.",
    tools=PURPLE_FLEA_TOOLS,
)
print(reply)

import time, requests
from requests.exceptions import Timeout, ConnectionError, HTTPError

PF_KEY = os.environ["PURPLEFLEA_API_KEY"]

def pf_request(method: str, url: str, retries: int = 3, **kwargs) -> dict:
    """
    Resilient HTTP helper.
    - Retries 3 times with exponential backoff on 429 / 5xx / network errors.
    - Returns a structured error dict instead of raising on final failure.
    """
    headers = {"Authorization": f"Bearer {PF_KEY}"}
    backoff = 1.0

    for attempt in range(retries):
        try:
            r = getattr(requests, method)(
                url, headers=headers, timeout=15, **kwargs
            )

            if r.status_code == 429:
                retry_after = float(r.headers.get("Retry-After", backoff))
                time.sleep(retry_after)
                backoff *= 2
                continue

            if r.status_code >= 500:
                time.sleep(backoff)
                backoff *= 2
                continue

            r.raise_for_status()
            return r.json()

        except (Timeout, ConnectionError) as e:
            if attempt == retries - 1:
                return {"error": "network_error", "message": str(e)}
            time.sleep(backoff); backoff *= 2

        except HTTPError as e:
            # 4xx errors are not retryable (bad args, auth failure, etc.)
            try:
                body = r.json()
            except Exception:
                body = {"raw": r.text}
            return {
                "error":   "api_error",
                "status":  r.status_code,
                "message": body.get("message", str(e)),
                "details": body,
            }

    return {"error": "max_retries_exceeded", "message": f"Failed after {retries} attempts."}


def dispatch(name: str, args: dict) -> dict:
    """Dispatcher with structured error passthrough to the model."""
    try:
        if name == "casino_place_bet":
            return pf_request("post", "https://api.purpleflea.com/v1/casino/bet", json=args)
        elif name == "trading_open_position":
            return pf_request("post", "https://api.purpleflea.com/v1/trading/positions", json=args)
        elif name == "wallet_get_balance":
            chain = args["chain"]
            return pf_request("get", f"https://wallet.purpleflea.com/v1/wallet/balance/{chain}")
        elif name == "wallet_swap":
            return pf_request("post", "https://wallet.purpleflea.com/v1/wallet/swap", json=args)
        elif name == "domains_search":
            return pf_request("get", "https://domains.purpleflea.com/v1/domains/search", params=args)
        elif name == "domains_register":
            return pf_request("post", "https://domains.purpleflea.com/v1/domains/register", json=args)
        else:
            return {"error": "unknown_tool", "message": f"No handler for tool '{name}'."}
    except Exception as e:
        # Last-resort catch — agent loop must never crash
        return {"error": "unexpected_error", "message": repr(e)}