SDK · Python
anoman-ai · Python
OpenAI-shaped client with typed access to the _anoman extension. Sync + async, full streaming, batch poll helper, every error class subclasses AnomanError.
Client init
Sync or async
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
)Chat completions
Same shape as OpenAI + typed _anoman
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"Full endpoint reference at /docs/endpoints/chat-completions.
Streaming
SSE with native iteration
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}")Batch
Enqueue + poll in one call
# 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}")Full batch lifecycle at /docs/endpoints/batch.
Errors
Typed exception hierarchy
Every error subclasses AnomanError. Branch on the specific subclass to decide whether to retry, escalate, or surface to the end user.
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}")Anoman-specific options
SDK shortcuts for headers
Common x-anoman-* headers are exposed as typed kwargs so you don’t have to remember the header name.
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
)