Python

mkdir tradeon-integration && cd tradeon-integration
python3 -m venv .venv && source .venv/bin/activate
pip install httpx flask
 
cat > .env <<'EOF'
TRADEON_API_BASE=https://api.tradeon.market
TRADEON_API_KEY=mer_live_REPLACE_ME
TRADEON_WEBHOOK_SECRET=whsec_REPLACE_ME
EOF

set -a; . ./.env; set +a

import os
import time
import sys
import httpx
 
API_BASE = os.environ.get("TRADEON_API_BASE", "https://api.tradeon.market")
API_KEY = os.environ["TRADEON_API_KEY"]  # KeyError on missing — fail fast
 
client = httpx.Client(
    base_url=API_BASE,
    headers={"Authorization": f"Bearer {API_KEY}"},
    timeout=15.0,
)
 
 
def buy(*, item_id: str, max_price: float, custom_id: str,
        target_steam_id: str, target_trade_url: str) -> dict:
    res = client.post(
        "/api/v1/merchant/buy",
        json={
            "item_id": item_id,
            "max_price": max_price,
            "custom_id": custom_id,
            "target_steam_id": target_steam_id,
            "target_trade_url": target_trade_url,
        },
    )
    body = res.json() if res.content else {}
 
    if res.status_code == 201:
        return {"kind": "created", "purchase": body}
 
    # Branch on error.code, not the HTTP status. See /getting-started/errors.
    code = (body.get("error") or {}).get("code")
    if res.status_code == 400:
        return {"kind": "invalid_request", "code": code, "error": body.get("error")}
    if res.status_code == 402:
        return {"kind": "insufficient_balance", "code": code, "error": body.get("error")}
    if res.status_code == 404:
        return {"kind": "item_not_found", "code": code, "error": body.get("error")}
    if res.status_code == 409:
        return {"kind": "conflict", "code": code, "error": body.get("error")}
    if res.status_code == 401:
        raise RuntimeError("API key missing or malformed Authorization header")
    if res.status_code == 403:
        raise RuntimeError("API key invalid or merchant blocked (rotate via the cabinet)")
    raise RuntimeError(f"unexpected {res.status_code}: {body!r}")
 
 
def poll_until_terminal(purchase_id: str, *, interval_s: float = 5.0,
                        timeout_s: float = 10 * 60) -> dict:
    deadline = time.monotonic() + timeout_s
    while time.monotonic() < deadline:
        res = client.get(f"/api/v1/merchant/purchases/{purchase_id}")
        if res.status_code != 200:
            raise RuntimeError(f"GET /purchases/{purchase_id} returned {res.status_code}")
        data = res.json()
        if data["status"] in ("completed", "failed"):
            return data
        time.sleep(interval_s)
    raise RuntimeError(f"timed out polling purchase {purchase_id}")
 
 
def main() -> int:
    outcome = buy(
        item_id="REPLACE_WITH_ITEM_ID",
        max_price=12.5,
        custom_id=f"order-{int(time.time() * 1000)}",  # unique per order; never random across retries
        target_steam_id="76561198000000000",
        target_trade_url=(
            "https://steamcommunity.com/tradeoffer/new/"
            "?partner=12345&token=abc123"
        ),
    )
    print(outcome)
    if outcome["kind"] == "created":
        terminal = poll_until_terminal(outcome["purchase"]["purchase_id"])
        print("terminal:", terminal)
    return 0
 
 
if __name__ == "__main__":
    sys.exit(main())

python buy.py

import hashlib
import hmac
import os
import time
from flask import Flask, request
 
app = Flask(__name__)
SECRET = os.environ["TRADEON_WEBHOOK_SECRET"].encode()
PORT = int(os.environ.get("PORT", "8080"))
 
# Process-local dedup. Replace with Redis SETNX or a DB unique index in prod —
# this set forgets everything on restart, so a retry across a deploy gets re-handled.
_seen_event_ids: set[str] = set()
 
 
@app.post("/webhooks/tradeon")
def tradeon_webhook():
    signature = request.headers.get("X-Signature", "")
    timestamp = request.headers.get("X-Timestamp", "")
    event_id = request.headers.get("X-Event-Id", "")
 
    if not (signature and timestamp and event_id):
        return ("missing signature headers", 400)
 
    # Reject anything skewed by more than ±5 minutes — see /webhooks/signature.
    try:
        ts = int(timestamp)
    except ValueError:
        return ("bad timestamp", 400)
    if abs(time.time() - ts) > 300:
        return ("stale timestamp", 400)
 
    # Use the exact bytes — re-encoding JSON would change whitespace / key order.
    raw = request.get_data()
    signed_payload = f"{timestamp}.".encode() + raw
    expected = hmac.new(SECRET, signed_payload, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, signature):
        return ("bad signature", 400)
 
    # Dedup: retries reuse X-Event-Id. Ack 2xx on dup — 4xx would mark the
    # event permanently failed.
    if event_id in _seen_event_ids:
        return ("", 200)
    _seen_event_ids.add(event_id)
 
    event = request.get_json(silent=True) or {}
    print(
        f"[webhook] {event.get('event')} "
        f"eid={event_id} data={event.get('data')}"
    )
 
    # TODO: dispatch by event["event"] — see /webhooks/events for the catalogue.
    # Forward-compat: ignore unknown event types (return 2xx); the API may add new ones.
 
    return ("", 200)
 
 
if __name__ == "__main__":
    print(f"tradeon webhook listener on http://localhost:{PORT}/webhooks/tradeon")
    app.run(host="0.0.0.0", port=PORT)

python webhook_server.py

# ngrok
ngrok http 8080
 
# Cloudflare Tunnel — free, no account needed for ad-hoc tunnels
cloudflared tunnel --url http://localhost:8080