AceData 平台 Python SDK 接入教程

acedatacloud 是 Ace Data Cloud 官方 Python SDK，把 api.acedata.cloud 上所有服务封装成类型化的 client.openai.chat.completions.create(...)、client.images.generate(...)、client.search.google(...) 等方法，同时提供同步和异步两套客户端。

底层基于 httpx，支持 SSE 流式、自动重试、类型化异常和 pydantic 类型校验。

源码与包地址：

• SDK 仓库：https://github.com/AceDataCloud/SDK
• PyPI：https://pypi.org/project/acedatacloud/

安装

pip install acedatacloud# 或 uv add / poetry add

如果需要 X402 链上付费（无 API Token 路径），再装一个：

pip install acedatacloud-x402

干净 venv 的版本检查输出：

$ python -c "import importlib.metadata as m; print(m.version('acedatacloud'))"2026.4.26.1$ python -c "from acedatacloud import AceDataCloud, AsyncAceDataCloud; print('ok')"ok

结果说明：

• 包版本是 2026.4.26.1（CalVer，2026 年 4 月 26 日修订）。
• AceDataCloud 是同步客户端，AsyncAceDataCloud 是 asyncio 异步客户端。
• 本 SDK 不依赖 pydantic，响应体统一返回 dict。这一点和 openai-python 不同，迁移时需要注意。

准备 API Token

参考 SDK 总览 - 申请 API Token 取到 token，然后在 shell 里 export：

export ACEDATACLOUD_API_TOKEN={token}

构造客户端时不传 api_token，SDK 会自动读取 ACEDATACLOUD_API_TOKEN 环境变量。如果你的环境里已经存了 ACEDATACLOUD_API_KEY（项目仓库约定），请显式传入：AceDataCloud(api_token=os.environ["ACEDATACLOUD_API_KEY"])。

示例 1：chat.completions（同步）

import os, time, jsonfrom acedatacloud import AceDataCloudclient = AceDataCloud(api_token=os.environ["ACEDATACLOUD_API_KEY"])t0 = time.time()res = client.openai.chat.completions.create(    model="gpt-4o-mini",    messages=[{"role": "user", "content": "Reply with exactly: ADC_PY_SDK_OK"}],    max_tokens=20,    temperature=0,)print("elapsed_ms", int((time.time() - t0) * 1000))print("id", res["id"])print("model", res["model"])print("content", res["choices"][0]["message"]["content"])print("usage", json.dumps({k: v for k, v in res["usage"].items()if k in ("prompt_tokens","completion_tokens","total_tokens")}))

程序运行结果：

elapsed_ms 2963id chatcmpl-DldFdnIlhSUXINpupgsUmZL78MnBumodel gpt-4o-minicontent ADC_PY_SDK_OKusage {"prompt_tokens": 17, "completion_tokens": 7, "total_tokens": 24}

结果说明：

• id 是上游响应 ID，可以在 使用历史 里搜到。
• content ADC_PY_SDK_OK 是模型真实返回的固定标识。
• res["usage"] 返回 dict，不是 pydantic model；一次调用大约消耗 24 token。

示例 2：chat.completions（SSE 流式）

stream=True 时 create 返回一个普通生成器，每次 yield 一个解析好的 chunk dict。

import os, timefrom acedatacloud import AceDataCloudclient = AceDataCloud(api_token=os.environ["ACEDATACLOUD_API_KEY"])t0 = time.time()first_chunk_ms = Nonechunks = 0collected = []for chunk in client.openai.chat.completions.create(    model="gpt-4o-mini",    messages=[{"role": "user", "content": "Count from 1 to 5, separated by single spaces, no extra text."}],    max_tokens=30,    temperature=0,    stream=True,):if first_chunk_ms isNone:        first_chunk_ms = int((time.time() - t0) * 1000)    chunks += 1    delta = (chunk.get("choices") or [{}])[0].get("delta", {}).get("content")if delta:        collected.append(delta)print("total_elapsed_ms", int((time.time() - t0) * 1000))print("first_chunk_ms", first_chunk_ms)print("chunks", chunks)print("collected", "".join(collected).strip())

程序运行结果：

total_elapsed_ms 2111first_chunk_ms 2104chunks 12collected 1 2 3 4 5

结果说明：

• 首帧延迟 2104 ms，后续 11 帧只用了 7 ms 就到齐——一旦上游开始流，本地是顺手就能消费。
• chunk 是普通 dict，按 OpenAI SSE 格式逐层 .get() 安全取值即可。
• 实际生产里推荐边 yield 边推 SSE 给前端，整体首字延迟接近 2 秒。

示例 3：AsyncAceDataCloud（异步）

AsyncAceDataCloud 的 API 跟同步版完全对称，只是所有 IO 方法返回 coroutine。适合 FastAPI / aiohttp / asyncio 服务。

import os, asyncio, timefrom acedatacloud import AsyncAceDataCloudasyncdefmain():    client = AsyncAceDataCloud(api_token=os.environ["ACEDATACLOUD_API_KEY"])    t0 = time.time()    res = await client.openai.chat.completions.create(        model="gpt-4o-mini",        messages=[{"role": "user", "content": "Reply with exactly: ADC_PY_ASYNC_OK"}],        max_tokens=20,        temperature=0,    )    print("elapsed_ms", int((time.time() - t0) * 1000))    print("id", res["id"])    print("content", res["choices"][0]["message"]["content"])await client.close()asyncio.run(main())

程序运行结果：

elapsed_ms 2392id chatcmpl-DldFwRlrgtYDBpIr0T55aDQI4GlbFcontent ADC_PY_ASYNC_OK

结果说明：

• 异步版本和同步版本走的是同一条 HTTP 路径，只是连接池实现不同（httpx.AsyncClient）。
• 退出时显式 await client.close() 把连接池关掉；长生命周期服务里只需要在进程退出前关一次。
• 单次延迟跟同步差不多，并发场景下异步才显出优势——一个 event loop 可以同时跑几十上百个 inflight 请求。

示例 4：images.generate（NanoBanana）

NanoBanana 是同步生成的图像服务，**不要传 wait**——SDK 调用就会一直阻塞到上游 200 返回。

import os, timefrom acedatacloud import AceDataCloudclient = AceDataCloud(api_token=os.environ["ACEDATACLOUD_API_KEY"])t0 = time.time()res = client.images.generate(    provider="nano-banana",    model="nano-banana",    prompt="A minimalist logo of a yellow banana on a white background, flat design",)print("elapsed_ms", int((time.time() - t0) * 1000))print("task_id", res.get("task_id"))print("trace_id", res.get("trace_id"))data = res.get("data") or []if data:    print("image_url", data[0].get("image_url"))

程序运行结果：

elapsed_ms 18977task_id 9e71f40f-1579-480d-adf4-07a95450904ftrace_id 5aa21d7f-af84-48e6-9ce0-9c6c36c8e5d9image_url https://platform.cdn.acedata.cloud/nanobanana/884e92df-a497-44e0-9681-35c7a00e0a6c.png

结果说明：

• image_url 是 CDN 上的稳定地址，可以直接下载或嵌入网页。
• 18.9 秒里几乎都是上游模型推理；本地 SDK 开销只有几毫秒。
• 对于 Midjourney、Sora、Veo、Suno 这种真正异步的任务，需要用 wait=True 或手动 TaskHandle.wait() 轮询，详见 SDK 任务轮询与流式响应。

示例 5：类型化错误处理

import osfrom acedatacloud import AceDataCloudfrom acedatacloud import AuthenticationError, RateLimitError, ValidationErrorbad = AceDataCloud(api_token="definitely-not-a-real-token")try:    bad.openai.chat.completions.create(        model="gpt-4o-mini",        messages=[{"role": "user", "content": "hi"}],        max_tokens=5,    )except AuthenticationError as err:    print("err_class", type(err).__name__)    print("status", err.status_code)    print("code", err.code)except RateLimitError as err:# 429，SDK 已经自动指数退避重试 2 次仍失败才会抛    print("rate limited:", err.code)except ValidationError as err:# 400，例如缺字段、模型名不存在    print("bad request:", err.code, err.message)

异常层级跟 TypeScript 一致：AuthenticationError (401)、TokenMismatchError (token 与服务不匹配)、InsufficientBalanceError (余额不足)、ResourceDisabledError (服务被禁用)、ValidationError (400)、RateLimitError (429)、ModerationError (403 内容审核)、APIError (兜底)、TimeoutError（超时）、TransportError（网络层）。

配置选项

from acedatacloud import AceDataCloudclient = AceDataCloud(# 必填之一：显式 token 或环境变量 ACEDATACLOUD_API_TOKEN    api_token="...",# 平台 API 根地址，默认 https://api.acedata.cloud    base_url="https://api.acedata.cloud",# 部分服务（如 dashboard 元数据）走 platform 域名    platform_base_url="https://platform.acedata.cloud",# 单次请求超时，秒；默认 300.0    timeout=300.0,# 自动重试次数，默认 2；重试条件：408 / 409 / 429 / 5xx / 网络错误    max_retries=2,# 自定义请求头    headers={"x-app": "my-service/1.0"},)

“

Python SDK 的 timeout 和 TaskHandle 的 poll_interval / max_wait 单位都是秒，TypeScript SDK 用的是毫秒，跨语言迁移时要特别注意。详见 SDK 任务轮询与流式响应。

“

SDK 默认读取 ACEDATACLOUD_API_TOKEN 环境变量；本文为了和 Claude Code VS Code 教程 等其它教程统一，示例里用的是 ACEDATACLOUD_API_KEY，需要 api_token=os.environ["ACEDATACLOUD_API_KEY"] 显式注入。

进阶：X402 支付钩子

from acedatacloud import AceDataCloudfrom acedatacloud_x402 import create_x402_payment_handlerclient = AceDataCloud(    payment_handler=create_x402_payment_handler(        network="base",        evm_signer=my_evm_signer,        prefer_scheme="exact",   # 或 "upto"    ))

完整流程和真实链上结果见 SDK + X402 支付钩子。

如何查看剩余额度

通过 Ace Data Cloud 控制台 - 应用列表，即可查看当前账户的剩余额度。

通过 Ace Data Cloud 控制台 - 使用历史 即可查看所有使用历史和扣费详情。

了解更多

• 🐍 acedatacloud on PyPI
• 🗂 SDK 源码
• 📘 TypeScript SDK 接入教程
• 🟦 Go SDK 接入教程
• 🔌 SDK + X402 支付钩子