Runway ML Text-zu-Video-API in Python: ein lauffähiges Beispiel und die Migration zu hiapi

Du möchtest ein Text-zu-Video-Modell aus Python aufrufen, und Runway Gen-3 ist der naheliegende Einstiegspunkt. Dieser Leitfaden zeigt ein minimales, lauffähiges Python-Beispiel, das das Runway-typische asynchrone Muster demonstriert (Task erstellen → Polling → Herunterladen), und überträgt anschließend exakt denselben Code auf hiapi wan2.7-video/text-to-video – eine direkte Alternative mit derselben /v1/tasks-Struktur.

Was du vorbereiten musst

• Python 3.9+ und die Bibliothek requests (pip install requests).
• Einen API-Schlüssel. Für hiapi: Hole einen aus deinem hiapi-Dashboard und exportiere ihn als HIAPI_API_KEY (er beginnt mit sk-).
• Einen Prompt. Videomodelle akzeptieren einen einzelnen Text-Prompt sowie wenige Parameter (Seitenverhältnis, Dauer in Sekunden).

Der Authentifizierungs-Header ist überall unten Authorization: Bearer sk-<key>. Wenn du HTTP 401 und permission_denied erhältst, ist der Schlüssel falsch oder hat keinen Zugriff auf das gewählte Modell.

Minimales lauffähiges Beispiel (Runway-typisches asynchrones Muster)

Text-zu-Video-Anbieter geben Videos nicht synchron zurück – die Generierung dauert 30–120 Sekunden. Die Standardstruktur ist:

1. POST /tasks mit Modell-ID + Eingabe → gibt eine taskId zurück.
2. GET /tasks/{id} → Polling, bis status success oder fail ist.
3. Lies output[0].url und lade das MP4 herunter (die URL ist nur kurzzeitig gültig).

Hier ist ein eigenständiges Python-Skript, das diese Schleife implementiert:

import os
import time
import requests

API_BASE = "https://api.runwayml.com/v1"  # oder jeder andere Anbieter mit derselben Struktur
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")

Dieses Skript ist bewusst einfach gehalten: kein SDK, kein asynchroner Klebstoff, nur requests. Sobald es läuft, kannst du denselben Kontrollfluss beibehalten und den Anbieter wechseln, indem du nur zwei Strings änderst.

Migration zu hiapi (wan2.7-video/text-to-video)

hiapi stellt denselben asynchronen Task-Vertrag unter https://api.hiapi.ai/v1/tasks bereit. Der einzige Unterschied im aufrufenden Code ist:

1. API_BASE wird zu https://api.hiapi.ai/v1.
2. Die Modell-ID ist wan2.7-video/text-to-video (nackte ID, keine URL).
3. Der Authentifizierungsschlüssel ist dein hiapi-sk-…-Schlüssel.

Konkret ändert sich nur create_task – wait_task und download bleiben identisch:

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",
            "input": {
                "prompt": prompt,
                "ratio": aspect_ratio,  # wan2.7 nennt es ratio: 16:9, 9:16, 1:1, 4:3, 3:4
                "duration": duration,   # 2-15 Sekunden
                "resolution": "720P",   # 720P $0.10/Sek.; Standard ist 1080P, $0.167/Sek.
            },
        },
        timeout=60,
    )
    resp.raise_for_status()
    return resp.json()["data"]["taskId"]

Die Antwortstruktur ist vollständig identisch: Beim Erstellen wird {"data": {"taskId": "..."}} zurückgegeben, beim Polling:

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

Daher funktionieren wait_task(...) und download(...) aus dem vorigen Abschnitt unverändert. Die vollständige Migration umfasst nur zwei Zeilen: API_BASE ändern und model ändern. Alles drumherum bleibt gleich.

output[].url ist kurzlebig (einige Stunden, die Antwort enthält expireAt). Behandle es wie eine Weiterleitung – lade die Bytes sofort herunter und speichere sie selbst (S3, R2, dein CDN). Bette niemals die Original-URL in eine Seite ein.

Produktionsreife Muster

Das minimale Beispiel oben eignet sich für Demonstrationen, aber vor dem Livegang brauchst du vier zusätzliche Dinge.

Verwende nach Möglichkeit Callbacks statt Polling

Alle 5 Sekunden zu pollen verschwendet Anfragen und hält einen Worker ein bis zwei Minuten lang beschäftigt. Wenn deine Umgebung eine öffentliche Webhook-URL hat, übergib sie an die API und lass sie dich zurückrufen:

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",
        "input": {"prompt": prompt, "ratio": "16:9", "duration": 5, "resolution": "720P"},
        "callback": {
            "url": "https://yourapp.com/hooks/hiapi",
            "when": "final",  # ruft uns nur bei status success oder fail zurück
        },
    },
    timeout=60,
)

Dein Webhook empfängt das vollständige Task-Objekt (mit derselben Struktur wie die Polling-Antwort), sodass der Download-Schritt vollständig identisch ist.

Wann Polling vs. Callback:

• Polling: Skripte, Notebooks, Batch-Jobs und überall dort, wo du keinen HTTP-Listener exponieren möchtest.
• Callback: Web-Apps, Queue-Worker und überall dort, wo ein öffentlicher Endpunkt existiert. Immer zusammen mit Idempotenz (nächster Abschnitt) – Anbieter können Wiederholungen versuchen.

Mache das Absenden idempotent

Wenn dein Code create_task wiederholt (Netzwerkflackern, Absturz mitten in der Schleife), ist es leicht, denselben Prompt doppelt einzureichen und zweimal zu bezahlen. Der günstige Fix: Hash die Eingabe zu einem stabilen Client-Schlüssel und dedupliziere vor dem Aufruf in deiner eigenen Tabelle:

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()

Sende diesen Schlüssel als Idempotency-Key-Header (max. 255 Byte). hiapi erstellt pro Konto und Schlüssel nur einen Task; Wiederholungen mit demselben Schlüssel geben die ursprüngliche taskId zurück, statt erneut abzurechnen. Speichere zusätzlich (submit_key → task_id) in deiner eigenen Datenbank – das hilft auch, wenn du den Task später nachschlagen musst.

Behandle Fehler präzise

Von hiapi siehst du zwei Arten von Fehlerhüllen:

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

… bei Authentifizierungs-/Berechtigungsfehlern (HTTP 401/403) sowie wenn ein Task letztlich fehlschlägt:

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

Behandle sie unterschiedlich: permission_denied beim Erstellen bedeutet, dass dein Schlüssel falsch ist oder das Modell nicht abdeckt – eine Wiederholung ist sinnlos. status: "fail" nach dem Polling kann ein vorübergehender Modellfehler sein, der eine erneute Wiederholung mit demselben Prompt wert ist. Protokolliere immer die request_id, damit der Support nachverfolgen kann.

Preise und Kontingente

Die Videogenerierung wird nach Ausgabesekunden abgerechnet, nicht nach Anfrage, und variiert je nach Auflösung und Modellstufe. Bevor du die Unit Economics berechnest, sieh dir die aktuellen Zahlen auf der hiapi-Preisseite an – hardcodiere keine Dollar-Werte im Repository.

Verwandte Lektüre

• Wan 2.7 Text-zu-Video-Modellseite – Modellkarte, unterstützte aspect_ratio/duration-Werte und ein Playground, der exakt dieselben /v1/tasks-Aufrufe ausführt.
• hiapi-Dokumentation – vollständige Referenz zum asynchronen Task-Vertrag (/v1/tasks-Erstellung, Polling, Callback-Schema, Fehlercodes).
• Modellkatalog – Bild-, Video- und Audiomodelle mit derselben Aufrufstruktur.

Häufig gestellte Fragen

F: Brauchen Video- und Bildmodelle unterschiedliche SDKs? A: Nein. Beide laufen über POST /v1/tasks mit {model, input} und geben beide {data:{taskId}} zurück. Nur die input-Felder unterscheiden sich je Modell – Video akzeptiert duration, Bild nicht. Die Polling-Schleife ist vollständig identisch.

F: Warum gibt output[].url nach einigen Stunden 404 zurück? A: Die URL ist signiert und läuft ab (expireAt in der Antwort ist ein Unix-Zeitstempel). Lade die Bytes sofort herunter, schiebe sie in deinen eigenen Speicher und liefere von dort aus.

F: Wie wechsle ich von Runway Gen-3 zu Wan 2.7, ohne neu zu schreiben? A: Wenn dein Anbieter dieselbe asynchrone Task-Struktur nutzt (Erstellen → Polling → Output-URL), ändere zwei Strings: API_BASE und die model-ID. Packe sie in Umgebungsvariablen (API_BASE, MODEL_ID), dann kannst du mit demselben Code A/B-Tests machen.

F: Ich erhalte permission_denied, obwohl mein Schlüssel korrekt aussieht. A: Prüfe zwei Dinge: (1) der Schlüssel hat Zugriff auf das spezifische Modell (manche Modelle erfordern eine kostenpflichtige Stufe), und (2) der Authorization-Header lautet genau Bearer sk-... mit Leerzeichen – nicht Bearer: sk-... und nicht der nackte Schlüssel. Die vollständige request_id im Fehlerrumpf wird für die Nachverfolgung durch den Support benötigt.

F: Soll ich pollen oder Callbacks verwenden? A: Verwende Callbacks in jeder Umgebung mit einem öffentlichen HTTP-Endpunkt (Web-Apps, Queue-Worker). Verwende Polling in Skripten, Cron-Jobs und Notebooks – überall dort, wo das Aufsetzen eines Webhooks übertrieben wäre. Speichere in beiden Fällen die taskId, damit ein abgestürzter Worker durch Polling wiederaufnehmen kann, statt neu einzureichen.