API Runway ML для генерации видео из текста в Python: рабочий пример и миграция на hiapi

Вы хотите вызывать модель генерации видео из текста из Python, и Runway Gen-3 — очевидная отправная точка. В этом руководстве показан минимальный рабочий пример на Python, демонстрирующий асинхронный шаблон в стиле Runway (создание задачи → опрос → скачивание), а затем тот же самый код переносится на wan2.7-video/text-to-video от hiapi — прямую замену с той же структурой /v1/tasks.

Что вам понадобится

• Python 3.9+ и библиотека requests (pip install requests).
• API-ключ. Для hiapi: получите его в панели управления hiapi и экспортируйте как HIAPI_API_KEY (он начинается с sk-).
• Промпт. Видеомодели принимают один текстовый промпт и несколько параметров (соотношение сторон, длительность в секундах).

Везде ниже заголовок аутентификации — Authorization: Bearer sk-<key>. Если вы получаете HTTP 401 и permission_denied, значит ключ неверен или у него нет доступа к выбранной модели.

Минимальный рабочий пример (асинхронный шаблон в стиле Runway)

Провайдеры генерации видео из текста не возвращают видео синхронно — генерация занимает 30–120 секунд. Стандартная структура такова:

1. POST /tasks с id модели + входными данными → возвращает taskId.
2. GET /tasks/{id} → опрашивайте, пока status не станет success или fail.
3. Прочитайте output[0].url и скачайте MP4 (этот URL действует ограниченное время).

Ниже — самодостаточный скрипт Python, реализующий этот цикл:

import os
import time
import requests

API_BASE = "https://api.runwayml.com/v1"  # или любой провайдер с той же структурой
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")

Этот скрипт намеренно прост: никакого SDK, никакого асинхронного клея — только requests. Как только он заработает, вы сможете сохранить ту же самую логику управления и переключить провайдера, заменив всего две строки.

Миграция на hiapi (wan2.7-video/text-to-video)

hiapi предоставляет тот же контракт асинхронных задач по адресу https://api.hiapi.ai/v1/tasks. Единственные отличия в коде вызова:

1. API_BASE становится https://api.hiapi.ai/v1.
2. Id модели — wan2.7-video/text-to-video (голый id, не URL).
3. Ключ аутентификации — ваш ключ hiapi sk-….

Конкретно, меняется только create_task — wait_task и download остаются идентичными:

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 называет это ratio: 16:9, 9:16, 1:1, 4:3, 3:4
                "duration": duration,   # 2-15 секунд
                "resolution": "720P",   # 720P $0.10/сек; по умолчанию 1080P, $0.167/сек
            },
        },
        timeout=60,
    )
    resp.raise_for_status()
    return resp.json()["data"]["taskId"]

Структура ответа полностью идентична: при создании возвращается {"data": {"taskId": "..."}}, при опросе:

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

Поэтому wait_task(...) и download(...) из предыдущего раздела работают без изменений. Полная миграция — это всего две строки: изменить API_BASE и изменить model. Всё остальное остаётся прежним.

output[].url действует ограниченное время (несколько часов, в ответе есть expireAt). Относитесь к нему как к редиректу — сразу скачивайте байты и храните их самостоятельно (S3, R2, ваш CDN). Никогда не встраивайте исходный URL в страницу.

Шаблоны для продакшена

Минимальный пример выше подходит для демонстрации, но перед запуском в продакшен вам понадобятся ещё четыре вещи.

Используйте callback вместо опроса, где это возможно

Опрос каждые 5 секунд расходует запросы и занимает worker на одну-две минуты. Если у вашей среды есть публичный webhook URL, передайте его API, и пусть он вызывает вас:

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",  # вызывать нас только при статусе success или fail
        },
    },
    timeout=60,
)

Ваш webhook получит полный объект задачи (с той же структурой, что и ответ опроса), поэтому шаг скачивания остаётся точно таким же.

Когда опрос, а когда callback:

• Опрос: скрипты, notebook, пакетные задачи и любые места, где вы не хотите открывать HTTP-листенер.
• Callback: веб-приложения, worker'ы очередей и любые места, где есть публичная конечная точка. Всегда в сочетании с идемпотентностью (следующий раздел) — провайдер может повторять попытки.

Сделайте отправку идемпотентной

Если ваш код повторяет create_task (сетевые сбои, падение посреди цикла), легко повторно отправить тот же промпт и заплатить дважды. Дешёвое решение: хешировать входные данные в стабильный клиентский ключ и дедуплицировать в своей таблице перед вызовом:

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

Отправьте этот ключ как заголовок Idempotency-Key (не более 255 байт). hiapi создаёт только одну задачу на аккаунт и ключ, и повторная попытка с тем же ключом возвращает исходный taskId, а не выставляет счёт повторно. Дополнительно сохраняйте (submit_key → task_id) в своей базе данных — это пригодится, когда позже понадобится найти задачу.

Обрабатывайте ошибки точно

Вы увидите от hiapi два вида обёрток ошибок:

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

… при ошибках аутентификации/прав (HTTP 401/403), а также когда задача в итоге завершается неудачей:

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

Относитесь к ним по-разному: permission_denied при создании означает, что ваш ключ неверен или не покрывает эту модель — повторять попытку бессмысленно. status: "fail" задачи после опроса может быть временной ошибкой модели, и стоит один раз повторить с тем же промптом. Всегда записывайте request_id, чтобы поддержка могла отследить проблему.

Цены и квоты

Генерация видео тарифицируется по секундам выходного видео, а не по запросам, и зависит от разрешения и уровня модели. Перед расчётом юнит-экономики смотрите актуальные цифры на странице цен hiapi — не хардкодьте долларовые значения в репозитории.

Связанные материалы

• Страница модели Wan 2.7 для генерации видео из текста — карточка модели, поддерживаемые значения aspect_ratio/duration и playground, который выполняет точно такие же вызовы /v1/tasks.
• Документация hiapi — полный справочник по контракту асинхронных задач (создание /v1/tasks, опрос, схема callback, коды ошибок).
• Каталог моделей — модели изображений, видео и аудио с общей структурой вызова.

Часто задаваемые вопросы

В: Нужны ли разные SDK для видеомоделей и моделей изображений? О: Нет. Обе используют POST /v1/tasks с {model, input} и обе возвращают {data:{taskId}}. Отличаются только поля input каждой модели — видео принимает duration, изображения нет. Цикл опроса абсолютно идентичен.

В: Почему output[].url через несколько часов возвращает 404? О: Этот URL подписан и истекает (expireAt в ответе — это временная метка Unix). Сразу скачайте байты, загрузите в своё хранилище и раздавайте оттуда.

В: Как переключиться с Runway Gen-3 на Wan 2.7 без переписывания? О: Если ваш провайдер использует ту же структуру асинхронных задач (создание → опрос → URL вывода), измените две строки: API_BASE и id model. Оберните их в переменные окружения (API_BASE, MODEL_ID), и вы сможете A/B-тестировать одним и тем же кодом.

В: Я получаю permission_denied, хотя мой ключ выглядит правильным. О: Проверьте две вещи: (1) у ключа есть доступ к конкретной модели (некоторые модели требуют платного уровня), и (2) заголовок Authorization — ровно Bearer sk-... с пробелом — не Bearer: sk-... и не голый ключ. Полный request_id в теле ошибки нужен поддержке для отслеживания.

В: Мне опрашивать или использовать callback? О: Используйте callback в любой среде с публичной HTTP-конечной точкой (веб-приложения, worker'ы очередей). Используйте опрос в скриптах, cron-задачах и notebook — там, где поднимать webhook кажется глупостью. В любом случае сохраняйте taskId, чтобы упавший worker мог восстановиться через опрос, а не отправлять задачу заново.