import os
import time
from urllib.parse import urlencode
import requests
from eth_account import Account
from eth_account.messages import encode_typed_data
from web3 import Web3
API_BASE = "https://rialto-trade-api.rialto.xyz"
CHAIN_ID = 4663
SELL_TOKEN = "0x5fc5360D0400a0Fd4f2af552ADD042D716F1d168"
BUY_TOKEN = "0xc93a8c440CEa26D7445dF01729f193b27965099f"
SELL_AMOUNT = "1"
SLIPPAGE_BPS = 50
ERC20_ABI = [
{
"name": "approve",
"type": "function",
"stateMutability": "nonpayable",
"inputs": [
{"name": "spender", "type": "address"},
{"name": "value", "type": "uint256"},
],
"outputs": [{"name": "", "type": "bool"}],
}
]
def require_env(name):
value = os.getenv(name)
if not value:
raise RuntimeError(f"missing {name}")
return value
def tx_fees(w3):
latest = w3.eth.get_block("latest")
priority = w3.to_wei("0.01", "gwei")
return {
"maxPriorityFeePerGas": priority,
"maxFeePerGas": int(latest["baseFeePerGas"]) * 2 + priority,
}
api_key = require_env("API_KEY")
private_key = require_env("PRIVATE_KEY")
w3 = Web3(Web3.HTTPProvider(require_env("RPC_URL")))
taker = Account.from_key(private_key).address
params = {
"sell_token": SELL_TOKEN,
"buy_token": BUY_TOKEN,
"sell_amount": SELL_AMOUNT,
"taker": taker,
"permit2_owner": taker,
"settlement": "permit2",
"slippage_bps": SLIPPAGE_BPS,
"chain_id": CHAIN_ID,
}
quote_response = requests.get(
f"{API_BASE}/quote?{urlencode(params)}",
headers={"Authorization": f"Bearer {api_key}"},
timeout=30,
)
quote_response.raise_for_status()
quote = quote_response.json()
if quote.get("settlement") != "permit2" or not quote.get("permit2"):
raise RuntimeError("gasless requires a Permit2 quote")
if quote.get("issues", {}).get("balance"):
raise RuntimeError(f"insufficient balance: {quote['issues']['balance']}")
allowance = quote.get("issues", {}).get("allowance")
if allowance:
token = Web3.to_checksum_address(quote["sell_token"])
spender = Web3.to_checksum_address(allowance["spender"])
amount = int(quote["sell_amount"])
approve_tx = w3.eth.contract(address=token, abi=ERC20_ABI).functions.approve(
spender, amount
).build_transaction(
{
"from": taker,
"chainId": CHAIN_ID,
"nonce": w3.eth.get_transaction_count(taker),
**tx_fees(w3),
}
)
approve_tx["gas"] = int(w3.eth.estimate_gas(approve_tx) * 1.2)
signed_approval = Account.sign_transaction(approve_tx, private_key)
approval_hash = w3.eth.send_raw_transaction(signed_approval.raw_transaction)
receipt = w3.eth.wait_for_transaction_receipt(approval_hash)
if receipt.status != 1:
raise RuntimeError(f"approval reverted: {approval_hash.hex()}")
typed_data = {
"domain": quote["permit2"]["domain"],
"types": quote["permit2"]["types"],
"primaryType": quote["permit2"]["primaryType"],
"message": quote["permit2"]["message"],
}
signed_permit = Account.sign_message(
encode_typed_data(full_message=typed_data),
private_key,
)
signature = "0x" + bytes(signed_permit.signature).hex()
submit_response = requests.post(
f"{API_BASE}/gasless/submit",
headers={"Authorization": f"Bearer {api_key}"},
json={
"quote_id": quote["quote_id"],
"signature": signature,
"idempotency_key": f"partner-{quote['quote_id']}",
},
timeout=60,
)
submit_response.raise_for_status()
relay = submit_response.json()
print("relay_id:", relay["relay_id"])
print("initial_status:", relay["status"])
terminal = {"confirmed", "failed", "expired", "rejected"}
while relay["status"] not in terminal:
time.sleep(5)
status_response = requests.get(
f"{API_BASE}/gasless/status/{relay['relay_id']}",
headers={"Authorization": f"Bearer {api_key}"},
timeout=30,
)
status_response.raise_for_status()
relay = status_response.json()
print("status:", relay["status"], "tx_hash:", relay.get("tx_hash"))
if relay["status"] != "confirmed":
raise RuntimeError(f"relay did not confirm: {relay}")
