Tool Development

Tools are async Python functions that the agent calls during trading cycles.

Basic Structure


import os
from typing import Any
import httpx
 
API_KEY = os.getenv("EXCHANGE_API_KEY", "")
 
async def get_ticker(symbol: str) -> dict[str, Any]:
    """Get current price and 24h stats for a trading pair.
 
    Args:
        symbol: Trading pair (e.g. "BTC/USDT").
 
    Returns:
        Ticker data with price, volume, and change.
    """
    async with httpx.AsyncClient(timeout=10.0) as client:
        resp = await client.get(f"https://api.exchange.com/ticker/{symbol}")
        resp.raise_for_status()
        return resp.json()

Requirements

Must be async — All tool functions use async def
Must return dict — JSON-serializable return value
Must have docstrings — The LLM reads these to understand tool purpose and parameters
Type hints on parameters — Helps the LLM call the tool correctly

Read vs Write Tools

Read Tools

Fetch data without side effects:


async def get_klines(
    symbol: str,
    interval: str = "1h",
    limit: int = 100,
) -> dict[str, Any]:
    """Fetch historical K-line (candlestick) data."""
    # ... fetch and return data

Write Tools

Place orders or modify positions:


async def place_order(
    symbol: str,
    side: str,
    amount: float,
    order_type: str = "market",
) -> dict[str, Any]:
    """Place a trading order on the exchange.
 
    Args:
        symbol: Trading pair.
        side: "buy" or "sell".
        amount: Order size in base currency.
        order_type: "market" or "limit".
 
    Returns:
        Order result with fill price and status.
    """
    # ... execute order

Environment Variables

Access credentials and configuration via os.getenv():


API_KEY = os.getenv("EXCHANGE_API_KEY", "")
API_SECRET = os.getenv("EXCHANGE_API_SECRET", "")
PLATFORM_URL = os.getenv("PLATFORM_URL", "https://api.orienai.xyz")

Declare required variables in orien.runtime.yaml under required_secrets. The agent runtime validates them at startup.

Error Handling

Use raise_for_status() on HTTP responses. The agent runtime catches exceptions and logs them without crashing the cycle.


async def get_data(symbol: str) -> dict:
    async with httpx.AsyncClient(timeout=15.0) as client:
        resp = await client.get(f"https://api.example.com/{symbol}")
        resp.raise_for_status()  # Raises on 4xx/5xx
        return resp.json()

Available Libraries

The agent runtime includes common data and trading packages:

httpx for async HTTP requests
numpy for numerical computation
pandas for data wrangling
ta for technical analysis indicators