p0aiapi

English | 简体中文

Python SDK for the P0 API. Synchronous and asynchronous clients, image and video generation, task queries, and automatic polling. Supports Python 3.10+ with httpx as the only runtime dependency.

Version: 0.2.0 · Default API: https://p0-api.inaiai.com

Installation

python -m pip install --upgrade p0aiapi==0.2.0
export P0_API_KEY='your-api-key'
export P0_API_SECRET_KEY='your-api-secret'

Use an API key and secret for v2 authentication. Credentials can also be passed to Client(api_key="...", secret_key="...") or AsyncClient(...). Omitting the secret preserves v1 compatibility; authentication selection and fallback are handled by the server.

Set P0_API_BASE_URL or pass base_url= to use another deployment. Explicit arguments take precedence over environment variables. The SDK preserves gateway path prefixes.

Quickstart

Replace the example URLs with your own accessible images. The source supplies the face; the target is the image to edit.

from p0aiapi import Client

with Client() as client:
    task = client.face_swap.image(
        "https://your-cdn.example.com/face.jpg",
        "https://your-cdn.example.com/portrait.jpg",
    )
    result = client.tasks.wait(task["task_id"])
    print(result.get("out_data"))

Resource methods work directly without loading a schema or looking up operation IDs. Generation returns an initial task dictionary; use its task_id to query or wait for completion. Context managers close client connections automatically.

Task queries

Use tasks.get() for one task, tasks.batch() for multiple tasks, and tasks.wait() for automatic polling.

from p0aiapi import Client

with Client() as client:
    task = client.tasks.get("task-1")
    print(task.get("status"), task.get("out_data"))

    for task in client.tasks.batch(["task-1", "task-2"]):
        print(task.get("task_id"), task.get("status"), task.get("out_data"))

    result = client.tasks.wait("task-1", timeout=600, poll_interval=10)

Async methods accept the same arguments:

import asyncio
from p0aiapi import AsyncClient


async def main():
    async with AsyncClient() as client:
        task = await client.tasks.get("task-1")
        tasks = await client.tasks.batch(["task-1", "task-2"])
        result = await client.tasks.wait("task-1", timeout=600, poll_interval=10)
        print(task, tasks, result.get("out_data"))


asyncio.run(main())

Pass the exact task_id returned by submission, rather than uuid or bid. IDs must contain at least one non-whitespace character and are preserved without trimming. Batch input must be nonempty. Batch responses are lists of task dictionaries: missing IDs are omitted, order is unspecified, and one ID may match multiple records. A missing single task or a batch with no matches raises NotFoundError (404).

Status 2 means completed, -1 means failed, and 0 / 1 / 3 continue polling. tasks.wait() defaults to a 600-second timeout and 10-second polling interval. It raises TaskFailedError on failure or TaskTimeoutError on timeout; use raise_on_failure=False to return a failed task. A timeout ends local polling without cancelling the server task. client.wait_for_task() remains available.

Resources

All methods below support both Client and AsyncClient.

Motion transfer and figurine generation:

from p0aiapi import Client

with Client() as client:
    video = client.videos.transfer_motion(
        "https://your-cdn.example.com/character.png",
        "https://your-cdn.example.com/walking.mp4",
        width=800,
        height=600,
        output_format="video",
        video_format="video/h264-mp4",
    )
    figurine = client.scenes.figurine(
        "https://your-cdn.example.com/character.png",
        "https://your-cdn.example.com/pose.png",
        prompt="Create a desktop figurine of the original character, using the reference pose.",
    )
    tasks = client.tasks.batch([video["task_id"], figurine["task_id"]])
    print(tasks)

Media inputs must be absolute HTTP(S) URLs accessible to the server. Local paths and URLs containing credentials are rejected; signed query strings are preserved. Optional **fields pass through supported API parameters such as bid, task_id, and notify_url. None is omitted; False and 0 are retained.

Use the named URL and prompt arguments with media methods. The underlying fields source_path, target_path, incoming_prompt, and scene_name cannot be overridden through **fields. scenes.figurine() uses the fixed person_model scene and produces images. The general scene list is a separate configuration source.

Types and errors

Task and TaskStatus provide type hints and editor support. Responses remain ordinary dictionaries, preserving unknown fields and the server's original out_data value.

HTTP errors inherit from APIError, with status_code and body available for inspection. Connection failures raise TransportError. Bodyless read requests retry up to twice by default; generation POSTs are never automatically retried. If the server returns submission_uncertain=true, query the task before submitting again.

Development

python -m pip install -e '.[dev]'
pytest
ruff check .
ruff format --check .

Tests use mock HTTP transports. Run python examples/offline_demo.py for an offline example.

Advanced usage · Changelog · Contributing · Publishing

The supported endpoints are based on YApi project 8 and backend contracts. Availability depends on the deployment. See backend compatibility and integration notes for verification scope.