Python SDK (`sociaro-ai`)

One OpenAI-compatible API for every provider behind the gateway — LLM chat (with streaming), image generation/editing, and async video — plus client-side fallback chains and typed errors.

pip install sociaro-ai

Requires Python 3.10+.

Quick start

from sociaro import Sociaro

client = Sociaro(api_key="gw_live_...")   # or set SOCIARO_API_KEY env var

response = client.chat.completions.create(
    model="openai/gpt-4o",                # always "provider/model"
    messages=[{"role": "user", "content": "What is the speed of light?"}],
)
print(response.choices[0].message.content)

model is always "provider/model". Supported providers: openai, anthropic, alibaba, byteplus, mulerouter. Anthropic chat is translated from the OpenAI shape to the Messages API by the SDK; everything else is passthrough.

Streaming

for chunk in client.chat.completions.create(
    model="openai/gpt-4o",
    messages=[{"role": "user", "content": "Count to 5."}],
    stream=True,
):
    print(chunk.choices[0].delta.content or "", end="", flush=True)

Images

img = client.images.generate(model="openai/dall-e-3", prompt="A red fox in snow")
print(img.data[0].url)

edited = client.images.edit(
    model="openai/dall-e-2",
    image=open("original.png", "rb"),
    prompt="Add a sunset",
)

Video (async jobs)

Blocking — the SDK drives the create→poll loop:

result = client.video.generate(
    model="byteplus/ep-...",
    prompt="A timelapse of cherry blossoms blooming",
    seconds=5,
    resolution="720p",
    on_progress=lambda status: print("status:", status),
)
print(result.url)

Non-blocking:

job = client.video.submit(model="byteplus/ep-...", prompt="Ocean waves", seconds=5)
print(job.task_id, job.status())
result = job.wait(timeout=300.0)

Fallback chains

Define logical model aliases backed by an ordered provider list; the SDK advances on retryable errors (5xx, timeouts, connection errors, and 429 by default):

client = Sociaro(
    api_key="gw_live_...",
    models={"smart": ["openai/gpt-4o", "anthropic/claude-3-5-sonnet"]},
)
response = client.chat.completions.create(model="smart", messages=[...])

Terminal errors (400/404/422, 401/403) raise immediately. If every provider fails, AllProvidersFailed aggregates the per-provider errors. Streaming fallback happens only before the first chunk is delivered.

Attribution

client = Sociaro(api_key="gw_live_...", default_attribution={"team": "product"})

client.chat.completions.create(
    model="openai/gpt-4o",
    messages=[...],
    attribution={"feature": "chat"},   # → X-Attr-Feature header
)

Tags show up as dimensions in GET /gw/stats.

Errors

SociaroError
├── APIError                    HTTP error from the gateway
│   ├── AuthError               401 / 403
│   ├── RateLimitError          429
│   ├── BadRequestError         400 / 404 / 422
│   └── ProviderUnavailableError  502 / 503 / 504 (after retries)
├── AsyncJobFailed              async media job reached terminal failure
├── AsyncJobTimeout             async media job did not finish within timeout
└── AllProvidersFailed          every provider in the fallback chain failed

Async client

AsyncSociaro mirrors Sociaro with async methods — use it with FastAPI, aiohttp, etc.:

from sociaro import AsyncSociaro

async with AsyncSociaro(api_key="gw_live_...") as client:
    response = await client.chat.completions.create(
        model="openai/gpt-4o",
        messages=[{"role": "user", "content": "Hello"}],
    )

Full reference

The complete SDK README (every parameter, return-value access styles, context managers) ships with the package: sdk/python/README.md.