argosvix 0.4.12a0

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="argk_...",  # 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.
capture_content	False	Opt-in plaintext prompt / completion capture (PII-redacted before send). See "Content capture" below.
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.

Content capture (opt-in)

By default only metadata leaves your process. Set capture_content=True to also record prompt and completion bodies — useful for quality review, eval datasets, and debugging:

client = wrap(
    OpenAI(),
    ArgosvixConfig(api_key="argk_...", capture_content=True),
)

• Coverage: non-streaming calls on all four providers (OpenAI / Anthropic / Gemini / Mistral). Streaming calls keep recording metadata as usual, but bodies are not captured.
• PII redaction before send: emails, credit-card numbers, phone numbers, etc. are replaced with [REDACTED_*] inside your process, before the record leaves it.
• Server-side consent gate: unless the account is on a paid plan (Pro or higher) and plaintext storage has been explicitly enabled in the dashboard settings (consent dialog), the backend discards the bodies. Flipping the SDK flag alone stores nothing.

See https://argosvix.com/en/docs/sdk-reference for details.

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="argk_...", 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.

Deployed prompts (resolve_prompt / with_prompt)

If you manage prompts in Argosvix (prompt registry + deployments), resolve_prompt fetches the currently deployed version of a prompt at runtime, and with_prompt tags every wrapped LLM call in the block with prompt = {name}@v{version} — so quality and cost can be compared per prompt version in the dashboard:

import os
from argosvix import resolve_prompt, with_prompt

p = resolve_prompt("support-bot", api_key=os.environ["ARGOSVIX_API_KEY"])
# p.template = prompt body / p.version = deployed version / p.tag = "support-bot@v3"

with with_prompt(p):
    # this call is tagged prompt=support-bot@v3 automatically
    client.chat.completions.create(
        model="gpt-5.5",
        messages=[
            {"role": "system", "content": p.template},
            {"role": "user", "content": user_input},
        ],
    )

• resolve_prompt(name, *, api_key, label="production", cache_ttl_ms=60000) resolves the current version for a deploy label. Results are cached in-memory with a 60-second TTL (0 disables), so it is safe on the hot path.
• Stale fallback: on network errors or 5xx an expired cache entry is returned instead of failing, so a transient backend outage doesn't stop your app. With no cached value the error propagates. 4xx (e.g. a deployment that doesn't exist) always raises ArgosvixPromptError.
• with_prompt accepts the resolve_prompt result, a {"name": ..., "version": ...} mapping / object, or a raw tag string. An explicit tags["prompt"] wins over the ambient tag.

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 capture is available via capture_content=True — PII redaction is applied before send, and the server discards bodies unless the account is on Pro or higher with explicit dashboard consent. See "Content capture" above and https://argosvix.com/en/docs/sdk-reference 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.