import os
import time
import requests
from eth_account import Account
from eth_account.messages import encode_defunct
from eth_utils import keccak
API_BASE = "https://rialto-trade-api.rialto.xyz"
CHAIN_ID = 4663
ACTION_CREATE_APPLICATION = "create_integrator_application"
ACTION_CREATE_API_KEY = "create_integrator_api_key"
def require_env(name):
value = os.getenv(name)
if not value:
raise RuntimeError(f"missing {name}")
return value
def optional(value):
return f"some:{value}" if value is not None else "none"
def payload_hash(fields):
canonical = ""
for key, value in fields:
value = str(value)
canonical += f"{key}={len(value.encode('utf-8'))}:{value}\n"
return "0x" + keccak(canonical.encode("utf-8")).hex()
def sign_message(private_key, message):
signed = Account.sign_message(encode_defunct(text=message), private_key)
return "0x" + bytes(signed.signature).hex()
def nonce(owner, action, hash_value=None):
body = {"chain_id": CHAIN_ID, "wallet": owner, "action": action}
if hash_value is not None:
body["payload_hash"] = hash_value
response = requests.post(f"{API_BASE}/integrators/nonce", json=body, timeout=30)
response.raise_for_status()
return response.json()
private_key = require_env("OWNER_PRIVATE_KEY")
owner = Account.from_key(private_key).address.lower()
display_name = "Example Partner"
slug = f"example-partner-{int(time.time())}"
contact_email = "ops@example.com"
telegram_handle = None
app_url = "https://example.com"
fee_recipient = owner
requested_max_fee_bps = 50
app_hash = payload_hash(
[
("action", ACTION_CREATE_APPLICATION),
("chain_id", CHAIN_ID),
("owner_wallet", owner),
("display_name", display_name),
("slug", slug),
("contact_email", optional(contact_email)),
("telegram_handle", optional(telegram_handle)),
("app_url", optional(app_url)),
("fee_recipient", fee_recipient),
("requested_max_fee_bps", requested_max_fee_bps),
]
)
app_nonce = nonce(owner, ACTION_CREATE_APPLICATION, app_hash)
app_body = {
"chain_id": CHAIN_ID,
"owner_wallet": owner,
"display_name": display_name,
"slug": slug,
"contact_email": contact_email,
"telegram_handle": telegram_handle,
"app_url": app_url,
"fee_recipient": fee_recipient,
"requested_max_fee_bps": requested_max_fee_bps,
"payload_hash": app_hash,
"nonce": app_nonce["nonce"],
"issued_at": app_nonce["issued_at"],
"expiration_time": app_nonce["expiration_time"],
"signature": sign_message(private_key, app_nonce["message"]),
}
app_response = requests.post(
f"{API_BASE}/integrators/applications",
json=app_body,
timeout=30,
)
app_response.raise_for_status()
application = app_response.json()
print("application:", application)
if application["status"] != "active":
print("Application is pending approval. Create the API key after approval.")
raise SystemExit(0)
integrator_id = application["integrator_id"]
label = "production-key"
key_hash = payload_hash(
[
("action", ACTION_CREATE_API_KEY),
("chain_id", CHAIN_ID),
("owner_wallet", owner),
("integrator_id", integrator_id),
("label", label),
]
)
key_nonce = nonce(owner, ACTION_CREATE_API_KEY, key_hash)
key_body = {
"chain_id": CHAIN_ID,
"owner_wallet": owner,
"integrator_id": integrator_id,
"label": label,
"payload_hash": key_hash,
"nonce": key_nonce["nonce"],
"issued_at": key_nonce["issued_at"],
"expiration_time": key_nonce["expiration_time"],
"signature": sign_message(private_key, key_nonce["message"]),
}
key_response = requests.post(
f"{API_BASE}/integrators/api-keys",
json=key_body,
timeout=30,
)
key_response.raise_for_status()
created_key = key_response.json()
print("raw API key; store now:", created_key["api_key"])
print("masked key:", created_key["masked_key"])
