テキスト/画像から動画を生成するREST APIを叩いてみる（非同期タスク連携の実装・2026年版）

テキストや画像を投げると動画（または画像）を生成してくれるAPIはいくつか出てきたが、その多くは「タスクを投げる → ポーリングで結果を取りに行く」という非同期ジョブ型のインターフェースになっている。同期で待たせると生成に数十秒〜数分かかってタイムアウトしてしまうので、当然といえば当然だ。

この記事では、その典型例として動画/画像生成のOpen API（REST）を題材に、タスク投入からポーリングで完成物のURLを受け取るところまでを、curl・Python・Node.jsで実装してみる。非同期タスク型のAPIを初めて触る人の足場になれば、という内容だ。

認証とAPIキーの取得

リクエストはすべてBearerトークン認証。ヘッダーに sk- から始まるキーを載せる。

Authorization: Bearer sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

キーは公式サイトにログインして、アカウントページの「Manage API Keys」から発行する。今回題材にしている Omni Video の場合、ログイン後のアカウントページで sk- キーを生成すると、そのままこのAPIに使える。キーはサーバーサイドに置き、フロントには絶対に出さないこと（クライアントに置くと残クレジットを丸ごと持っていかれる）。

使えるモデル

model_id は以下が受け付けられる。動画系と画像系で、レスポンスに入ってくるURLのフィールドが変わる点に注意。

model_id	用途	出力
omni-flash	テキスト/画像 → 動画	video_url
omni-pro	テキスト/画像 → 動画	video_url
seedance-2	テキスト/画像 → 動画	video_url
gpt-image-2	テキスト/画像 → 画像	image_url
nano-banana-2	テキスト/画像 → 画像	image_url

タスクを作成する

POST /api/v1/tasks/create にJSONを投げる。必須は model_id と prompt。image_urls（画像を起点にする場合）と aspect_ratio は任意。

curl -X POST https://omnivideo.net/api/v1/tasks/create \
  -H "Authorization: Bearer sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" \
  -H "Content-Type: application/json" \
  -d '{
    "model_id": "gpt-image-2",
    "prompt": "a serene zen garden at sunrise, ultra detailed",
    "aspect_ratio": "16:9"
  }'

レスポンスはこうなる。ここで返ってくる task_id を握っておく。

{
  "code": 200,
  "msg": "提交成功，等待生成",
  "data": {
    "task_id": "abcdef123456",
    "request_id": "kie_xxxxxxxxxxxx",
    "credits": 15
  }
}

クレジットは投入時点で引かれる。ただし生成が失敗（後述の task_status: 4）したタスクは自動で返金される仕様なので、失敗ぶんを自分で勘定する必要はない。

タスクをポーリングする

GET /api/v1/tasks/{task_id} を task_status が 3（成功）か 4（失敗）になるまで叩く。

curl https://omnivideo.net/api/v1/tasks/abcdef123456 \
  -H "Authorization: Bearer sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"

{
  "code": 200,
  "msg": "ok",
  "data": {
    "task_id": "abcdef123456",
    "task_status": 3,
    "task_type": "image",
    "model_id": "gpt-image-2",
    "image_url": "https://your-cdn.com/...",
    "video_url": null,
    "audio_url": null,
    "credits": 15,
    "created_at": 1730000000
  }
}

task_status の意味は次の4つ。

• 1 = キュー待ち
• 2 = 生成中
• 3 = 成功
• 4 = 失敗

動画モデルなら video_url、画像モデルなら image_url に完成物のURLが入る。逆側のフィールドは null のままなので、model_id の種別で読むフィールドを切り替えると安全だ。

Pythonでの実装例

投入してからポーリングで取り切るまでを一本にするとこうなる。

import os
import time
import requests

BASE = "https://omnivideo.net/api/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['OMNI_API_KEY']}"}

def create_task(model_id: str, prompt: str, aspect_ratio: str = "16:9"):
    r = requests.post(
        f"{BASE}/tasks/create",
        headers=HEADERS,
        json={"model_id": model_id, "prompt": prompt, "aspect_ratio": aspect_ratio},
        timeout=30,
    )
    r.raise_for_status()
    body = r.json()
    if body["code"] != 200:
        raise RuntimeError(f"create failed: {body['msg']}")
    return body["data"]["task_id"]

def wait_for_result(task_id: str, interval: int = 5, timeout: int = 600):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        r = requests.get(f"{BASE}/tasks/{task_id}", headers=HEADERS, timeout=30)
        r.raise_for_status()
        data = r.json()["data"]
        status = data["task_status"]
        if status == 3:
            return data["video_url"] or data["image_url"]
        if status == 4:
            raise RuntimeError(f"generation failed (task {task_id})")
        time.sleep(interval)  # 1=キュー、2=実行中 の間は待つ
    raise TimeoutError(f"task {task_id} did not finish in {timeout}s")

if __name__ == "__main__":
    tid = create_task("gpt-image-2", "a serene zen garden at sunrise")
    print("result:", wait_for_result(tid))

ポーリング間隔は5秒くらいから始めて、長くかかるモデルでは指数バックオフにしてもいい。Webhookは用意されていないので、結果取得はこのポーリング一択になる。

Node.js（fetch）での実装例

const BASE = "https://omnivideo.net/api/v1";
const headers = { Authorization: `Bearer ${process.env.OMNI_API_KEY}` };

async function createTask(modelId, prompt, aspectRatio = "16:9") {
  const res = await fetch(`${BASE}/tasks/create`, {
    method: "POST",
    headers: { ...headers, "Content-Type": "application/json" },
    body: JSON.stringify({ model_id: modelId, prompt, aspect_ratio: aspectRatio }),
  });
  const body = await res.json();
  if (body.code !== 200) throw new Error(`create failed: ${body.msg}`);
  return body.data.task_id;
}

async function waitForResult(taskId, interval = 5000, timeout = 600000) {
  const deadline = Date.now() + timeout;
  while (Date.now() < deadline) {
    const res = await fetch(`${BASE}/tasks/${taskId}`, { headers });
    const { data } = await res.json();
    if (data.task_status === 3) return data.video_url ?? data.image_url;
    if (data.task_status === 4) throw new Error(`generation failed: ${taskId}`);
    await new Promise((r) => setTimeout(r, interval));
  }
  throw new Error(`task ${taskId} timed out`);
}

エラーハンドリング

エラーは大きく3パターン。

• code: 200 … 成功。data を読む。
• code: 0 … ビジネス側の失敗。理由は msg に入るので、ログにはここを残す。
• HTTP 401 … キーが無効、または未指定。Authorization ヘッダーを疑う。

HTTPステータスとレスポンスの code は別レイヤーなので、raise_for_status() だけで安心せず、code も必ず見ること。ここを素通りさせると「200が返ってるのに data が空」で詰まる。

ハマりどころ

• クレジットは投入時に引かれる。 ループのテスト中にうっかり大量投入すると一気に減る。失敗ぶんは自動返金されるが、成功ぶんは戻らない。開発中は安いモデルで通すのが無難。
• 同期で待てると思わないこと。 tasks/create は即レスで task_id を返すだけ。完成物はポーリングで取りに行く前提の設計になっている。
• 出力フィールドの出し分け。 動画モデルなのに image_url を読んでいてずっと null、という凡ミスをやりがち。model_id で分岐させておくと事故らない。

まとめ

「投げて、ポーリングして、URLを受け取る」という非同期タスク型は、生成系APIではほぼ定番のパターンだ。一度この形でラッパーを書いておけば、モデルを差し替えるだけで動画にも画像にも使い回せる。キーはアカウントページから発行して環境変数に逃がす、code フィールドも必ず見る、この2点だけ押さえておけば最初の連携でハマることはまず無いはずだ。