ase-sdk-python

ASE 服务的同步 Python SDK，支持 AIpaas 和 AIaaS 两种请求格式，以及 HTTP 和 WebSocket 调用。

实现参考 iflytek/ase-sdk-go，固定基线为 cf7591bcb340a30c78a0036c30f8a572ee84b9e0。Python 发行包名是 ase-sdk-python，导入名是 ase_sdk。

安装

需要 Python 3.10 或更新版本：

python -m pip install ase-sdk-python

在源码目录中开发：

python3 -m venv .venv
source .venv/bin/activate
python -m pip install -e '.[dev]'
python -m unittest discover -s tests -v

凭证与服务地址

以下示例从环境变量读取配置：

export ASE_APP_ID='你的 app_id'
export ASE_API_KEY='你的 api_key'
export ASE_API_SECRET='你的 api_secret'
export ASE_HOST='实际服务域名或主机:端口'
export ASE_URI='/实际服务路径'

host 不包含协议或路径，uri 以 / 开头。默认 tls=True，HTTP 使用 HTTPS，WebSocket 使用 WSS；需要访问明确提供明文协议的本地服务时，可显式设置 tls=False。

地址必须使用 ASCII：国际化域名先转换为 IDNA，非 ASCII 路径先进行百分号编码；uri 不包含查询参数或片段。客户端会先规范化地址，再对实际发送的主机名和路径签名。

app_id 是请求体中的应用标识；api_key 用于标识签名凭证；api_secret 仅用于本地计算 HMAC，不作为请求字段发送。SDK 会在缺省时填充 header.app_id（AIpaas）或 common.app_id（AIaaS）。若请求显式提供了与客户端不一致的 app_id，会抛出 ValueError。

HTTP 单次调用

AIpaas

import base64
import os

from ase_sdk import Client

request = {
    "header": {},
    "parameter": {},
    "payload": {
        "input": {  # 按服务协议替换参数名和内容
            "text": base64.b64encode("你好".encode("utf-8")).decode("ascii"),
            "status": 3,
        }
    },
}

with Client(
    app_id=os.environ["ASE_APP_ID"],
    api_key=os.environ["ASE_API_KEY"],
    api_secret=os.environ["ASE_API_SECRET"],
    host=os.environ["ASE_HOST"],
    uri=os.environ["ASE_URI"],
    mode="aipaas",  # 默认值
) as client:
    response = client.once(request)
    print(response)

AIpaas 单次 HTTP 调用默认补充缺失的 header.status=3。显式传入的状态会保留。请求的其他业务参数遵循目标服务的协议。

AIaaS

import os

from ase_sdk import Client

request = {
    "common": {},
    "business": {},  # 按服务协议填写业务参数
    "data": {},      # 按服务协议填写数据和状态
}

with Client(
    app_id=os.environ["ASE_APP_ID"],
    api_key=os.environ["ASE_API_KEY"],
    api_secret=os.environ["ASE_API_SECRET"],
    host=os.environ["ASE_HOST"],
    uri=os.environ["ASE_URI"],
    mode="aiaas",
) as client:
    response = client.once(request)
    print(response)

mode="aiaas" 时，once() 使用 AIaaS 请求格式和 HTTP 鉴权。也可以调用 once_aiaas(request)，为本次 HTTP 调用显式选用 AIaaS 格式和鉴权。AIaaS 不自动补充状态字段。

SDK 不会自动对媒体内容进行 Base64 编码或解码，也不推断音频格式、采样率、帧大小或业务字段。示例中的编码由调用方完成；如果目标服务接收明文，应直接传明文。

WebSocket 调用

import os

from ase_sdk import Client

with Client(
    app_id=os.environ["ASE_APP_ID"],
    api_key=os.environ["ASE_API_KEY"],
    api_secret=os.environ["ASE_API_SECRET"],
    host=os.environ["ASE_HOST"],
    uri=os.environ["ASE_URI"],
    mode="aipaas",
) as client:
    client.connect()
    client.send({
        "header": {"status": 2},
        "parameter": {},
        "payload": {},  # 按服务协议填入单帧请求数据
    })
    while True:
        response = client.receive()
        print(response)
        header = response.get("header", {})
        if header.get("code", 0) != 0 or header.get("status") == 2:
            break

上述结束条件是常见的 AIpaas 响应约定；请使用实际服务定义的状态和错误字段。AIaaS 通过 mode="aiaas" 配合 send()，或直接使用 send_aiaas()，发送 common / business / data 格式。AIaaS 响应通常需要按服务协议检查 code、data.status 等字段。

两种模式的 WebSocket 握手都使用与 Go SDK 一致的 GET 签名 URL；AIaaS 的 HTTP Digest 鉴权不用于 WebSocket。

with 负责退出时清理资源，不主动建立 WebSocket。connect() 可显式建立连接；send() 和 receive() 也会在需要时首次连接。WebSocket 状态由调用方提供，SDK 不自动添加首帧或末帧标志。

同一个客户端支持一个发送线程与一个接收线程并发，适合边上传音频边读取结果。不要在同一连接上创建多个接收线程；多个独立会话应使用各自的 Client。SDK 不自动重连或重放 WebSocket 帧。

源码包的 examples/ 目录提供 http_aipaas.py、http_aiaas.py 和 websocket_stream.py，支持从 JSON 文件读取目标服务的真实请求；流式示例使用 JSONL 文件逐帧发送，同时接收结果。

API

Client(
    app_id, api_key, api_secret, host, uri,
    *, mode="aipaas", tls=True,
    timeout=30, retries=0, retry_backoff=0.5,
    handshake_timeout=10, read_timeout=30, write_timeout=30, close_timeout=5,
    max_size=16 * 1024 * 1024, connect_headers=None,
)

参数	用途
mode	aipaas 或 aiaas，决定默认请求格式和 HTTP 鉴权
tls	是否使用 HTTPS / WSS，默认开启并校验证书
timeout	HTTP 请求超时，单位秒
retries	HTTP 请求失败后的最大重试次数，默认 0
retry_backoff	HTTP 重试的初始退避秒数
handshake_timeout	WebSocket 连接握手超时，单位秒
read_timeout	等待单条 WebSocket 消息的超时，单位秒
write_timeout	发送单条 WebSocket 消息的超时，单位秒；超时后中断连接
close_timeout	WebSocket 关闭超时，单位秒；超时后中断底层连接
max_size	WebSocket 接收单条消息的最大字节数，默认 16 MiB
connect_headers	WebSocket 握手时附加的请求头，不能覆盖鉴权字段或协议握手字段

超时参数可设为 None 以禁用对应超时，max_size=None 可禁用接收大小限制。HTTP 会保留环境代理配置，但不会读取 .netrc 中的 Basic 鉴权覆盖 ASE 签名；WebSocket 与 Go SDK 一样直接连接服务。

方法	返回值 / 行为
once(request)	按客户端模式发送 HTTP POST，返回解析后的 JSON 字典
once_raw(request)	同上，返回原始响应 bytes
once_aiaas(request)	显式使用 AIaaS HTTP POST，返回 JSON 字典
once_aiaas_raw(request)	显式使用 AIaaS HTTP POST，返回 bytes
connect()	建立 WebSocket 连接
send(request)	按客户端模式发送一条 WebSocket JSON 消息
send_aiaas(request)	显式发送一条 AIaaS WebSocket JSON 消息
receive()	读取一条 WebSocket 消息并解析为 JSON 字典
receive_raw()	读取一条 WebSocket 消息，返回 bytes
close() / destroy()	关闭连接并释放客户端资源

字典是最直接的请求表达方式，也可使用导出的 Request / AIPAASRequest、AIaaSRequest / AIAASRequest、RequestHeader、TextPayload、AudioPayload、ImagePayload 和 DataStatus 模型。状态值与 Go SDK 一致：首帧 0、中间帧 1、末帧 2、单次请求 3。

错误与重试

• HTTPError：HTTP 非成功响应；status_code 和 body 提供状态码及原始响应正文。
• TransportError：网络、连接或超时等传输错误。
• ProtocolError：响应不是预期的 JSON 对象等协议错误。
• ClientClosedError：客户端已经关闭。
• 上述 SDK 异常均继承 ASEError；无效参数可抛出 ValueError 或 TypeError。

服务业务错误码不为零时，SDK 仍然返回响应，由调用方按实际响应协议判断。原始响应可能包含业务数据，异常的字符串表示不会自动输出签名 URL 或响应正文。

默认不重试 HTTP POST。显式增加 retries 可能造成服务重复执行或重复计费，只有在调用方能够接受或处理重复执行时才开启。WebSocket 不自动重试。

鉴权与 Go SDK 的对应关系

AIpaas HTTP，以及两种模式的 WebSocket

待签名字符串使用换行符 \n，末尾不额外增加换行：

host: {host}
date: {date}
{method} {uri} HTTP/1.1

HTTP 的 method 为 POST，WebSocket 为 GET。先计算 Base64(HMAC-SHA256(api_secret, 签名原串))，再组装：

api_key="{api_key}", algorithm="hmac-sha256", headers="host date request-line", signature="{signature}"

将整个鉴权描述再次 Base64 编码作为 authorization，连同 date 和 host 一起作为 URL 查询参数进行 URL 编码。

AIaaS HTTP

先对实际发送的 JSON 请求体字节计算：

Digest: SHA-256={Base64(SHA256(body_bytes))}

待签名字符串为：

host: {host}
date: {date}
POST {uri} HTTP/1.1
digest: SHA-256={body_sha256_base64}

计算 Base64(HMAC-SHA256(api_secret, 签名原串)) 后，在请求头中发送：

Authorization: hmac api_key="{api_key}", algorithm="hmac-sha256", headers="host date request-line digest", signature="{signature}"
Host: {host}
Date: {date}
Digest: SHA-256={body_sha256_base64}
Content-Type: application/json

该 Authorization 值包含 hmac 前缀，不做第二层 Base64 编码。SDK 对请求体只序列化一次，签名和发送使用相同字节，避免 JSON 空白、字段顺序或字符编码差异引发验签失败。日期采用 UTC 时间；服务端通常要求客户端时钟偏差不超过 300 秒。

构建和发布

发布流程参考同工作区 AIGES 的 pyaiges，产出源码包和 wheel，并运行严格元数据检查：

python -m unittest discover -s tests -v
PYTHON=.venv/bin/python ./publish_pypi.sh --build-only

脚本要求指定解释器中已安装 build 和 twine，不会自动修改全局 Python 环境。验证通过的产物会复制到 dist/。上传仅使用本次构建的两个产物，不会上传 dist/ 中的历史版本。

凭证可配置在 ~/.pypirc，也可由 PYPI_TOKEN 或 TWINE_USERNAME=__token__ / TWINE_PASSWORD 环境变量提供。令牌通过环境变量传给 Twine，不放入命令行参数。配置好凭证后执行：

# 正式 PyPI
PYTHON=.venv/bin/python ./publish_pypi.sh

# TestPyPI 使用独立的 TestPyPI 凭证
PYTHON=.venv/bin/python ./publish_pypi.sh --test

测试覆盖固定 Go 鉴权向量以及本地 HTTP / WebSocket 服务的真实传输；接入实际引擎时还需使用目标服务的业务参数和凭证验证。PyPI 同一版本的发行文件不可覆盖；后续发布需要更新 pyproject.toml、SDK 版本和 CHANGELOG.md。

许可证

Apache License 2.0，完整许可文本随发行包的 LICENSE 文件提供。