Build with the
RoyalTechno API

#!/usr/bin/env python3"""RoyalTechno AI - reference API client (Python 3.9+, `requests` only).​Follows every rule in https://royaltechno.ai/api-docs#rules, so copy it as-is:a sliding window of MAX_IN_FLIGHT jobs, ONE poll loop for all jobs withIf-None-Match, Retry-After + exponential backoff on 429/503, stop on 4xx,one Idempotency-Key per job, and it never resubmits a job that is in flight.​    RT_API_KEY=rt_live_... python rt_client.py --model nano-banana-pro \\        --prompt "a royal crown made of glass, studio lighting" --count 2​Live numbers for YOUR key: GET /v1/account -> "limits" (they can be higherthan the defaults below after a per-key allowance)."""import argparseimport osimport randomimport timeimport uuid​import requests​API = "https://api.royaltechno.cc/v1"MAX_IN_FLIGHT = 20                  # per API key: GET /v1/account -> limits.recommended_max_in_flightPOLL_SEC = {"image": 5, "video": 10, "chain": 30}BACKOFF_BASE_SEC, BACKOFF_MAX_SEC, MAX_ATTEMPTS = 2, 60, 6RETRYABLE = {429, 503}              # honour Retry-After, then back offFATAL = {400, 401, 402, 403, 404}   # never clears by retrying: fix the request, key, balance or planTIMEOUT_SEC = 30​session = requests.Session()session.headers["Authorization"] = f"Bearer {os.environ['RT_API_KEY']}"​​class Fatal(Exception):    """A refusal that retrying can never fix."""​​def backoff(attempt):    """2 s doubling to 60 s, +/- 20 % jitter so retries never line up."""    return min(BACKOFF_MAX_SEC, BACKOFF_BASE_SEC * 2 ** attempt) * random.uniform(0.8, 1.2)​​def request(method, path, **kw):    """One API call with the rules baked in: 4xx raises, 429/503 sleep for    max(Retry-After, backoff) and retry, at most MAX_ATTEMPTS tries.    A `queue_full` 429 is returned to the caller instead of retried: the    right reaction is to wait for one of YOUR jobs to finish, not to POST again."""    for attempt in range(MAX_ATTEMPTS):        r = session.request(method, f"{API}{path}", timeout=TIMEOUT_SEC, **kw)        if r.status_code in FATAL:            raise Fatal(r.json()["error"])        if r.status_code not in RETRYABLE:            return r        if r.json().get("error", {}).get("code") == "queue_full":            return r        time.sleep(max(float(r.headers.get("Retry-After", 0)), backoff(attempt)))    raise RuntimeError(f"gave up after {MAX_ATTEMPTS} attempts: {method} {path}")​​def submit(model, job_input):    """POST one job -> (job, 0) or (None, seconds to wait) when the queue is full."""    headers = {"Idempotency-Key": str(uuid.uuid4())}  # one UUID per job, reused only for its retries    r = request("POST", "/jobs", json={"model": model, "input": job_input}, headers=headers)    if r.status_code == 429:        return None, float(r.headers.get("Retry-After", 60))    return r.json(), 0.0​​def poll(job_id, etag):    """GET one job with If-None-Match -> (job or None on 304, etag)."""    r = request("GET", f"/jobs/{job_id}", headers={"If-None-Match": etag} if etag else {})    if r.status_code == 304:        return None, etag    return r.json(), r.headers.get("ETag", etag)​​def run(jobs, kind="image"):    """Sliding window: <= MAX_IN_FLIGHT in flight, one poll loop, never resubmits.    `jobs` is a list of (model, input) tuples; returns the finished job dicts."""    todo, inflight, done = list(jobs), {}, []    interval, next_submit = POLL_SEC[kind], 0.0    while todo or inflight:        now = time.time()        while todo and len(inflight) < MAX_IN_FLIGHT and now >= next_submit:            job, wait = submit(*todo[0])            if job is None:                       # queue_full: stop filling, wait Retry-After                next_submit = now + wait                break            todo.pop(0)            inflight[job["id"]] = {"etag": None, "next_poll": now + interval}        for job_id, st in list(inflight.items()):            if st["next_poll"] > now:                continue            job, st["etag"] = poll(job_id, st["etag"])            st["next_poll"] = now + interval            if job and job["status"] in ("succeeded", "failed", "canceled"):                done.append(job)                del inflight[job_id]        wakeups = [st["next_poll"] for st in inflight.values()] + ([next_submit] if todo else [])        if wakeups:            time.sleep(max(0.2, min(wakeups) - time.time()))    return done​​def main():    ap = argparse.ArgumentParser(description="RoyalTechno AI reference client")    ap.add_argument("--model", default="nano-banana-pro")    ap.add_argument("--prompt", required=True)    ap.add_argument("--count", type=int, default=1, help="how many separate jobs to submit")    ap.add_argument("--kind", choices=sorted(POLL_SEC), default="image", help="picks the poll cadence")    a = ap.parse_args()    for job in run([(a.model, {"prompt": a.prompt})] * a.count, kind=a.kind):        detail = (job.get("output") or {}).get("url") or (job.get("error") or {}).get("message")        print(job["id"], job["status"], detail)​​if __name__ == "__main__":    main()

curl https://api.royaltechno.cc/v1/jobs \  -H "Authorization: Bearer $RT_API_KEY" \  -H "Content-Type: application/json" \  -d '{    "model": "eleven-v3",    "input": {      "prompt": "Привіт! Це озвучка, згенерована через RoyalTechno API.",      "voice": "Ntd0iVwICtUtA6Fvx27M",      "language": "uk"    }  }'​# Poll GET /v1/jobs/{id} like an image job. When it succeeds:#   "output": { "type": "audio", "url": "https://cdn.royaltechno.cc/…/job_….mp3",#               "mime_type": "audio/mpeg", "duration_sec": 4, "chars": 54 }