ainative-ai-sdk-provider

Python SDK for multi-provider AI inference — chat completions, streaming, and agentic tool-calling over one interface, with server-side routing and automatic fallback across Cerebras, Meta Llama, NVIDIA NIM & DigitalOcean.

AINative's backend routes every chat completion across Cerebras, Meta Llama, NVIDIA NIM, DigitalOcean and BYOK based on your plan tier, live per-provider rate-limit state, daily token and cost caps, and model availability — then transparently falls back through its chain when a provider is exhausted or geo-blocked. This SDK is the client for that. You ask for a model; the platform picks the upstream.

pip install ainative-ai-sdk-provider

Quick Start

from ainative_ai_sdk_provider import AINativeProvider

provider = AINativeProvider()  # reads AINATIVE_API_KEY

result = provider.complete(
    messages=[{"role": "user", "content": "Explain SSE in one sentence."}],
    model="deepseek-4-flash",
)
print(result.text)
print(result.total_tokens, "tokens")

Why there is no provider= parameter

This is the deliberate design of the SDK, not a missing feature.

The server-side router picks an upstream per request. Pinning a provider from the client would switch that off — no tier-aware selection, no rate-limit awareness, no cost-aware ordering, no automatic fallback when an upstream 429s or geo-blocks you. So the SDK never sends a provider field, which is exactly the signal the endpoint needs to do its job.

Two consequences worth knowing:

• result.provider is informational. The backend intentionally reports several internal upstreams as "ainative". Don't branch on it.
• The SDK does not retry rate limits for you. The backend already walked its own fallback chain before returning a 429 or 503, and tells you how long to wait. Retrying immediately would fight its circuit breaker — honour err.retry_after instead.

Authentication

# 1. Environment variable (recommended)
#    export AINATIVE_API_KEY=your_key
provider = AINativeProvider()

# 2. Explicit
provider = AINativeProvider(api_key="your_key")

# 3. Point at another environment
provider = AINativeProvider(api_key="k", base_url="http://localhost:8000/api/v1")
#    or: export AINATIVE_BASE_URL=...

The client sends your key as both X-API-Key and Authorization: Bearer, since the platform's unified auth accepts either — so the same client works against the public gateway and a directly-addressed backend without reconfiguration.

Streaming

for chunk in provider.stream(
    messages=[{"role": "user", "content": "Count to five."}],
    model="deepseek-4-flash",
):
    print(chunk.content, end="", flush=True)

Streaming is served by upstreams that speak OpenAI SSE natively. If the router lands your request on one that doesn't, the API returns a normal JSON body and this method yields a single chunk with the full completion — so your loop keeps working either way.

If an upstream fails mid-stream, the API injects an error frame; the SDK raises UpstreamUnavailableError rather than silently truncating your output.

Async

Every method has an async counterpart. acomplete mirrors complete, astream mirrors stream.

import asyncio
from ainative_ai_sdk_provider import AINativeProvider

async def main():
    async with AINativeProvider() as provider:
        result = await provider.acomplete(
            messages=[{"role": "user", "content": "Hello!"}],
        )
        print(result.text)

        async for chunk in provider.astream(
            messages=[{"role": "user", "content": "Count to three."}],
        ):
            print(chunk.content, end="")

asyncio.run(main())

Tool calling

The API runs an agentic tool-calling loop server-side, up to max_iterations.

tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Get current weather for a city",
        "parameters": {
            "type": "object",
            "properties": {"city": {"type": "string"}},
            "required": ["city"],
        },
    },
}]

result = provider.complete(
    messages=[{"role": "user", "content": "Weather in Oakland?"}],
    model="deepseek-4-flash",
    tools=tools,
    tool_choice="auto",
    max_iterations=3,
)

if result.tool_calls:
    for call in result.tool_calls:
        print(call["function"]["name"], call["function"]["arguments"])
else:
    print(result.text)

tool_subset=["get_weather"] restricts which registered server-side tools get loaded for the request.

Error handling

Errors map onto the API's real failure modes, each carrying the payload the API supplies with it.

from ainative_ai_sdk_provider import (
    AINativeAPIError,
    AuthenticationError,
    InsufficientCreditsError,
    ModelDeprecatedError,
    RateLimitError,
    RegionUnavailableError,
    UpstreamUnavailableError,
)

try:
    result = provider.complete(messages=[{"role": "user", "content": "hi"}])
except InsufficientCreditsError as err:
    # 402 — chat completions need a paid plan
    print(err.details["upgrade"]["action"])
except RateLimitError as err:
    # 429 — the backend already tried its fallback chain
    print(f"Retry in {err.retry_after}s")
except RegionUnavailableError as err:
    # 403 — model geo-blocked; the API names alternatives
    print("Try instead:", err.details["suggested_models"])
except ModelDeprecatedError as err:
    # 410 — model decommissioned, migrate
    print(err.message)
except UpstreamUnavailableError as err:
    # 502/503/504 — upstream overloaded, erroring, or timed out
    print(f"Retry in {err.retry_after}s")
except AINativeAPIError as err:
    print(err.status_code, err.code, err.message)

Exception	Status	Meaning
AuthenticationError	401, 403	Missing, invalid, or unauthorized API key
InvalidRequestError	400, 404, 422	Malformed or rejected request
InsufficientCreditsError	402	Chat completions require a paid plan
RegionUnavailableError	403	Model geo-blocked; backend fallback also failed
ModelDeprecatedError	410	Model decommissioned — migrate
RateLimitError	429	Rate limited; honour retry_after
UpstreamUnavailableError	502, 503, 504	Upstream overloaded/erroring/timed out
ServerError	500	Unexpected server-side failure
AINativeConnectionError	—	API unreachable (DNS/TCP/TLS)
AINativeTimeoutError	—	Client-side timeout

All of them subclass AINativeError; all HTTP ones subclass AINativeAPIError, which exposes .status_code, .code, .message, .details, .retry_after and .response.

API reference

AINativeProvider(api_key=None, *, base_url=None, timeout=180.0, client=None, async_client=None)

Pass client / async_client to reuse your own httpx client (connection pooling, proxies, custom transports). A client you supply is yours to close — the SDK won't close it for you.

The 180s default outlasts the endpoint's own 150s budget across provider fallbacks, so you receive the API's structured timeout error instead of the client aborting first.

complete(messages, *, model=None, temperature=None, max_tokens=None, system=None, tools=None, tool_choice=None, tool_subset=None, max_iterations=None, user_id=None, extra_body=None) -> ChatCompletion

messages is required and non-empty. temperature is bounded 0.0-2.0 and max_iterations 1-10, both validated locally so mistakes fail before a round trip. Unset parameters are omitted entirely, letting the API's own defaults apply. extra_body merges extra fields for endpoint parameters newer than this SDK version.

stream(...) -> Iterator[StreamChunk]

Same parameters minus the tool-calling ones — use complete for agentic loops.

acomplete(...) / astream(...)

Async equivalents.

ChatCompletion

Field	Description
text	Assistant text from the first choice ("" on a tool-call-only reply)
tool_calls	Tool calls the model requested, or []
id, model, created	Completion identity
provider	Provider label — informational only, see above
finish_reason	stop, tool_calls, …
prompt_tokens, completion_tokens, total_tokens	Usage
iterations, total_tokens_used	Tool-loop metadata, when reported
model_substituted	Set when the backend swapped your model for an equivalent
choices, usage, raw	Unprocessed response data

StreamChunk

content (text fragment), finish_reason (final frame only), raw.

Scope (v0.1)

Implemented: chat completions — non-streaming and streaming, sync and async, server-side tool calling, the full error surface of the completions endpoint, and typed results.

Not yet: embeddings, the chat session endpoints (/chat/sessions/... persistence and history), client-side BYOK key passthrough, and a published model catalog. Those are separate endpoints and a separate surface — see core#8066 for what a v2 would cover.

Development

pip install -e ".[dev]"
pytest tests/ -v --cov=ainative_ai_sdk_provider --cov-report=term-missing

Tests are fully mocked with respx — no live API calls and no API key needed.

License

MIT