audacity-sdk

Python client for the Audacity Investments 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 audacity-sdk

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

Quick start

Non-streaming (Converse)

from audacity import Audacity

client = Audacity(api_key="audacity_api_…")   # or set AUDACITY_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 audacity import Audacity

client = Audacity()

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 — Audacity SDK
from audacity import Audacity
client = Audacity(api_key="audacity_api_…")   # only line that changes

response = client.converse(
    modelId="gpt-5.4-mini",                   # use Audacity 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 — Audacity 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 (Audacity 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.

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.

Error handling

from audacity import Audacity
from audacity.exceptions import (
    MissingApiKeyError,
    AccessDeniedException,
    ThrottlingException,
    ServiceQuotaExceededException,
    ModelStreamErrorException,
    SdkError,
)

client = Audacity()

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

client = Audacity(
    api_key="audacity_api_…",
    base_url="https://portal.audacityinvestments.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.