SDK · Python
anoman-ai · Python
Klien berbentuk OpenAI dengan akses typed ke ekstensi _anoman. Sync + async, streaming penuh, helper poll batch, setiap kelas error merupakan subclass dari AnomanError.
Inisialisasi klien
Sync atau 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
Bentuk sama seperti OpenAI + _anoman yang typed
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"Referensi endpoint lengkap di /docs/endpoints/chat-completions.
Streaming
SSE dengan iterasi native
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 dalam satu panggilan
# 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}")Siklus batch lengkap di /docs/endpoints/batch.
Error
Hierarki exception yang typed
Setiap error merupakan subclass dari AnomanError. Bercabang berdasarkan subclass spesifik untuk menentukan apakah harus retry, eskalasi, atau menampilkannya ke 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}")Opsi khusus Anoman
Shortcut SDK untuk header
Header x-anoman-* yang umum diekspos sebagai kwargs bertipe sehingga Anda tidak perlu mengingat nama header-nya.
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
)