HiAPI
Blogue

Tutorial

API Runway ML de texto para vídeo em Python: um exemplo executável e a migração para hiapi

Jun 26, 2026
API Runway ML de texto para vídeo em Python: um exemplo executável e a migração para hiapi

Você quer chamar um modelo de texto para vídeo a partir do Python, e o Runway Gen-3 é o ponto de partida óbvio. Este guia mostra um exemplo mínimo executável em Python que demonstra o padrão assíncrono no estilo Runway (criar tarefa → consultar → baixar) e, em seguida, mapeia exatamente o mesmo código para o wan2.7-video/text-to-video da hiapi — uma alternativa direta que usa a mesma estrutura de /v1/tasks.

O que você precisa

  • Python 3.9+ e a biblioteca requests (pip install requests).
  • Uma chave de API. Para a hiapi: obtenha uma no seu painel de controle hiapi e exporte-a como HIAPI_API_KEY (ela começa com sk-).
  • Um prompt. Os modelos de vídeo aceitam um único prompt de texto e alguns parâmetros (proporção de tela, duração em segundos).

O cabeçalho de autenticação em todos os lugares abaixo é Authorization: Bearer sk-<key>. Se você receber HTTP 401 e permission_denied, a chave está errada ou não tem acesso ao modelo selecionado.

Exemplo mínimo executável (padrão assíncrono no estilo Runway)

Os provedores de texto para vídeo não retornam vídeos de forma síncrona — a geração leva de 30 a 120 segundos. A estrutura padrão é:

  1. POST /tasks com o id do modelo + entrada → retorna um taskId.
  2. GET /tasks/{id} → consulte até que status seja success ou fail.
  3. Leia output[0].url e baixe o MP4 (essa URL é válida por pouco tempo).

Abaixo está um script Python autocontido que implementa esse loop:

import os
import time
import requests

API_BASE = "https://api.runwayml.com/v1"  # ou qualquer provedor que use a mesma estrutura
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 foi mantido deliberadamente simples: sem SDK, sem código de cola assíncrono, apenas requests. Depois que ele funcionar, você pode manter exatamente o mesmo fluxo de controle e trocar de provedor alterando apenas duas strings.

Migração para hiapi (wan2.7-video/text-to-video)

A hiapi expõe o mesmo contrato de tarefa assíncrona em https://api.hiapi.ai/v1/tasks. As únicas diferenças no código de chamada são:

  1. API_BASE passa a ser https://api.hiapi.ai/v1.
  2. O id do modelo é wan2.7-video/text-to-video (id puro, não uma URL).
  3. A chave de autenticação é a sua chave hiapi sk-….

Concretamente, apenas create_task muda — wait_task e download permanecem idênticos:

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 chama isso de ratio: 16:9, 9:16, 1:1, 4:3, 3:4
                "duration": duration,   # 2-15 segundos
                "resolution": "720P",   # 720P $0.10/s; o padrão é 1080P, $0.167/s
            },
        },
        timeout=60,
    )
    resp.raise_for_status()
    return resp.json()["data"]["taskId"]

A estrutura da resposta é idêntica: na criação retorna {"data": {"taskId": "..."}}, e na consulta retorna:

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

Portanto, wait_task(...) e download(...) da seção anterior funcionam sem alterações. A migração completa tem apenas duas linhas: alterar API_BASE e alterar model. Todo o resto permanece igual.

output[].url é de curta duração (algumas horas; a resposta traz expireAt). Trate-a como um redirecionamento — baixe os bytes imediatamente e armazene por conta própria (S3, R2, seu CDN). Nunca incorpore a URL original em uma página.

Padrões para produção

O exemplo mínimo acima serve para demonstrações, mas antes de ir para produção você precisa de mais quatro coisas.

Use callbacks em vez de polling sempre que possível

Consultar a cada 5 segundos desperdiça requisições e mantém um worker ocupado por um ou dois minutos. Se o seu ambiente tiver uma URL de webhook pública, entregue-a à API e deixe que ela chame você de volta:

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",  # só nos chame quando o status for success ou fail
        },
    },
    timeout=60,
)

Seu webhook recebe o objeto de tarefa completo (mesma estrutura da resposta de consulta), então a etapa de download é exatamente igual.

Quando usar polling vs callback:

  • Polling: scripts, notebooks, tarefas em lote e qualquer lugar onde você não queira expor um listener HTTP.
  • Callback: aplicações web, workers de fila e qualquer lugar onde exista um endpoint público. Sempre combine com idempotência (próxima seção) — o provedor pode tentar novamente.

Torne o envio idempotente

Se o seu código tentar novamente create_task (instabilidade de rede, falha no meio do loop), é fácil enviar o mesmo prompt em duplicidade e pagar duas vezes. A correção barata: faça o hash da entrada em uma chave de cliente estável e deduplique na sua própria tabela antes de chamar:

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

Envie essa chave como cabeçalho Idempotency-Key (máximo de 255 bytes). A hiapi cria apenas uma tarefa por conta e chave; novas tentativas com a mesma chave retornam o taskId original em vez de cobrar novamente. Além disso, armazenar (submit_key → task_id) no seu próprio banco de dados também ajuda quando você precisar consultar a tarefa mais tarde.

Trate os erros com precisão

Você verá dois formatos de erro da hiapi:

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

…que aparece em falhas de autenticação/permissão (HTTP 401/403), e quando a tarefa falha definitivamente:

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

Trate-os de forma diferente: um permission_denied na criação significa que sua chave está errada ou não cobre aquele modelo — tentar novamente não faz sentido. Já um status: "fail" após a consulta pode ser um erro transitório do modelo e vale a pena tentar novamente uma vez com o mesmo prompt. Sempre registre o request_id para que o suporte possa rastrear.

Preços e cotas

A geração de vídeo é cobrada por segundo de saída, não por requisição, e varia conforme a resolução e o nível do modelo. Antes de calcular a economia unitária, consulte os números atualizados na página de preços da hiapi — não codifique valores em dólares no repositório.

Leituras relacionadas

Perguntas frequentes

P: Modelos de vídeo e de imagem precisam de SDKs diferentes? R: Não. Ambos usam POST /v1/tasks com {model, input} e ambos retornam {data:{taskId}}. Apenas os campos de input de cada modelo diferem — vídeo aceita duration, imagem não. O loop de consulta é exatamente o mesmo.

P: Por que output[].url retorna 404 depois de algumas horas? R: Essa URL é assinada e expira (o expireAt na resposta é um timestamp Unix). Baixe os bytes imediatamente, envie para o seu próprio armazenamento e sirva a partir de lá.

P: Como faço para trocar do Runway Gen-3 para o Wan 2.7 sem reescrever nada? R: Se o seu provedor usa a mesma estrutura de tarefa assíncrona (criar → consultar → URL de saída), altere duas strings: API_BASE e o id do model. Coloque-as em variáveis de ambiente (API_BASE, MODEL_ID) e você poderá fazer testes A/B com o mesmo código.

P: Recebo permission_denied mesmo com a chave aparentemente correta. R: Verifique duas coisas: (1) se a chave tem acesso ao modelo específico (alguns modelos exigem um nível pago) e (2) se o cabeçalho Authorization é exatamente Bearer sk-... com espaço — não Bearer: sk-... e nem a chave crua. O request_id completo no corpo do erro é necessário para o suporte rastrear.

P: Devo usar polling ou callback? R: Use callback em qualquer ambiente que tenha um endpoint HTTP público (aplicações web, workers de fila). Use polling em scripts, tarefas cron e notebooks — qualquer lugar onde montar um webhook seja desnecessário. De qualquer forma, armazene o taskId para que um worker que falhou possa se recuperar consultando em vez de reenviar.

Artigos relacionados

Ver todos os artigos →

Comece a criar com artefactos de IA persistentes

Obtenha a sua chave de API e chame modelos de imagem, vídeo e áudio através de uma única API baseada em tasks.

Ler a documentação da API