Python SDK.

pip install anoman-ai

poetry add anoman-ai

uv pip install anoman-ai

import os
from anoman import AnomanClient
 
client = AnomanClient(
    api_key=os.environ["ANOMAN_API_KEY"],
    # base_url defaults to https://api.anoman.io
    timeout=120.0,   # seconds — applies to every HTTP call
    max_retries=3,   # retried only on 429/503/504
)

import os
from anoman import AsyncAnomanClient
 
client = AsyncAnomanClient(api_key=os.environ["ANOMAN_API_KEY"])
 
# Use as a context manager for clean shutdown of httpx pools.
async with client:
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Hi"}],
    )

response = client.chat.completions.create(
    model="claude-sonnet-4-6",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "What's the capital of Indonesia?"},
    ],
    temperature=0.7,
    max_tokens=200,
)
 
# Standard fields — identical to OpenAI SDK
print(response.choices[0].message.content)
print(response.usage.total_tokens)
 
# Anoman extension — typed access
print(response.anoman.cost_usd)              # Decimal("0.000041")
print(response.anoman.guardrails.injection)  # GuardrailResult(status="pass", score=0.02)
print(response.anoman.cache.hit)             # False
print(response.anoman.routing.region)        # "id"

import json
 
def get_weather(city: str) -> dict:
    return {"city": city, "temp_c": 31, "condition": "humid"}
 
response = client.chat.completions.create(
    model="claude-sonnet-4-6",
    messages=[{"role": "user", "content": "Weather in Jakarta?"}],
    tools=[{
        "type": "function",
        "function": {
            "name": "get_weather",
            "parameters": {
                "type": "object",
                "properties": {"city": {"type": "string"}},
                "required": ["city"],
            },
        },
    }],
)
 
if response.choices[0].message.tool_calls:
    call = response.choices[0].message.tool_calls[0]
    args = json.loads(call.function.arguments)
    print(get_weather(**args))

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{
        "role": "user",
        "content": [
            {"type": "text", "text": "What's in this image?"},
            {"type": "image_url",
             "image_url": {"url": "https://example.com/chart.png"}},
        ],
    }],
)

stream = client.chat.completions.create(
    model="claude-sonnet-4-6",
    messages=[{"role": "user", "content": "Tell me a short story"}],
    stream=True,
)
 
for chunk in stream:
    delta = chunk.choices[0].delta.content or ""
    print(delta, end="", flush=True)
 
# After the loop, the final _anoman frame is available on stream.anoman_meta
print(f"\nCost: {stream.anoman_meta.cost_usd}")
print(f"Guardrails: {stream.anoman_meta.guardrails.injection.status}")

async with AsyncAnomanClient(api_key="anm-sk-...") as client:
    stream = await client.chat.completions.create(
        model="claude-sonnet-4-6",
        messages=[{"role": "user", "content": "Tell me a story"}],
        stream=True,
    )
    async for chunk in stream:
        print(chunk.choices[0].delta.content or "", end="", flush=True)

import time
 
stream = client.chat.completions.create(
    model="claude-sonnet-4-6",
    messages=[{"role": "user", "content": "Write a long essay"}],
    stream=True,
)
 
deadline = time.time() + 5  # cancel after 5s
for chunk in stream:
    print(chunk.choices[0].delta.content or "", end="", flush=True)
    if time.time() > deadline:
        stream.close()  # closes the HTTP connection cleanly
        break

# Enqueue a batch job (returns 202)
job = client.chat.completions.create(
    model="deepseek-v3",
    messages=[{"role": "user", "content": "Summarize this 50-page doc..."}],
    prefer_batch=True,        # SDK kwarg = x-anoman-prefer-batch header
)
print(f"Queued {job.id}, SLA {job.sla_minutes}m")
 
# Poll to completion — SDK handles 202/200 transitions + Retry-After
result = client.poll_batch(
    job.id,
    deadline_minutes=30,
    on_progress=lambda r: print(f"  remaining {r.sla_remaining_minutes}m"),
)
print(result.choices[0].message.content)
print(f"Saved: ${result.anoman.savings_usd}")

client.batch.cancel(job.id)  # raises if already executing
print(client.batch.get(job.id).status)  # "cancelled"

from anoman import AnomanClient
from anoman import (
    AnomanError,
    AuthError,            # 401, 403 auth_*
    BudgetExceededError,  # 402
    GuardrailError,       # 403 guardrail_*
    RateLimitError,       # 429
    ProviderError,        # 503, 504
)
 
client = AnomanClient(api_key="anm-sk-...")
 
try:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Hi"}],
    )
except GuardrailError as e:
    # Don't retry — fix the input or escalate to manual review.
    print(f"Blocked: {e.code} — {e.message}")
except RateLimitError as e:
    # Honor the server-suggested wait.
    print(f"Wait {e.retry_after}s")
except ProviderError as e:
    # Transient — exponential backoff is appropriate.
    print(f"Upstream {e.status_code}: {e.message}")
except BudgetExceededError:
    # Top up or wait for monthly reset.
    print("Out of budget")
except AuthError:
    # Key was revoked or never valid.
    print("Auth failed")
except AnomanError as e:
    # Catch-all for anything else.
    print(f"Unexpected: {e}")

response = client.chat.completions.create(
    model="claude-sonnet-4-6",
    messages=[...],
    session_id="conv-7k4mP",       # x-anoman-session-id
    agent_id="support-bot-v3",     # x-anoman-agent-id
    metadata={"customer_tier": "enterprise"},  # surfaces in traces
)

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[...],
    force_realtime=True,   # x-anoman-realtime
)

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[...],
    no_cache=True,   # x-anoman-no-cache
)