Seedance API in Python and Node.js: a complete tutorial

HTTP/1.1 202 Accepted { "success": true, "data": { "job": { "id": "3f1c0a9e-...", "mediaType": "video", "modelId": "kreator-video-seedance-2", "status": "queued", ... }, "balance_tokens": 389, "quoted_token_cost": 11, "estimated_token_cost": 11 } }

# kreatorflow_video.py: create a Seedance video, wait for it, save the MP4. # pip install requests (Python 3.9+) import os import sys import time import uuid import requests BASE = "https://kreatorflow.ai/v1" KEY = os.environ["KREATORFLOW_API_KEY"] AUTH = {"Authorization": f"Bearer {KEY}"} FINAL = {"completed", "failed", "canceled", "blocked"} class KreatorFlowError(Exception): def __init__(self, status, code, message): super().__init__(f"{status} {code}: {message}") self.status, self.code = status, code def _json(res): body = res.json() if res.status_code >= 400 or not body.get("success"): raise KreatorFlowError(res.status_code, body.get("code"), body.get("error")) return body["data"] def create_video(prompt, *, model="kreator-video-seedance-2", duration=5, aspect_ratio="16:9", resolution="720p", webhook_url=None, **extra): # extra: any other POST /v1/videos field, e.g. first_frame_url for image to video. payload = {"model": model, "prompt": prompt, "duration": duration, "aspect_ratio": aspect_ratio, "resolution": resolution, **extra} if webhook_url: payload["webhook_url"] = webhook_url # One Idempotency-Key per clip: resending the same request never creates a second job. headers = {**AUTH, "Idempotency-Key": str(uuid.uuid4())} data = _json(requests.post(f"{BASE}/videos", json=payload, headers=headers, timeout=60)) print(f"job {data['job']['id']} quoted at {data['quoted_token_cost']} tokens") return data["job"] def get_job(job_id): res = requests.get(f"{BASE}/jobs/{job_id}", headers=AUTH, timeout=30) if res.status_code == 429: time.sleep(int(res.headers.get("Retry-After", "5"))) return get_job(job_id) return _json(res)["job"] def wait_for_job(job, every=10, timeout=1800): deadline = time.time() + timeout while job["status"] not in FINAL: if time.time() > deadline: raise TimeoutError(f"job {job['id']} still {job['status']}") time.sleep(every) job = get_job(job["id"]) return job def download(job_id, path): with requests.get(f"{BASE}/jobs/{job_id}/output", headers=AUTH, stream=True, timeout=120) as res: res.raise_for_status() with open(path, "wb") as out: for chunk in res.iter_content(chunk_size=1 << 20): out.write(chunk) return path if __name__ == "__main__": prompt = " ".join(sys.argv[1:]) or "A paper boat drifting down a rainy street at dusk, low angle" try: job = wait_for_job(create_video(prompt)) except KreatorFlowError as err: sys.exit(f"request failed: {err}") if job["status"] == "completed": print("saved", download(job["id"], f"{job['id']}.mp4")) else: sys.exit(f"{job['status']}: {job.get('errorCode')} {job.get('errorMessage')}")

// kreatorflow-video.mjs: create a Seedance video, wait for it, save the MP4. // Node 18+ (fetch is built in). Run: node kreatorflow-video.mjs "your prompt" import { createWriteStream } from "node:fs"; import { Readable } from "node:stream"; import { pipeline } from "node:stream/promises"; const BASE = "https://kreatorflow.ai/v1"; const AUTH = { Authorization: `Bearer ${process.env.KREATORFLOW_API_KEY}` }; const FINAL = new Set(["completed", "failed", "canceled", "blocked"]); const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); export class KreatorFlowError extends Error { constructor(status, code, message) { super(`${status} ${code}: ${message}`); this.status = status; this.code = code; } } async function json(res) { const body = await res.json(); if (!res.ok || !body.success) throw new KreatorFlowError(res.status, body.code, body.error); return body.data; } // extra: any other POST /v1/videos field, e.g. { first_frame_url } for image to video. export async function createVideo(prompt, { model = "kreator-video-seedance-2", duration = 5, aspectRatio = "16:9", resolution = "720p", webhookUrl, extra = {} } = {}) { const res = await fetch(`${BASE}/videos`, { method: "POST", // One Idempotency-Key per clip: resending the same request never creates a second job. headers: { ...AUTH, "Content-Type": "application/json", "Idempotency-Key": crypto.randomUUID() }, body: JSON.stringify({ model, prompt, duration, aspect_ratio: aspectRatio, resolution, ...extra, ...(webhookUrl ? { webhook_url: webhookUrl } : {}) }), }); const data = await json(res); console.log(`job ${data.job.id} quoted at ${data.quoted_token_cost} tokens`); return data.job; } export async function getJob(id) { const res = await fetch(`${BASE}/jobs/${id}`, { headers: AUTH }); if (res.status === 429) { await sleep(Number(res.headers.get("Retry-After") ?? 5) * 1000); return getJob(id); } return (await json(res)).job; } export async function waitForJob(job, { every = 10_000, timeout = 30 * 60_000 } = {}) { const deadline = Date.now() + timeout; while (!FINAL.has(job.status)) { if (Date.now() > deadline) throw new Error(`job ${job.id} still ${job.status}`); await sleep(every); job = await getJob(job.id); } return job; } export async function download(id, path) { const res = await fetch(`${BASE}/jobs/${id}/output`, { headers: AUTH }); if (!res.ok) throw new KreatorFlowError(res.status, "download_failed", await res.text()); await pipeline(Readable.fromWeb(res.body), createWriteStream(path)); return path; } const prompt = process.argv.slice(2).join(" ") || "A paper boat drifting down a rainy street at dusk, low angle"; try { const job = await waitForJob(await createVideo(prompt)); if (job.status === "completed") console.log("saved", await download(job.id, `${job.id}.mp4`)); else console.error(job.status, job.errorCode, job.errorMessage); } catch (err) { console.error("request failed:", err.message); process.exitCode = 1; }

# Python: animate a picture (same client, one extra field) from kreatorflow_video import create_video, download, wait_for_job job = wait_for_job(create_video( "The camera slowly pushes in. Waves roll against the rocks and the lighthouse beam sweeps the sea.", aspect_ratio="adaptive", first_frame_url="https://example.com/lighthouse.jpg", # a public HTTPS image )) if job["status"] == "completed": download(job["id"], "lighthouse.mp4") // Node.js: the same job with the ES module client import { createVideo, download, waitForJob } from "./kreatorflow-video.mjs"; const job = await waitForJob(await createVideo("The camera slowly pushes in. Waves roll against the rocks.", { aspectRatio: "adaptive", extra: { first_frame_url: "https://example.com/lighthouse.jpg" }, })); if (job.status === "completed") await download(job.id, "lighthouse.mp4");

# batch.py: submit several prompts at once, then collect every clip. from concurrent.futures import ThreadPoolExecutor from kreatorflow_video import create_video, download, wait_for_job prompts = [ "A red fox trotting through fresh snow at sunrise, side tracking shot", "Steam rising from a bowl of ramen, slow push in, warm light", "A skateboarder kickflips over a puddle at golden hour, low angle", ] jobs = [create_video(p, model="kreator-video-seedance-2-fast") for p in prompts] # submit all first with ThreadPoolExecutor(max_workers=len(jobs)) as pool: for job in pool.map(wait_for_job, jobs): if job["status"] == "completed": print("saved", download(job["id"], f"{job['id']}.mp4")) else: print(job["id"], job["status"], job.get("errorCode"))

# webhook.py: receive the signed "job finished" call instead of polling. # pip install flask Run: flask --app webhook run --port 8000 import hashlib import hmac import os import time from flask import Flask, abort, request app = Flask(__name__) SECRET = os.environ["KREATORFLOW_WEBHOOK_SECRET"].encode() seen = set() # use your database in production @app.post("/kreatorflow/webhook") def kreatorflow_webhook(): raw = request.get_data(as_text=True) sig = request.headers.get("X-KreatorFlow-Signature", "") parts = dict(p.split("=", 1) for p in sig.split(",") if "=" in p) expected = hmac.new(SECRET, f'{parts.get("t", "")}.{raw}'.encode(), hashlib.sha256).hexdigest() fresh = abs(time.time() - int(parts.get("t", "0"))) < 300 if not (fresh and hmac.compare_digest(expected, parts.get("v1", ""))): abort(401) event = request.get_json() if event["id"] not in seen: seen.add(event["id"]) job = event["data"]["job"] print(event["type"], job["id"], job["status"]) # queue the download here return "ok", 200

// webhook.mjs: receive the signed "job finished" call (no framework needed). import { createHmac, timingSafeEqual } from "node:crypto"; import { createServer } from "node:http"; const SECRET = process.env.KREATORFLOW_WEBHOOK_SECRET; const seen = new Set(); // use your database in production function verify(raw, header = "") { const parts = Object.fromEntries(header.split(",").map((part) => part.split("="))); const expected = createHmac("sha256", SECRET).update(`${parts.t}.${raw}`).digest("hex"); const fresh = Math.abs(Date.now() / 1000 - Number(parts.t)) < 300; return fresh && typeof parts.v1 === "string" && parts.v1.length === expected.length && timingSafeEqual(Buffer.from(parts.v1), Buffer.from(expected)); } createServer(async (req, res) => { if (req.method !== "POST" || req.url !== "/kreatorflow/webhook") return res.writeHead(404).end(); let raw = ""; for await (const chunk of req) raw += chunk; if (!verify(raw, req.headers["x-kreatorflow-signature"])) return res.writeHead(401).end("bad signature"); const event = JSON.parse(raw); if (!seen.has(event.id)) { seen.add(event.id); console.log(event.type, event.data.job.id, event.data.job.status); // queue the download here } res.writeHead(200).end("ok"); }).listen(8000);