aireserve-sdk

Python client for the AI Reserve LLM gateway. Exposes the same Amazon Bedrock Converse surface so teams migrating off Bedrock can swap the client constructor and keep the rest of their call-sites unchanged.

Installation

pip install aireserve-sdk

Renamed from audacity-sdk / @audacity/sdk in 0.7.0. The old package names stop receiving updates. from audacity import Audacity and new Audacity(...) keep working for one release with a deprecation warning; migrate to aireserve / Aireserve. The legacy AUDACITY_API_KEY and AUDACITY_BASE_URL environment variables remain accepted as fallbacks.

Zero runtime dependencies — stdlib only. Requires Python 3.9+.

Quick start

Non-streaming (Converse)

from aireserve import Aireserve

client = Aireserve(api_key="aireserve_api_…")   # or set AIRESERVE_API_KEY

response = client.converse(
    modelId="gpt-5.4-mini",
    messages=[{"role": "user", "content": [{"text": "What is 2+2?"}]}],
    inferenceConfig={"maxTokens": 256, "temperature": 0.0},
)

print(response["output"]["message"]["content"][0]["text"])
print(response["stopReason"])   # "end_turn"
print(response["usage"])        # {"inputTokens": …, "outputTokens": …, "totalTokens": …}
print(response["metrics"])      # {"latencyMs": …}

Streaming (ConverseStream)

from aireserve import Aireserve

client = Aireserve()

stream_response = client.converse_stream(
    modelId="gpt-5.4-mini",
    messages=[{"role": "user", "content": [{"text": "Write me a haiku."}]}],
)

for event in stream_response["stream"]:         # boto3 parity: response["stream"]
    if "contentBlockDelta" in event:
        delta = event["contentBlockDelta"]["delta"]
        print(delta.get("text", ""), end="", flush=True)

print()  # newline at end

Migrating from boto3 bedrock-runtime

# BEFORE — boto3
import boto3
client = boto3.client("bedrock-runtime", region_name="us-east-1")

response = client.converse(
    modelId="anthropic.claude-3-sonnet-20240229-v1:0",
    messages=[{"role": "user", "content": [{"text": "Hi"}]}],
)

# AFTER — AI Reserve SDK
from aireserve import Aireserve
client = Aireserve(api_key="aireserve_api_…")   # only line that changes

response = client.converse(
    modelId="gpt-5.4-mini",                   # use AI Reserve model ID
    messages=[{"role": "user", "content": [{"text": "Hi"}]}],
)

Streaming diff:

# BEFORE — boto3
stream_resp = client.converse_stream(modelId=…, messages=…)
for event in stream_resp["stream"]:
    …

# AFTER — AI Reserve SDK (identical call-site)
stream_resp = client.converse_stream(modelId=…, messages=…)
for event in stream_resp["stream"]:
    …

Tool use

response = client.converse(
    modelId="gpt-5.4-mini",
    messages=[{"role": "user", "content": [{"text": "What's the weather in NYC?"}]}],
    toolConfig={
        "tools": [{
            "toolSpec": {
                "name": "get_weather",
                "description": "Get current weather for a city",
                "inputSchema": {
                    "json": {
                        "type": "object",
                        "properties": {"city": {"type": "string"}},
                        "required": ["city"],
                    }
                },
            }
        }],
        "toolChoice": {"auto": {}},
    },
)

# Assistant responds with a tool call
tool_use = response["output"]["message"]["content"][0]["toolUse"]
print(tool_use["name"])   # "get_weather"
print(tool_use["input"])  # {"city": "NYC"}

# Send tool result back
response2 = client.converse(
    modelId="gpt-5.4-mini",
    messages=[
        {"role": "user",  "content": [{"text": "What's the weather in NYC?"}]},
        {"role": "assistant", "content": response["output"]["message"]["content"]},
        {"role": "user",  "content": [
            {"toolResult": {
                "toolUseId": tool_use["toolUseId"],
                "content": [{"text": "Sunny, 72°F"}],
            }}
        ]},
    ],
)

Images (vision models)

Bedrock-style image content blocks are supported in user messages. Pass raw bytes (encoded for you) or a URL (AI Reserve extension):

with open("chart.png", "rb") as f:
    image_bytes = f.read()

response = client.converse(
    modelId="gpt-5.5",
    messages=[{
        "role": "user",
        "content": [
            {"text": "What does this chart show?"},
            {"image": {"format": "png", "source": {"bytes": image_bytes}}},
        ],
    }],
)

# Or reference a hosted image directly (not available in Bedrock):
# {"image": {"format": "jpeg", "source": {"url": "https://example.com/photo.jpg"}}}

format is one of png, jpeg, gif, webp. Use a vision-capable model.

Video (Gemini models)

Bedrock-style video content blocks are supported in user messages. Video input is only available on the Gemini family (gemini-2.5-flash, gemini-2.5-pro, gemini-3-flash-preview); every other model rejects video with an HTTP 400.

with open("demo.mp4", "rb") as f:
    video_bytes = f.read()

response = client.converse(
    modelId="gemini-2.5-flash",
    messages=[{
        "role": "user",
        "content": [
            {"text": "Summarize what happens in this video."},
            {"video": {"format": "mp4", "source": {"bytes": video_bytes}}},
        ],
    }],
)

format is one of mp4, mov, mkv, webm, flv, mpeg, mpg, wmv, three_gp. Inline (bytes) video is base64-encoded into the request — keep those files under ~20 MB.

For larger files (up to 1 GB), upload once and reference by URI:

upload = client.files.upload("large-demo.mp4", content_type="video/mp4")

response = client.converse(
    modelId="gemini-2.5-flash",
    messages=[{
        "role": "user",
        "content": [
            {"text": "Summarize this video."},
            {"video": {"format": "mp4", "source": {"uri": upload["uri"]}}},
        ],
    }],
)

client.files.upload(data, content_type=…) accepts raw bytes or a file path and returns {"file_id", "upload_url", "uri", "expires_at"}. Uploads are resumable: the helper streams the file in 8 MB chunks over a GCS resumable session and automatically resumes from the last confirmed byte after a network drop (bounded retries). Uploaded files are transient inference inputs: they expire after ~24 hours and are scoped to your API key's organization; the signed upload URL itself is valid ~15 minutes.

To control video token cost, set mediaResolution on the request — "low" processes video at ~4x fewer tokens (Gemini models; ignored elsewhere):

response = client.converse(
    modelId="gemini-2.5-flash",
    mediaResolution="low",
    messages=[…],
)

Image generation

Generate images from a text prompt with client.images.generate. With response_format="b64_json" the image bytes come back inline:

import base64

result = client.images.generate(
    model="gpt-image-1",
    prompt="A watercolor painting of a fox in a snowy forest",
    size="1024x1024",
    response_format="b64_json",
)

with open("fox.png", "wb") as f:
    f.write(base64.b64decode(result["data"][0]["b64_json"]))

With response_format="url" (the default) the gateway stores the image and returns a signed download link that expires after ~24 hours:

import urllib.request

result = client.images.generate(
    model="gemini-2.5-flash-image",
    prompt="A watercolor painting of a fox in a snowy forest",
)

urllib.request.urlretrieve(result["data"][0]["url"], "fox.png")

Optional parameters: n (1–10 images), size ("WxH", model-dependent), quality (e.g. "standard", "hd"), and user. The response dict carries created, data (each entry has url or b64_json, plus revised_prompt when the provider rewrites your prompt) and optional usage token counts. Errors map to the same exception classes as converse (401 → AccessDeniedException, 429 → ThrottlingException, spend cap → ServiceQuotaExceededException).

Image models

Per-image models bill a flat rate per generated image; token-based models report token counts in the response usage field. Each request's cost is recorded against your key like any other API call.

Reliability note. Upstream image backends occasionally stall with a 503 for a few minutes. There is deliberately no automatic fallback to a different image model (silently swapping models would change output style and quality) — the SDK already retries 503s with backoff up to max_retries, and callers should retry beyond that rather than switch models.

Prompt caching

Place a Bedrock-style cachePoint block after the stable prefix you want the provider to cache (system prompt, large documents). Everything up to the cache point is cached provider-side on Claude models; OpenAI/Gemini models cache automatically and ignore the marker. At most 4 cache points per request.

response = client.converse(
    modelId="claude-sonnet-4-5",
    system=[
        {"text": long_system_prompt},
        {"cachePoint": {"type": "default"}},
    ],
    messages=[{
        "role": "user",
        "content": [
            {"text": big_reference_document},
            {"cachePoint": {"type": "default"}},
            {"text": "Summarise the key risks."},
        ],
    }],
)

# Cache activity is reported in usage (Bedrock names):
print(response["usage"]["cacheReadInputTokens"])   # tokens served from cache
print(response["usage"]["cacheWriteInputTokens"])  # tokens written to cache

A cachePoint with nothing before it in the same message is silently ignored.

OpenAI & Anthropic wire formats (pass-through)

Prefer the OpenAI or Anthropic request shapes over Bedrock Converse? The same client exposes both gateway-native formats directly — same key, same retry policy, same exceptions, no shape translation (requests sent verbatim, responses returned raw). Both formats work with every gateway model; the gateway bridges the wire format for you.

# OpenAI format → POST /v1/chat/completions
response = client.chat.completions.create(
    model="gpt-5.4-mini",
    messages=[{"role": "user", "content": "Hello!"}],
    max_tokens=256,
)
print(response["choices"][0]["message"]["content"])

# Streaming: raw OpenAI chunks
for chunk in client.chat.completions.create(
    model="claude-sonnet-4-6",   # any gateway model
    messages=[{"role": "user", "content": "Write a haiku."}],
    stream=True,
):
    delta = chunk["choices"][0]["delta"] if chunk.get("choices") else {}
    print(delta.get("content", ""), end="", flush=True)

# Anthropic format → POST /v1/messages (Claude Code's wire format)
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=256,
    messages=[{"role": "user", "content": "Hello!"}],
)
print(response["content"][0]["text"])

# Streaming: raw Anthropic events (message_start … message_stop)
for event in client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=256,
    messages=[{"role": "user", "content": "Write a haiku."}],
    stream=True,
):
    if event["type"] == "content_block_delta":
        print(event["delta"].get("text", ""), end="", flush=True)

# Free token counting → POST /v1/messages/count_tokens
count = client.messages.count_tokens(
    model="claude-sonnet-4-6",
    messages=[{"role": "user", "content": "How many tokens is this?"}],
)
print(count["input_tokens"])

Fields are passed through untouched, so anything the gateway supports works — no SDK release needed for new request fields. (Using the official openai or anthropic packages instead also works: point their base_url at the gateway — see the developer docs.)

Error handling

from aireserve import Aireserve
from aireserve.exceptions import (
    MissingApiKeyError,
    AccessDeniedException,
    ThrottlingException,
    ServiceQuotaExceededException,
    ModelStreamErrorException,
    SdkError,
)

client = Aireserve()

try:
    response = client.converse(modelId="gpt-5.4-mini", messages=[…])
except client.exceptions.ThrottlingException as e:
    print(f"Rate limited: {e.message}, retry after {e.retry_after_seconds}s")
except client.exceptions.AccessDeniedException as e:
    print(f"Auth error [{e.status_code}]: {e.message}")
except client.exceptions.ServiceQuotaExceededException as e:
    print(f"Budget exhausted: {e.message}")
except SdkError as e:
    print(f"Network/decode error: {e.message}")

Streaming errors surface when you iterate the stream:

try:
    for event in stream_response["stream"]:
        …
except client.exceptions.ModelStreamErrorException as e:
    print(f"Stream broken: {e.message}")

Configuration & environment variables

The legacy AUDACITY_API_KEY / AUDACITY_BASE_URL names still work as fallbacks; when both are set, the AIRESERVE_* names win.

client = Aireserve(
    api_key="aireserve_api_…",
    base_url="https://api.aireserve.com",
    timeout=60.0,
    max_retries=3,
)

Retry behaviour

The SDK automatically retries transient failures with jittered exponential backoff (capped at 20 s):

• Retried: ThrottlingException (429), ModelTimeoutException (408), ServiceUnavailableException (502/503/504), InternalServerException (500), network errors.
• Never retried: AccessDeniedException, ValidationException, ResourceNotFoundException, ServiceQuotaExceededException (including BUDGET_EXCEEDED at 429/402), or any 4xx except 408/429.
• Streaming: retries apply only before the first SSE byte; after that, connection drops raise ModelStreamErrorException.