p0aiapi

P0 API 的 Python SDK，支持 Python 3.10+、同步 / asyncio 调用、图像与视频任务、任务查询、SSE 和 OpenAPI JSON 动态调用。运行时仅依赖 httpx。

发行包和导入名称均为 p0aiapi，版本 0.1.2。默认 API 地址为 https://p0-api.inaiai.com。

下面的服务接口均见于 YApi 项目 8。包内接口配置结合 YApi 与本地 backend 源码维护；接口是否可用仍取决于实际部署版本。SDK 不会把文档路径自动替换成其他接口。

安装与配置

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

本地开发可使用 python -m pip install -e .。

需要连接其他部署时，设置 P0_API_BASE_URL，或传入 Client(base_url="https://your-api.example.com", api_key="...", secret_key="...")。显式参数优先于环境变量，未设置地址时使用 https://p0-api.inaiai.com。地址填写服务根地址；反向代理的路径前缀会被保留。

推荐使用鉴权 v2，同时配置 API key 和 secret。 P0_API_KEY / api_key= 发送 apikey 请求头，P0_API_SECRET_KEY / secret_key= 发送 X-SECRET-KEY。也可显式传入 Client(api_key="...", secret_key="...")；AsyncClient 用法相同。鉴权版本选择与回退由后端处理，是否已启用以实际部署为准。

未设置 secret 时兼容原来的 apikey 调用方式。可选的 token= / P0_API_TOKEN 额外发送 Authorization: Bearer ...，不替代 API key 和 secret。SDK 不内置密钥，初始化不会访问服务器。使用 with 管理连接，或在结束时调用 close()。

查询任务

一个 task_id 用 client.tasks.get()，多个 task_id 用 client.tasks.batch()。两种方法直接查询任务状态和结果，不需要加载 OpenAPI 配置。

使用提交任务时返回的 task_id，按原值精确匹配；不能用内部 uuid 或业务 bid 替代，也不要去掉或增加首尾空白。

同步查询

将示例中的任务 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"))

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

异步查询

import asyncio
from p0aiapi import AsyncClient


async def main():
    async with AsyncClient() as client:
        task = await client.tasks.get("task-1")
        print(task.get("status"), task.get("out_data"))

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


asyncio.run(main())

异步客户端在同一个 asyncio 事件循环中复用，结束时使用 async with 或 await client.aclose()。

单独查询使用 /api/public/task，成功响应为任务对象；同一 task_id 存在多条记录时，返回记录由服务端决定。

批量查询使用 YApi 3469 的 /api/public/task/details，发送裸 JSON 字符串数组。传入非空 task_id 列表，成功响应直接返回任务对象数组。未命中的 ID 被省略，结果顺序可能与输入不同，同一 ID 可能返回多条记录；请按 task_id 关联结果。

单独查询未命中，或批量查询的所有 ID 均未命中时，SDK 抛出 NotFoundError（HTTP 404）。HTTP 200 表示查询成功，任务是否完成仍需检查 status：2 成功，-1 失败，0 / 1 / 3 继续等待。out_data 保留服务端原始值。

自动等待任务结束

需要自动轮询时使用 wait_for_task()。它轮询单任务查询接口检查状态，直到成功、失败或超时；任务查询可能缓存约 10 秒，建议 poll_interval=10。

from p0aiapi import Client

with Client() as client:
    result = client.wait_for_task("task-1", timeout=600, poll_interval=10)
    print(result.get("out_data"))

异步用法为 await client.wait_for_task(task_id, timeout=600, poll_interval=10)。超时停止本地等待，后端任务仍继续执行。

普通换脸图片和视频

使用已获授权的普通非色情图片和视频。source_path 是提供人脸的图片 URL；target_path 是待替换人脸的目标图片或视频 URL。URL 必须能被 backend 访问，不能传本机文件路径。

包内配置提供 YApi 2520 的换脸图片和 2522 的换脸视频。两者通过 data= 发送表单，返回初始任务记录：

from p0aiapi import Client, load_catalog

with Client(openapi=load_catalog()) as client:
    image_task = client.call(
        "yapi_2520",
        data={
            "source_path": "https://your-cdn.example.com/authorized-face.jpg",
            "target_path": "https://your-cdn.example.com/authorized-portrait.jpg",
        },
    )
    video_task = client.call(
        "yapi_2522",
        data={
            "source_path": "https://your-cdn.example.com/authorized-face.jpg",
            "target_path": "https://your-cdn.example.com/authorized-scene.mp4",
            "output_format": "video",
        },
    )
    print(client.tasks.get(image_task["task_id"]))
    print(client.tasks.batch([image_task["task_id"], video_task["task_id"]]))

将示例 URL 替换成你的授权素材。单独查询或批量查询返回当前状态，需要等待最终结果时使用 wait_for_task()。异步客户端使用相同参数，并对 call() 和查询方法加 await。

场景列表、手办图片与动作迁移视频

通过 YApi 2512 获取通用场景列表，响应为字典数组。列表来自部署的场景配置，字段以实际返回为准：

from p0aiapi import Client, load_catalog

with Client(openapi=load_catalog()) as client:
    scenes = client.call("yapi_2512")
    print("场景记录数：", len(scenes))

按具体生成接口核对支持的普通场景，不要自动选列表第一项。通用列表与 Qwen 手办配置是不同的集合，列表项不保证能直接传给 Qwen 接口。

YApi 2515 使用已确认的 person_model 场景生成手办图片。source_path 是原始角色图片，target_path 是姿势参考图片，两者均为 backend 可访问的 URL。需要明确传入有效的 scene_name；该接口的路径虽然含 videos，实际输出是图片任务。

from p0aiapi import Client, load_catalog

with Client(openapi=load_catalog()) as client:
    task = client.call(
        "yapi_2515",
        data={
            "source_path": "https://your-cdn.example.com/original-character.png",
            "target_path": "https://your-cdn.example.com/pose-reference.png",
            "scene_name": "person_model",
            "incoming_prompt": "将原创插画角色制作成桌面收藏手办，参考第二张图片的姿势。",
        },
    )
    print(client.tasks.get(task["task_id"]))

需要生成普通动作视频时，可使用 YApi 2509 的动作迁移：source_path 为角色图片 URL，target_path 为动作参考视频 URL。例如使用已有授权的行走视频：

from p0aiapi import Client, load_catalog

with Client(openapi=load_catalog()) as client:
    task = client.call(
        "yapi_2509",
        data={
            "source_path": "https://your-cdn.example.com/original-character.png",
            "target_path": "https://your-cdn.example.com/walking-reference.mp4",
            "width": 800,
            "height": 600,
            "output_format": "video",
            "video_format": "video/h264-mp4",
        },
    )
    print(client.tasks.get(task["task_id"]))

上述生成操作均返回任务记录，后续用 tasks.get() / tasks.batch() 查询 status 和 out_data，或用 wait_for_task() 自动等待。

常用接口

同步和异步客户端的方法同名；异步请求方法需要 await。

表中的 call() 使用包内配置，初始化时传入 openapi=load_catalog()。

方法	用途
client.call("yapi_2520", data=...)	普通换脸图片
client.call("yapi_2522", data=...)	普通换脸视频
client.call("yapi_2512")	通用场景列表
client.call("yapi_2515", data=...)	person_model 手办图片
client.call("yapi_2509", data=...)	普通动作迁移视频
client.images.generate(prompt, **fields)	文生图
client.novels.generate(description, **fields)	提交小说生成任务
client.novels.continue_from(file_url, **fields)	从公网 TXT / Markdown 文件续写
client.audio.list_voices()	TTS v1 音色列表
client.audio.generate(text, **fields)	TTS v1 语音生成
client.audio.upload_voice(spk_id, prompt_text, path)	上传音色样本及对应文本
client.tasks.get(task_id)	查询任务记录
client.tasks.batch([task_id, ...])	批量查询任务记录，发送裸 JSON 数组
client.tasks.events([task_id, ...])	可选的 SSE 任务事件监听
client.wait_for_task(task_id, timeout=600)	本地轮询，等待任务结束

生成方法使用后端要求的表单编码。**fields 透传可选参数并省略 None，例如具体接口支持的 task_id、bid、app_id、notify_url。实际字段和默认值以对应接口为准，响应保留原始结构及新增字段。

音色上传使用 multipart，SDK 会关闭自己打开的文件；直接调用 request(files=...) 时由调用方管理文件句柄。

with Client() as client:
    voice = client.audio.upload_voice("my-voice", "音频中的原文", "sample.wav")
    print(voice)

OpenAPI 接口配置

使用包内配置

load_catalog() 离线加载随包安装的接口配置，使用 yapi_接口ID 作为 operationId。配置只包含本版本支持的 YApi 接口，请求参数按本地 backend 签名校正，不包含文档中的密钥和示例值。

from p0aiapi import Client, load_catalog

with Client(openapi=load_catalog()) as client:
    voices = client.call("yapi_2526")
    tasks = client.call_endpoint("POST", "/api/public/task/details", json=["task-1", "task-2"])
    print(voices, tasks)

p0aiapi operations
p0aiapi export-catalog --output openapi.catalog.json

call_endpoint() 通过方法和路径模板定位已加载的操作。动态调用显式区分 path_params、params（query）、headers、cookies、json、data（form）和 files。表单接口使用 data，不能改用 json。动态调用共用客户端认证、超时和错误处理。

下载部署配置

p0aiapi fetch-schema --output openapi.json
p0aiapi operations openapi.json

CLI 同样使用默认 API 地址和 P0_API_BASE_URL / P0_API_KEY / P0_API_SECRET_KEY / P0_API_TOKEN。文档路径默认 /openapi.json，可通过 --path 指定。下载检查版本和操作目录结构后原子替换文件，失败保留原文件。

from p0aiapi import Client

with Client(openapi="openapi.json") as client:
    print(list(client.schema.operations))

也可传入字典或 OpenAPISchema。client.load_openapi("new.json") 从本地更新；client.refresh_openapi() / await client.refresh_openapi() 显式从当前部署下载。加载或刷新会替换当前配置，验证失败时保留原配置；部署文档可能隐藏部分路由，不自动与包内目录合并。

升级前比较

p0aiapi fetch-schema --output openapi.next.json
p0aiapi diff openapi.json openapi.next.json --fail-on-change

输出 added / removed / changed operationId 列表，包含引用模型变化。--fail-on-change 遇到任何差异返回退出码 2，但不自动判定业务兼容性。生产项目建议把快照与代码一起版本控制，测试后升级。

支持 OpenAPI 3.0 / 3.1 JSON、本地 $ref、JSON / URL-encoded / multipart 请求、简单路径参数和普通 query 数组。不支持 YAML、Swagger 2.0 或 deepObject 等复杂序列化；不做完整 JSON Schema 校验，也不生成接口专属类型。便利方法不依赖 OpenAPI 配置，响应不按文档强制转换类型。

request_raw() 返回完整读取的 httpx.Response；request() 根据响应类型返回 JSON、文本、字节或 None。API 请求地址由客户端的 base_url 决定，schema 的 servers 不覆盖它。

可选：SSE 任务事件监听

任务状态和结果推荐通过上面的单独或批量查询获取。需要监听事件时，使用 tasks.events()、stream_events() 或 stream_call() 增量读取。普通 request() / call() 会等待响应结束。流式读取不自动重试或重连，timeout=60 可设置读取空闲超时。

from contextlib import closing
from p0aiapi import Client

with Client() as client:
    with closing(client.tasks.events(["task-1", "task-2"], timeout=60)) as events:
        for event in events:
            print(event.event, event.data)

异步版本返回异步迭代器，创建时不加 await：

from contextlib import aclosing
from p0aiapi import AsyncClient


async def watch():
    async with AsyncClient() as client:
        async with aclosing(client.tasks.events(["task-1"])) as events:
            async for event in events:
                print(event.data)

迭代结束、异常或显式 close() / aclose() 会释放连接；提前退出循环时使用上面的上下文管理器。ServerSentEvent 保留原始 data、事件名称、id 和 retry；仅在数据确实为 JSON 时调用 event.json()。服务端事件名称可能与任务终态不同，最终结果使用任务 status 判断或调用 wait_for_task()。

错误、重试和任务状态

from p0aiapi import APIError, Client, TaskFailedError, TaskTimeoutError

with Client() as client:
    try:
        task = client.audio.generate("你好，世界。")
        result = client.wait_for_task(task["task_id"], timeout=600, poll_interval=10)
    except APIError as exc:
        print(exc.status_code)
        if isinstance(exc.body, dict) and exc.body.get("submission_uncertain"):
            print("任务可能已入队，请先查询：", exc.body.get("task_id"))
    except TaskFailedError as exc:
        print(exc.task.get("error"))
    except TaskTimeoutError as exc:
        print("停止本地等待，后端任务仍在执行", exc.last_task)

• HTTP 错误：AuthenticationError（401/403）、NotFoundError（404）、ValidationError（422）、RateLimitError（429）、ServerError（5xx）均继承 APIError，保留 .body、.response、.request_id。默认异常文字不回显凭据或响应内容。
• 连接及 HTTP 超时为 TransportError；坏 JSON、解压失败或不适用的响应为 P0ResponseError。这些错误及任务错误继承 P0Error。本地配置错误和 SchemaError 继承 ValueError。
• 无请求体的 GET/HEAD/OPTIONS 默认最多重试 2 次，覆盖连接故障和 408/429/500/502/503/504；读取 Retry-After 并退避，单次最多等待 30 秒。POST 等写请求不自动重试。若返回 submission_uncertain=true，应先查询任务，避免重复提交。
• 状态 0 等待、1 处理、3 水印处理中，继续等待；2 成功；-1 抛出 TaskFailedError，或通过 raise_on_failure=False 返回失败记录。数字字符串也受支持，未知状态继续等待至超时。
• 轮询在总预算内处理临时连接故障及上述可重试 HTTP 状态；404 和认证错误立即抛出。任务查询可能缓存约 10 秒，建议 poll_interval=10。超时只停止本地等待，不取消远端任务。
• 同步 HTTP 超时按网络阶段计算，正在执行的请求可能超过轮询截止时间；异步等待使用总 deadline，可被调用方取消。
• API 路径必须为 /...；不接受外部 URL，也不自动跟随跳转。下载外部结果请使用独立的 HTTP 客户端。

开发和发布

python -m pip install -e '.[dev]'
pytest
ruff check .
ruff format --check .
python -m build
python -m twine check dist/p0aiapi-0.1.2-py3-none-any.whl dist/p0aiapi-0.1.2.tar.gz

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

发布步骤 · 架构及接口维护。GitHub Actions 包含测试矩阵和手动 TestPyPI / PyPI 发布流程。