foya-aiapi-sdk

用于集成 Foya backend API 的 Python SDK。支持 Python 3.10+、同步 / asyncio 调用、任务等待、文件与音色上传、SSE，以及 OpenAPI JSON 动态调用。运行时仅依赖 httpx，无需安装 backend、数据库、模型或消息队列。

发行包名称：foya-aiapi-sdk；导入名称：foya_aiapi。当前版本 0.1.0，已准备打包和发布流程，不代表已上传 PyPI。

已对照 YApi 项目 8 的全部 55 条记录与本地 backend 源码：内置配置包含 27 条后端已注册的通用接口（其中 2 条被后端 OpenAPI 隐藏）；8 条文档接口在当前后端缺失；20 条裸体化或显式性化接口仅保留审计元信息，不提供调用封装。核对日期为 2026-10-01，逐项结果见 覆盖报告。源码注册不代表线上部署已经联调通过。

安装与快速开始

当前从本地安装：

python -m pip install /path/to/foya-aiapi-sdk

发布后，其他项目可通过 python -m pip install foya-aiapi-sdk 安装。

export FOYA_API_BASE_URL='http://localhost:8089'
export FOYA_API_KEY='你的 API key'

from foya_aiapi import Client

with Client() as client:
    task = client.images.generate(
        prompt="雪山脚下的一座木屋，清晨的柔和光线",
        width=1024,
        height=1024,
        batch_size=1,
    )
    result = client.wait_for_task(task["task_id"], timeout=600, poll_interval=5)
    print(result.get("out_data"))

也可直接使用 Client(base_url="https://api.example.com/aiapi", api_key="...", timeout=30, max_retries=2)。显式参数优先于环境变量。地址填服务根地址，不额外附加 /api；若反向代理使用 /aiapi 等前缀，放在 base_url 中，SDK 会保留它。

认证采用后端实际要求的 apikey 请求头。可选 token= 或 FOYA_API_TOKEN 会额外发送 Authorization: Bearer ...，不替代 apikey。初始化不会访问服务器，SDK 不内置服务地址或密钥。复用客户端连接池，在结束时 close()，或使用 with。

异步调用

import asyncio
from foya_aiapi import AsyncClient


async def main():
    async with AsyncClient() as client:
        task = await client.images.generate(prompt="海边灯塔的水彩画", batch_size=1)
        result = await client.wait_for_task(task["task_id"], timeout=600)
        print(result)


asyncio.run(main())

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

常用接口

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

方法	用途
client.health()	存活检查
client.images.generate(prompt, **fields)	文生图，使用后端要求的表单编码
client.images.transform(source_path, prompt, **fields)	图生图
client.novels.generate(description, **fields)	提交小说生成任务
client.novels.continue_from(file_url, **fields)	从公网 TXT / Markdown 文件续写
client.audio.list_voices() / generate(text, **fields)	TTS v1 音色列表与语音生成
client.audio.upload_voice(spk_id, prompt_text, path)	上传音色样本及对应文本
client.tts.list_speakers() / generate(text, spk_id, **fields)	TTS v2 音色列表与语音生成
client.tts.upload_speaker(spk_id, spk_name, path)	上传 TTS v2 音色，名称可传 None
client.tasks.get(task_id)	查询该 task_id 最新记录
client.tasks.get_by_uuid(uuid) / get_by_bid(bid)	按内部 ID / 业务编号查询
client.tasks.batch([task_id, ...])	批量查询，发送裸 JSON 数组
client.tasks.list(page=1, page_size=20, **filters)	分页任务列表
client.tasks.retry(task_id)	明确要求后端重发任务
client.tasks.sync_status(task_id)	根据已有结果修复后端任务状态
client.tasks.events([task_id, ...])	逐条读取 SSE 事件，异步使用 async for
client.wait_for_task(task_id, timeout=600)	等待任务结束
client.files.upload(path, permanent=False)	上传本地文件
client.files.output_url(filename, permanent=False)	构造文件 URL，本地操作，无需 await

**fields 透传新增参数，省略 None，默认值由后端决定。例如通用任务字段 task_id、bid、app_id、notify_url，以及水印中间件字段 is_watermark。实际接受的字段以具体接口为准；小说续写的 task_id 由服务端生成。响应保留原始字典及新增字段。

上传返回的 file_url 是文件名，不是完整 URL：

with Client() as client:
    uploaded = client.files.upload("input.png")
    image_url = client.files.output_url(uploaded["file_url"])
    task = client.images.transform(image_url, prompt="将这张照片转为水彩风格")

image_url 必须能被 backend 访问。永久文件地址可能受 apikey 保护，不能假设下游能匿名读取。上传方法关闭自己打开的文件；直接用 request(files=...) 时由调用方管理句柄。

用 OpenAPI JSON 跟进 API 更新

推荐 便利方法 + 固定版本的 OpenAPI 快照。backend 是 FastAPI，默认文档在 /openapi.json，Swagger UI 在 /docs。SDK 读取 JSON 契约，不解析 Swagger 网页。

使用包内接口配置

无需运行 backend 或访问 YApi，即可加载随包安装的接口配置。它依据 YApi 列表和当前 backend 源码核对，使用 yapi_接口ID 作为稳定 operationId；参数按后端签名修正，认证在 OpenAPI security 中声明，不包含文档中的密钥和示例值。

from foya_aiapi import Client, load_catalog

with Client(openapi=load_catalog()) as client:
    voices = client.call("yapi_2526")
    task = client.call_endpoint("POST", "/api/public/generate/audio", data={"text": "你好，世界"})
    print(voices, task)

foya-aiapi operations
foya-aiapi export-catalog --output openapi.catalog.json

call_endpoint() 按方法和路径模板定位已加载的操作，不需要知道 operationId。包内目录是经过核对的 27 条接口子集，其他常用便利方法可独立使用。缺失的 8 条路由（包含 4 条 webhook 配置接口）只记入覆盖报告；配置文件不能让后端凭空支持旧接口，也不会将旧路径自动改写为相似接口。

下载和查看

foya-aiapi fetch-schema --base-url http://localhost:8089 --output openapi.json
foya-aiapi operations openapi.json

CLI 同样读取认证环境变量。文档路径可通过 --path /your/openapi.json 指定。下载先检查版本和操作目录结构，再原子替换目标文件，失败保留原文件；这不是完整 OpenAPI 规范校验。

加载后调用

from foya_aiapi import Client

with Client(openapi="openapi.json") as client:
    task = client.call_endpoint(
        "POST",
        "/api/public/generate/text2image",
        data={"prompt": "森林中的小溪", "batch_size": 1},
    )
    print(task)

动态调用显式区分 path_params、params（query）、headers、cookies、json、data（form）和 files。不要把表单接口改用 JSON。client.call() / await client.call() 共用客户端认证、超时和错误处理。

也可传入字典或 OpenAPISchema 实例。client.load_openapi("new.json") 从本地更新，无网络请求；client.refresh_openapi() / await client.refresh_openapi() 显式从同一部署下载最新文档。新文档验证失败时保留旧 schema。

加载或刷新会完整替换当前配置。实际部署的 /openapi.json 仍可能缺少隐藏路由；它不会自动与包内目录合并。要统一维护完整公开契约，需要后端开放希望集成的路由并声明认证。

升级前比较

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

输出 added / removed / changed operationId 列表，包含引用模型变化。默认只输出结果；--fail-on-change 遇到任何差异返回退出码 2，可用于 CI 提醒检查，但不自动判断是否兼容。

新增接口和字段通常可更新快照后直接使用；路径、必填参数、业务含义的破坏性变更仍需调整调用代码。便利方法的路径变化需要升级 SDK。生产项目建议把快照与代码一起版本控制，测试后升级，不要每次请求自动拉取文档。

支持 OpenAPI 3.0 / 3.1 JSON、本地 $ref、JSON / URL-encoded / multipart 请求、简单路径参数和普通 query 数组。Swagger 2.0 需先转为 OpenAPI 3；YAML、deepObject 等复杂参数序列化不支持。操作解析或差异检查遇到需解析的外部/缺失引用会报错；未参与解析的正文和响应 schema 原样保留，加载时不会遍历验证其中所有引用。没有 operationId 时使用 METHOD /path，重复 ID 报错。此层不做完整 JSON Schema 校验，也不生成接口专属类型。

backend 部分接口隐藏于文档，中间件字段可能未声明，部分响应模型的 status 类型与真实返回不同，因此不强制按文档转换响应。固定便利方法不依赖 schema，文档未覆盖的接口也可直接请求：

with Client() as client:
    audio = client.audio.generate("你好，世界")

request_raw() 返回完整读取的 httpx.Response；request() 根据响应类型返回 JSON、文本、字节或 None。WebSocket 和通用流式下载尚未封装。

SSE 任务事件

使用 stream_events()、stream_call() 或 tasks.events() 逐条读取，普通 request() / call() 会等到响应完整结束。流式读取不自动重试或重连。超时默认沿用客户端设置，可通过 timeout=60 调整读取空闲超时。

from contextlib import closing
from foya_aiapi 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 foya_aiapi import AsyncClient, load_catalog


async def watch():
    async with AsyncClient(openapi=load_catalog()) as client:
        async with aclosing(client.stream_call("yapi_2530", params={"ids": ["task-1"]})) as events:
            async for event in events:
                print(event.data)

迭代结束、异常或显式 close() / aclose() 会释放连接；提前退出循环时使用上面的上下文管理器。ServerSentEvent 保留原始 data 文本、事件名称、id 和 retry。只有数据确实为 JSON 时才调用 event.json()，当前后端可能发送 Python 字典的文本表示。

当前任务队列会把状态 3（水印处理中）归入失败事件分支，因此 SSE 事件名称不能直接作为最终成功/失败依据；等待最终结果使用 wait_for_task()。SDK 保留服务端原始事件，不擅自解释或改写状态。

错误、重试和任务状态

from foya_aiapi import APIError, Client, TaskFailedError, TaskTimeoutError

with Client() as client:
    try:
        task = client.images.generate(prompt="远山", batch_size=1)
        result = client.wait_for_task(task["task_id"], timeout=600)
    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。默认异常文字不回显凭据或响应内容；body 可能含任务内容，仅在受控日志中检查。
• 连接及 HTTP 超时为 TransportError，坏 JSON 或无法解压的响应为 FoyaResponseError，SSE 的压缩响应读取失败也使用同一异常。无法解压的响应不自动重试，避免重复处理坏数据。以上错误及任务错误继承 FoyaError。本地配置错误和 SchemaError 继承 ValueError。
• 无请求体的 GET/HEAD/OPTIONS 默认最多重试 2 次，覆盖连接故障和 408/429/500/502/503/504，退避并读取 Retry-After，单次等待最多 30 秒。POST 等写请求不自动重试。
• 后端 503 + submission_uncertain=true 表示任务可能已入队，应先查询 task_id，不要直接重发。tasks.retry() 只在你明确调用时发送。
• 状态 0 等待、1 处理、3 水印处理中，继续等；2 成功；-1 抛出 TaskFailedError，或通过 raise_on_failure=False 返回失败记录。兼容数字字符串，未知状态继续等至超时。
• 轮询在总预算内继续处理临时连接故障及上述可重试 HTTP 状态，404 和认证错误立即抛出。后端查询约有 10 秒缓存。超时只停止等待，不取消远端任务。
• 同步 HTTP 超时按网络阶段计算，正在执行的请求可能超过轮询截止时间；异步等待使用总 deadline，可被调用方取消。
• 服务地址由 base_url 控制，schema 的 servers 不覆盖它；API 路径必须是 /...，不接受外部 URL，不自动跟随跳转，避免带着 apikey 跳往其他服务。下载外部结果请用单独的 HTTP 客户端。

开发和发布

python -m pip install -e '.[dev]'
pytest
ruff check .
ruff format --check .
python -m build
python -m twine check dist/*

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

发布步骤 · 架构及后端维护建议。已配置 GitHub Actions 测试矩阵及手动 TestPyPI / PyPI 发布，需关联你自己的仓库和 PyPI 项目。

参考：FastAPI 文档 URL、HTTPX、Python 打包指南、PyPI Trusted Publishers。