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 など）で走らせるべきです。以下の例は同期ポーリングを使っています—スクリプトや試作には十分ですが、多数の同時ユーザーを扱うには不向きです。

モデルを選ぶ

出典**: CometAPI のモデルページ（2026年5月）。注: 「Sora 2」は CometAPI のモデルの識別子*です—基盤モデルの詳細は彼らのモデルページを参照してください。

• Veo 3 Fast はテキストから動画と画像から動画の両方に対応。秒単価が最安で、最初の選択肢として適しています。
• 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:    """Submit a Veo 3 Fast text-to-video task. Returns task_id."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY environment variable is not set")​    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("A paper kite drifting above a wheat field on a windy afternoon")print(f"Task submitted: {task_id}")

結果をポーリングする

import time​def poll_veo_task(task_id: str, interval: int = 10, max_wait: int = 600) -> str:    """Poll until Veo task completes. Returns video URL."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY environment variable is not set")​    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 {task_id} failed with status '{status}': "                f"{result.get('error', 'no error detail returned')}"            )​        time.sleep(interval)        elapsed += interval​    raise TimeoutError(f"Task {task_id} did not complete within {max_wait} seconds")​​video_url = poll_veo_task(task_id)print(f"Video ready: {video_url}")

より細かく制御したい場合は Kling Video を使う

Kling は異なるエンドポイント構造を持ち、JSON を使用します。Kling の終端ステータスは "succeed"（"succeeded" ではない）である点に注意してください—これは API の実際のレスポンス仕様に一致します:

def submit_kling_task(prompt: str, duration: str = "5", mode: str = "std") -> str:    """Submit a Kling text-to-video task. Returns task_id."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY environment variable is not set")​    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" or "pro"            "aspect_ratio": "16:9",            "duration": duration  # "5" or "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:    """Poll Kling task until complete. Returns video URL."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY environment variable is not set")​    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 uses "succeed", not "succeeded"            return result["data"]["task_result"]["videos"][0]["url"]        elif status == "failed":            error_detail = result.get("data", {}).get("task_result", "no detail")            raise RuntimeError(                f"Kling task {task_id} failed: {error_detail}"            )​        time.sleep(interval)        elapsed += interval​    raise TimeoutError(f"Kling task {task_id} timed out after {max_wait}s")

出典**: CometAPI Kling Video docs

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

Runway は画像から動画のみに対応します。さらに追加ヘッダー（X-Runway-Version）が必要です:

def submit_runway_task(image_url: str, motion_prompt: str, duration: int = 5) -> str:    """Submit a Runway image-to-video task. Returns task_id."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY environment variable is not set")​    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,  # must be a stable 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:    """Poll Runway task. Returns video URL when done."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY environment variable is not set")​    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-specific: task is still initializing, retry after a few seconds            time.sleep(interval)            elapsed += interval            continue        elif status == "succeeded":            return result["output"][0]        elif status in ("failed", "cancelled"):            raise RuntimeError(f"Runway task {task_id} failed: {result.get('error', 'no detail')}")​        time.sleep(interval)        elapsed += interval​    raise TimeoutError(f"Runway task {task_id} timed out after {max_wait}s")

出典**: CometAPI Runway docs

URL が失効する前に動画を保存する

生成 API からの動画 URL は一時的です。すぐにファイルをダウンロードし、あなたが管理する場所に保存しましょう:

import requestsimport pathlib​def download_video(url: str, output_path: str) -> None:    """Download video from URL to local file using streaming."""    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"Saved to {output_path}")​​# Full flowtask_id = submit_veo_task("A timelapse of clouds moving over a city skyline")video_url = poll_veo_task(task_id)download_video(video_url, "output/city_timelapse.mp4")

本番では、ローカルへの書き込みの代わりに S3、Cloudflare R2、または選択したストレージへアップロードしてください。ストリーミングのパターンは同じです—動画全体をメモリに読み込むのではなく、バイト列を直接パイプします。

失敗を処理する

出典: CometAPI video generation docs

Node.js 版

Node.js 18+ には fetch と FormData がネイティブに含まれます。次の例は 4 つすべてのモデルをカバーします:

// Node.js 18+ — no extra packages needed​const API_KEY = process.env.COMETAPI_KEY;if (!API_KEY) throw new Error('COMETAPI_KEY is not set');​// --- 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 submit failed: ${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(`Poll failed: ${res.status}`);    const result = await res.json();​    if (result.status === 'succeeded') return result.output[0];    if (['failed', 'cancelled'].includes(result.status)) {      throw new Error(`Task ${taskId} failed: ${result.error ?? 'no detail'}`);    }    await new Promise(r => setTimeout(r, intervalMs));    elapsed += intervalMs;  }  throw new Error(`Task ${taskId} timed out`);}​// --- 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 submit failed: ${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 poll failed: ${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 task ${taskId} failed: ${JSON.stringify(result.data.task_result ?? 'no detail')}`);    }    await new Promise(r => setTimeout(r, intervalMs));    elapsed += intervalMs;  }  throw new Error(`Kling task ${taskId} timed out`);}​// --- Runway (image-to-video) ---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 submit failed: ${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 poll failed: ${res.status}`);    const result = await res.json();    const status = result.status;​    if (status === 'task_not_exist') {      // CometAPI-specific: task still initializing      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 task ${taskId} failed: ${result.error ?? 'no detail'}`);    }    await new Promise(r => setTimeout(r, intervalMs));    elapsed += intervalMs;  }  throw new Error(`Runway task ${taskId} timed out`);}​// Usage exampleconst taskId = await submitVeoTask('A paper kite drifting above a wheat field');const videoUrl = await pollVeoTask(taskId);console.log('Video ready:', videoUrl);

次のステップ

4 つの動画モデルを動かすコード、失敗に対処するポーリングループ、生成コンテンツを失わないためのダウンロード手順が揃いました。

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

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

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/秒）、高速で、呼び出しも簡単です。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 サーバーで使っても安全ですか？

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