앱에 동영상 생성을 추가하는 것은 이미지 생성을 추가하는 것과는 다릅니다. API 호출은 즉시 반환되지만 — 동영상은 아직 준비되지 않습니다. 작업 ID를 받고, 완료될 때까지 “다 됐나요?”라고 계속 확인해야 합니다.
대부분의 개발자는 처음으로 비디오 API를 호출할 때 응답 본문에서 비디오 URL을 기대하지만, 대신 작업 ID를 받게 됩니다. 이 가이드는 전체 플로우를 안내합니다: 작업 제출, 결과 폴링, 실패 처리, 그리고 URL이 만료되기 전에 출력 저장까지.
이번에 구축할 것
텍스트 프롬프트 또는 이미지를 받아 비디오 생성 작업을 제출하고, 완료될 때까지 폴링한 후 최종 비디오 URL을 반환하는 백엔드 서비스. 하나의 API 키로 Veo 3 Fast, Sora 2, Kling Video, Runway — 네 가지 모델을 다룹니다.
사전 준비 사항:
- Python 3.8+ 또는 Node.js 18+
- CometAPI 키
- REST API에 대한 기본 이해
동영상 생성이 다른 이유 이해하기
이미지 생성은 요청을 보내면 같은 응답에서 이미지를 바로 받습니다. 동영상 생성은 비동기 작업 큐를 사용합니다:
- Submit 생성 요청을 보냄 →
task_id를 받음 - 몇 초 간격으로 상태 엔드포인트를 Poll
- 상태가 종료 상태에 도달하면 비디오 URL을 받음
- 비디오를 Download and store — URL은 일시적임
동영상 생성을 이미지 생성처럼 취급하고, 첫 응답에 비디오가 포함되길 기다리면 매번 요청이 타임아웃됩니다.
프로덕션 웹 서비스에서는 이 폴링 루프를 요청 처리기에서 돌리지 말고 백그라운드 워커(Celery, Bull 등)에서 실행해야 합니다. 아래 예제는 동기 폴링을 사용합니다 — 스크립트나 프로토타입에는 괜찮지만, 동시 사용자를 처리하기에는 적합하지 않습니다.
모델 선택
| Model | Provider | Max duration | Price (via CometAPI) | Best for |
|---|---|---|---|---|
| Veo 3 Fast | 8 sec | $0.05/sec | 빠른 프로토타이핑, 소셜 클립 | |
| Sora 2 | OpenAI (via CometAPI model ID) | ~10 sec | $0.08/sec | 고품질 크리에이티브 쇼츠 |
| Kling Video | Kuaishou | 10 sec | $0.13–$2.64/task | 마케팅 콘텐츠, 세밀한 제어 |
| Runway Gen-3A Turbo | Runway | 5 or 10 sec | $0.32/task | 이미지 투 비디오, 커머셜 콘텐츠 |
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는 텍스트 투 비디오와 이미지 투 비디오를 모두 지원합니다. 초당 비용이 가장 낮아 시작점으로 적합합니다.
- Sora 2는 비디오와 함께 오디오를 네이티브로 생성합니다 — 별도의 TTS 없이 대사, 앰비언트 사운드, 효과음을 포함합니다.
- Kling Video는
negative_prompt,cfg_scale, 카메라 무브먼트 설정,pro모드를 제공합니다. 네 가지 중 제어가 가장 세밀합니다. - Runway는 CometAPI를 통한 이미지 투 비디오 전용입니다. 정지 이미지를 주고 모션 설명을 추가하면 이를 애니메이션화합니다.
Veo 작업 제출
Veo는 multipart/form-data를 사용합니다. Python requests에서는 올바르게 전송하려면 files=를 사용하세요 — data=dict는 application/x-www-form-urlencoded로 전송되며, 이는 동일하지 않습니다:
import requestsimport osfrom dotenv import load_dotenvload_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 timedef 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")
Source*:* 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")
Source*:* CometAPI Runway docs
URL이 만료되기 전에 비디오 저장
생성 API의 비디오 URL은 일시적입니다. 즉시 파일을 다운로드하고, 제어 가능한 저장소에 보관하세요:
import requestsimport pathlibdef 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 등 원하는 스토리지로 업로드하세요. 스트리밍 패턴은 동일합니다 — 전체 비디오를 메모리에 올리지 말고 바이트 스트림을 바로 파이프하세요.
실패 처리
| Symptom | Likely cause | Fix |
|---|---|---|
| Task stuck in queued for 10+ min | 서버 부하 또는 모델 사용 불가 | 다른 모델로 재시도 |
| task_not_exist on first Runway poll | 작업이 아직 초기화 중 | 5초 대기 후 재시도 — CometAPI의 문서화된 동작 |
| failed with no error message | 프롬프트가 콘텐츠 필터에 걸림 | 프롬프트를 바꿔 표현 |
| Video URL returns 403 | 다운로드 전에 URL이 만료됨 | URL을 받은 직후 바로 다운로드 |
| Timeout after 10 min | 생성에 너무 오래 걸림 | max_wait 증가 또는 Veo 3 Fast로 전환 |
| Kling returns "succeed" not "succeeded" | Kling API의 비표준 상태 문자열 | 정상입니다 — 위의 Kling 폴링 코드를 참고하세요 |
Source: CometAPI video generation docs
Node.js 버전
Node.js 18+에는 fetch와 FormData가 기본 포함됩니다. 이 예제는 네 가지 모델을 모두 다룹니다:
// Node.js 18+ — no extra packages neededconst 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);
다음 단계
네 가지 비디오 모델에 대한 동작하는 코드, 실패를 처리하는 폴링 루프, 그리고 생성된 콘텐츠를 잃지 않도록 하는 다운로드 단계까지 준비되었습니다.
대부분의 개발자가 다음에 겪는 문제는 하나의 모델을 하드코딩해 두었기 때문에 더 저렴하거나 빠른 옵션으로 전환하려면 여러 파일을 수정해야 한다는 점입니다. 다음 글에서는 코드를 다시 쓰지 않고 모델 간에 요청을 라우팅하는 방법을 다룹니다.
Next: 코드를 다시 쓰지 않고 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/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: 동기 폴링 루프를 웹 서버에서 사용해도 안전한가요?
아니요. 이 가이드의 폴링 루프는 수 분 동안 스레드를 블로킹합니다. 동시 사용자가 있는 실제 웹 서비스에서는 폴링을 백그라운드 워커(Celery for Python, Bull for Node.js)에서 실행하세요. 요청 처리기에서 작업을 제출하고, 클라이언트에게 작업 ID를 반환한 뒤, 워커가 비디오 준비 완료 시 클라이언트에 알리도록 하세요.