Argosvix Python SDK

AI agent observability (cost / latency / tokens / errors) for OpenAI / Anthropic / Gemini / Mistral / xAI Grok / Moonshot Kimi / DeepSeek / Alibaba Qwen. Sync + async + streaming wrap. 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 extras at once
pip install "argosvix[all]"
# Grok / Kimi / DeepSeek need no extra — they use the openai client

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:

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 every supported provider. 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)

Grok, Kimi, and DeepSeek reuse the OpenAI wrapper because they serve an OpenAI-compatible API — wrap the same OpenAI client you already have. The record carries the real provider, so cost is priced with that provider's rates. An unreadable or unknown base_url falls back to openai (this also covers Azure OpenAI).

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.