Call the Varosity API from Python

Call the Varosity API from Python

There is no Varosity Python package to install — the API is plain HTTPS + JSON, so the requests library is all an agent needs. This guide shows the whole loop: submit → poll → download.

Looking for the old "Agent Video SDK" page? The varosity_sdk package it described was never published. Everything it promised is covered by the REST calls below (or by the MCP server and the CLI).

Before you start

1. Issue an API key (vsk_…) at https://varosity.ai/app/keys/api-keys. Give it the generate:video scope (add generate:image, generate:voice, generate:music if you need them). 2. Export it — never hard-code it:

export VAROSITY_API_KEY="vsk_..."
pip install requests

Every request goes to https://varosity.ai with an Authorization: Bearer <key> header.

The three calls

The submit response's status is normally rendering (a fast model may already report succeeded). When polling, status is one of queued, rendering, succeeded, failed, or canceled. outputUrl is null until the job has succeeded.

Held renders. On Varosity Credits, a render that finishes when your balance can't cover it is held: the poll keeps returning rendering (or queued) with code: "insufficient_credits" and an error message naming the amount. Waiting won't release it. Add credits at /app/credits, then poll again: the next poll charges the render and returns succeeded with the outputUrl. A held render is kept for 24 hours. The poll loop below stops and reports this instead of waiting.

Request body for POST /api/v1/video/generate

Complete example

import os
import time

import requests

BASE = "https://varosity.ai" HEADERS = { "Authorization": f"Bearer {os.environ['VAROSITY_API_KEY']}", "Content-Type": "application/json", }

def submit_video(prompt, model_id="kling-3.0", aspect_ratio="16:9", duration_sec=5, reference_image_url=None): """Submit a render. Returns the response JSON (contains jobId).""" body = { "prompt": prompt, "modelId": model_id, "aspectRatio": aspect_ratio, "durationSec": duration_sec, } if reference_image_url: body["referenceImageUrl"] = reference_image_url

res = requests.post(f"{BASE}/api/v1/video/generate", headers=HEADERS, json=body, timeout=90) data = res.json() if not res.ok or not data.get("ok"): raise RuntimeError( f"submit failed ({res.status_code}): " f"{data.get('code')} — {data.get('error')}" ) return data

def wait_for_job(job_id, interval_sec=5, timeout_sec=900): """Poll until the job succeeds or fails. Returns the final job JSON.""" deadline = time.time() + timeout_sec while time.time() < deadline: res = requests.get(f"{BASE}/api/v1/jobs/{job_id}", headers=HEADERS, timeout=60) job = res.json() if not res.ok or not job.get("ok"): raise RuntimeError(f"poll failed ({res.status_code}): {job.get('error')}") if job["status"] == "succeeded": return job if job["status"] in ("failed", "canceled"): raise RuntimeError(f"job {job['status']}: {job.get('error')}") if job.get("code") == "insufficient_credits": # Finished but held until paid: top up, then poll again. raise RuntimeError(f"job {job_id} held: {job.get('error')}") time.sleep(interval_sec) raise TimeoutError(f"job {job_id} still running after {timeout_sec}s")

def download(url, path): with requests.get(url, stream=True, timeout=120) as res: res.raise_for_status() with open(path, "wb") as fh: for chunk in res.iter_content(chunk_size=1 << 16): fh.write(chunk) return path

if __name__ == "__main__": submitted = submit_video( "Wide cinematic shot of a coastal cliff at golden hour, slow drone push-in", model_id="kling-3.0", duration_sec=5, ) print("jobId:", submitted["jobId"])

Some models finish inside the submit call; most return "rendering".

job = submitted if submitted["status"] == "succeeded" else wait_for_job(submitted["jobId"]) print("outputUrl:", job["outputUrl"]) download(job["outputUrl"], "shot.mp4") ```

Poll every 3–5 seconds. Most clips land in 30–120 seconds; long-form avatar models can take several minutes.

Several shots at once

Each POST /api/v1/video/generate call is independent, so submit every shot first and then poll them together:

shots = [
    ("Wide establishing shot of a mountain lodge at dawn", "kling-3.0"),
    ("Close-up of hands pouring coffee, steam rising", "veo-3.1"),
    ("Skier carving through powder, tracking shot", "seedance-1.0-pro"),
]
job_ids = [submit_video(prompt, model_id=m)["jobId"] for prompt, m in shots]
urls = [wait_for_job(job_id)["outputUrl"] for job_id in job_ids]

Each REST call creates its own single-shot project, so join the clips yourself (for example with ffmpeg), or build the shots inside one project through the MCP server (create_project → generate_video with project_id → render_project) to get a server-side stitch.

Errors worth handling

Failures come back as { "ok": false, "error": "…", "code": "…" }.

The daily cap is set per API key at /app/keys/api-keys. Each call reserves its estimated cost. Media-driven models (OmniHuman, lip-sync, LivePortrait, Wan Animate, HeyGen, D-ID) bill the real length of the supplied audio or video, and so do transcription, dubbing and the voice changer; on Varosity Credits that real charge is added to the key's day when the job finishes — so the next call can be refused even though this one was accepted.

To make a retry safe, send an Idempotency-Key header (any unique string per logical request) on the POST — a repeat with the same key returns the original response instead of starting a second render. On POST /v1/video/generate and /v1/images a repeat sent while the first is still running gets 409 idempotency_in_flight (reusing a key on a different endpoint gets 409 idempotency_key_reused on any route that reads the header); on /v1/sfx, /tts, /music, /v1/audio/assemble, /v1/projects/{projectId}/stitch and the run routes a repeat returns the stored response only once the first has finished. Other POSTs ignore the header, so check before retrying a voice clone, transcription or dub.

Images, voice, and music

The same key and the same pattern work for the synchronous endpoints — no polling, the URL is in the response:

def post(path, body):
    res = requests.post(f"{BASE}{path}", headers=HEADERS, json=body, timeout=300)
    data = res.json()
    if not res.ok or not data.get("ok"):
        raise RuntimeError(f"{path} failed ({res.status_code}): {data.get('error')}")
    return data

image = post("/api/v1/images", { "prompt": "Product hero shot of a matte black espresso machine, studio light", "modelId": "flux-1.1-pro", "aspectRatio": "16:9", })["imageUrl"]

voice = post("/api/v1/tts", { "text": "Welcome back. Here's what's new this week.", "voiceId": "21m00Tcm4TlvDq8ikWAM", })["audioUrl"]

music = post("/api/v1/music", { "modelId": "elevenlabs-music", "prompt": "cinematic synth, slow build, hopeful, instrumental", "durationSec": 30, })["audioUrl"] ```

durationSec for music is 5–300 (each model has its own maximum; Lyria 2 always makes one 30-second clip). Some music models run asynchronously and return { trackId, providerJobId, sync: false } with no audioUrl yet — poll POST /api/music/poll with { "trackId": "…" } until it returns status: "succeeded" and an audioUrl. elevenlabs-music returns the URL directly.

See also

• Agent mode (MCP) — the same capabilities as MCP tools
• Using the CLI — varosity generate video … from a shell
• Picking the right model
• Provider options — per-model providerOptions
• OpenAPI spec