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

Console mode (FRUGAL_METRICS_EXPORTER=console) needs only FRUGAL_PROJECT_ID; every other mode needs both. If a required variable is unset, the wrapper silently returns the client unmodified:

Optional

The export endpoint and authentication come only from FRUGAL_-prefixed variables, so the SDK never crosses wires with the host's own OTel pipeline: the generic OTEL_EXPORTER_OTLP_ENDPOINT and OTEL_EXPORTER_OTLP_HEADERS are never read. (The OTel exporter library still reads its own generic TLS, compression, timeout and temporality variables.)

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, frugal.instrument on the otel_cardinality_overflow stage, and frugal.customer_id / frugal.custom_label when set; frugal.custom_label never on the setup_failed / emission stages):

• 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; both draw on one shared registry and become <overflow> past FRUGAL_MAX_MODELS distinct model strings
• 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 (with a once-per-value warning). custom_label=None means no label; a non-string value 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,
    )

For an enclosing agent invocation, set operation="invoke_agent" and report the number of model-response steps performed inside it:

with track_ai_call(
    system="anthropic",
    operation="invoke_agent",
    request_model="claude-sonnet-4-20250514",
) as span:
    result = my_agent_framework_call(prompt)
    span.set_usage(
        input_tokens=result.usage.input_tokens,
        output_tokens=result.usage.output_tokens,
        agent_turns=result.turn_count,
    )

• Provider wrappers still measure individual model calls and never set agent_turns.
• Agent frameworks such as the Claude Agent SDK, OpenAI Agents SDK, and LangGraph do not have dedicated wrappers yet; track them manually as shown.
• frugal.gen_ai.history_depth is the request message count (messages.length), not a turn count.
• Missing, non-integer, non-positive, or otherwise invalid agent_turns values are silently dropped. They never raise or affect customer execution.
• Turn data is emitted only on success, like token usage.
• Avoid double-counting tokens. Only report aggregate input_tokens / output_tokens on the invoke_agent call when the agent's inner model-client calls are not already wrapped by a supported provider wrapper. If they are wrapped, those wrappers already emit token usage per inner call — report agent_turns only on the invoke_agent call in that case, or the same tokens are double-counted.

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 except the frugal.gen_ai.internal_errors setup_failed / emission points; 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, agent_turns, 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 FRUGAL_METRICS_EXPORTER=console to print the metrics to stdout instead of exporting them. Console mode needs only FRUGAL_PROJECT_ID; the API key, endpoint and header variables are ignored. An empty endpoint does not select the console exporter: it falls through to the default Frugal endpoint.

export FRUGAL_PROJECT_ID=my-project
export FRUGAL_METRICS_EXPORTER=console
export FRUGAL_METRIC_EXPORT_INTERVAL_MS=5000
python my_script.py

Or point the library at a local OTLP/HTTP collector to see what it exports:

export FRUGAL_API_KEY=test-key
export FRUGAL_PROJECT_ID=my-project
export FRUGAL_OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318/v1/metrics
export FRUGAL_METRIC_EXPORT_INTERVAL_MS=5000
python my_script.py

The collector should receive metric records with gen_ai.* and frugal.* attributes every 5 seconds.