cost-monitor 0.1.0

Python client for cost-monitor: report LLM events and read back a project's events feed and spend metrics.

cost-monitor (Python)

Python client for cost-monitor — report LLM events and read back a project's events feed and spend metrics. A full-parity port of the TypeScript @cost-monitor/sdk: sync and async, batching with retry, and auto-instrumentation for OpenAI and Anthropic.

pip install cost-monitor
# with auto-instrumentation for your provider:
pip install "cost-monitor[openai]"      # or [anthropic]

Auto-tracking in 3 lines

Wrap your provider client; keep calling it exactly as before — every completion is metered to cost-monitor. The response (and stream) you get back is untouched; the server is the source of truth for cost.

from openai import OpenAI
from cost_monitor import wrap_openai

client = wrap_openai(
    OpenAI(),
    api_key="cm_live_...",                 # a cost-monitor PROJECT key
    base_url="https://your-cost-monitor",  # the API origin (host of /v1)
    default_feature="chat",                # optional default attribution
)

client.chat.completions.create(model="gpt-4o-mini", messages=[...])  # tracked

wrap_anthropic works the same for an Anthropic / AsyncAnthropic client.

Per-call attribution

Attribute a single call to a customer / feature / user with the reserved cost_monitor= keyword. It is always stripped before the provider call, so the provider never sees it. Per-call values override the wrapper defaults.

client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[...],
    cost_monitor={"customer_id": "cust_123", "feature": "summarize"},
)

Typing note: cost_monitor= is runtime-safe but is not part of the provider SDK's static signature, so a type checker will flag it as an unexpected keyword. For strictly-typed codebases, set attribution via the wrapper defaults (default_feature / default_customer_id / default_user_id), or report the call with a manual CostMonitorClient.track(...) instead.

Async

from openai import AsyncOpenAI
from cost_monitor import wrap_openai

client = wrap_openai(AsyncOpenAI(), api_key="cm_live_...", base_url="https://...")
await client.chat.completions.create(model="gpt-4o-mini", messages=[...])

Flushing before exit

A wrapper with an SDK-owned buffer exposes a control handle. Flush/close it on shutdown so the last batch is delivered (the buffer also flushes periodically):

client.cost_monitor.close()          # sync: final flush + stop the timer
await client.cost_monitor.aclose()   # async

Or pass your own buffer and own its lifecycle (mirrors the TS { sink } form):

from cost_monitor import CostMonitorClient, EventBuffer, wrap_openai

buffer = EventBuffer(CostMonitorClient(api_key="cm_live_...", base_url="https://..."))
client = wrap_openai(OpenAI(), sink=buffer, default_feature="chat")
# ... at shutdown:
buffer.close()

Pass exactly one of sink= or (api_key + base_url).

Manual reporting

No provider wrapper needed — report any event yourself (works for any LLM / raw HTTP):

from cost_monitor import CostMonitorClient

cm = CostMonitorClient(api_key="cm_live_...", base_url="https://...")
cm.track(
    provider="openai", model="gpt-4o-mini",
    input_tokens=130, output_tokens=39, status="success",
    feature="chat", customer_id="cust_123",
)

Read back a project's data: cm.get_events(), cm.get_metrics(period="30d"), cm.get_event(id), cm.get_margin_series(), and the export_*_csv() methods.

Reliability

• Telemetry never breaks your app. Wrapper/buffer failures are swallowed and routed to an on_error callback; the provider call always completes.
• Batching + retry. Events are sent in batches (default 50, or every 5s); transient failures (5xx / 429 / network) retry with exponential backoff; a 413 or 4xx is dropped, not retried.
• Idempotent. Each event gets a stable client_event_id (uuid4) so a retry dedups server-side.

Requires Python 3.9+. openai (>=1.0.0) and anthropic (>=0.30.0) are optional extras — install only the one you use.

Released under the MIT license.