Frugal SDK for Python

Instrument AI API calls with one line of code. Get per-call-site metrics — token usage, latency, call counts — exported via OpenTelemetry to any collector.

Install

pip install frugal-sdk-python

# Plus whichever provider SDKs you use:
pip install openai anthropic

Quick start

from frugal_metrics import wrap_openai, wrap_anthropic, wrap_bedrock
from openai import OpenAI
from anthropic import Anthropic
import boto3

client = wrap_openai(OpenAI())
anthropic_client = wrap_anthropic(Anthropic())
bedrock_client = wrap_bedrock(boto3.client("bedrock-runtime", region_name="us-west-2"))

# Every call now emits metrics — no other code changes needed
response = client.chat.completions.create(model="gpt-4o", messages=[...])

Configuration

All configuration is via environment variables. No init() call, no configuration objects.

Required

Set both or the wrapper silently returns the client unmodified:

Optional

What gets captured

Every wrapped call emits OTel metrics tagged with the caller's source file and function, the provider, and the model:

The analyzer-sampled metrics fire on a fraction of calls (FRUGAL_PROMPT_ANALYSIS_SAMPLE_RATE, default 10%); the token-side metrics, history_depth, and duplicate_requests are 100% — never sampled.

Every metric carries this base attribute set (except frugal.gen_ai.internal_errors, which is emitted from failure paths where these attributes may not exist yet and carries only frugal.project_id, the stage/type, and frugal.customer_id / frugal.custom_label when set):

• frugal.project_id — from your env vars (frugal.customer_id too when set); frugal.component is added on the per-component series when FRUGAL_COMPONENTS is set
• frugal.caller_file, frugal.caller_function — auto-detected from the call stack
• gen_ai.system + gen_ai.provider.name — the provider, dual-emitted under both the legacy and current OTel attribute (openai, anthropic, aws.bedrock, vertex_ai/gcp.vertex_ai, …, or custom). Set via the system kwarg; an unrecognized value is coerced to custom with a one-time warning. See docs/metrics-reference.md.
• gen_ai.request.model, gen_ai.response.model — the model used
• gen_ai.operation.name — chat, text_completion, embeddings, generate_content, execute_tool, create_agent, invoke_agent, or other (the sentinel an unrecognized operation coerces to)
• frugal.batched — true for batch-API paths (OpenAI Batch API submissions; Anthropic Message Batches submissions and results), false otherwise
• frugal.custom_label — only on track_ai_call / start_ai_call calls that pass custom_label=...; the value must be in the FRUGAL_CUSTOM_LABEL allowlist or it is coerced to other

The "Additional attributes" column above lists what each metric carries on top of the base set. See docs/metrics-reference.md for the values each attribute takes, sentinel values for caller fields, and the full internal-error-stage list.

Metrics are aggregated per export interval (not sent per call), so network overhead is minimal regardless of call volume.

Supported providers

wrap_bedrock covers all four bedrock-runtime operations (invoke_model, invoke_model_with_response_stream, converse, converse_stream) across every model on Bedrock — Anthropic, OpenAI gpt-oss, Amazon Nova, Meta Llama, Mistral, Cohere, AI21, DeepSeek, Qwen, Kimi, etc. Cross-region inference profile prefixes (us., eu., jp., apac., au., global.) are preserved in gen_ai.request.model. Async clients (aiobotocore / aioboto3) are detected automatically. See docs/quickstart.md for the full Bedrock walkthrough.

Custom call sites

For AI calls that don't go through a wrapped client — an HTTP call to a model gateway, a CLI shell-out, or an internal helper that returns raw usage — mark the call site yourself and feed in the usage you have. You get the same metrics and the same caller attribution as the auto-wrappers.

from frugal_metrics import track_ai_call

with track_ai_call(system="anthropic", request_model="claude-sonnet-4-20250514") as span:
    res = my_custom_llm_call(prompt)
    span.set_usage(
        input_tokens=res.usage.input_tokens,
        output_tokens=res.usage.output_tokens,
        response_model=res.model,
    )

The with block times the call and emits metrics on exit — or an error metric if the block raises (the exception always propagates). Works the same around an await in async code. Call span.set_usage(...) once the response is back; if you never call it, you still get a call-count and latency point.

For calls that span function boundaries — where a single with block won't fit — use the manual handle:

from frugal_metrics import start_ai_call

span = start_ai_call(system="anthropic", request_model="claude-sonnet-4-20250514")
try:
    res = my_custom_llm_call(prompt)
    span.success(
        input_tokens=res.usage.input_tokens,
        output_tokens=res.usage.output_tokens,
        response_model=res.model,
    )
except BaseException as exc:
    span.error(exc)
    raise

success() and error() finalize the call exactly once and are idempotent — extra calls are ignored.

Metadata — track_ai_call / start_ai_call kwargs (all optional): system (→ gen_ai.system, default "custom"), operation (default "chat"), request_model (default "<unknown>"), custom_label (→ the frugal.custom_label label on every metric of the call; trimmed, and coerced to other unless in the FRUGAL_CUSTOM_LABEL allowlist). To also run the sampled prompt analyzer, pass any of messages / system_prompt / tools / response_format / max_tokens.

Usage — set_usage(...) / success(...) kwargs (all optional): input_tokens, output_tokens, cache_read_tokens, cache_write_tokens, reasoning_tokens, response_model.

The same env-var configuration, safety guarantees, and exported metrics apply. See docs/quickstart.md for the full custom call-site walkthrough.

Streaming

Streaming works with no extra setup. The wrapper returns a drop-in replacement that passes events through and emits metrics when the stream finishes.

# OpenAI — opt in to include_usage for token counts in streams
stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[...],
    stream=True,
    stream_options={"include_usage": True},
)
for chunk in stream:
    ...

# Anthropic — token counts available natively, no opt-in needed
stream = anthropic_client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[...],
    stream=True,
)
for event in stream:
    ...

Internal library attribution

If your code calls an internal helper library that wraps the AI SDK (e.g. acme_ai_helpers), add it to the block list so attribution lands on your business code:

FRUGAL_PATH_BLOCK_LIST="src/acme_ai_helpers,libs/llm_utils"

Safety guarantees

• If any required env var is unset, wrap_openai / wrap_anthropic return the client completely unmodified with zero overhead.
• If any part of our instrumentation fails at runtime (OTel init, stack walking, metric emission), the original SDK call executes normally. We never raise our own exceptions into your code.
• Wrapping is idempotent — calling wrap_openai(client) twice on the same client is safe.

Debugging

The SDK is silent by design — when misconfigured it degrades to a no-op. To see why it is doing nothing (missing env var, instrument-build failure, cardinality-cap hit, a metric export that keeps failing), register a debug callback:

from frugal_metrics import set_debug_callback

def on_debug(level, message, context=None):
    # level: "info" | "warn" | "error"
    # message: human-readable string
    # context: optional dict of structured detail, e.g. {"stage": "setup_failed"}
    getattr(my_logger, level if level != "warn" else "warning")(
        "[frugal-metrics] %s", message, extra=context or {}
    )

set_debug_callback(on_debug)

This is a structural-diagnostics hook, not a per-call trace. It only fires for structural or persistent problems — missing/invalid configuration, setup failures, cardinality-cap hits, and consecutive export failures (one de-duped error, re-armed on the next success). Ordinary transient per-call API errors are never emitted — those are your own API errors and already visible to you.

Notes:

• A throwing callback is swallowed. If your callback raises, the exception never escapes into the SDK or your code.
• Issues are replayed. Configuration runs lazily and may happen before you register, so init/config diagnostics emitted beforehand are held in a small bounded buffer and replayed in order the moment you call set_debug_callback. Register once at startup to catch them all.
• The callback is global and thread-safe. Pass None to clear it.

Flushing & graceful shutdown

Metrics export on a periodic interval (default 60s). A short-lived process — a CLI, a job, a serverless invocation, a test run — can exit before the next export, silently dropping its final window of metrics. Use flush / shutdown to avoid that:

from frugal_metrics import flush, shutdown

flush()     # force an immediate export; SDK keeps recording afterwards
shutdown()  # flush + tear down the provider (rebuilt lazily on next wrap)

Both are safe no-ops when the SDK was never initialized and never raise.

For processes that exit on their own, opt in to automatic flush-on-exit:

from frugal_metrics import enable_auto_shutdown

enable_auto_shutdown()  # flush + shutdown via atexit, plus SIGTERM / SIGINT

enable_auto_shutdown always registers an atexit hook and, when handle_signals=True (the default), installs SIGTERM/SIGINT handlers that flush+shutdown and then chain to whatever handler was previously installed (so your own handler still runs). Off the main thread, signal registration is unavailable and it falls back to atexit-only. It is idempotent — repeat calls register nothing extra.

It is opt-in because installing signal handlers is intrusive. If you run your own signal handling, do not call enable_auto_shutdown(handle_signals=True); prefer calling shutdown() from inside your own handler instead.

Verifying the setup

Set the OTLP endpoint to empty to print metrics to the console instead of exporting them:

export FRUGAL_API_KEY=local
export FRUGAL_PROJECT_ID=my-project
export FRUGAL_OTEL_EXPORTER_OTLP_ENDPOINT=   # empty → ConsoleMetricExporter
export FRUGAL_METRIC_EXPORT_INTERVAL_MS=5000
python my_script.py

You should see JSON metric records with gen_ai.* and frugal.* attributes every 5 seconds.