witness-sdk 0.1.0

The Python SDK for the Witness Application.

This package does not work yet. Do not install until V1.

Witness Python SDK

The Python SDK for the Witness Application.

Repository: github.com/witness-sdk/python-sdk

Witness captures inference telemetry on a side channel: events are enqueued in-process and delivered by a background thread to witness.sh. Your inference hot path is never blocked, and delivery failures never raise into your application — if ingestion is down, your models still answer.

Install

pip install witness-sdk

The import name is witness. Requires Python 3.10+. The only runtime dependency is httpx.

Quickstart

import witness
from openai import OpenAI

witness.init(api_key="w_live_...", project="prod-chat")
witness.watch(OpenAI)

client = OpenAI(
    api_key="gsk_...",
    base_url="https://api.groq.com/openai/v1",
)

response = client.chat.completions.create(
    model="llama-3.3-70b-versatile",
    messages=[{"role": "user", "content": "Hello"}],
)

# inference → your provider, direct
# telemetry → witness.sh, async

witness.watch() patches an OpenAI-compatible client class once at startup. Every chat.completions.create, embeddings.create, messages.create (Anthropic-style), and streaming call on instances of that class is traced automatically. Provider is inferred from the client base_url (Groq, OpenAI, Together, etc.). Async clients are not supported yet — watch(AsyncOpenAI) raises rather than recording wrong data.

Prefer explicit control? Use @witness.log() instead:

@witness.log(provider="groq")
def chat(messages):
    return client.chat.completions.create(
        model="llama-3.3-70b-versatile",
        messages=messages,
    )

witness.log() wraps any callable that performs an inference call. It times the call, extracts model and token usage from the response by duck-typing (any OpenAI-compatible response shape works), and enqueues an event. The wrapped function's return value and exceptions pass through untouched — failed calls are recorded with status: "error" and re-raised.

Provider and model are read from the response when possible; decorator keyword arguments (provider=, model=) fill in whatever the response cannot answer.

Opting out of telemetry

Skip reporting for specific calls:

# Block scope — works with watch() and @witness.log()
with witness.ignore():
    client.chat.completions.create(model="...", messages=[...])

# Single call — works with both; stripped before your code / the provider sees it
client.chat.completions.create(model="...", messages=[...], witness_ignore=True)

To keep only a fraction of events on high-volume paths, set a sample rate:

witness.init(api_key="...", project="...", sample_rate=0.1)  # keep ~10%

Configuration

witness.init() accepts:

Parameter	Default	Notes
api_key	WITNESS_API_KEY env var	Required (argument or env)
project	WITNESS_PROJECT env var	Required (argument or env)
base_url	https://witness.sh/api/v1	Override for staging / self-hosted
queue_size	10000	Bounded in-memory event queue
sample_rate	1.0	Fraction of calls to record (per-call decision)

If no API key or project is configured, init() logs a warning and the SDK stays disabled: instrumented code runs normally and nothing is sent. Your observability layer never crashes your app over a missing env var.

witness.flush(timeout=5.0) blocks until queued events are delivered and keeps the worker running — useful in notebooks and short scripts. witness.shutdown() flushes and stops; an atexit handler is registered automatically by init().

Privacy

Witness sends metadata only: provider, model, latency, token counts, and status. Prompt and completion text are never transmitted.

Prefork servers (gunicorn, uwsgi)

Background threads do not survive fork(). If you run a prefork server, call witness.init() in a post-fork hook so each worker gets its own transport:

# gunicorn.conf.py
def post_fork(server, worker):
    import witness
    witness.init()  # reads WITNESS_API_KEY / WITNESS_PROJECT

Wire contract

The SDK posts {"events": [...]} batch envelopes to POST {base_url}/events with a Bearer API key. The full contract lives in src/witness/schema/v1/event.json and is versioned via the schema_version field on every event.

Roadmap

• Async @witness.log for coroutines
• Worker-side batching (the wire format already supports it)
• Time-to-last-token on streamed responses

Development

python -m venv .venv && .venv/bin/pip install -e ".[dev]"
.venv/bin/python -m pytest tests/

License

MIT