picx-ai

Official Python client for the PicX image and video generation API.

• Sync (PicX) and async (AsyncPicX) clients with the same surface
• Typed results and a typed exception hierarchy; ships py.typed, passes mypy --strict
• Automatic retries with exponential backoff + jitter, Retry-After aware
• Idempotency-Key support on the calls that spend credits
• One runtime dependency: httpx

This is not the CLI. This package is a library you install into your own application. The admin-cli/ project in this organisation builds the internal picx-admin command line binary and is unrelated to this SDK.

Install

pip install picx-ai
# or
uv add picx-ai

Requires Python 3.9+.

Quickstart

import os
from picx import PicX

picx = PicX(os.environ["PICX_API_KEY"])
job = picx.video.create(prompt="sneaker on marble, slow orbit", duration=12)
asset = job.wait()
print(asset.url)

PicX() falls back to the PICX_API_KEY environment variable when no key is passed, so PicX() works once the variable is set. Use it as a context manager to close the connection pool deterministically:

with PicX() as picx:
    asset = picx.images.generate("a red sneaker on white marble", size="2K", aspect_ratio="1:1")
    print(asset.url, asset.credits_used)

Configuration

picx = PicX(
    api_key="pxsk_...",  # defaults to os.environ["PICX_API_KEY"]
    base_url="https://api.picxstudio.com/v1",  # or the PICX_BASE_URL env var
    timeout=60.0,  # seconds, per request
    max_retries=2,  # retries for 429 / 5xx / connection errors
)

API

picx.images is also available as picx.image, and picx.video as picx.videos.

Images

asset = picx.images.generate("a red sneaker on white marble", size="2K", aspect_ratio="1:1")
asset.id, asset.url, asset.model, asset.size, asset.aspect_ratio, asset.credits_used

edited = picx.images.edit("put it on a wet street at night", [asset.url], size="4K")

Managed uploads (assets)

images.edit and the video modes need public URLs — they can't read a local file. Upload it first:

asset = picx.assets.create("portrait.jpg", metadata={"source": "onboarding"})
print(asset.id, asset.url)

# Now use the URL in an edit or video call:
avatar = picx.images.edit(
    "turn this into a naive doodle avatar",
    image_urls=[asset.url],
)

file accepts a path (string or Path), an open binary file, or raw bytes:

asset = picx.assets.create(b"\x89PNG...", filename="photo.png", content_type="image/png")

List, get, and delete:

result = picx.assets.list(kind="image", limit=50)
for a in result:
    print(a.id, a.url)

single = picx.assets.get(asset.id)
picx.assets.delete(asset.id)  # soft delete — existing URLs keep working

The async client has the same surface:

async with AsyncPicX() as picx:
    asset = await picx.assets.create("portrait.jpg")

Videos (202 Accepted, then poll)

POST /videos/generate is asynchronous server-side, so create() returns a job:

job = picx.video.create(
    prompt="sneaker on marble, slow orbit",
    duration=12,  # server default 5
    resolution="720p",  # "480p" | "720p" | "1080p"; server default "720p"; must be priced for the model
    sound=True,  # server default True
    aspect_ratio="16:9",
    mode="image",  # see the mode table below
    image_url="https://…/ref.png",
    callback_url="https://example.com/webhook",
)

job.id, job.status  # "gen_…", "queued"
generation = job.wait(timeout=900, poll_interval=5)
print(generation.url)  # alias for .output_url

Modes

mode= selects how the video is sourced. Each mode has its own required fields, validated client-side (raising ValidationError) so a bad call fails before it spends a request:

lipsync is the one mode that does not require a prompt — the spoken content comes from audio_url. Since 0.4.1 you can omit prompt entirely; an empty string still works, and either way a blank prompt is dropped from the request body for lipsync (and only for lipsync). Every other mode still raises ValidationError when the prompt is missing:

# First frame → last frame interpolation
morph = picx.video.create(
    "the logo unfolds into a paper crane",
    mode="frames",
    start_frame_url="https://cdn.example.com/logo.png",
    end_frame_url="https://cdn.example.com/crane.png",  # optional
)

# Continue an existing clip
longer = picx.video.create(
    "the camera keeps pulling back to reveal the skyline",
    mode="extend",
    source_video_url="https://cdn.example.com/clip.mp4",
)

# Drive a face with an audio track — no prompt needed
talking = picx.video.create(
    mode="lipsync",
    source_video_url="https://cdn.example.com/portrait.mp4",
    audio_url="https://cdn.example.com/voice.mp3",
)

# Composite an image into a source clip
composited = picx.video.create(
    "place the product on the table",
    mode="edit",
    source_video_url="https://cdn.example.com/scene.mp4",
    image_url="https://cdn.example.com/product.png",
)

wait() polls GET /generations/{id} until a terminal status (succeeded, completed, failed, error, cancelled, canceled). It raises JobFailedError on a failed generation — pass raise_on_failure=False to get the Generation back instead — and JobTimeoutError if the timeout elapses first. job.refresh() polls once.

Anything left unset is omitted from the request body so the server applies its own defaults (model fal-ai/bytedance/seedance/v2, duration 5, resolution 720p, sound on).

Streaming progress (SSE)

picx.generations.events(id) streams live progress over Server-Sent Events — cheaper and more responsive than polling generations.get() in a loop. The sync client yields an iterator; AsyncPicX yields an async iterator. Each GenerationEvent carries the SSE event name (event), the parsed JSON payload (data, or None for a non-JSON keep-alive such as ping), and the undecoded raw_data so a non-JSON frame is never lost. Convenience properties event.status and event.is_terminal read the status out of data for you:

job = picx.video.create("sneaker on marble, slow orbit")

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

async with AsyncPicX() as picx:
    job = await picx.video.create("sneaker on marble, slow orbit")
    async for event in picx.generations.events(job.id):
        if event.is_terminal:
            break

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

Templates

picx.templates searches the PicX prompt-template catalogue — ready-made image and video prompts, tuned per model and tagged by topic. Public content, but it still needs an API key like the rest of /v1.

result = picx.templates.list(
    media_type="video",
    topic="product",
    featured=True,
    limit=20,
)
for template in result:  # TemplateList is iterable and sized
    print(template.id, template.title, template.target_model)

# Tags are matched as a set — a template must carry every tag you pass.
cinematic = picx.templates.list(tags=["cinematic", "portrait"])

one = picx.templates.get("1234")
print(one.prompt, one.target_model)

Three catalogue behaviours are deliberate, not bugs:

• total is an estimate, not a count. The catalogue is large (~50k rows) and a COUNT over it is avoided on purpose. total is offset + page length + 1 when more rows exist. Don't treat it as an exact figure — page until a short page comes back:

  offset = 0
  while True:
      page = picx.templates.list(q="sneaker", limit=100, offset=offset)
      for t in page:
          process(t)
      if len(page) < page.limit:  # short page → done
          break
      offset += page.limit
  

• topic filters, but template.topic is always None. The topic param works as a query-time keyword bucket; there is no per-row topic column, so the field on every returned template is None. Filter on it, don't read it back.

• prompt is None for gated rows by design. Premium/gated templates redact their prompt text publicly. prompt is None means gated, not missing — the rest of the metadata (title, preview_url, tags, …) is still present.

Models, usage, account

for model in picx.models.list(type="video"):
    print(model.id, model.name, model.credits)

usage = picx.account.usage(period=30)
print(usage.total_requests, usage.credits_used, usage.total_cost_usd, usage.model_breakdown)

me = picx.account.me()
print(me.email, me.is_active, me.credits)

tier = picx.account.tier()
print(tier.plan_code, tier.requests_per_minute, tier.allowed_model_types)

account.tier() returns the rate limits and quotas for the key's plan — plan_code, requests_per_minute, requests_per_day, concurrent_limit, max_credits_per_day and allowed_model_types. Read it to pace requests under the plan's caps instead of discovering them through 429s.

GET /models is the only public endpoint, so PicX(api_key=None).models.list() works without credentials.

Every result object also keeps the untouched response on .raw, so new API fields are reachable before the SDK models them.

Async

import asyncio, os
from picx import AsyncPicX


async def main() -> None:
    async with AsyncPicX(os.environ["PICX_API_KEY"]) as picx:
        job = await picx.video.create(prompt="sneaker on marble, slow orbit", duration=12)
        asset = await job.wait()
        print(asset.url)

        image = await picx.images.generate("a red sneaker on white marble")
        print(image.url)


asyncio.run(main())

The async surface mirrors the sync one method for method — only await and aclose()/async with differ. A parity test enforces this.

Error handling

from picx import (
    PicXError,  # base class: .status_code, .request_id, .body, .message
    ValidationError,  # 400 / 422, and invalid arguments caught locally
    AuthenticationError,  # 401, or no API key configured
    PermissionDeniedError,  # 403, the key lacks the required scope
    NotFoundError,  # 404
    RateLimitError,  # 429, exposes .retry_after
    ServerError,  # 5xx
    APIConnectionError,  # DNS/TLS/socket failure
    APITimeoutError,  # subclass of APIConnectionError
    JobFailedError,  # a generation ended failed/cancelled
    JobTimeoutError,  # wait() timed out, generation still running
)

try:
    asset = picx.images.generate("a red sneaker")
except RateLimitError as exc:
    print("slow down for", exc.retry_after, "seconds")
except PermissionDeniedError:
    print("this key is missing the images:generate scope")
except PicXError as exc:
    print(exc.status_code, exc.request_id, exc)

Errors are parsed from both response shapes the API uses — FastAPI's {"detail": …} (including the 422 list form) and {"error": …, "detail": …}.

Retries and idempotency

Retries apply to 429, 5xx and connection/timeout failures only — never to other 4xx. Backoff is exponential with full jitter and honours Retry-After (capped at 60s). max_retries defaults to 2; set max_retries=0 to disable.

A POST is not replayed unless you supply an idempotency key, because replaying it could charge twice:

import uuid

asset = picx.images.generate("a red sneaker", idempotency_key=str(uuid.uuid4()))

The key is sent as the Idempotency-Key header, which the backend honours on calls that spend credits. idempotency_key is available on images.generate, images.edit and video.create.

API keys are never logged

The key is redacted from repr()/str() of the client and from every exception message and response body (anything matching pxsk_… becomes pxsk_***REDACTED***). Read it back deliberately with picx.api_key if you really need it. This is covered by tests.

Examples

export PICX_API_KEY=pxsk_...
python examples/generate_image.py "a red sneaker on white marble"
python examples/generate_video.py "sneaker on marble, slow orbit"
python examples/generate_video.py --async

Development

uv venv --python 3.13
uv pip install -e ".[dev]"
uv run mypy
uv run pytest

The test suite runs entirely against httpx.MockTransport — no network access, no API key required.

License

MIT