Python SDK

官方 Python SDK，封装 AI 应用开发平台的 OpenAPI（/v1/openapi/*）。提供同步 Client 与异步 AsyncClient，内置鉴权、重试、SSE 流式与 WebSocket 语音流。

Python 3.11+
同步 Client + 异步 AsyncClient，接口对称
请求/响应均为 Pydantic v2 强类型模型
类型化异常树（APIKeyExpiredError 等）—— catch 类型，不写字符串匹配
SSE 流式 —— 强类型事件 + match 模式匹配
HITL Resume —— ConfirmResponse / ChoiceResponse 等响应类型
语音 WebSocket 流式为可选扩展（aiadp[voice]）

安装

包发布在公司私有 Nexus 仓库，需指定私有源安装。

pip：

bash

pip install aiadp --index-url https://repo.i.huaxisy.com/repository/pypi-group/simple/

# 含 voice WebSocket 流式扩展
pip install "aiadp[voice]" --index-url https://repo.i.huaxisy.com/repository/pypi-group/simple/

uv：

bash

uv add aiadp --index-url https://repo.i.huaxisy.com/repository/pypi-group/simple/

依赖分层：

核心：chat / files / knowledge / workflow / apps / thirdparty + 非流式 voice
[voice] 扩展：voice WebSocket 流式（TTS / STT）

要求 Python 3.11+。

快速开始

python

from aiadp import Client
from aiadp.chat.models import SendMessagesRequest

with Client("sk-xxx", "https://runtime-api.invalid/v1/openapi") as client:
    res = client.chat.send_messages_blocking(
        SendMessagesRequest(app_id="<your-app-id>", user_id="user-1", query="你好"))
    print(res.output.content)

业务能力挂在 client 的属性命名空间，每个属性惰性创建并缓存：

属性	用途
`client.apps`	应用列表 / 详情 / 运行参数
`client.chat`	Agent 对话（send-messages、resume、stop、会话 / 消息历史）
`client.files`	Agent 工作空间文件操作
`client.knowledge`	知识库列表 / 详情 / 检索
`client.workflow`	工作流执行（blocking / streaming）
`client.thirdparty`	三方服务列表 / 工具列表 / 调用
`client.voice`	语音合成 / 识别 / 音色列表（含 WebSocket 流式）

Client 实现上下文管理协议，用 with 自动释放底层 httpx.Client。

配置

python

Client(api_key, base_url,
       timeout=30.0,
       stream_connect_timeout=10.0,
       max_retries=3,
       retry_base_backoff=1.0,
       user_agent="aiadp-python/x.y.z",
       http_client=None,    # 可选：自带 httpx.Client
       logger=None)         # 可选：实现 SDK Logger 协议的对象

api_key 须以 sk- 开头，base_url 须为 http(s)——构造期即校验，失败抛 InvalidAPIKeyError / InvalidBaseURLError。
重试覆盖网络错误与 502/503/504，4xx 不重试；退避为指数 + jitter。
SDK 自动注入 Authorization: Bearer <api_key>，调用方无需手动设置鉴权请求头。

异步

AsyncClient 与 Client 接口对称，方法均为协程。注意流式方法返回协程，需先 await 再进入 async with。

python

import asyncio
from aiadp import AsyncClient
from aiadp.chat.models import SendMessagesRequest

async def main():
    async with AsyncClient("sk-xxx", "https://runtime-api.invalid/v1/openapi") as client:
        async with await client.chat.send_messages_stream(
                SendMessagesRequest(app_id="a", user_id="u", query="你好")) as stream:
            async for ev in stream:
                ...

asyncio.run(main())

下文以同步 Client 为例，异步用法把方法 await、迭代换成 async for 即可。

应用 apps

python

def list(req: ListRequest | None = None) -> ListResult
def detail(req: DetailRequest) -> DetailResult
def parameters(req: ParametersRequest) -> ParametersResult

python

from aiadp.apps.models import ListRequest, DetailRequest, ParametersRequest

# 应用列表
result = client.apps.list(ListRequest(page=1, page_size=20))
for a in result.record:
    print(a.id, a.name, a.app_type)

# 应用详情：含完整 app_schema（模型、提示词、工具、知识库、记忆、交互配置等）
detail = client.apps.detail(DetailRequest(app_id=app_id))
print(detail.app_schema.model_info.model_id,
      detail.app_schema.interaction_config.chat_prologue.prologue)

# 运行参数：扁平化的交互配置（开场白、建议问题、语音 TTS/ASR、运行约束）
params = client.apps.parameters(ParametersRequest(app_id=app_id))
print(params.voice.tts.code, params.voice.asr.provider, params.runtime_setting.max_steps)

DetailResult.app_schema 是应用的完整配置快照，ParametersResult 是面向运行时的精简交互配置；DetailRequest / ParametersRequest 都支持 version 指定历史版本。

对话 chat

python

def send_messages_blocking(req: SendMessagesRequest) -> SendMessagesResult
def send_messages_stream(req: SendMessagesRequest) -> ChatStream
def resume_blocking(req: ResumeRequest) -> SendMessagesResult
def resume_stream(req: ResumeRequest) -> ChatStream
def stop(task_id: str) -> None
def conversations(req: ConversationsRequest) -> ConversationsResult
def chat_history(req: HistoryRequest) -> list[HistoryItem]

请求与响应字段：

python

class SendMessagesRequest:
    app_id: str
    user_id: str
    query: str
    username: str = ""
    conversation_id: str = ""
    files: list[FileItem] = []
    input: dict[str, Any] | None = None
    with_tts: bool = False
    tts_format: str = ""
    select_model_id: str = ""

class SendMessagesResult:
    conversation_id: str
    message_id: str
    task_id: str
    status: str
    duration: int
    output: BlockingOutput   # content / finish_reason / tokens

阻塞式：

python

from aiadp.chat.models import SendMessagesRequest

res = client.chat.send_messages_blocking(
    SendMessagesRequest(app_id=app_id, user_id="user-1", query="你好"))
print(res.output.content)

response_mode 由 SDK 自动注入，请勿自行设置。

流式响应

流式对话返回事件迭代器，事件为强类型 Pydantic 模型，可用 match 分发：

python

from aiadp.events.chat import Chunk, Completed, ThinkingStart, Thinking, ThinkingEnd
from aiadp.events.interrupt import Interrupt

with client.chat.send_messages_stream(req) as stream:
    for ev in stream:
        match ev:
            case ThinkingStart(): print("<think>", end="")
            case Thinking():      print(ev.data.content, end="")
            case ThinkingEnd():   print("</think>", end="")
            case Chunk():         print(ev.data.content, end="")
            case Interrupt():     handle_hitl(ev)   # HITL，见下方 Resume
            case Completed():     pass

ChatStream 同时实现上下文管理协议与迭代协议，with 退出时自动关闭底层连接。事件还覆盖工具调用（ToolCall / ToolStart / ToolResult / ToolError）、子 Agent（SubAgentStarted / SubAgentCompleted / SubAgentFailed）、沙箱生命周期、Token 用量（TokenUsage）等，未知事件归为 Unknown。

HITL 中断与恢复

对话流出现 Interrupt 事件时，会带回 data.checkpoint_id 与 data.interrupts[].interrupt_id。按 interrupt ID 给出应答后调用 resume_stream / resume_blocking：

python

from aiadp.chat.models import ResumeRequest
from aiadp.events.interrupt import Interrupt
from aiadp.events.resume import ConfirmResponse

checkpoint_id = conversation_id = interrupt_id = ""
with client.chat.send_messages_stream(req) as stream:
    for ev in stream:
        if isinstance(ev, Interrupt):
            conversation_id = ev.common.conversation_id
            checkpoint_id = ev.data.checkpoint_id
            interrupt_id = ev.data.interrupts[0].interrupt_id
            break

resume = ResumeRequest(
    app_id=app_id, user_id="user-1",
    conversation_id=conversation_id, checkpoint_id=checkpoint_id,
    responses={interrupt_id: ConfirmResponse(confirm=True)})

with client.chat.resume_stream(resume) as stream:
    for ev in stream:
        ...

responses 的值是 ResumeResponse 的实现类型，对应不同交互：ConfirmResponse / TextResponse / ChoiceResponse / ChoicesResponse / FormDataResponse / EditedContentResponse。

停止生成

从流中 Start 事件拿到 task_id 后随时可停：

python

client.chat.stop(task_id)

会话与消息历史

python

from aiadp.chat.models import ConversationsRequest, HistoryRequest

# 会话列表（按应用，可选按用户过滤、分页）
conv = client.chat.conversations(
    ConversationsRequest(app_id=app_id, user_id="demo-user", page=1, page_size=20))
for c in conv.record:
    print(c.id, c.name)

# 某会话的消息历史
items = client.chat.chat_history(
    HistoryRequest(app_id=app_id, conversation_id=conversation_id, page=1, page_size=20))
for m in items:
    print(m.query, "->", m.content)

HistoryItem 含 query / content / thinking / tool_calls / items（子 Agent 与工具调用明细）等完整执行轨迹。

文件 files

python

def upload(req: UploadRequest) -> FileObject
def list_directory(req: ListRequest) -> ListResult
def list_tree(req: TreeRequest) -> TreeResult
def edit(req: EditRequest) -> None
def copy(req: CopyRequest) -> FileObject
def move(req: MoveRequest) -> FileObject
def delete(req: DeleteRequest) -> None
def mkdir(req: MkdirRequest) -> FileObject
def search(req: SearchRequest) -> SearchResult

所有文件接口都需要 user_id 定位 Agent 工作空间。UploadRequest.file 接受 bytes、pathlib.Path 或任意二进制流（BinaryIO）：

python

from pathlib import Path
from aiadp.files.models import UploadRequest, ListRequest

# bytes
obj = client.files.upload(UploadRequest(
    user_id="demo-user", path="/data/hello.txt",
    filename="hello.txt", file=b"hello world"))
print(obj.path, obj.url)

# Path
client.files.upload(UploadRequest(
    user_id="demo-user", filename="hello.txt", file=Path("./hello.txt")))

# 列目录
ls = client.files.list_directory(ListRequest(user_id="demo-user", path="/data"))
for f in ls.files:
    print(f.path, f.size)

move 对应网关的 rename 接口（old_path → new_path），同目录即重命名、跨目录即移动。list_tree 递归列出文件树，max_depth 为 0 表示不限深度。

知识库 knowledge

python

def list(req: ListRequest | None = None) -> ListResult
def detail(knowledge_id: str) -> Detail
def retrieve(knowledge_id: str, req: RetrieveRequest) -> RetrieveResult

python

from aiadp.knowledge.models import ListRequest, RetrieveRequest

result = client.knowledge.list(ListRequest(page=1, page_size=10))
first = result.items[0]

detail = client.knowledge.detail(first.id)
print(detail.retriever_config.retriever_type, detail.retriever_config.top_k)

res = client.knowledge.retrieve(first.id,
    RetrieveRequest(query="高血压患者能不能吃布洛芬?"))
for r in res.records:
    print(f"score={r.score:.3f} {r.content}")

可选地用 RetrieveRequest.retrieval_model（RetrievalModelOverride）覆盖单次检索的 top_k / score_threshold 等参数。

注意类型差异：详情响应里 score_threshold_enabled 是 int（0/1），而检索覆盖参数里的同名字段是 bool。

工作流 workflow

python

def invoke_blocking(req: InvokeRequest) -> InvokeResult
def invoke_stream(req: InvokeRequest) -> WorkflowStream

python

from aiadp.workflow.models import InvokeRequest
from aiadp.events.workflow import WorkflowMessage, WorkflowDone, WorkflowError

# 阻塞式
res = client.workflow.invoke_blocking(
    InvokeRequest(workflow_id=workflow_id, input={"input": "华西"}))
print(res.final_output, res.token)

# 流式
with client.workflow.invoke_stream(
        InvokeRequest(workflow_id=workflow_id, input={"input": "华西"})) as stream:
    for ev in stream:
        match ev:
            case WorkflowMessage(): print(ev.content, end="")
            case WorkflowDone():    print(f"\n[done] status={ev.status}")
            case WorkflowError():   print(f"\n[error] {ev.error_message}")

WorkflowStream 产出强类型工作流事件（WorkflowMessage / WorkflowNodeStart / WorkflowNodeEnd / WorkflowDone / WorkflowError / WorkflowInterrupt 等）。

三方服务 thirdparty

python

def list_services() -> list[Service]
def list_tools(service_slug: str) -> ToolsResult
def invoke(service_slug: str, tool_slug: str, req: InvokeRequest) -> InvokeResult

python

from aiadp.thirdparty.models import InvokeRequest

services = client.thirdparty.list_services()
for s in services:
    print(s.name, "configured=", s.is_configured)

tools = client.thirdparty.list_tools(services[0].name)
print(tools.service.name, len(tools.tools))

res = client.thirdparty.invoke(
    services[0].name, tools.tools[0].name, InvokeRequest(inputs={}))
print(res.output)   # 不同工具结构不同

不同工具返回结构不同，InvokeResult.output 是 Any，按需自行处理。

语音 voice

python

# 非流式（核心包即可）
def synthesize(req: SynthesizeRequest) -> TTSResult
def recognize(req: RecognizeRequest) -> STTResult
def list_voices(provider: str) -> list[VoiceOption]

# WebSocket 流式（需 aiadp[voice]）
def tts_stream(req: SynthesizeRequest) -> TTSStream
def stt_stream(req: RecognizeRequest) -> STTStream

音频在请求/响应中以 base64 字符串传输（audio_data 字段）。语音合成：

python

import base64
from aiadp.voice.models import SynthesizeRequest

voices = client.voice.list_voices("aliyun")
for v in voices:
    print(v.name, v.value, v.language, v.scene)

res = client.voice.synthesize(SynthesizeRequest(
    provider="aliyun", text="你好，欢迎使用语音合成。",
    voice_code=voices[0].value, format="mp3"))
audio = base64.b64decode(res.audio_data)
print(res.format, res.duration_ms, len(audio))

语音识别（非流式）：

python

from aiadp.voice.models import RecognizeRequest

res = client.voice.recognize(RecognizeRequest(
    provider="aliyun",
    audio_data=base64.b64encode(data).decode(),
    format="pcm", sample_rate=16000))
print(res.text, res.confidence)

WebSocket 流式

流式语音走 WebSocket，需安装 aiadp[voice] 扩展，否则抛 MissingDependencyError。

TTS：迭代 tts_stream 得到二进制音频块，文本 final 帧标记结束。
STT：双向流，send_audio(data) 推送音频帧，迭代读取 STTStreamResult（含 is_partial / is_final）；推送完毕调用 close_send()，迭代自然结束。

python

# 流式 TTS
with client.voice.tts_stream(req) as stream:
    for chunk in stream:
        out.write(chunk)

两个流都实现上下文管理协议，with 退出时自动释放连接。STT 流式时 RecognizeRequest 的非音频字段作为首帧配置，不要带 audio_data。

错误处理

所有 API 错误派生自 APIError；鉴权类错误（如密钥过期、工作区禁用）有专门子类，可精准捕获：

python

from aiadp import APIError, APIKeyExpiredError, TransportError

try:
    client.chat.send_messages_blocking(req)
except APIKeyExpiredError:
    regenerate_api_key()
except APIError as e:
    print(e.http_status, e.code, e.message, e.trace_id)
except TransportError as e:
    # 网络 IO / 编解码错误
    ...

异常类	触发条件
`InvalidAPIKeyError` / `InvalidBaseURLError`	构造期凭证校验失败
`APIKeyExpiredError` / `APIKeyDisabledError` / `InvalidAPIKeyTokenError`	401 + 对应 message
`MissingAPIKeyError` / `InvalidAuthFormatError`	401 + 对应 message
`InvalidPathError`	403 + 对应 message
`AuthServiceUnavailableError` / `AuthServiceInitError`	503 / 500 + 特定 message
`APIError` 兜底	其他 4xx/5xx 或业务 `code != 200`
`TransportError`	网络 IO / 编解码
`MissingDependencyError`	用流式语音但未安装 `aiadp[voice]`

错误码与文案以网关/服务实际返回为准，详见错误处理。

示例

完整可运行示例见 SDK 仓库的 examples/，每个文件覆盖一个能力命名空间：

示例	功能
`apps_list`	应用列表 / 详情 / 参数
`chat_blocking` / `chat_streaming`	对话两种模式
`chat_resume`	HITL 中断 + resume
`chat_conversations`	会话列表 + 消息历史
`async_chat`	异步对话
`files_ops`	上传 / 列目录 / 编辑 / 搜索 / 删除
`knowledge_retrieve`	list / detail / retrieve
`workflow_invoke`	blocking + streaming
`thirdparty_invoke`	services / tools / invoke
`voice_tts` / `voice_stt`	语音合成 / 识别（含流式）

example 从环境变量读取凭证。复制 .env.example 为 .env 填入真实值后加载并运行：

bash

cp .env.example .env          # 填入 AIADP_API_KEY / BASE_URL 等
set -a && source .env && set +a
uv run python examples/chat_blocking.py

各 example 所需变量见 .env.example：chat / apps 类需 AIADP_APP_ID，workflow_invoke 需 AIADP_WORKFLOW_ID，voice_stt 需 AIADP_AUDIO_FILE。

Python SDK ​

安装 ​

快速开始 ​

配置 ​

异步 ​

应用 apps ​

对话 chat ​

流式响应 ​

HITL 中断与恢复 ​

停止生成 ​

会话与消息历史 ​

文件 files ​

知识库 knowledge ​

工作流 workflow ​

三方服务 thirdparty ​

语音 voice ​

WebSocket 流式 ​

错误处理 ​

示例 ​

Python SDK

安装

快速开始

配置

异步

应用 apps

对话 chat

流式响应

HITL 中断与恢复

停止生成

会话与消息历史

文件 files

知识库 knowledge

工作流 workflow

三方服务 thirdparty

语音 voice

WebSocket 流式

错误处理

示例