Runway ML Text-to-Video API in Python: A Working Example and Migrating to hiapi

You want to call a text-to-video model from Python, and Runway Gen-3 is the obvious starting point. This guide shows a minimal working Python example for the Runway-style async pattern (create task → poll → download), then maps the exact same code onto hiapi's wan2.7-video/text-to-video@pro — a drop-in alternative that uses the same /v1/tasks shape.

What you need

• Python 3.9+ and the requests library (pip install requests).
• An API key. For hiapi: grab one from your hiapi dashboard and export it as HIAPI_API_KEY (it starts with sk-).
• A prompt. Video models accept a single text prompt plus a few knobs (aspect ratio, duration in seconds).

Auth header is Authorization: Bearer sk-<key> everywhere below. If you get HTTP 401 with permission_denied, the key is wrong or doesn't have access to the selected model.

The minimal working example (Runway-style async pattern)

Text-to-video providers don't return a video synchronously — generation takes 30–120s. The standard shape is:

1. POST /tasks with model id + input → returns a taskId.
2. GET /tasks/{id} → poll until status is success or fail.
3. Read output[0].url and download the MP4 (the URL is short-lived).

Here's a self-contained Python script that implements that loop:

import os
import time
import requests

API_BASE = "https://api.runwayml.com/v1"  # or any provider using the same shape
API_KEY = os.environ["RUNWAY_API_KEY"]

def create_task(prompt: str, aspect_ratio: str = "16:9", duration: int = 5) -> str:
    resp = requests.post(
        f"{API_BASE}/tasks",
        headers={"Authorization": f"Bearer {API_KEY}",
                 "Content-Type": "application/json"},
        json={
            "model": "gen3a_turbo",
            "input": {
                "prompt": prompt,
                "aspect_ratio": aspect_ratio,
                "duration": duration,
            },
        },
        timeout=60,
    )
    resp.raise_for_status()
    return resp.json()["data"]["taskId"]

def wait_task(task_id: str, timeout_s: int = 600, interval: int = 5) -> dict:
    deadline = time.time() + timeout_s
    while time.time() < deadline:
        resp = requests.get(
            f"{API_BASE}/tasks/{task_id}",
            headers={"Authorization": f"Bearer {API_KEY}"},
            timeout=30,
        )
        resp.raise_for_status()
        task = resp.json()["data"]
        if task["status"] == "success":
            return task
        if task["status"] == "fail":
            raise RuntimeError(f"task failed: {task.get('error')}")
        time.sleep(interval)
    raise TimeoutError(f"task {task_id} not done after {timeout_s}s")

def download(url: str, path: str) -> None:
    with requests.get(url, stream=True, timeout=120) as r:
        r.raise_for_status()
        with open(path, "wb") as f:
            for chunk in r.iter_content(chunk_size=1 << 20):
                f.write(chunk)

if __name__ == "__main__":
    tid = create_task("a cat walking on a beach at sunset, cinematic")
    print(f"task {tid} created, polling…")
    task = wait_task(tid)
    download(task["output"][0]["url"], "out.mp4")
    print("wrote out.mp4")

That script is intentionally boring: no SDK, no async glue, just requests. Once it works, you can keep this exact control flow and only swap two strings to change providers.

Migrating to hiapi (wan2.7-video/text-to-video@pro)

hiapi exposes the same async task contract at https://api.hiapi.ai/v1/tasks. The only differences in the calling code:

1. API_BASE becomes https://api.hiapi.ai/v1.
2. The model id is wan2.7-video/text-to-video@pro (a bare id, not a URL).
3. The auth key is your hiapi sk-… key.

Concretely, only create_task changes — wait_task and download are identical:

import os
import requests

API_BASE = "https://api.hiapi.ai/v1"
API_KEY = os.environ["HIAPI_API_KEY"]

def create_task(prompt: str, aspect_ratio: str = "16:9", duration: int = 5) -> str:
    resp = requests.post(
        f"{API_BASE}/tasks",
        headers={"Authorization": f"Bearer {API_KEY}",
                 "Content-Type": "application/json"},
        json={
            "model": "wan2.7-video/text-to-video@pro",
            "input": {
                "prompt": prompt,
                "aspect_ratio": aspect_ratio,  # 16:9, 9:16, 1:1, …
                "duration": duration,           # seconds
            },
        },
        timeout=60,
    )
    resp.raise_for_status()
    return resp.json()["data"]["taskId"]

The response shape is identical: {"data": {"taskId": "..."}} on create, and on poll:

{
  "data": {
    "status": "success",
    "output": [{"url": "https://…/result.mp4", "expireAt": 1750000000}],
    "error": null
  }
}

So wait_task(...) and download(...) from the previous section work unchanged. The full migration is two-line: change API_BASE and change model. Everything around it stays.

The output[].url is short-lived (a few hours, the response carries expireAt). Treat it as a redirect — download the bytes immediately, store them yourself (S3, R2, your CDN). Never embed the raw URL on a page.

Production-grade patterns

The minimal example above works for a demo, but you'll want four more things before shipping it.

Use callbacks instead of polling (when you can)

Polling every 5s wastes requests and keeps a worker pinned for a minute or two. If your environment has a public webhook URL, hand it to the API and let it call you:

resp = requests.post(
    f"{API_BASE}/tasks",
    headers={"Authorization": f"Bearer {API_KEY}",
             "Content-Type": "application/json"},
    json={
        "model": "wan2.7-video/text-to-video@pro",
        "input": {"prompt": prompt, "aspect_ratio": "16:9", "duration": 5},
        "callback": {
            "url": "https://yourapp.com/hooks/hiapi",
            "when": "final",  # only call us when status is success or fail
        },
    },
    timeout=60,
)

Your webhook receives the full task object (same shape as the poll response), so the download step is identical.

When to poll vs callback:

• Polling: scripts, notebooks, batch jobs, anywhere you don't want to expose an HTTP listener.
• Callbacks: web apps, queue workers, anywhere a public endpoint exists. Always pair with idempotency (next section) — providers may retry.

Make submissions idempotent

If your code retries create_task (network blip, crash mid-loop), you can easily submit the same prompt twice and pay twice. Cheap fix: hash the inputs into a stable client-side key and dedupe in your own table before calling:

import hashlib, json

def submit_key(prompt: str, aspect_ratio: str, duration: int) -> str:
    payload = json.dumps({"p": prompt, "a": aspect_ratio, "d": duration},
                         sort_keys=True)
    return hashlib.sha256(payload.encode()).hexdigest()

Store (submit_key → task_id) in your DB before issuing the POST. On retry, look up the key first; if it's there, just poll the existing task.

Handle errors precisely

Two error envelopes you'll see from hiapi:

{ "error": {
    "code": "permission_denied",
    "message": "This API key cannot use the selected model.",
    "request_id": "20260626…",
    "type": "hiapi_error"
} }

…on auth/permission failures (HTTP 401/403), and on terminal task failure:

{ "data": { "status": "fail",
            "error": {"code": "...", "message": "..."} } }

Treat them differently: a permission_denied on create means your key is wrong or doesn't cover the model — no point retrying. A task status: "fail" after polling may be a transient model error and is worth one retry with the same prompt. Always log the request_id so support can trace it.

Pricing and quotas

Video generation is priced per output second, not per request, and varies by resolution and model tier. Check the live numbers on the hiapi pricing page before you compute your unit economics — don't hard-code dollar values in your repo.

Related reading

• Wan 2.7 text-to-video model page — model card, supported aspect_ratio/duration values, and a playground that issues the exact same /v1/tasks call.
• hiapi docs — full reference for the async task contract (/v1/tasks create, poll, callback schema, error codes).
• Model catalog — image, video, and audio models that share the same call shape.

FAQ

Q: Do I need a different SDK for video vs image models? A: No. Both go through POST /v1/tasks with {model, input}, and both come back as {data:{taskId}}. Only the input fields differ per model — videos take duration, images don't. The polling loop is identical.

Q: Why is output[].url 404ing after a few hours? A: The URL is signed and expires (expireAt in the response is a Unix timestamp). Download the bytes immediately, push to your own storage, and serve from there.

Q: How do I switch from Runway Gen-3 to Wan 2.7 without rewriting? A: If your provider uses the same async task shape (create → poll → output URL), change two strings: the API_BASE and the model id. Wrap them in env vars (API_BASE, MODEL_ID) and you can A/B against the same code.

Q: I'm getting permission_denied even though my key looks right. A: Check two things: (1) the key has access to the specific model (some models require a paid tier), and (2) the Authorization header is exactly Bearer sk-... with the space — not Bearer: sk-... and not a raw key. The full request_id in the error body is what support needs to trace it.

Q: Should I poll or use callbacks? A: Use callbacks in any environment that has a public HTTP endpoint (web apps, queue workers). Use polling in scripts, cron jobs, and notebooks — anywhere standing up a webhook would be silly. Either way, store the taskId so a crashed worker can resume by polling instead of resubmitting.