A Practical Guide to the Ace Data Cloud Python SDK

If your AI prototype has grown from one curl command into a small service, the next pain point is usually not the model — it is handling retries, streaming, async calls, task results, and errors without filling your codebase with one-off HTTP wrappers.

The Ace Data Cloud Python SDK is designed for that middle ground: you still call familiar API shapes such as chat completions and image generation, but you get a Python client with synchronous and asynchronous modes, SSE streaming support, retries, typed exceptions, and plain dict responses.

This guide turns the Python SDK documentation into a practical setup you can reuse in a CLI tool, FastAPI backend, automation script, or internal agent service.

What you can do

The acedatacloud package wraps services on api.acedata.cloud into typed methods. The documentation calls out examples such as:

• client.openai.chat.completions.create(...)
• client.images.generate(...)
• client.search.google(...)

Under the hood it is built on httpx. The SDK supports SSE streaming, automatic retries, typed exceptions, and pydantic-style validation behavior, while response bodies are returned as regular Python dict objects.

That last part is worth noticing if you are migrating from openai-python: the docs explicitly note that the response body uniformly returns dict, not a pydantic model.

Install and prepare the token

Install the SDK from PyPI:

pip install acedatacloud
# or uv add / poetry add

If you use the X402 payment path instead of the API token path, the documentation lists a separate package:

pip install acedatacloud-x402

For the normal token-based path, export your token in the shell:

export ACEDATACLOUD_API_TOKEN={token}

The SDK reads ACEDATACLOUD_API_TOKEN automatically when you construct a client without passing api_token. The docs also mention a common repository convention where teams store ACEDATACLOUD_API_KEY; in that case, pass it explicitly:

import os
from acedatacloud import AceDataCloud

client = AceDataCloud(api_token=os.environ["ACEDATACLOUD_API_KEY"])

Call chat completions synchronously

A simple synchronous request is the best smoke test because it verifies authentication, model routing, and response parsing in one call.

import os, time, json
from acedatacloud import AceDataCloud

client = 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")}))

Because res is a dict, you access fields with normal dictionary indexing. The example also prints usage, which is useful in real services for logging token consumption alongside your own request ID.

In production, I would wrap this in a small function that accepts messages, model, and max_tokens, then logs res["id"] plus your user or job identifier. That gives you a clean trail for debugging later.

Stream responses for UI feedback

When stream=True, the SDK returns a regular generator. Each yielded item is a parsed chunk dictionary following the OpenAI SSE format.

import os, time
from acedatacloud import AceDataCloud

client = AceDataCloud(api_token=os.environ["ACEDATACLOUD_API_KEY"])

t0 = time.time()
first_chunk_ms = None
chunks = 0
collected = []

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 is None:
        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("first_chunk_ms", first_chunk_ms)
print("chunks", chunks)
print("collected", "".join(collected).strip())

This pattern maps nicely to web apps: read each delta, forward it to your frontend SSE endpoint, and keep the final joined text for persistence or audit logs.

Use the async client in services

For FastAPI, aiohttp, or any asyncio service, use AsyncAceDataCloud. The API is symmetrical, but I/O methods return coroutines.

import os, asyncio, time
from acedatacloud import AsyncAceDataCloud

async def main():
    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())

The docs recommend explicitly closing the async client to shut down the connection pool. In a long-running service, create the client once during startup and close it once during shutdown rather than creating a new client per request.

Generate an image with images.generate

The Python SDK also exposes image generation. The documentation includes a NanoBanana example and notes that NanoBanana is synchronous, so you should not pass wait; the call blocks until the upstream returns 200.

import os, time
from acedatacloud import AceDataCloud

client = 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"))

For longer-running asynchronous services such as Midjourney, Sora, Veo, and Suno, the docs point to task polling with wait=True or TaskHandle.wait().

Handle errors deliberately

The SDK includes typed exceptions such as AuthenticationError, RateLimitError, and ValidationError.

from acedatacloud import AceDataCloud
from acedatacloud import AuthenticationError, RateLimitError, ValidationError

bad = 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:
    print("rate limited:", err.code)
except ValidationError as err:
    print("bad request:", err.code, err.message)

A useful production rule is to treat these differently: authentication errors should alert configuration owners, rate limits should trigger backoff or queueing, and validation errors should usually be fixed before retrying.

Configure timeouts, retries, and base URLs

The documented client options include api_token, base_url, platform_base_url, timeout, max_retries, and custom headers:

from acedatacloud import AceDataCloud

client = AceDataCloud(
    api_token="...",
    base_url="https://api.acedata.cloud",
    platform_base_url="https://platform.acedata.cloud",
    timeout=300.0,
    max_retries=2,
    headers={"x-app": "my-service/1.0"},
)

One migration footnote from the docs: Python SDK timeout values and task polling values are in seconds, while the TypeScript SDK uses milliseconds.

A simple service shape

For most projects, I would start with three files:

1. settings.py reads ACEDATACLOUD_API_TOKEN or your explicit secret name.
2. ai_client.py creates one AceDataCloud or AsyncAceDataCloud instance.
3. Feature code calls small wrapper functions for chat, streaming, or image generation.

That keeps the SDK boundary obvious and makes it easy to swap sync for async later.

Read the full Python SDK documentation here: https://platform.acedata.cloud/documents/sdk-python