API Runway ML di text-to-video in Python: un esempio eseguibile e la migrazione a hiapi

Vuoi chiamare un modello text-to-video da Python e Runway Gen-3 è il punto di partenza ovvio. Questa guida mostra un esempio Python minimo ed eseguibile che illustra il pattern asincrono in stile Runway (crea attività → polling → download), per poi mappare esattamente lo stesso codice su wan2.7-video/text-to-video di hiapi — un sostituto diretto che usa la stessa struttura /v1/tasks.

Cosa ti serve

• Python 3.9+ e la libreria requests (pip install requests).
• Una chiave API. Per hiapi: ottienine una dalla tua dashboard hiapi ed esportala come HIAPI_API_KEY (inizia con sk-).
• Un prompt. I modelli video accettano un singolo prompt testuale più pochi parametri (rapporto d'aspetto, durata in secondi).

L'header di autenticazione usato ovunque qui sotto è Authorization: Bearer sk-<key>. Se ricevi HTTP 401 con permission_denied, la chiave è sbagliata oppure non ha accesso al modello scelto.

Esempio minimo eseguibile (pattern asincrono in stile Runway)

I fornitori di text-to-video non restituiscono il video in modo sincrono — la generazione richiede 30–120 secondi. La struttura standard è:

1. POST /tasks con model id + input → restituisce un taskId.
2. GET /tasks/{id} → esegui il polling finché status è success o fail.
3. Leggi output[0].url e scarica l'MP4 (l'URL è a validità breve).

Qui sotto c'è uno script Python autonomo che implementa questo ciclo:

import os
import time
import requests

API_BASE = "https://api.runwayml.com/v1"  # o qualsiasi provider con la stessa struttura
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")

Lo script è volutamente semplice: nessun SDK, nessun codice di colla asincrono, solo requests. Una volta che funziona, puoi mantenere esattamente lo stesso flusso di controllo e cambiare provider modificando solo due stringhe.

Migrazione a hiapi (wan2.7-video/text-to-video)

hiapi espone lo stesso contratto di attività asincrona su https://api.hiapi.ai/v1/tasks. Le uniche differenze nel codice chiamante sono:

1. API_BASE diventa https://api.hiapi.ai/v1.
2. Il model id è wan2.7-video/text-to-video (id nudo, non un URL).
3. La chiave di autenticazione è la tua chiave hiapi sk-….

Concretamente, cambia solo create_task — wait_task e download restano identici:

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 lo chiama ratio: 16:9, 9:16, 1:1, 4:3, 3:4
                "duration": duration,   # 2-15 secondi
                "resolution": "720P",   # 720P $0.10/secondo; il default è 1080P, $0.167/secondo
            },
        },
        timeout=60,
    )
    resp.raise_for_status()
    return resp.json()["data"]["taskId"]

La struttura della risposta è identica: alla creazione restituisce {"data": {"taskId": "..."}}, al polling restituisce:

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

Quindi wait_task(...) e download(...) della sezione precedente funzionano senza modifiche. La migrazione completa richiede solo due righe: cambiare API_BASE e cambiare model. Tutto il resto resta invariato.

output[].url è a validità breve (poche ore, la risposta include expireAt). Trattalo come un redirect — scarica subito i byte e archiviali tu stesso (S3, R2, il tuo CDN). Non incorporare mai l'URL originale in una pagina.

Pattern per la produzione

L'esempio minimo sopra va bene per una demo, ma prima di andare in produzione servono altre quattro cose.

Usa i callback al posto del polling quando puoi

Fare polling ogni 5 secondi spreca richieste e tiene occupato un worker per uno o due minuti. Se il tuo ambiente dispone di un URL webhook pubblico, passalo all'API e lascia che ti richiami:

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",  # richiamaci solo quando lo stato è success o fail
        },
    },
    timeout=60,
)

Il tuo webhook riceve l'oggetto attività completo (con la stessa struttura della risposta di polling), quindi il passo di download è identico.

Quando usare polling vs callback:

• Polling: script, notebook, attività batch e qualsiasi contesto in cui non vuoi esporre un listener HTTP.
• Callback: applicazioni web, worker in coda e qualsiasi contesto in cui esista un endpoint pubblico. Abbinalo sempre all'idempotenza (sezione successiva) — i provider possono ritentare.

Rendi idempotente l'invio

Se il tuo codice ritenta create_task (instabilità di rete, crash a metà ciclo), è facile inviare due volte lo stesso prompt e pagare due volte. La soluzione economica: calcola l'hash dell'input in una chiave client stabile e deduplica nella tua tabella prima della chiamata:

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

Invia quella chiave come header Idempotency-Key (massimo 255 byte). hiapi crea una sola attività per account e chiave, e i ritentativi con la stessa chiave restituiscono il taskId originale invece di addebitare di nuovo. Memorizza inoltre la coppia (submit_key → task_id) nel tuo database, utile anche per recuperare l'attività in seguito.

Gestisci gli errori con precisione

Da hiapi vedrai due tipi di involucri di errore:

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

……che appare in caso di errore di autenticazione/permessi (HTTP 401/403), e quando l'attività fallisce definitivamente:

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

Trattali in modo diverso: un permission_denied alla creazione significa che la chiave è sbagliata o non copre quel modello — ritentare non ha senso. Un status: "fail" dopo il polling può essere un errore transitorio del modello, e vale la pena ritentare una volta con lo stesso prompt. Registra sempre il request_id, così il supporto può tracciare la richiesta.

Prezzi e quota

La generazione video si paga in base ai secondi di output, non per richiesta, e varia in base alla risoluzione e al livello del modello. Prima di calcolare l'economia unitaria, controlla i valori aggiornati sulla pagina dei prezzi di hiapi — non codificare a mano i valori in dollari nel repository.

Letture correlate

• Pagina del modello Wan 2.7 text-to-video — model card, valori supportati di aspect_ratio/duration e un playground che effettua esattamente le stesse chiamate a /v1/tasks.
• Documentazione di hiapi — riferimento completo del contratto delle attività asincrone (creazione /v1/tasks, polling, schema dei callback, codici di errore).
• Catalogo dei modelli — modelli di immagine, video e audio che condividono la stessa struttura di chiamata.

Domande frequenti

D: I modelli video e quelli immagine richiedono SDK diversi? R: No. Entrambi usano POST /v1/tasks con {model, input} e restituiscono {data:{taskId}}. Cambiano solo i campi input di ciascun modello — i video accettano duration, le immagini no. Il ciclo di polling è identico.

D: Perché output[].url restituisce 404 dopo qualche ora? R: L'URL è firmato e scade (expireAt nella risposta è un timestamp Unix). Scarica subito i byte, caricali sul tuo storage e servi il file da lì.

D: Come passo da Runway Gen-3 a Wan 2.7 senza riscrivere il codice? R: Se il tuo provider usa la stessa struttura di attività asincrona (crea → polling → URL di output), cambia due stringhe: API_BASE e l'id model. Racchiudile in variabili d'ambiente (API_BASE, MODEL_ID) e potrai fare A/B test con lo stesso codice.

D: Ricevo permission_denied anche se la mia chiave sembra corretta. R: Controlla due cose: (1) la chiave ha accesso al modello specifico (alcuni modelli richiedono un piano a pagamento) e (2) l'header Authorization è esattamente Bearer sk-... con uno spazio — non Bearer: sk-... e non la chiave grezza. Il request_id completo nel corpo dell'errore è necessario al supporto per tracciare la richiesta.

D: Devo usare il polling o i callback? R: Usa i callback in qualsiasi ambiente con un endpoint HTTP pubblico (applicazioni web, worker in coda). Usa il polling in script, attività cron e notebook — ovunque impostare un webhook risulterebbe inutile. In entrambi i casi conserva il taskId, così un worker andato in crash può riprendersi tramite polling invece di reinviare.