argosvix 0.4.4a0

Argosvix Python SDK = AI agent observability (cost / latency / tokens / errors) for OpenAI / Anthropic / Gemini / Mistral

Argosvix Python SDK

AI agent observability (cost / latency / tokens / errors) for OpenAI / Anthropic / Gemini / Mistral. Sync + async + streaming wrap for all 4 providers. Prompt-caching cost/savings is captured automatically.

Install

pip install argosvix
# OR include a specific provider SDK as extra
pip install "argosvix[openai]"
pip install "argosvix[anthropic]"
pip install "argosvix[gemini]"
pip install "argosvix[mistral]"
# all 4 at once
pip install "argosvix[all]"

Quickstart

from openai import OpenAI
from argosvix import wrap, ArgosvixConfig

client = wrap(
    OpenAI(),
    ArgosvixConfig(
        api_key="argosvix_live_...",  # get from https://dashboard.argosvix.com/api-keys
        tags={"service": "my-app", "env": "prod"},
    ),
)

resp = client.chat.completions.create(
    model="gpt-5.5",
    messages=[{"role": "user", "content": "Hello"}],
)
# The call is automatically recorded (cost / tokens / latency / model) and
# batched to https://ingest.argosvix.com/v1/ingest within 5 seconds.

Visit https://dashboard.argosvix.com after a few seconds to see the call appear.

Configuration

ArgosvixConfig accepts:

Field	Default	Description
api_key	None	Argosvix API key. Required for record submission.
endpoint	https://ingest.argosvix.com/v1/ingest	Ingest endpoint.
tags	{}	Tags attached to every record (e.g. {"service": "bot"}).
disabled	False	Disable record submission entirely (e.g. local dev).
flush_interval_ms	5000	Buffer flush interval.
buffer_max_size	100	Max records before auto-flush.
flush_retry_attempts	2	Total retry attempts including the initial try.
provider	None	Explicit provider override ("openai" / etc). Auto-detected from client class name.
trace_id	None	OTel-subset trace ID. Attached to all records from this client.
span_id	None	OTel-subset span ID.
parent_span_id	None	OTel-subset parent span ID.

Short-lived processes (Lambda / Cron / CLI)

The SDK auto-registers atexit to flush remaining records when the process exits. But for Lambda / Edge Functions / Workers-style short-lived runtimes where atexit may not fire, explicitly flush:

from argosvix import get_recorder

rec = get_recorder(client)
if rec is not None:
    rec.flush_blocking()  # blocks until all buffered records are POSTed

Supported providers (Phase 4)

Provider	Sync	Async	Streaming	Notes
OpenAI	✅	✅	✅	client.chat.completions.create (sync + AsyncOpenAI). For token/cost on streams, pass stream_options={"include_usage": True} (OpenAI only emits usage then).
Anthropic	✅	✅	✅	client.messages.create(stream=True). The client.messages.stream() context-manager helper is not yet recorded (a warning is logged when present).
Google Gemini	✅	✅	✅	generate_content + generate_content_stream (sync client.models + async client.aio.models, google-genai).
Mistral	✅	✅	➖	client.chat.complete + complete_async. The separate client.chat.stream helper is not yet recorded (a warning is logged when present).

Streaming notes: argosvix wraps the returned stream transparently and records once on completion (or on the error / early-exit path). Usage tokens arrive at stream completion, so a stream you create but never consume is not recorded. OpenAI Responses API support is backlog. Need a provider or helper sooner? File an issue at https://github.com/argosvix/Argosvix/issues.

Multi-provider example

from openai import OpenAI
from anthropic import Anthropic
from google import genai
from mistralai import Mistral
from argosvix import wrap, ArgosvixConfig

cfg = ArgosvixConfig(api_key="argosvix_live_...", tags={"app": "comparison-bot"})
oa = wrap(OpenAI(), cfg)
an = wrap(Anthropic(), cfg)
gm = wrap(genai.Client(), cfg)
ms = wrap(Mistral(api_key="..."), cfg)

# All calls are recorded to the same Argosvix account, distinguishable by provider.
oa.chat.completions.create(model="gpt-5.5", messages=[{"role": "user", "content": "Hi"}])
an.messages.create(model="claude-opus-4", messages=[{"role": "user", "content": "Hi"}], max_tokens=512)
gm.models.generate_content(model="gemini-2.5-flash", contents="Hi")
ms.chat.complete(model="mistral-large-latest", messages=[{"role": "user", "content": "Hi"}])

Trace correlation

The easiest way to group related calls is with_trace — wrap a unit of work and every LLM call inside it joins one trace automatically (no manual trace_id), each as its own span:

from argosvix import wrap, with_trace

client = wrap(OpenAI(), ArgosvixConfig(api_key="..."))

with with_trace():
    # both calls share one auto-generated trace; each is its own span
    client.chat.completions.create(model="gpt-5.5", messages=[...])
    client.chat.completions.create(model="gpt-5.5", messages=[...])

Use with_span to record non-LLM steps (retrieval / tool / agent / chain) and nest the LLM calls inside them, so the trace shows the full agent tree:

from argosvix import with_trace, with_span

with with_trace():
    with with_span("retrieval", "vector_search", metadata={"docCount": len(docs)}):
        docs = search(query)
        client.chat.completions.create(model="gpt-5.5", messages=build_prompt(docs))

with_span records latency/status/error automatically. Keep metadata to non-sensitive structured attributes (counts, sizes) — don't put raw documents or args there.

Built on contextvars, so it follows await / asyncio.Task automatically. Precedence: explicit config.trace_id > ambient with_trace > none; opt out with auto_context=False. (contextvars does not cross into run_in_executor / threads — use contextvars.copy_context().run(...) if you offload a wrapped call to a thread.)

You can still pin a fixed trace_id on the client for the simple one-trace-per-client case:

import uuid

client = wrap(OpenAI(), ArgosvixConfig(api_key="...", trace_id=uuid.uuid4().hex))
# All calls from this client share trace_id in the dashboard's traces waterfall view.

Privacy

The SDK records metadata only (= tokens, cost, latency, model name, error info, your tags). Prompts and completions are NOT recorded by default. Opt-in plain-text storage (with PII redaction + AES-256 encryption + 7-30 day retention + 1-click delete) is planned for v1.5 — see https://argosvix.com/privacy for details.

Pricing table

PRICING is a snapshot updated quarterly from each provider's official pricing page. Unknown models return 0.0 cost + a warning. To verify a model is known:

from argosvix import calculate_cost

cost = calculate_cost("openai", "gpt-5.5", prompt_tokens=1000, completion_tokens=500)
print(cost)  # 0.0125 USD

Development

# install with dev deps
pip install -e ".[dev]"

# run tests
pytest

# lint
ruff check argosvix tests

License

MIT © Yuto Makihara (Argosvix). See LICENSE.