
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.
requests library (pip install requests).HIAPI_API_KEY (it starts with sk-).Auth header is
Authorization: Bearer sk-<key>everywhere below. If you get HTTP 401 withpermission_denied, the key is wrong or doesn't have access to the selected model.
Text-to-video providers don't return a video synchronously — generation takes 30–120s. The standard shape is:
POST /tasks with model id + input → returns a taskId.GET /tasks/{id} → poll until status is success or fail.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.
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:
API_BASE becomes https://api.hiapi.ai/v1.wan2.7-video/text-to-video@pro (a bare id, not a URL).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[].urlis short-lived (a few hours, the response carriesexpireAt). Treat it as a redirect — download the bytes immediately, store them yourself (S3, R2, your CDN). Never embed the raw URL on a page.
The minimal example above works for a demo, but you'll want four more things before shipping it.
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:
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.
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.
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.
aspect_ratio/duration values, and a playground that issues the exact same /v1/tasks call./v1/tasks create, poll, callback schema, error codes).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.