Cara Menambahkan Pembuatan Video Berbasis AI ke Aplikasi SaaS

Menambahkan pembuatan video ke aplikasi Anda tidak sama dengan menambahkan pembuatan gambar. Panggilan API akan langsung mengembalikan respons — tetapi videonya belum siap. Anda mendapatkan sebuah task ID, dan Anda harus terus menanyakan “sudah selesai?” sampai memang selesai.

Kebanyakan developer mengalami ini saat pertama kali memanggil video API: menunggu body respons berisi URL video, tetapi yang didapat justru task ID. Panduan ini menjelaskan alur lengkapnya: mengirim tugas, melakukan polling untuk hasil, menangani kegagalan, dan menyimpan output sebelum URL kedaluwarsa.

Apa yang akan Anda bangun

Sebuah layanan backend yang menerima prompt teks atau gambar, mengirim tugas pembuatan video, melakukan polling hingga selesai, lalu mengembalikan URL video final. Anda akan bekerja dengan empat model — Veo 3 Fast, Sora 2, Kling Video, dan Runway — semuanya memakai satu API key.

Prasyarat:

• Python 3.8+ atau Node.js 18+
• Kunci CometAPI
• Pemahaman dasar REST API

Pahami mengapa pembuatan video berbeda

Dengan pembuatan gambar, Anda mengirim permintaan dan mendapatkan gambar pada respons yang sama. Pembuatan video memakai antrean tugas async:

1. Submit permintaan pembuatan → mendapat task_id
2. Poll endpoint status setiap beberapa detik
3. Saat status mencapai kondisi terminal, Anda mendapatkan URL video
4. Unduh dan simpan video — URL bersifat sementara

Jika Anda memperlakukan pembuatan video seperti pembuatan gambar dan menunggu respons pertama berisi video, permintaan Anda akan selalu time out.

Di layanan web produksi, loop polling ini sebaiknya berjalan di background worker (Celery, Bull, atau sejenisnya), bukan di request handler Anda. Contoh di bawah menggunakan polling sinkron — cocok untuk skrip dan prototipe, tetapi tidak untuk menangani pengguna secara bersamaan.

Pilih model

Source*: Halaman model CometAPI, Mei 2026. Catatan: "Sora 2" adalah* identifier model CometAPI — lihat halaman model mereka untuk detail model yang mendasarinya.

• Veo 3 Fast mendukung text-to-video dan image-to-video. Termurah per detik, titik awal yang baik.
• Sora 2 menghasilkan audio secara native bersama video — dialog, suara ambient, dan efek tanpa langkah TTS terpisah.
• Kling Video memberi Anda negative_prompt, cfg_scale, pengaturan pergerakan kamera, dan mode pro. Kontrol paling lengkap di antara keempatnya.
• Runway hanya image-to-video via CometAPI. Beri gambar statis dan deskripsi gerakan, dan itu akan menganimasikannya.

Kirim tugas Veo

Veo menggunakan multipart/form-data. Gunakan files= pada Python requests agar terkirim dengan benar — data=dict mengirim application/x-www-form-urlencoded, yang berbeda:

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}")

Lakukan polling untuk hasil

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}")

Gunakan Kling Video untuk kontrol lebih lanjut

Kling memiliki struktur endpoint berbeda dan menggunakan JSON. Perhatikan bahwa string status terminal Kling adalah "succeed" (bukan "succeeded") — ini sesuai format respons API-nya:

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

Source*:* CometAPI Kling Video docs

Animasikan gambar statis dengan Runway

Runway hanya image-to-video. Ia juga memerlukan header tambahan (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")

Source*:* CometAPI Runway docs

Simpan video sebelum URL kedaluwarsa

URL video dari API pembuatan bersifat sementara. Segera unduh file dan simpan di tempat yang Anda kendalikan:

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

Di lingkungan produksi, ganti penulisan file lokal dengan unggah ke S3, Cloudflare R2, atau penyimpanan pilihan Anda. Pola streaming tetap sama — alirkan byte secara langsung alih-alih memuat seluruh video ke memori.

Tangani kegagalan

Sumber: CometAPI video generation docs

Versi Node.js

Node.js 18+ sudah menyertakan fetch dan FormData secara native. Contoh ini mencakup keempat model:

// 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);

Langkah berikutnya

Anda kini memiliki kode yang berfungsi untuk empat model video, loop polling yang menangani kegagalan, dan langkah pengunduhan yang mencegah hilangnya konten yang dihasilkan.

Masalah berikut yang paling sering ditemui developer: mereka menghardcode satu model, dan beralih ke opsi yang lebih murah atau lebih cepat berarti menyentuh banyak file. Artikel berikutnya membahas cara merutekan permintaan lintas model tanpa menulis ulang kode Anda.

Berikutnya: Cara Berpindah Antar Model AI Tanpa Menulis Ulang Kode Anda

FAQ

T: Mengapa saya mendapatkan task ID alih-alih video pada respons API?

Pembuatan video bersifat async — model seperti Veo, Sora, Kling, dan Runway memerlukan 2–5 menit untuk merender. API segera mengembalikan task ID agar permintaan Anda tidak time out. Anda melakukan polling ke endpoint status terpisah sampai tugas mencapai status terminal (succeeded, succeed, failed).

T: Berapa lama URL video yang dihasilkan tetap valid?

URL video dari API pembuatan bersifat sementara. Segera unduh file setelah mendapatkan URL dan simpan di penyimpanan Anda sendiri (S3, Cloudflare R2, dll.). Jangan menyimpan URL dan berharap masih berfungsi beberapa jam kemudian.

T: Apa perbedaan Veo 3 Fast dan Kling Video?

Veo 3 Fast lebih murah ($0.05/detik), lebih cepat, dan lebih sederhana dipanggil. Kling Video memberi Anda lebih banyak kontrol: negative_prompt, cfg_scale, pengaturan pergerakan kamera, dan mode kualitas pro. Jika Anda perlu menyetel keluaran secara detail, gunakan Kling. Jika Anda perlu kecepatan dan biaya rendah, gunakan Veo 3 Fast.

T: Bisakah saya membuat video dari gambar alih-alih prompt teks?

Ya. Veo mendukung image-to-video dengan melewatkan file input_reference. Kling mendukungnya melalui endpoint /kling/v1/videos/image2video dengan parameter image (URL atau base64). Runway hanya image-to-video — tidak menerima prompt teks saja melalui CometAPI.

T: Mengapa Runway mengembalikan task_not_exist pada polling pertama?

Ini adalah perilaku CometAPI yang terdokumentasi — tugas masih dalam tahap inisialisasi di backend. Tunggu beberapa detik dan coba lagi. Ini bukan error. Kode polling di atas menanganinya secara otomatis.

T: Mengapa Kling menggunakan "succeed" alih-alih "succeeded"?

Itu adalah format respons API Kling yang sebenarnya. Bukan typo. Veo dan Runway menggunakan "succeeded" — Kling menggunakan "succeed". Jika Anda membuat wrapper polling terpadu, Anda perlu menangani kedua string tersebut.

T: Apakah loop polling sinkron aman digunakan di web server?

Tidak. Loop polling dalam panduan ini memblokir thread selama beberapa menit. Pada layanan web nyata dengan pengguna bersamaan, jalankan polling di background worker (Celery untuk Python, Bull untuk Node.js). Kirim tugas di request handler, kembalikan task ID ke klien, dan biarkan worker memberi tahu klien saat video siap.