> ## Documentation Index
> Fetch the complete documentation index at: https://docs.vidgen.id/llms.txt
> Use this file to discover all available pages before exploring further.

# Python

> A Vigen API client using requests: create a job, upload a reference, poll for the result.

Requires Python 3.9+ and `requests`.

```bash theme={null} theme={null}
pip install requests
```

```python vigen.py theme={null} theme={null}
import os
import time
from typing import Any

import requests

BASE = "https://vigen.id/api/v1"


class VigenError(RuntimeError):
    def __init__(self, status: int, body: dict[str, Any] | None):
        self.status = status
        self.body = body or {}
        super().__init__(self.body.get("error") or f"HTTP {status}")


class Vigen:
    def __init__(self, api_key: str | None = None):
        self.session = requests.Session()
        self.session.headers["Authorization"] = f"Bearer {api_key or os.environ['VIGEN_API_KEY']}"

    def _request(self, method: str, path: str, **kwargs: Any) -> dict[str, Any]:
        res = self.session.request(method, f"{BASE}{path}", timeout=kwargs.pop("timeout", 120), **kwargs)
        try:
            payload = res.json()
        except ValueError:
            payload = None
        if not res.ok:
            raise VigenError(res.status_code, payload)
        return payload

    def generate(self, model: str, prompt: str, **kwargs: Any) -> dict[str, Any]:
        body = {"model": model, "prompt": prompt, **kwargs}
        return self._request("POST", "/generate", json=body)["job"]

    def job(self, job_id: str) -> dict[str, Any]:
        return self._request("GET", f"/jobs/{job_id}")["job"]

    def upload(self, path: str, model: str | None = None) -> dict[str, Any]:
        files = {"file": open(path, "rb")}
        data = {"model": model} if model else None
        try:
            return self._request("POST", "/upload", files=files, data=data)["asset"]
        finally:
            files["file"].close()

    def credits(self) -> dict[str, Any]:
        return self._request("GET", "/user/credits")

    def models(self) -> dict[str, Any]:
        return self._request("GET", "/models")["models"]

    def wait_for_job(
        self,
        job_id: str,
        interval: float = 5.0,
        cap: float = 30.0,
        timeout: float = 2 * 60 * 60,
    ) -> dict[str, Any]:
        deadline = time.monotonic() + timeout
        delay = interval
        while True:
            job = self.job(job_id)
            if job["status"] in {"succeeded", "failed", "cancelled"}:
                return job
            if time.monotonic() + delay > deadline:
                return job
            time.sleep(delay)
            delay = min(delay * 1.6, cap)
```

## Usage

```python theme={null} theme={null}
import uuid

import requests

from vigen import Vigen, VigenError

api = Vigen()

asset = api.upload("karakter.png", model="seedance-2-5-flash")

job = api.generate(
    "seedance-2-5-flash",
    prompt="karakter ini melambaikan tangan ke kamera, latar studio lembut",
    references=[asset],
    parameters={"duration": "5", "ratio": "16:9", "resolution": "720p"},
    idempotencyKey=f"batch-7-{uuid.uuid4().hex[:12]}",
)

hasil = api.wait_for_job(job["id"])

if hasil["status"] == "succeeded":
    with open(f"hasil-{hasil['id']}.mp4", "wb") as f:
        f.write(requests.get(hasil["result"]["url"], timeout=300).content)
```

> [!WARNING]
>
> The field name in the JSON body is `idempotencyKey` (camelCase), not `idempotency_key`. A typo means the server generates its own key and your retry is unprotected.

## Batch pattern

```python theme={null} theme={null}
def generate_batch(prompts: list[str], model: str) -> list[dict]:
    sisa = api.credits()["limits"]["dailyVideosRemaining"]
    aktif = api.credits()["limits"]["currentActiveJobs"]
    jatah = min(sisa, max(0, api.credits()["limits"]["maxConcurrentJobs"] - aktif))

    jobs = []
    for prompt in prompts[:jatah]:
        try:
            jobs.append(api.generate(model, prompt=prompt, parameters={"duration": "5"}))
        except VigenError as err:
            if err.status in (402, 409):
                break
            if err.status == 429:
                time.sleep(30)
            else:
                raise
    return jobs
```

> [!TIP]
>
> For async services (FastAPI, Django), call `wait_for_job` in a separate worker, not in the request handler — two hours of polling would exhaust your connections.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.