SaaSアプリにAI動画生成機能を追加する方法

アプリに動画生成を追加することは、画像生成を追加するのとは同じではありません。API 呼び出しはすぐに返りますが、動画はまだ準備できていません。タスク ID が返されるので、「もう終わった？」と完了まで問い合わせ続ける必要があります。

ほとんどの開発者は、初めて動画 API を呼び出したときにレスポンスボディに動画 URL があるのを待ち、代わりにタスク ID を受け取って戸惑います。本ガイドでは、タスクの送信、結果のポーリング、失敗時の処理、URL の有効期限が切れる前の出力保存まで、フルフローを順に説明します。

何を作るか

テキストプロンプトまたは画像を受け取り、動画生成タスクを送信し、完了までポーリングして、最終的な動画 URL を返すバックエンドサービスを作ります。1 つの API キーで、Veo 3 Fast、Sora 2、Kling Video、Runway の 4 モデルを扱います。

前提条件:

• Python 3.8+ または Node.js 18+
• CometAPI キー
• REST API の基礎知識

なぜ動画生成は違うのか理解する

画像生成では、リクエストを送ると同じレスポンスで画像が返ってきます。動画生成は非同期タスクキューを使います。

1. 生成リクエストを「送信」→ task_id が返る
2. 数秒ごとにステータスエンドポイントを「ポーリング」
3. ステータスが終端状態になったら、動画 URL を取得
4. 動画を「ダウンロードして保存」— URL は一時的

動画生成を画像生成と同じように扱って、最初のレスポンスに動画が含まれるのを待つと、毎回タイムアウトします。

本番の Web サービスでは、このポーリングループはリクエストハンドラではなくバックグラウンドワーカー（Celery、Bull など）で実行すべきです。以下の例は同期ポーリングを使っています。スクリプトやプロトタイプには適していますが、多数のユーザーを同時に扱う用途には向きません。

モデルを選ぶ

Source*: CometAPI model pages, May 2026. Note: "Sora 2" is CometAPI's model* identifier — refer to their model page for the underlying model details.

• Veo 3 Fast はテキスト→動画と画像→動画の両方に対応。1 秒あたり最安で、最初の選択肢に適しています。
• Sora 2 は動画と同時に音声もネイティブ生成します。セリフ、環境音、効果音を別の TTS ステップなしで生成。
• Kling Video は negative_prompt、cfg_scale、カメラ動作設定、pro モードなどを提供。4 つの中で最も細かな制御が可能。
• Runway は CometAPI 経由では画像→動画のみに対応。静止画像とモーション説明を与えるとアニメーション化します。

Veo タスクを送信する

Veo は multipart/form-data を使います。Python requests では正しく送るために files= を使ってください。data=dict は application/x-www-form-urlencoded になり、これは別物です。

import requestsimport osfrom dotenv import load_dotenv​load_dotenv()​def submit_veo_task(prompt: str, size: str = "16x9") -> str:    """Veo 3 Fast のテキスト→動画タスクを送信します。task_id を返します。"""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY 環境変数が設定されていません")​    response = requests.post(        "https://api.cometapi.com/v1/videos",        headers={"Authorization": f"Bearer {api_key}"},        files={            "prompt": (None, prompt),            "model": (None, "veo3-fast"),            "size": (None, size)        },        timeout=30    )    response.raise_for_status()    return response.json()["id"]​​task_id = submit_veo_task("風の強い午後、小麦畑の上を漂う紙の凧")print(f"タスクを送信しました: {task_id}")

結果をポーリングする

import time​def poll_veo_task(task_id: str, interval: int = 10, max_wait: int = 600) -> str:    """Veo タスクが完了するまでポーリングします。動画 URL を返します。"""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY 環境変数が設定されていません")​    headers = {"Authorization": f"Bearer {api_key}"}    url = f"https://api.cometapi.com/v1/videos/{task_id}"    elapsed = 0​    while elapsed < max_wait:        response = requests.get(url, headers=headers, timeout=30)        response.raise_for_status()        result = response.json()        status = result.get("status")​        if status == "succeeded":            return result["output"][0]        elif status in ("failed", "cancelled"):            raise RuntimeError(                f"タスク {task_id} はステータス '{status}' で失敗しました: "                f"{result.get('error', 'エラー詳細は返されませんでした')}"            )​        time.sleep(interval)        elapsed += interval​    raise TimeoutError(f"タスク {task_id} は {max_wait} 秒以内に完了しませんでした")​​video_url = poll_veo_task(task_id)print(f"動画の準備ができました: {video_url}")

より細かな制御には Kling Video を使う

Kling は異なるエンドポイント構成で JSON を使用します。Kling の終端ステータス文字列は "succeed"（"succeeded" ではない）で、これは API の実際のレスポンス形式に一致します。

def submit_kling_task(prompt: str, duration: str = "5", mode: str = "std") -> str:    """Kling のテキスト→動画タスクを送信します。task_id を返します。"""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY 環境変数が設定されていません")​    response = requests.post(        "https://api.cometapi.com/kling/v1/videos/text2video",        headers={            "Authorization": f"Bearer {api_key}",            "Content-Type": "application/json"        },        json={            "model_name": "kling-v1-6",            "prompt": prompt,            "negative_prompt": "blurry, low quality, watermark",            "cfg_scale": 0.5,            "mode": mode,         # "std" または "pro"            "aspect_ratio": "16:9",            "duration": duration  # "5" または "10"        },        timeout=30    )    response.raise_for_status()    return response.json()["data"]["task_id"]​​def poll_kling_task(task_id: str, interval: int = 10, max_wait: int = 600) -> str:    """Kling タスクが完了するまでポーリングします。動画 URL を返します。"""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY 環境変数が設定されていません")​    headers = {"Authorization": f"Bearer {api_key}"}    url = f"https://api.cometapi.com/kling/v1/videos/text2video/{task_id}"    elapsed = 0​    while elapsed < max_wait:        response = requests.get(url, headers=headers, timeout=30)        response.raise_for_status()        result = response.json()        status = result["data"]["task_status"]​        if status == "succeed":  # Kling は "succeed"（"succeeded" ではない）を使用            return result["data"]["task_result"]["videos"][0]["url"]        elif status == "failed":            error_detail = result.get("data", {}).get("task_result", "詳細なし")            raise RuntimeError(                f"Kling タスク {task_id} が失敗しました: {error_detail}"            )​        time.sleep(interval)        elapsed += interval​    raise TimeoutError(f"Kling タスク {task_id} は {max_wait}s 後にタイムアウトしました")

Source*:* CometAPI Kling Video docs

静止画像を Runway でアニメーション化する

Runway は画像→動画のみです。さらに追加ヘッダー（X-Runway-Version）が必要です。

def submit_runway_task(image_url: str, motion_prompt: str, duration: int = 5) -> str:    """Runway の画像→動画タスクを送信します。task_id を返します。"""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY 環境変数が設定されていません")​    response = requests.post(        "https://api.cometapi.com/runwayml/v1/image_to_video",        headers={            "Authorization": f"Bearer {api_key}",            "X-Runway-Version": "2024-11-06",            "Content-Type": "application/json"        },        json={            "model": "gen3a_turbo",            "promptImage": image_url,  # 安定した HTTPS URL である必要があります            "promptText": motion_prompt,            "duration": duration,            "ratio": "1280:720",            "watermark": False        },        timeout=30    )    response.raise_for_status()    return response.json()["id"]​​def poll_runway_task(task_id: str, interval: int = 5, max_wait: int = 600) -> str:    """Runway タスクをポーリングします。完了時に動画 URL を返します。"""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY 環境変数が設定されていません")​    headers = {        "Authorization": f"Bearer {api_key}",        "X-Runway-Version": "2024-11-06"    }    url = f"https://api.cometapi.com/runwayml/v1/tasks/{task_id}"    elapsed = 0​    while elapsed < max_wait:        response = requests.get(url, headers=headers, timeout=30)        response.raise_for_status()        result = response.json()        status = result.get("status")​        if status == "task_not_exist":            # CometAPI 固有: バックエンドでタスクが初期化中。数秒後に再試行            time.sleep(interval)            elapsed += interval            continue        elif status == "succeeded":            return result["output"][0]        elif status in ("failed", "cancelled"):            raise RuntimeError(f"Runway タスク {task_id} が失敗しました: {result.get('error', '詳細なし')}")​        time.sleep(interval)        elapsed += interval​    raise TimeoutError(f"Runway タスク {task_id} は {max_wait}s 後にタイムアウトしました")

Source*:* CometAPI Runway docs

URL の有効期限が切れる前に動画を保存する

生成 API から返る動画の URL は一時的です。URL を受け取ったらすぐにダウンロードし、自分が管理する場所に保存しましょう。

import requestsimport pathlib​def download_video(url: str, output_path: str) -> None:    """ストリーミングで URL からローカルファイルへ動画をダウンロードします。"""    out = pathlib.Path(output_path)    if out.parent != pathlib.Path("."):        out.parent.mkdir(parents=True, exist_ok=True)​    with requests.get(url, stream=True, timeout=60) as r:        r.raise_for_status()        with open(out, "wb") as f:            for chunk in r.iter_content(chunk_size=8192):                f.write(chunk)    print(f"{output_path} に保存しました")​​# フルフローtask_id = submit_veo_task("都市のスカイライン上を雲が動くタイムラプス")video_url = poll_veo_task(task_id)download_video(video_url, "output/city_timelapse.mp4")

本番ではローカルファイルへの書き込みの代わりに、S3、Cloudflare R2、または任意のストレージにアップロードします。ストリーミングのパターンは同じです—動画全体をメモリに読み込まず、バイトを直接パイプします。

失敗の扱い

Source: CometAPI video generation docs

Node.js 版

Node.js 18+ は fetch と FormData をネイティブで含みます。この例は 4 モデルすべてをカバーします。

// Node.js 18+ — 追加パッケージ不要​const API_KEY = process.env.COMETAPI_KEY;if (!API_KEY) throw new Error('COMETAPI_KEY が設定されていません');​// --- Veo 3 Fast ---async function submitVeoTask(prompt, size = '16x9') {  const form = new FormData();  form.append('prompt', prompt);  form.append('model', 'veo3-fast');  form.append('size', size);​  const res = await fetch('https://api.cometapi.com/v1/videos', {    method: 'POST',    headers: { 'Authorization': `Bearer ${API_KEY}` },    body: form  });  if (!res.ok) throw new Error(`Veo の送信に失敗: ${res.status}`);  return (await res.json()).id;}​async function pollVeoTask(taskId, intervalMs = 10000, maxWaitMs = 600000) {  let elapsed = 0;  while (elapsed < maxWaitMs) {    const res = await fetch(`https://api.cometapi.com/v1/videos/${taskId}`, {      headers: { 'Authorization': `Bearer ${API_KEY}` }    });    if (!res.ok) throw new Error(`ポーリングに失敗: ${res.status}`);    const result = await res.json();​    if (result.status === 'succeeded') return result.output[0];    if (['failed', 'cancelled'].includes(result.status)) {      throw new Error(`タスク ${taskId} が失敗: ${result.error ?? '詳細なし'}`);    }    await new Promise(r => setTimeout(r, intervalMs));    elapsed += intervalMs;  }  throw new Error(`タスク ${taskId} はタイムアウトしました`);}​// --- Kling Video ---async function submitKlingTask(prompt, duration = '5', mode = 'std') {  const res = await fetch('https://api.cometapi.com/kling/v1/videos/text2video', {    method: 'POST',    headers: {      'Authorization': `Bearer ${API_KEY}`,      'Content-Type': 'application/json'    },    body: JSON.stringify({      model_name: 'kling-v1-6',      prompt,      negative_prompt: 'blurry, low quality, watermark',      cfg_scale: 0.5,      mode,      aspect_ratio: '16:9',      duration    })  });  if (!res.ok) throw new Error(`Kling の送信に失敗: ${res.status}`);  return (await res.json()).data.task_id;}​async function pollKlingTask(taskId, intervalMs = 10000, maxWaitMs = 600000) {  let elapsed = 0;  while (elapsed < maxWaitMs) {    const res = await fetch(      `https://api.cometapi.com/kling/v1/videos/text2video/${taskId}`,      { headers: { 'Authorization': `Bearer ${API_KEY}` } }    );    if (!res.ok) throw new Error(`Kling のポーリングに失敗: ${res.status}`);    const result = await res.json();    const status = result.data.task_status;​    if (status === 'succeed') return result.data.task_result.videos[0].url;    if (status === 'failed') {      throw new Error(`Kling タスク ${taskId} が失敗: ${JSON.stringify(result.data.task_result ?? '詳細なし')}`);    }    await new Promise(r => setTimeout(r, intervalMs));    elapsed += intervalMs;  }  throw new Error(`Kling タスク ${taskId} はタイムアウトしました`);}​// --- Runway (画像→動画) ---async function submitRunwayTask(imageUrl, motionPrompt, duration = 5) {  const res = await fetch('https://api.cometapi.com/runwayml/v1/image_to_video', {    method: 'POST',    headers: {      'Authorization': `Bearer ${API_KEY}`,      'X-Runway-Version': '2024-11-06',      'Content-Type': 'application/json'    },    body: JSON.stringify({      model: 'gen3a_turbo',      promptImage: imageUrl,      promptText: motionPrompt,      duration,      ratio: '1280:720',      watermark: false    })  });  if (!res.ok) throw new Error(`Runway の送信に失敗: ${res.status}`);  return (await res.json()).id;}​async function pollRunwayTask(taskId, intervalMs = 5000, maxWaitMs = 600000) {  let elapsed = 0;  while (elapsed < maxWaitMs) {    const res = await fetch(      `https://api.cometapi.com/runwayml/v1/tasks/${taskId}`,      { headers: { 'Authorization': `Bearer ${API_KEY}`, 'X-Runway-Version': '2024-11-06' } }    );    if (!res.ok) throw new Error(`Runway のポーリングに失敗: ${res.status}`);    const result = await res.json();    const status = result.status;​    if (status === 'task_not_exist') {      // CometAPI 固有: タスクはまだ初期化中      await new Promise(r => setTimeout(r, intervalMs));      elapsed += intervalMs;      continue;    }    if (status === 'succeeded') return result.output[0];    if (['failed', 'cancelled'].includes(status)) {      throw new Error(`Runway タスク ${taskId} が失敗: ${result.error ?? '詳細なし'}`);    }    await new Promise(r => setTimeout(r, intervalMs));    elapsed += intervalMs;  }  throw new Error(`Runway タスク ${taskId} はタイムアウトしました`);}​// 使用例const taskId = await submitVeoTask('風の強い午後、小麦畑の上を漂う紙の凧');const videoUrl = await pollVeoTask(taskId);console.log('動画の準備ができました:', videoUrl);

次のステップ

4 つの動画モデルで動作するコード、失敗も扱うポーリングループ、生成コンテンツを失わないためのダウンロード手順が手に入りました。

多くの開発者が次に直面する問題は、1 つのモデルをハードコードしてしまい、安価または高速なオプションに切り替えるたびに複数ファイルを触る必要があることです。次の記事では、コードを書き換えることなくモデル間でリクエストをルーティングする方法を扱います。

次へ: モデルをコードを書き換えずに切り替える方法

FAQ

Q: なぜ API レスポンスで動画ではなくタスク ID が返るのですか？

動画生成は非同期です。Veo、Sora、Kling、Runway のようなモデルはレンダリングに 2〜5 分かかります。リクエストがタイムアウトしないよう、API はすぐにタスク ID を返します。タスクが終端状態（succeeded、succeed、failed）に達するまで、別のステータスエンドポイントをポーリングします。

Q: 生成された動画 URL はどのくらいの間有効ですか？

生成 API の動画 URL は一時的です。URL を取得したらすぐにファイルをダウンロードし、（S3、Cloudflare R2 など）自分のストレージに保存してください。URL を保存して数時間後に動作することを期待しないでください。

Q: Veo 3 Fast と Kling Video の違いは何ですか？

Veo 3 Fast は安価（$0.05/sec）で高速かつ呼び出しが簡単です。Kling Video は negative_prompt、cfg_scale、カメラ動作設定、pro 品質モードなど、より細かな制御が可能です。出力を細かく調整したいなら Kling、速度と低コストを重視するなら Veo 3 Fast を使いましょう。

Q: テキストプロンプトではなく画像から動画を生成できますか？

はい。Veo は input_reference ファイルを渡すことで画像→動画に対応しています。Kling は image パラメータ（URL または base64）を使う /kling/v1/videos/image2video エンドポイントで対応。Runway は画像→動画のみで、CometAPI 経由ではテキストのみのプロンプトは受け付けません。

Q: なぜ Runway は最初のポーリングで task_not_exist を返すのですか？

これは CometAPI の仕様で、バックエンドでタスクがまだ初期化中であることを示します。数秒待って再試行してください。エラーではありません。上のポーリングコードはこれを自動で処理します。

Q: なぜ Kling は "succeed" を使い、"succeeded" ではないのですか？

Kling の実際の API レスポンス形式だからです。誤記ではありません。Veo と Runway は "succeeded" を使いますが、Kling は "succeed" を使います。統一ポーリングラッパーを作るなら、両方の文字列を扱う必要があります。

Q: 同期ポーリングループを Web サーバで使っても安全ですか？

いいえ。本ガイドのポーリングループは数分間スレッドをブロックします。多数のユーザーがいる実サービスでは、ポーリングはバックグラウンドワーカー（Python なら Celery、Node.js なら Bull）で実行してください。タスクの送信はリクエストハンドラで行い、タスク ID をクライアントに返し、動画の準備ができたらワーカーがクライアントに通知します。