# Python SDK

> Call the PicX Studio API from Python with the official picx-ai package, sync or async.

The official Python SDK wraps the PicX Studio REST API in a typed client: `picx.images.generate(...)`, `picx.video.create(...)`, `picx.models.list(...)`. It handles auth headers, JSON encoding, retries with backoff, timeouts, and video job polling, and it raises typed exceptions you can catch individually. It is fully type-hinted, and ships `AsyncPicX` for asyncio applications.

> [!NOTE]
> The SDK is not the CLI. The [PicX CLI](https://github.com/Type-Think-AI/picx-cli) is a separate tool you install globally and run in your terminal for one-off jobs and scripting. The SDK is a library you add as a dependency to call the API from inside your own application. They are independent — use either, or both.

## Install

```bash
pip install picx-ai
```

```bash
uv add picx-ai
```

## Quickstart

Pass your API key to the constructor, or leave it out and the SDK reads `PICX_API_KEY` from the environment.

```python
import os
from picx import PicX

picx = PicX(os.environ["PICX_API_KEY"])
# or simply PicX() — falls back to the PICX_API_KEY env var

asset = picx.images.generate(
    prompt="a beautiful sunset over the ocean, cinematic lighting",
    size="2K",
    aspect_ratio="16:9",
)

print(asset.url)
print(asset.id, asset.model, asset.credits_used)
```

> [!WARNING]
> Never use an API key in a browser or any client-side bundle. `pxsk_…` keys are server-side credentials that carry your billing and full key scopes; anything shipped to a browser is public. Keep the SDK on the server — a web backend, a worker, a notebook you control — and expose your own endpoint to the frontend. That includes not embedding a key in a Python app compiled to WebAssembly or shipped to end users. A key that reaches a browser must be treated as leaked and revoked.

## Generate an image

`images.generate` is synchronous — it returns the finished asset. Only `prompt` is required.

```python
asset = picx.images.generate(
    prompt="a red leather sneaker on white marble, studio lighting",
    model="gemini-3.1-flash-image-preview",
    size="2K",
    aspect_ratio="16:9",
)

print(asset.url)           # hosted CDN URL
print(asset.credits_used)  # credits billed for this call
```

| Argument | Type | Required | Description |
| --- | --- | --- | --- |
| `prompt` | `str` | Yes | What to generate. |
| `model` | `str` | No | Model ID. Defaults to the current image model. |
| `size` | `str` | No | Output size, for example `1K`, `2K`, `4K`. |
| `aspect_ratio` | `str` | No | Aspect ratio, for example `16:9`, `1:1`. |

## Edit an image

`images.edit` takes an instruction plus one to five source image URLs. It is also synchronous.

```python
edited = picx.images.edit(
    instruction="make the colors more vibrant and add a subtle glow effect",
    image_urls=["https://cdn.picxstudio.com/api/generated/image_abc123.png"],
    model="gemini-3.1-flash-image-preview",
    size="1K",
)

print(edited.url)
```

Pass several URLs to composite or use additional references:

```python
composite = picx.images.edit(
    instruction="place the product from the second image onto the background of the first",
    image_urls=[
        "https://example.com/background.jpg",
        "https://example.com/product.png",
    ],
)
```

> [!TIP]
> `image_urls` accepts 1–5 URLs. More than five raises `ValidationError` before the request is sent, so you fail fast instead of burning a round trip.

## Generate a video

Video generation is asynchronous. `video.create` returns a **job**, not an asset. Call `job.wait()` to poll until the video is ready and get the finished asset back.

```python
job = picx.video.create(
    prompt="a cat walking in the rain, cinematic",
    duration=5,
    resolution="720p",
    sound=True,
)

print(job.id, job.status)  # "pending"

asset = job.wait()         # polls with backoff until terminal
print(asset.url)
```

`job.wait()` returns when the generation completes and raises if it fails, so you do not need to inspect `status` yourself.

| Argument | Type | Required | Description |
| --- | --- | --- | --- |
| `prompt` | `str` | Yes | What to generate. |
| `model` | `str` | No | Video model ID. |
| `duration` | `int` | No | Length in seconds. |
| `resolution` | `str` | No | For example `720p`, `1080p`. |
| `sound` | `bool` | No | Generate an audio track. |

### Bind a webhook to a video job

Receive a notification when the job finishes instead of blocking:

```python
job = picx.video.create(
    prompt="a cat jumping off a table",
    duration=5,
    resolution="480p",
    webhook_url="https://your-server.com/webhook",
    webhook_events=["generation.completed"],
)
# job.raw["webhook"]["mode"] == "inline"
```

Or bind by a registered webhook ID:

```python
job = picx.video.create(
    prompt="a cat jumping",
    webhook_id="your-webhook-uuid",
)
```

### Poll manually

If you would rather not block — for example you store the job ID and check it from a Celery task or a webhook handler — poll `generations.get` yourself.

```python
import time

job = picx.video.create(prompt="a cat walking in the rain")

# Persist job.id, then later, from anywhere:
generation = picx.generations.get(job.id)

while generation.status not in ("completed", "failed"):
    time.sleep(5)
    generation = picx.generations.get(job.id)

if generation.status == "failed":
    raise RuntimeError(generation.error)

print(generation.url)
```

## Video modes

`mode` selects how a video is sourced. `text` is the default; the other modes take existing media as input. Each mode has its own required fields, and the SDK checks them **client-side before the request goes out** — a missing `audio_url` on a lipsync call raises `ValidationError` locally with a message naming the field, so you never spend a round trip on a 422. Requires `picx-ai` 0.4.0 or later.

| `mode` | Required fields | Needs `prompt`? |
| --- | --- | --- |
| `text` (default) | — | Yes |
| `image` | `image_url` | Yes |
| `reference` | `reference_urls` | Yes |
| `frames` | `start_frame_url` (`end_frame_url` optional) | Yes |
| `extend` | `source_video_url` | Yes |
| `lipsync` | `source_video_url` **and** `audio_url` | No |
| `edit` | `source_video_url` **and** `image_url` | Yes |

`video.create` takes `prompt` as its first positional argument for **every** mode, including `lipsync`. Lipsync does not use the prompt — the spoken content comes from `audio_url` — but the argument is still required by the signature, so pass an empty string: `picx.video.create("", mode="lipsync", ...)`.

### Animate between two frames (`frames`)

Give a start frame and, optionally, an end frame. The model interpolates a clip between them.

```python
job = picx.video.create(
    "smooth dolly-in on the product",
    mode="frames",
    start_frame_url="https://cdn.example.com/first.png",
    end_frame_url="https://cdn.example.com/last.png",  # optional
    duration=5,
)

asset = job.wait()
print(asset.url)
```

### Extend an existing clip (`extend`)

Continue a video past its last frame.

```python
job = picx.video.create(
    "the camera keeps pulling back to reveal the skyline",
    mode="extend",
    source_video_url="https://cdn.example.com/clip.mp4",
)

asset = job.wait()
print(asset.url)
```

### Lip-sync a clip to audio (`lipsync`)

Drive the lips in a source video from an audio track. This is the one mode that does **not** take a prompt — pass `""` to satisfy the required positional argument.

```python
job = picx.video.create(
    "",  # required positional; lipsync ignores it, speech comes from audio_url
    mode="lipsync",
    source_video_url="https://cdn.example.com/face.mp4",
    audio_url="https://cdn.example.com/voice.mp3",
)

asset = job.wait()
print(asset.url)
```

### Edit a clip with a reference image (`edit`)

Apply the content of an image to a source video.

```python
job = picx.video.create(
    "restyle the jacket to match the reference",
    mode="edit",
    source_video_url="https://cdn.example.com/clip.mp4",
    image_url="https://cdn.example.com/reference.png",
)

asset = job.wait()
print(asset.url)
```

> [!TIP]
> Client-side validation is a real time saver here: `picx.video.create("…", mode="edit", source_video_url="…")` raises `ValidationError` immediately because `image_url` is missing, rather than queuing a job that the server would reject. The message names the exact field the mode needs.

## Prompt templates

The template catalogue is a searchable library of about 50,000 ready-made image and video prompts, tuned per model and tagged by topic. `templates.list` searches it and `templates.get` fetches one by id. Both need a valid API key like the rest of `/v1`, but return public catalogue content. Requires `picx-ai` 0.4.0 or later.

```python
results = picx.templates.list(
    q="sneaker",
    media_type="video",
    featured=True,
    limit=20,
)

print(results.total)
for t in results:  # TemplateList is iterable and sized
    print(t.id, t.title, t.target_model)
```

All filters are optional and combine with AND. `tags` is matched as a set — a template must carry every tag you pass.

| Argument | Type | Description |
| --- | --- | --- |
| `q` | `str` | Free-text search across the catalogue. |
| `media_type` | `str` | `"image"` or `"video"`. |
| `topic` | `str` | Restrict to a single topic (see the note below). |
| `tags` | `list[str]` | Match templates carrying **every** tag. |
| `target_model` | `str` | Restrict to templates tuned for one model. |
| `featured` | `bool` | Only editorially featured templates. |
| `trending` | `bool` | Only currently trending templates. |
| `limit` | `int` | Page size, 1–100. Defaults to 30. |
| `offset` | `int` | Rows to skip. Defaults to 0. |

Fetch one template by id:

```python
template = picx.templates.get("tpl_abc123")
# Feed it straight into a generation:
if template.prompt is not None:
    asset = picx.images.generate(
        prompt=template.prompt,
        model=template.target_model,  # None is fine — server picks the default
    )
    print(asset.url)
```

Three behaviours of the catalogue will bite you if you assume otherwise:

> [!WARNING]
> - **`total` is an estimate, not an exact count.** Do not page to `offset >= total` and expect the last page to line up. Page until a request returns fewer rows than `limit` — that short (or empty) page is the real end.
> - **The `topic` filter works, but the `topic` field is always `None`.** You can filter by topic, but you cannot read back which topic a returned template belongs to.
> - **A `None` `prompt` means the row is premium/gated, not that data is missing.** Skip such rows (or surface them as locked) rather than treating them as broken.

Paging correctly, given that `total` is only an estimate:

```python
offset = 0
page_size = 100
while True:
    page = picx.templates.list(q="cinematic", limit=page_size, offset=offset)
    for t in page:
        if t.prompt is None:
            continue  # gated/premium row
        process(t)
    if len(page) < page_size:
        break  # short page = real end
    offset += page_size
```

## Account tier and limits

`account.tier()` returns the rate limits and quotas for your key's plan. Read it at start-up to size your own client-side throttling, or to check which model families the key may use before you submit a generation.

```python
tier = picx.account.tier()

print(tier.plan_code)            # e.g. "free", "pro"
print(tier.requests_per_minute)  # per-minute request cap
print(tier.requests_per_day)     # per-day request cap
print(tier.concurrent_limit)     # max generations running at once
print(tier.max_credits_per_day)  # daily credit ceiling
print(tier.allowed_model_types)  # e.g. ["image", "video"]

# Gate a video submit on the plan before spending a request.
if "video" not in tier.allowed_model_types:
    raise RuntimeError(f"Plan {tier.plan_code} cannot generate video.")
```

## Stream generation progress

Rather than polling `generations.get` in a loop, subscribe to the live Server-Sent Events stream. `generations.events(id)` yields one event per frame as the server pushes status updates, then a terminal `completed`/`failed` event. It is cheaper and more responsive than polling. Requires `picx-ai` 0.4.0 or later and the `generations:read` scope.

The sync client returns an iterator:

```python
job = picx.video.create("a cat walking in the rain")

for event in picx.generations.events(job.id):
    print(event.event, event.status)
    if event.is_terminal:
        break
```

Each event carries the SSE frame's fields, plus two convenience properties:

| Attribute | Type | Description |
| --- | --- | --- |
| `event` | `str` | SSE event name, e.g. `status`, `completed`, `failed`, `ping`. |
| `data` | `dict \| None` | Parsed JSON payload, or `None` when the frame's `data:` text was not JSON (e.g. a `ping` keep-alive). |
| `raw_data` | `str` | The undecoded `data:` text — always present, so a non-JSON frame is never lost. |
| `id` | `str \| None` | The SSE `id:` field, when the frame carried one. |
| `retry` | `int \| None` | The SSE `retry:` reconnection hint in ms, when present. |
| `status` | `str \| None` | Convenience: the `status` inside `data`, or `None`. |
| `is_terminal` | `bool` | Convenience: `True` once the streamed status is terminal. |

The generator holds the connection open, so consume it promptly (or break out of the loop) to release it. Streaming requests are never retried; `timeout` bounds connect/read-idle time, not total stream duration — pass `timeout=None` for no read timeout on a long render.

`AsyncPicX` exposes the same stream as an async iterator — the surface is identical, only the consumption loop changes to `async for`:

```python
import asyncio
from picx import AsyncPicX

async def watch(job_id: str):
    async with AsyncPicX() as picx:
        async for event in picx.generations.events(job_id):
            print(event.event, event.status)
            if event.is_terminal:
                print("done:", event.data)
                break

asyncio.run(watch("generation-uuid"))
```

## List and manage generations

```python
# List recent generations, optionally filtered
gens = picx.generations.list(type="video", limit=10)
for g in gens:
    print(g.id, g.status, g.type)

# Cancel a pending generation
picx.generations.cancel("generation-uuid")

# View webhook deliveries for a specific generation
deliveries = picx.generations.deliveries("generation-uuid")
```

## List models

`models.list` is public — it needs no API key, which makes it a convenient connectivity check.

```python
from picx import PicX

picx = PicX()  # no key required for this call

all_models = picx.models.list()
video_models = picx.models.list(type="video")

for model in video_models:
    print(model.id, model.type, model.credits)
```

## Async images and webhooks

Both image calls are synchronous by default. Pass `callback_url` (or a webhook binding) and the API returns `202` immediately, the SDK hands back a `GenerationJob`, and the finished image is POSTed to your URL. Requires `picx-ai` 0.3.0 or later.

```python
import os
from picx import PicX, GenerationJob

picx = PicX(os.environ["PICX_API_KEY"])

# No target -> ImageAsset, exactly as before.
image = picx.images.generate("a cat")
print(image.url)

# With a target -> GenerationJob. Nothing is held open.
job = picx.images.generate("a cat", callback_url="https://your-server.com/hooks/picx")
print(isinstance(job, GenerationJob), job.id, job.status)

# Editing takes the same fields.
edit_job = picx.images.edit(
    "replace the sky with a sunset",
    ["https://example.com/photo.jpg"],
    webhook_url="https://your-server.com/hooks/picx",
)

# Always have a fallback — poll if a delivery is missed.
generation = job.wait(poll_interval=3, timeout=600)
print(generation.status, generation.output_url)
```

The async client is identical — `AsyncPicX(...).images.generate(...)` takes the same
parameters and returns an `AsyncGenerationJob` whose `wait()` is awaitable.

See [Async Image Generation](/docs/developer-tools/async-image-generation) for the payload shape, signature verification and retry behaviour.

## Webhook management and deliveries

> [!NOTE]
> `create`, `list`, `test` and `delete` are session-authenticated in the console and raise `NotFoundError` with an API key. The delivery-inspection methods below **do** work with an API key, from `picx-ai` 0.3.1.

```python
# Registered-webhook management — console session only, see the note above.
wh = picx.webhooks.create(
    url="https://your-server.com/webhook",
    events=["generation.completed", "generation.failed"],
)
print(wh.id, wh.secret)  # save the secret
picx.webhooks.list()
picx.webhooks.test(wh.id)
picx.webhooks.delete(wh.id)
```

Delivery inspection works with an API key — this is how you debug a webhook that never arrived:

```python
# Every delivery attempted for one generation, including inline/callback_url ones.
log = picx.generations.deliveries(job.id)
print(log.total)
for d in log.deliveries:
    print(d.event_type, d.outcome, d.status_code, d.target_url)
    print(d.attempts)  # per-attempt history

# Per-webhook history, filtered to the failures.
failed = picx.webhooks.deliveries(webhook_id, outcome="exhausted", limit=50)

# Replay one after fixing your endpoint. Reuses the original payload and event id.
result = picx.webhooks.redeliver(delivery_id)
print(result.outcome, result.status_code)
```

Binding a generation to an already-registered webhook by id also works with an API key:

```python
job = picx.images.generate("a cat", webhook_id="0b9c1e42-…")
```

## Check usage and account

```python
me = picx.account.me()
print(me.name, me.credits["balance"])

usage = picx.account.usage(period="30d")
print(usage)
```

## Async client

`AsyncPicX` exposes exactly the same surface with awaitable methods. Use it in FastAPI, aiohttp, or any asyncio application so a generation does not block the event loop.

```python
import asyncio
from picx import AsyncPicX

async def main():
    async with AsyncPicX() as picx:  # reads PICX_API_KEY
        asset = await picx.images.generate(
            prompt="a beautiful sunset over the ocean",
            size="2K",
        )
        print(asset.url)

        job = await picx.video.create(prompt="a cat walking in the rain")
        video = await job.wait()
        print(video.url)

asyncio.run(main())
```

Because the surface is identical, fanning out concurrent generations is just `asyncio.gather`:

```python
import asyncio
from picx import AsyncPicX

async def batch(prompts):
    async with AsyncPicX() as picx:
        return await asyncio.gather(
            *(picx.images.generate(prompt=p, size="2K") for p in prompts)
        )

assets = asyncio.run(batch([
    "a red sneaker on marble",
    "a blue sneaker on concrete",
    "a white sneaker on sand",
]))

for asset in assets:
    print(asset.url)
```

> [!TIP]
> Use the async client as a context manager (`async with`) so the underlying connection pool is closed cleanly. If you construct it directly, call `await picx.close()` when you are done. Mind your account's concurrency limit when using `gather` — see [Rate Limits](/docs/getting-started/rate-limits).

## Handle errors

Every failure raises a subclass of `PicXError`, so you can catch the failure mode instead of inspecting status codes.

```python
import os
from picx import (
    PicX,
    PicXError,
    AuthenticationError,
    PermissionDeniedError,
    NotFoundError,
    ValidationError,
    RateLimitError,
    ServerError,
    NetworkError,
)

picx = PicX(os.environ["PICX_API_KEY"])

try:
    asset = picx.images.generate(prompt="a sunset")
    print(asset.url)
except AuthenticationError:
    ...  # 401 — key missing, malformed, or revoked
except PermissionDeniedError:
    ...  # 403 — the key is valid but lacks the required scope
except NotFoundError:
    ...  # 404 — no such generation or model
except ValidationError as err:
    ...  # 422 — bad parameters; str(err) says which
except RateLimitError as err:
    print(f"Rate limited, retry in {err.retry_after}s")
except ServerError:
    ...  # 5xx — already retried
except NetworkError:
    ...  # connection failure or timeout
except PicXError:
    ...  # anything else from the SDK
```

| Exception | Status | Meaning |
| --- | --- | --- |
| `AuthenticationError` | 401 | Key missing, malformed, or revoked. |
| `PermissionDeniedError` | 403 | Key lacks the scope for this operation. |
| `NotFoundError` | 404 | Generation, model, or resource does not exist. |
| `ValidationError` | 422 | Invalid parameters. Not retried. |
| `RateLimitError` | 429 | Limit exceeded. Exposes `retry_after` in seconds. |
| `ServerError` | 5xx | Upstream failure. Retried. |
| `NetworkError` | — | Connection failure or timeout. Retried. |

The same exceptions are raised by `AsyncPicX`, so error handling ports between the two clients unchanged.

> [!NOTE]
> `RateLimitError` and `ServerError` only surface after the SDK has exhausted its retries. If you catch one, backing off further is the right move — see [Rate Limits](/docs/getting-started/rate-limits).

## Configure retries and timeouts

The constructor takes keyword arguments alongside the key. `AsyncPicX` accepts the same ones.

```python
from picx import PicX

picx = PicX(
    api_key=os.environ["PICX_API_KEY"],
    base_url="https://api.picxstudio.com/v1",
    timeout=120.0,   # seconds, per request
    max_retries=3,
)
```

| Argument | Default | Description |
| --- | --- | --- |
| `api_key` | `PICX_API_KEY` env var | Your `pxsk_…` key. |
| `base_url` | `https://api.picxstudio.com/v1` | Override for staging or a proxy. |
| `timeout` | SDK default | Per-request timeout in seconds. |
| `max_retries` | `3` | Retry attempts before raising. Set `0` to disable. |

The SDK retries **only** on 429, 5xx, and network errors, with exponential backoff that honors `Retry-After`. It never retries 400, 401, 403, 404, or 422 — those will not succeed on a second attempt, so retrying only wastes time.

### Idempotency keys

Pass a per-call idempotency key on any POST so a retry — the SDK's or yours — cannot bill you twice for the same generation.

```python
asset = picx.images.generate(
    prompt="a red sneaker on marble",
    idempotency_key="order-4417-hero",
)
```

Reusing a key returns the original result instead of generating again. Use a stable value derived from your own data, such as an order or row ID.

## Prefer raw HTTP?

The SDK is optional. Every endpoint is a plain REST call, so if you would rather not add a dependency — a locked-down environment, a strict dependency policy, or a script that already uses `requests` — call the API directly.

```python
import os
import requests

resp = requests.post(
    "https://api.picxstudio.com/v1/images/generate",
    headers={"Authorization": f"Bearer {os.environ['PICX_API_KEY']}"},
    json={
        "prompt": "a beautiful sunset over the ocean",
        "size": "2K",
        "aspect_ratio": "16:9",
    },
)
resp.raise_for_status()
result = resp.json()
print(result["url"], result["credits_used"])
```

Going direct means handling retries, backoff, and video polling yourself. See [cURL Examples](/docs/code-examples/curl-examples) for every endpoint, and [Rate Limits](/docs/getting-started/rate-limits) for the retry rules to implement.