LLM Cost Observability SDK

Client-side SDK for provider response interception, usage extraction, request buffering, and telemetry flush to the ModelMetre backend cost pipeline. The package is named ModelMetre and supports Python 3.9+.

What This SDK Does

• Wraps provider SDK methods without changing normal response behavior.
• Extracts usage metadata from provider responses locally.
• Buffers request details in-memory.
• Flushes buffered batches on size/time thresholds.
• Optionally sends flushed batches to backend telemetry endpoint.
• Supports lazy API-key verification for direct server API calls via CostAnalyticsClient.

What This SDK Does Not Do On Client

• Does not perform authoritative upstream pricing sync on the client.
• Does not do final backend-grade cost computation in the main intercept-and-flush path.
• Does not require provider credentials to leave the client environment.

Architecture (Current)

1. Your app calls provider SDK (Anthropic/OpenAI/custom).
2. Wrapped method runs, returns original provider response.
3. Interceptor extracts usage/model/stop reason.
4. RequestDetailsBuffer stores request details locally.
5. Buffer flushes when:
   • FLUSH_BATCH_SIZE reached (default 50), or
   • timer hits FLUSH_INTERVAL_SECONDS (default 30).
6. If telemetry is configured, flushed batch is POSTed to backend (/v1/telemetry/flush by default).
7. Backend resolves pricing and computes final costs.

Package Layout (Relevant)

• src/ModelMetre/sdk.py: CostAnalyticsSDK facade.
• src/ModelMetre/pricing/interceptor.py: wrapping and extraction pipeline.
• src/ModelMetre/pricing/extractors.py: provider extractors.
• src/ModelMetre/pricing/aggregator.py: request buffer + flush triggers.
• src/ModelMetre/api/telemetry.py: telemetry sender with failed flush retention/retry behavior.
• src/ModelMetre/client.py: authenticated API client (CostAnalyticsClient) with lazy key verification.
• src/ModelMetre/auth/config.py: env-based API key loading (CA_API_KEY).

Install

pip install ModelMetre

For local development and the full test toolchain:

pip install -e ".[dev]"

Quick Start

1) Wrap a Provider Client

from anthropic import Anthropic
from ModelMetre import CostAnalyticsSDK

sdk = CostAnalyticsSDK(
    api_key="ca_live_...",
    client_id="my-service",
    server_url="https://telemetry.example.com",
)

client = Anthropic()
client = sdk.wrap_client(
    client=client,
    provider="anthropic",
    method_path="messages.create",
)

response = client.messages.create(
    model="claude-3-haiku-20240307",
    max_tokens=128,
    messages=[{"role": "user", "content": "Hello"}],
)

print(sdk.get_metrics())

2) Wrap Other Client Shapes

from ModelMetre import CostAnalyticsSDK

sdk = CostAnalyticsSDK(
    api_key="ca_live_...",
    client_id="my-service",
    server_url="https://telemetry.example.com",
)

client = sdk.wrap_client(
    client=client,
    provider="custom-provider",
    method_path="responses.create",
)

3) Manual Flush

from ModelMetre import RequestDetailsBuffer

buffer = RequestDetailsBuffer()
buffer.flush()

Core APIs

CostAnalyticsSDK (src/ModelMetre/sdk.py)

• CostAnalyticsSDK(api_key: str, client_id: str, server_url: str = DEFAULT_SERVER_URL)
• wrap_client(client, provider, method_path, response_to_dict=None, metadata=None)
• process_response(response, provider, request_id=None, metadata=None)
• get_metrics()
• get_pending_requests()
• flush_buffer()
• resolve_identity(actor_id, actor_type)
• get_identity()
• shutdown()

Identity Hierarchy

The identity endpoint resolves the full attribution chain for an authenticated SDK key:

organization
  |
  +-- department
    |
    +-- project
      |
      +-- environment (optional)
        |
        +-- sdk_key
          |
          +-- actor (user or service)

The server is the source of truth for the tenant portion of this hierarchy. The authenticated CA_API_KEY selects the project_id, optional environment_id, and sdk_key_id. The server derives department_id and organization_id from project_id; they cannot be overridden in an SDK payload. The caller supplies only the per-call actor identity:

Field	Source	Notes
organization_id	project_id -> department_id -> organization_id	Derived by the server.
department_id	Project relationship	Derived by the server.
project_id	Authenticated API key	The key must be scoped on the website.
environment_id	Authenticated API key	null means all environments in the project.
sdk_key_id	Authenticated API key	The API key record ID.
actor_id	SDK or host application	May be a profile UUID, a username, or a service name.
actor_type	SDK or host application	Must be user or service.

Resolve and cache identity with CostAnalyticsSDK

Resolve identity explicitly after constructing the SDK. The result is cached on that SDK instance and is used by subsequent telemetry processing. Resolve again when the acting user or service changes.

from ModelMetre import CostAnalyticsSDK

sdk = CostAnalyticsSDK(
    api_key="ca_live_...",       # Scoped to a project on the server
    client_id="checkout-worker",
    server_url="https://api.example.com",
)

identity = sdk.resolve_identity(
    actor_id="checkout-worker",
    actor_type="service",
)

print(identity.project_id)
print(identity.environment_id)  # None when the key covers all environments
print(identity.actor_known)     # Informational; False is valid for new actors
assert sdk.get_identity() == identity

actor_id is self-asserted and does not need to exist in profiles. When it happens to be a profile UUID, actor_known is True; an unknown actor is not rejected. CostAnalyticsSDK keeps this resolved identity instance-scoped, so different SDK instances do not share identity state.

Resolve identity with CostAnalyticsClient

Use the lower-level client when the application needs the resolved fields without the buffering facade:

from ModelMetre import CostAnalyticsClient

client = CostAnalyticsClient(
    api_key="ca_live_...",
    server_url="https://api.example.com",
)

identity = client.resolve_identity(
    actor_id="alice@example.com",
    actor_type="user",
)

print(identity.organization_id)
print(identity.department_id)
print(identity.project_id)
print(identity.sdk_key_id)

The endpoint is POST /v1/identity/resolve with this request body:

{
  "actor_id": "alice@example.com",
  "actor_type": "user"
}

The server returns organization_id, department_id, project_id, environment_id, sdk_key_id, actor_id, actor_type, and actor_known. It returns key_not_scoped (403) when the API key has no project, project_not_found (409) when the key points to a deleted project, and invalid_actor_type (422) when actor_type is not user or service. These are exposed by the client as IdentityResolutionError.error_code.

Attribute manually processed telemetry

After resolving an identity, normal CostAnalyticsSDK.process_response() calls use the cached actor. A single call can override the actor without changing the cached identity, which is useful for a shared worker:

sdk.resolve_identity(actor_id="worker-01", actor_type="service")

sdk.process_response(
    response,
    provider="openai",
    actor_id="user-123",
    actor_type="user",
    trace_id="trace-456",
)

For telemetry flushes, tenant fields are always injected from the authenticated key on the server. Client payloads may include actor_id, actor_type, provider_credential_id, and trace_id, but must not attempt to provide trusted organization, department, project, environment, or SDK-key fields.

get_metrics() currently returns buffer-oriented metrics:

{
  "buffer_size": 12,
  "pending_requests": 12
}

Buffer (src/ModelMetre/pricing/aggregator.py)

• FLUSH_BATCH_SIZE = 50
• FLUSH_INTERVAL_SECONDS = 30
• get_request_buffer() / get_cost_aggregator()
• RequestDetailsBuffer.record_request(...)
• RequestDetailsBuffer.flush()
• RequestDetailsBuffer.get_pending_requests()

Telemetry (src/ModelMetre/api/telemetry.py)

TelemetryClient behavior:

• Sends flush payload to POST {server_url}/v1/telemetry/flush (default endpoint path).
• Includes headers:
  • Content-Type: application/json
  • X-Client-ID: <uuid>
  • required Authorization: Bearer <api_key>
• Requires a non-empty API key before sending a flush request.
• On flush failure, retains up to last 5 failed batches in-memory.
• If failure looks like "request not received", retries once immediately.
• Failed batches are retried on subsequent flush attempts.

Auth for SDK-to-Server API Calls

CostAnalyticsClient (src/ModelMetre/client.py) supports direct authenticated calls to server APIs.

• Reads CA_API_KEY when api_key not passed.
• Performs lazy auth verification against GET /v1/auth/verify (default path).
• Sends request metadata headers:
  • Authorization
  • X-CA-Key-Id
  • X-CA-User-Id
  • X-Request-Id
  • X-CA-Provider
  • X-CA-Model

Example:

from ModelMetre import CostAnalyticsClient

client = CostAnalyticsClient(
    api_key="ca_live_...",
    server_url="https://api.example.com",
)

resp = client.request("GET", "/v1/costs")
print(resp.status_code)

Environment Variables

• CA_API_KEY: client API key for CostAnalyticsClient.
• CA_SERVER_URL: optional base URL override for CostAnalyticsClient and get_sdk().
• CA_CLIENT_ID: optional stable client ID used by get_sdk().

Note: backend services may require additional variables (for example server HMAC secret) that are configured in the server repository.

Backend Contract

Telemetry flush endpoint should accept payload in this shape:

{
  "client_id": "uuid",
  "batch": [
    {
      "timestamp": "2026-01-01T00:00:00.000000",
      "request_id": "req-123",
      "model": "claude-3-haiku-20240307",
      "provider": "anthropic",
      "input_tokens": 100,
      "output_tokens": 50,
      "cache_read_tokens": 0,
      "cache_creation_tokens": 0,
      "stop_reason": "end_turn",
      "metadata": {"method": "messages.create"}
    }
  ]
}

Other SDK-facing backend routes are:

• POST /v1/telemetry/flush: authenticated batch ingestion and persistence.
• POST /v1/pricing/custom: authenticated account-scoped custom pricing.
• POST /v1/identity/resolve: resolves the authenticated key's project chain for a caller-supplied actor_id and actor_type (user or service).

The server derives tenant fields such as project_id, environment_id, and sdk_key_id from the authenticated key. They must not be supplied by the client as trusted payload fields.

Auth verification endpoint expected by CostAnalyticsClient:

• GET /v1/auth/verify
• Response:

{
  "user_id": "...",
  "api_key_id": "..."
}

Testing

pip install -e ".[dev]"
pytest tests/ -v

Equivalent Make targets are make test, make test-unit, make test-integration, make test-auth, make lint, and make build.

Notes on Compatibility Surface

The SDK exposes one generic extractor/interceptor path. Provider names are metadata values passed to wrap_client; authoritative pricing resolution lives on the backend.