AceDataCloud Python SDK

Official Python client for the AceDataCloud API.

Installation

pip install acedatacloud

Quick Start

from acedatacloud import AceDataCloud

client = AceDataCloud(api_token="your-token")

# OpenAI-compatible chat completions
response = client.openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
)
print(response["choices"][0]["message"]["content"])

# Streaming
for chunk in client.openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Tell me a story"}],
    stream=True,
):
    print(chunk.get("choices", [{}])[0].get("delta", {}).get("content", ""), end="")

Async Usage

import asyncio
from acedatacloud import AsyncAceDataCloud


async def main():
    client = AsyncAceDataCloud(api_token="your-token")
    result = await client.search.google(query="Python SDK")
    print(result)
    await client.close()


asyncio.run(main())

Resources

Image Generation (with Task Polling)

# Default provider (nano-banana)
task = client.images.generate(prompt="A sunset over mountains")
result = task.wait()  # polls until complete
print(result["image_url"])

# Use a specific provider
flux_task = client.images.generate(prompt="A sunset over mountains", provider="flux")

Multi-Provider Support

Image, video, and audio resources support a provider parameter to switch between services:

# Video — default is 'sora'
client.video.generate(prompt="A cat playing piano", provider="kling")
client.video.generate(prompt="Ocean waves", provider="luma")

# Audio — default is 'suno'
client.audio.generate(prompt="A jazz song", provider="producer")

Available providers:

Error Handling

from acedatacloud import AceDataCloud, AuthenticationError, RateLimitError

client = AceDataCloud(api_token="your-token")
try:
    client.search.google(query="test")
except AuthenticationError:
    print("Invalid or expired token")
except RateLimitError:
    print("Too many requests, slow down")

Configuration

client = AceDataCloud(
    api_token="your-token",
    base_url="https://x402.acedata.cloud",  # API gateway (default)
    platform_base_url="https://platform.acedata.cloud",  # Management plane
    timeout=300.0,  # Request timeout in seconds
    max_retries=2,  # Retry count for transient errors
)

The token can also be set via the ACEDATACLOUD_API_TOKEN environment variable.

Paying with X402 Instead of a Bearer Token

The SDK supports a pluggable payment_handler that is invoked when the API returns 402 Payment Required. Any callable (or async callable) that returns the required headers is accepted:

from acedatacloud import AceDataCloud, PaymentHandlerContext, PaymentHandlerResult


def my_payment_handler(ctx: PaymentHandlerContext) -> PaymentHandlerResult:
    # ctx["accepts"] is the list of payment requirements returned by the server.
    # Your job: sign the preferred one and return the retry header.
    x_payment = sign_and_encode(ctx["accepts"])  # your implementation
    return {"headers": {"X-Payment": x_payment}}


client = AceDataCloud(payment_handler=my_payment_handler)

# Normal SDK calls — the handler is only invoked when the server returns 402.
result = client.openai.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Hi"}],
)

The AsyncAceDataCloud client accepts either a synchronous or asynchronous handler. For a ready-made signer, install acedatacloud-x402 and pass its handler directly:

from acedatacloud import AceDataCloud
from acedatacloud_x402 import EVMAccountSigner, create_x402_payment_handler

client = AceDataCloud(
    payment_handler=create_x402_payment_handler(
        network="base",
        evm_signer=EVMAccountSigner.from_private_key("0x..."),
        prefer_scheme="upto",
    )
)

See AceDataCloud/X402Client for Base, Solana, and SKALE setup details.

License

MIT