Supported Tokens: Python Reference

pip install requests

export API_KEY="<your-rialto-api-key>"

python partner_supported_tokens_example.py \
  --sell-token 0x... \
  --buy-token 0x... \
  --sell-amount 10

#!/usr/bin/env python3
"""Minimal Partner example for caching Rialto's supported-token list."""

from __future__ import annotations

import argparse
import json
import os
import re
import time
from dataclasses import dataclass, field

import requests

API_URL = os.getenv("RIALTO_API_URL", "https://rialto-trade-api.rialto.xyz").rstrip("/")
API_KEY = os.getenv("API_KEY", "")
CHAIN_ID = 4663
TAKER = "0x0000000000000000000000000000000000000001"
TIMEOUT_SECONDS = 30
REFRESH_SECONDS = 15  # Refresh the list every 15 seconds.

ADDRESS_PATTERN = re.compile(r"0x[0-9a-fA-F]{40}")


def normalize_address(address: str) -> str:
    if not ADDRESS_PATTERN.fullmatch(address):
        raise ValueError(f"invalid token address: {address}")
    return address.lower()


HEADERS = {
    "Accept": "application/json",
    "Accept-Encoding": "gzip",
    "User-Agent": "partner-rialto-supported-token-example/1.0",
}


@dataclass
class SupportedTokens:
    addresses: set[str] = field(default_factory=set)
    etag: str | None = None

    def refresh(self) -> None:
        headers = {"If-None-Match": self.etag} if self.etag else {}
        response = requests.get(
            f"{API_URL}/tokens/supported-tokens",
            params={"chain_id": CHAIN_ID},
            headers=HEADERS | headers,
            timeout=TIMEOUT_SECONDS,
        )

        if response.status_code == 304:
            print(f"supported tokens: unchanged ({self.etag})")
            return

        response.raise_for_status()
        payload = response.json()
        addresses = {normalize_address(address) for address in payload["addresses"]}

        if payload["chain_id"] != CHAIN_ID or payload["count"] != len(addresses):
            raise RuntimeError("invalid supported-token snapshot")

        self.addresses = addresses
        self.etag = response.headers.get("ETag")
        print(f"supported tokens: loaded {len(addresses):,} ({self.etag})")

    def supports(self, token: str) -> bool:
        return normalize_address(token) in self.addresses


def quote(sell_token: str, buy_token: str, sell_amount: str) -> requests.Response:
    if not API_KEY:
        raise RuntimeError("API_KEY is missing from the environment")

    return requests.get(
        f"{API_URL}/quote",
        params={
            "chain_id": CHAIN_ID,
            "sell_token": sell_token,
            "buy_token": buy_token,
            "sell_amount": sell_amount,
            "taker": TAKER,
            "slippage_bps": 50,
        },
        headers=HEADERS | {"Authorization": f"Bearer {API_KEY}"},
        timeout=TIMEOUT_SECONDS,
    )


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--sell-token", required=True)
    parser.add_argument("--buy-token", required=True)
    parser.add_argument("--sell-amount", default="10")
    parser.add_argument(
        "--report-unsupported-demand",
        action="store_true",
        help="Send genuine unsupported demand to Rialto instead of skipping it locally.",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()

    sell_token = normalize_address(args.sell_token)
    buy_token = normalize_address(args.buy_token)

    supported = SupportedTokens()
    supported.refresh()
    time.sleep(REFRESH_SECONDS)
    supported.refresh()

    pair_supported = supported.supports(sell_token) and supported.supports(buy_token)
    print(f"pair supported: {pair_supported}")

    if not pair_supported and not args.report_unsupported_demand:
        print("quote skipped locally")
        return 0

    # Genuine unsupported requests help Rialto discover demand. Repeated Partner
    # demand starts asynchronous validation; it does not guarantee admission.
    # These probes use a separate allowance of up to 10x the normal quote rate.
    response = quote(sell_token, buy_token, args.sell_amount)

    print(f"quote: HTTP {response.status_code}")
    print(json.dumps(response.json(), indent=2))

    if pair_supported:
        return 0 if response.ok else 1
    return 0 if response.status_code in {400, 404, 422} else 1


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (requests.RequestException, RuntimeError, ValueError) as error:
        raise SystemExit(f"error: {error}") from error