frugal-sdk-python 0.2.0

Instrumentation for Cost-to-Code attribution.

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:

Variable	Purpose
FRUGAL_API_KEY	Per-customer bearer token issued by Frugal. Builds the export Authorization header. (Or supply the bearer yourself via FRUGAL_OTEL_EXPORTER_OTLP_HEADERS.)
FRUGAL_PROJECT_ID	The Frugal project this deployment belongs to. Stats are attributed per project, so the same repo across microservices no longer collides.

Optional

Variable	Purpose
FRUGAL_CUSTOMER_ID	Customer ID issued by Frugal. When unset, the tenant label (auto-injected server-side by VMAuth from the bearer token) identifies the customer and frugal.customer_id is omitted.
FRUGAL_REPO_ROOT	Absolute path to the source root; the strip prefix for caller_file. Defaults to the process working directory.
FRUGAL_OTEL_EXPORTER_OTLP_ENDPOINT	OTLP collector URL for Frugal metrics. Defaults to https://metrics.frugal.co/opentelemetry/v1/metrics. Falls back to OTEL_EXPORTER_OTLP_ENDPOINT.
FRUGAL_OTEL_EXPORTER_OTLP_HEADERS	Export auth headers (e.g. Authorization=Bearer ...). Overrides the header built from FRUGAL_API_KEY. Falls back to OTEL_EXPORTER_OTLP_HEADERS.
FRUGAL_COMPONENTS	Comma-separated logical components (e.g. ai-handler,reco). When set, each metric is ALSO emitted once per component (tagged frugal.component) on top of the base project series, so usage can be viewed for the whole project or split by component.
FRUGAL_PATH_BLOCK_LIST	Comma-separated repo-relative paths to skip for caller attribution (see below).
FRUGAL_METRIC_EXPORT_INTERVAL_MS	Export cadence in milliseconds. Default 60000.
FRUGAL_MAX_CALL_SITES	Cardinality cap — max distinct (caller_file, caller_function) pairs to attribute. Default 500. Set to 0 to disable.
FRUGAL_PROMPT_ANALYSIS_SAMPLE_RATE	Fraction of calls (0..1) that run the lightweight prompt analyzer. Default 0.1. Set to 0 to disable, 1 for every call.
FRUGAL_DISABLED	Set to 1 to disable all instrumentation without uninstalling.

What gets captured

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

Metric	Kind	Description	Additional attributes
gen_ai.client.operation.duration	histogram	Call latency in seconds	—
gen_ai.client.token.usage	histogram	Input and output token counts	gen_ai.token.type
frugal.gen_ai.calls	counter	Number of calls per site	—
frugal.gen_ai.cache_read_tokens	histogram	Prompt cache hits	—
frugal.gen_ai.cache_write_tokens	histogram	Prompt cache writes	—
frugal.gen_ai.reasoning_tokens	histogram	OpenAI reasoning tokens (o-series), Anthropic thinking tokens	—
frugal.gen_ai.errors	counter	Errors by exception type	frugal.error.type
frugal.gen_ai.internal_errors	counter	frugal-metrics internal failures (cardinality cap, bugs)	frugal.internal_error.stage
frugal.gen_ai.output_cap_utilization	histogram	output_tokens / max_tokens ratio when max_tokens is set	frugal.structured_output_mode
frugal.gen_ai.history_depth	histogram	messages.length (analyzer-sampled)	—
frugal.gen_ai.few_shot_count	histogram	Few-shot exemplar pairs before final user (analyzer-sampled)	—
frugal.gen_ai.noise_ratio	histogram	HTML/base64/whitespace fraction of input (analyzer-sampled)	—
frugal.gen_ai.cache_prefix_stable	counter	Per-block prompt-cache stability (analyzer-sampled)	frugal.block, frugal.stable
frugal.gen_ai.structured_output_hint	counter	"Return JSON" prompts missing response_format/tools	—

The bottom block of analyzer-sampled metrics fires on a fraction of calls (FRUGAL_PROMPT_ANALYSIS_SAMPLE_RATE, default 10%); the token-side metrics are 100% — never sampled.

Every metric carries this base attribute 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 — openai, anthropic, aws.bedrock, or vertex_ai
• gen_ai.request.model, gen_ai.response.model — the model used
• gen_ai.operation.name — chat, text_completion, or embeddings
• frugal.batched — true for batch-API submissions, false otherwise

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

Provider	Usage
OpenAI	wrap_openai(OpenAI())
Anthropic	wrap_anthropic(Anthropic())
Claude on AWS Bedrock (via Anthropic SDK)	wrap_anthropic(AnthropicBedrock())
AWS Bedrock — any model, AWS SDKs directly	wrap_bedrock(boto3.client("bedrock-runtime"))
Claude on Google Vertex	wrap_anthropic(AnthropicVertex())

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>"). 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.