Python 中的 Runway ML 文本生成视频 API：可运行的示例与迁移至 hiapi

Python からテキスト生成動画モデルを呼び出したいなら、Runway Gen-3 が明白な出発点です。本ガイドでは、最小限の実行可能な Python サンプルを通じて Runway スタイルの非同期パターン（タスク作成 → ポーリング → ダウンロード）を示し、その後まったく同じコードを hiapi の wan2.7-video/text-to-video にマッピングします —— 同じ /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 のみ。一度動作すれば、まったく同じ制御フローを維持したまま、2 つの文字列を変更するだけでプロバイダーを切り替えられます。

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(...) は変更なしで動作します。完全な移行はたった 2 行：API_BASE の変更と model の変更です。周囲のすべては変わりません。

output[].url は短期間のみ有効です（数時間、レスポンスに expireAt が含まれます）。これをリダイレクトとして扱ってください —— バイトを直ちにダウンロードし、自分で保存します（S3、R2、あなたの CDN）。生の URL をページに埋め込まないでください。

本番レベルのパターン

上記の最小サンプルはデモには適していますが、本番公開前に追加で 4 つの事項が必要です。

可能な限りポーリングではなくコールバックを使う

5 秒ごとのポーリングはリクエストを浪費し、ワーカーを 1〜2 分間占有します。環境に公開 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 は完全なタスクオブジェクトを受け取ります（ポーリングレスポンスと同じ構造）。したがってダウンロード手順はまったく同じです。

ポーリング vs コールバックの使い分け：

• ポーリング：スクリプト、notebook、バッチタスク、および HTTP リスナーを公開したくないあらゆる場所。
• コールバック：Web アプリ、キューワーカー、および公開エンドポイントがあるあらゆる場所。常に冪等性（次節）と組み合わせてください —— プロバイダーは再試行する可能性があります。

送信を冪等にする

コードが create_task を再試行する場合（ネットワークの揺らぎ、ループ途中のクラッシュ）、同じプロンプトを重複送信して 2 回課金されやすくなります。安価な修正方法：入力を安定したクライアントキーにハッシュ化し、呼び出し前に自身のテーブルで重複排除します：

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 はアカウントとキーごとに 1 つのタスクのみを作成し、同じキーでの再試行は重複課金ではなく元の taskId を返します。さらにそれを基に (submit_key → task_id) を自身のデータベースに保存しておけば、後でタスクを検索する必要があるときに役立ちます。

エラーを正確に処理する

hiapi からは 2 種類のエラー封筒が返ります：

{ "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" は一時的なモデルエラーの可能性があり、同じプロンプトで 1 回再試行する価値があります。常に request_id をログに記録し、サポート担当が追跡できるようにしてください。

料金とクォータ

動画生成はリクエスト単位ではなく出力秒数単位で課金され、解像度とモデル階層によって異なります。ユニットエコノミクスを計算する前に、hiapi 料金ページ で最新の数値を確認してください —— リポジトリにドル値をハードコードしないでください。

関連記事

• Wan 2.7 テキスト生成動画モデルページ —— モデルカード、サポートされる aspect_ratio/duration の値、そしてまったく同じ /v1/tasks 呼び出しを行う playground。
• hiapi ドキュメント —— 非同期タスク契約の完全なリファレンス（/v1/tasks の作成、ポーリング、コールバックスキーマ、エラーコード）。
• モデルカタログ —— 同じ呼び出し構造を共有する画像、動画、音声モデル。

よくある質問

Q：動画モデルと画像モデルには異なる SDK が必要ですか？ A：いいえ。どちらも POST /v1/tasks に {model, input} を付けて送信し、どちらも {data:{taskId}} を返します。異なるのはモデルごとの input フィールドのみです —— 動画は duration を受け取りますが、画像は受け取りません。ポーリングループはまったく同じです。

Q：output[].url が数時間後に 404 になるのはなぜですか？ A：その URL は署名付きで期限切れになります（レスポンスの expireAt は Unix タイムスタンプです）。バイトを直ちにダウンロードし、自身のストレージにプッシュして、そこから配信してください。

Q：書き直しなしで Runway Gen-3 から Wan 2.7 に切り替えるにはどうすればよいですか？ A：プロバイダーが同じ非同期タスク構造（作成 → ポーリング → 出力 URL）を使用していれば、2 つの文字列を変更するだけです：API_BASE と model id。それらを環境変数（API_BASE、MODEL_ID）にラップすれば、同じコードで A/B テストができます。

Q：キーは正しいように見えるのに permission_denied を受け取ります。 A：2 つのことを確認してください：(1) キーが特定のモデルにアクセスできるか（一部のモデルは有料階層が必要）、および (2) Authorization ヘッダーがスペース付きの Bearer sk-... に正確になっているか —— Bearer: sk-... でも生のキーでもありません。エラー本文の完全な request_id はサポート担当の追跡に必要です。

Q：ポーリングとコールバックのどちらを使うべきですか？ A：公開 HTTP エンドポイントがある環境（Web アプリ、キューワーカー）ではコールバックを使用してください。スクリプト、cron タスク、notebook —— webhook の設定が馬鹿げて見えるあらゆる場所ではポーリングを使用してください。いずれの場合も taskId を保存し、クラッシュしたワーカーが再送信ではなくポーリングによって復旧できるようにしてください。