p0aiapi

English | 简体中文

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.1 · Default API: https://p0-api.inaiai.com

Installation

python -m pip install --upgrade p0aiapi==0.2.1
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.

Method	Purpose
client.face_swap.image(source_url, target_url)	Replace a face in a target image using a source face image
client.face_swap.video(source_url, target_url, output_format="video")	Replace a face in a target video using a source face image
client.videos.transfer_motion(source_url, motion_url)	Animate a source image using a reference motion video
client.scenes.list()	Retrieve scene metadata as a list of dictionaries
client.scenes.figurine(source_url, reference_url, prompt=...)	Generate a figurine image using a character image and pose reference
client.images.generate(prompt, **fields)	Generate images from text
client.novels.generate(description, **fields)	Submit novel generation
client.novels.continue_from(file_url, **fields)	Continue writing from a public TXT or Markdown document
client.audio.list_voices() / client.audio.generate(text, **fields)	List voices and generate speech
client.audio.upload_voice(spk_id, prompt_text, path)	Upload a voice sample with its transcript

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 .
python scripts/check_readme.py

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

Full documentation, the changelog, and contribution guidelines are included in the source distribution: docs/advanced.md, CHANGELOG.md, CONTRIBUTING.md, and docs/publishing.md.

The supported endpoints are based on YApi project 8 and backend contracts. Availability depends on the deployment. Verification scope is recorded in docs/backend-compatibility.md and docs/live-verification-2026-10-09.md in the source distribution.

---

Chinese

P0 API 的 Python SDK，提供同步和 asyncio 客户端、普通图像与视频任务、任务查询及自动等待。支持 Python 3.10+，运行时仅依赖 httpx。

发行包与导入名均为 p0aiapi，当前版本 0.2.1。默认服务地址为 https://p0-api.inaiai.com。以下服务接口见于 YApi 项目 8，实际可用能力取决于部署版本。

安装与鉴权

python -m pip install --upgrade p0aiapi==0.2.1
export P0_API_KEY='你的 API key'
export P0_API_SECRET_KEY='你的 API secret'

推荐同时配置 key 和 secret 使用鉴权 v2，分别发送 apikey 和 X-SECRET-KEY。也可传入 Client(api_key="...", secret_key="...")，AsyncClient 用法相同。未提供 secret 时兼容 v1，鉴权版本选择及回退由后端处理。SDK 不内置凭据。

其他部署可设置 P0_API_BASE_URL 或 base_url=，显式参数优先于环境变量；反向代理路径前缀会保留。使用 with / async with 管理连接。

快速开始

使用已获授权的普通非色情素材，将示例地址替换为 backend 能访问的图片 URL。第一张提供人脸，第二张是目标图片：

from p0aiapi import Client

# 替换为你的授权素材 URL。
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"))

普通方法可以直接使用，无需加载接口配置或查找 operationId。生成方法返回初始任务字典，通过 task_id 查询或等待结果。

查询任务

一个任务用 tasks.get()，多个任务用 tasks.batch()；需要自动轮询时用 tasks.wait()。 将下面的 ID 换成提交任务时返回的 task_id。

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)

异步方法使用相同参数，并加 await：

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())

task_id 必须是非空且不全为空白的字符串，SDK 保留原值，不自动去掉空白；不能用内部 uuid 或业务编号 bid 替代。batch() 接收非空任务 ID 列表或其他可迭代对象，发送裸 JSON 数组，返回任务字典数组。未命中的 ID 被省略，结果不保证输入顺序，同一 ID 可能有多条记录；单独查询未命中，或批量全部未命中时抛出 NotFoundError（404）。

状态 2 成功、-1 失败；0 / 1 / 3 继续等待。tasks.wait() 默认等待 600 秒、每 10 秒查询一次，失败抛出 TaskFailedError，超时抛出 TaskTimeoutError；raise_on_failure=False 可返回失败记录。超时停止本地等待，后端任务继续执行。原来的 client.wait_for_task() 仍可使用。

常用能力

下面的方法同时支持 Client 和 AsyncClient，不依赖 OpenAPI 配置。

方法	用途与输入
client.face_swap.image(source_url, target_url)	换脸图片：人脸图片 → 目标图片
client.face_swap.video(source_url, target_url, output_format="video")	换脸视频：人脸图片 → 目标视频
client.videos.transfer_motion(source_url, motion_url)	动作迁移：角色图片 + 动作参考视频 → 视频
client.scenes.list()	查询通用场景元信息，返回字典数组
client.scenes.figurine(source_url, reference_url, prompt=...)	角色原图 + 姿势参考图 → 手办图片，固定 person_model 场景
client.images.generate(prompt, **fields)	文生图
client.novels.generate(description, **fields)	小说生成
client.novels.continue_from(file_url, **fields)	从公网 TXT / Markdown 文件续写
client.audio.list_voices() / generate(text, **fields)	音色列表与语音生成
client.audio.upload_voice(spk_id, prompt_text, path)	上传音色样本及对应文本

动作迁移和手办生成示例：

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="将原创角色制作成桌面收藏手办，参考第二张图片的姿势。",
    )
    tasks = client.tasks.batch([video["task_id"], figurine["task_id"]])
    print(tasks)

媒体方法接收绝对 HTTP(S) URL，拒绝本机路径和含用户名、密码的 URL，保留签名查询参数。SDK 在本地校验 URL 格式，素材是否能下载由 backend 检查。

生成方法接受 **fields 传入具体接口支持的可选字段，例如 bid、task_id、notify_url。None 省略，False 和 0 保留，默认值由服务端决定。新媒体方法使用 source_url、target_url / motion_url / reference_url 和 prompt 参数；不要在 **fields 中重复传旧的 source_path、target_path、incoming_prompt 或 scene_name。

scenes.list() 是通用场景清单，与手办配置属于不同集合。scenes.figurine() 固定使用已核对的 person_model，生成图片；无需从清单中选择模板。

类型与错误

公开的 Task 类型和 TaskStatus 枚举用于类型提示与编辑器补全。任务仍是服务端返回的原始字典，保留未知字段；out_data 的形态取决于任务类型，不进行强制转换。

HTTP 错误继承 APIError，可检查 status_code 和 body；连接错误为 TransportError。任务失败和等待超时分别为 TaskFailedError、TaskTimeoutError。默认无请求体的读取请求最多重试 2 次，生成 POST 不自动重试；返回 submission_uncertain=true 时先查询任务，避免重复提交。

完整进阶文档、设计及适配检查记录随源码发行包提供，位于 docs/ 目录。

开发

python -m pip install -e '.[dev]'
pytest
ruff check .
ruff format --check .
python scripts/check_readme.py

测试使用模拟 HTTP 验证请求合同和错误处理，不提交真实生成任务。离线演示：python examples/offline_demo.py。

源码发行包同时包含 CHANGELOG.md、CONTRIBUTING.md、docs/publishing.md 和独立的 README.zh-CN.md。