API de texto a video de Runway ML en Python: un ejemplo ejecutable y la migración a hiapi

Quieres llamar a un modelo de texto a video desde Python y Runway Gen-3 es el punto de partida obvio. Esta guía muestra un ejemplo mínimo ejecutable en Python que demuestra el patrón asíncrono al estilo de Runway (crear tarea → sondear → descargar) y luego traslada exactamente el mismo código a wan2.7-video/text-to-video de hiapi, un sustituto directo que usa la misma estructura de /v1/tasks.

Qué necesitas preparar

• Python 3.9+ y la librería requests (pip install requests).
• Una clave de API. Para hiapi: obtén una desde tu panel de control de hiapi y expórtala como HIAPI_API_KEY (empieza por sk-).
• Un prompt. Los modelos de video aceptan un único prompt de texto y unos pocos parámetros (relación de aspecto, duración en segundos).

La cabecera de autenticación en todo lo que sigue es Authorization: Bearer sk-<key>. Si recibes HTTP 401 y permission_denied, la clave es incorrecta o no tiene acceso al modelo elegido.

Ejemplo mínimo ejecutable (patrón asíncrono al estilo de Runway)

Los proveedores de texto a video no devuelven el video de forma síncrona: la generación tarda entre 30 y 120 segundos. La estructura estándar es:

1. POST /tasks con el id del modelo + la entrada → devuelve un taskId.
2. GET /tasks/{id} → sondear hasta que status sea success o fail.
3. Leer output[0].url y descargar el MP4 (esa URL es de vigencia corta).

A continuación, un script de Python autocontenido que implementa este bucle:

import os
import time
import requests

API_BASE = "https://api.runwayml.com/v1"  # o cualquier proveedor que use la misma estructura
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")

Este script se mantiene deliberadamente simple: sin SDK, sin código de pegamento asíncrono, solo requests. Una vez que funcione, puedes conservar exactamente el mismo flujo de control y cambiar de proveedor cambiando solo dos cadenas.

Migración a hiapi (wan2.7-video/text-to-video)

hiapi expone el mismo contrato de tareas asíncronas en https://api.hiapi.ai/v1/tasks. Las únicas diferencias en el código de llamada son:

1. API_BASE pasa a ser https://api.hiapi.ai/v1.
2. El id del modelo es wan2.7-video/text-to-video (id simple, no una URL).
3. La clave de autenticación es tu clave sk-… de hiapi.

En concreto, solo cambia create_task; wait_task y download son idénticas:

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 llama ratio: 16:9, 9:16, 1:1, 4:3, 3:4
                "duration": duration,   # 2-15 segundos
                "resolution": "720P",   # 720P $0.10/segundo; por defecto es 1080P, $0.167/segundo
            },
        },
        timeout=60,
    )
    resp.raise_for_status()
    return resp.json()["data"]["taskId"]

La estructura de respuesta es exactamente la misma: al crear devuelve {"data": {"taskId": "..."}}, y al sondear devuelve:

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

Por tanto, wait_task(...) y download(...) de la sección anterior funcionan sin modificaciones. La migración completa son solo dos líneas: cambiar API_BASE y cambiar model. Todo lo demás permanece igual.

output[].url es de vigencia corta (unas horas; la respuesta incluye expireAt). Trátala como una redirección: descarga los bytes de inmediato y guárdalos por tu cuenta (S3, R2, tu CDN). Nunca incrustes la URL original en una página.

Patrones de nivel de producción

El ejemplo mínimo anterior sirve para una demostración, pero antes de lanzar a producción necesitas cuatro cosas adicionales.

Usa callbacks en lugar de sondeo siempre que sea posible

Sondear cada 5 segundos desperdicia peticiones y mantiene ocupado a un worker durante uno o dos minutos. Si tu entorno tiene una URL de webhook pública, pásasela a la API para que te llame de vuelta:

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",  # solo nos llama cuando el estado es success o fail
        },
    },
    timeout=60,
)

Tu webhook recibe el objeto de tarea completo (con la misma estructura que la respuesta de sondeo), por lo que el paso de descarga es idéntico.

Cuándo sondear frente a cuándo usar callback:

• Sondeo: scripts, notebooks, tareas por lotes y cualquier lugar donde no quieras exponer un listener HTTP.
• Callback: aplicaciones web, workers de colas y cualquier lugar donde exista un endpoint público. Acompáñalo siempre de idempotencia (siguiente sección), ya que el proveedor puede reintentar.

Haz que el envío sea idempotente

Si tu código reintenta create_task (fallos de red, caída a mitad del bucle), es fácil enviar dos veces el mismo prompt y pagar dos veces. La solución barata: aplicar un hash a la entrada para obtener una clave cliente estable y deduplicar en tu propia tabla antes de llamar:

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

Envía esa clave como cabecera Idempotency-Key (máximo 255 bytes). hiapi crea una sola tarea por cuenta y clave; los reintentos con la misma clave devuelven el taskId original en lugar de cobrar de nuevo. Guardar (submit_key → task_id) en tu propia base de datos también ayuda cuando más adelante necesites localizar la tarea.

Trata los errores con precisión

Verás dos envoltorios de error de hiapi:

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

…en fallos de autenticación o permisos (HTTP 401/403), y cuando una tarea termina en fallo:

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

Trátalos de forma distinta: un permission_denied al crear significa que tu clave es incorrecta o no cubre ese modelo, así que reintentar no tiene sentido. Un status: "fail" de la tarea tras el sondeo puede ser un error transitorio del modelo que merece un reintento con el mismo prompt. Registra siempre el request_id para que el soporte pueda rastrearlo.

Precios y cuotas

La generación de video se factura por segundos de salida, no por petición, y varía según la resolución y el nivel del modelo. Antes de calcular la economía unitaria, consulta las cifras vigentes en la página de precios de hiapi; no codifiques valores en dólares en el repositorio.

Lecturas relacionadas

• Página del modelo Wan 2.7 de texto a video: ficha del modelo, valores admitidos de aspect_ratio/duration y un playground que emite exactamente la misma llamada a /v1/tasks.
• Documentación de hiapi: referencia completa del contrato de tareas asíncronas (creación en /v1/tasks, sondeo, esquema de callback, códigos de error).
• Catálogo de modelos: modelos de imagen, video y audio que comparten la misma estructura de llamada.

Preguntas frecuentes

P: ¿Los modelos de video y de imagen necesitan SDK diferentes? R: No. Ambos usan POST /v1/tasks con {model, input} y ambos devuelven {data:{taskId}}. Solo cambian los campos de input de cada modelo: el video acepta duration, la imagen no. El bucle de sondeo es idéntico.

P: ¿Por qué output[].url devuelve 404 a las pocas horas? R: La URL va firmada y caduca (el expireAt de la respuesta es una marca de tiempo Unix). Descarga los bytes de inmediato, súbelos a tu propio almacenamiento y sirve el contenido desde ahí.

P: ¿Cómo cambio de Runway Gen-3 a Wan 2.7 sin reescribir nada? R: Si tu proveedor usa la misma estructura de tareas asíncronas (crear → sondear → URL de salida), cambia dos cadenas: API_BASE y el id de model. Envuélvelas en variables de entorno (API_BASE, MODEL_ID) y podrás hacer pruebas A/B con el mismo código.

P: Recibo permission_denied aunque mi clave parece correcta. R: Comprueba dos cosas: (1) que la clave tenga acceso al modelo concreto (algunos modelos requieren un nivel de pago), y (2) que la cabecera Authorization sea exactamente Bearer sk-... con un espacio, no Bearer: sk-... ni la clave sin más. El request_id completo del cuerpo del error es lo que necesita el soporte para rastrearlo.

P: ¿Debo sondear o usar callbacks? R: Usa callbacks en cualquier entorno con un endpoint HTTP público (aplicaciones web, workers de colas). Usa sondeo en scripts, tareas cron y notebooks, es decir, donde montar un webhook sería absurdo. En ambos casos, guarda el taskId para que un worker caído pueda recuperarse sondeando en lugar de reenviar la tarea.