API Reference

Initialization

dexcost.init()

def init(
    api_key: str | None = None,
    storage: str | None = None,
    buffer_path: str | None = None,
    batch_size: int = 100,
    flush_interval: float = 5.0,
    auto_instrument: list[str] | None = None,
    redact_fields: list[str] | None = None,
    hash_customer_id: bool = False,
    track_http: bool = True,
    service_catalog_url: str | None = None,
    environment: str | None = None,
    enable_retry_heuristics: bool = False,
    retry_heuristic_window: float | None = None,
    retry_heuristic_threshold: float | None = None,
    track_network: bool = True,
    network_event_threshold_bytes: int = 102_400,
    network_event_on_error: bool = True,
    network_event_latency_ms: int = 0,
    compute_billing_overrides: dict[str, str] | None = None,
    k8s_node_aware: bool = False,
) -> DexcostConfig

Initialize the SDK. Call once at process startup. Returns the DexcostConfig created. If called a second time without an intervening dexcost.close(), the second call is a no-op and the existing config is returned (a warning is logged).

dexcost.set_context()

def set_context(
    customer_id: str | None = None,
    project_id: str | None = None,
    metadata: dict[str, Any] | None = None,
    agent: str | None = None,
) -> None

Set the attribution context for subsequent LLM calls and tasks in the current async task or thread.

dexcost.get_context()

def get_context() -> DexcostContext | None

Return the current DexcostContext, or None if set_context() has not been called.

dexcost.clear_context()

def clear_context() -> None

Remove the current attribution context.

Tasks

dexcost.task()

@contextmanager
def task(
    task_type: str = "",
    metadata: dict[str, Any] | None = None,
) -> Generator[TrackedTask, None, None]

Sync context manager for grouping costs into a single business task. Supports with only (decorated with @contextmanager). For async code, use CostTracker.task() which returns a dual sync/async context manager. Reads customer_id and project_id from set_context().

Yields a TrackedTask. On clean exit, status is set to "success". On exception, status is "failed" and events are still persisted. (Tasks begin with status "pending" while running.)

Raises RuntimeError if dexcost.init() has not been called.

dexcost.get_current_task()

def get_current_task() -> Task | None

Return the Task currently active in this context, or None.

dexcost.set_current_task()

def set_current_task(task: Task | None) -> contextvars.Token[Task | None]

Set the active task and return a token for later restoration via ContextVar.reset().

dexcost.task_context() / dexcost.async_task_context()

@contextmanager
def task_context(task: Task) -> Generator[None, None, None]

@asynccontextmanager
async def async_task_context(task: Task) -> AsyncGenerator[None, None]

Low-level context managers used internally by CostTracker. Set and restore _current_task for the duration of the block.

Recording costs

dexcost.record_cost()

def record_cost(
    service: str,
    cost_usd: Decimal | str,
    *,
    event_type: str = "external_cost",
    cost_confidence: str = "exact",
    pricing_source: str = "manual",
    pricing_version: str | None = None,
    details: dict[str, Any] | None = None,
) -> Event

Record a non-LLM cost against the current active task. Use cost_usd as Decimal or string — never float.

Raises RuntimeError if init() has not been called or no active task exists.

dexcost.set_api_key()

def set_api_key(new_key: str) -> bool

Update the SDK's API key and resume sync after an authentication failure. When the Control Layer returns 401/403, the SyncWorker permanently stops. set_api_key() updates the global config, clears the stop signal, and restarts the worker thread if it has already terminated — without requiring a process restart.

Returns True on success. Returns False (and logs a warning) if init() has not been called.

dexcost.flush()

def flush() -> None

Force an immediate sync of buffered events to the Control Layer. No-op in local-only mode or if init() has not been called.

dexcost.close()

def close() -> None

Stop the SyncWorker and flush pending events. Safe to call even if init() has not been called (no-op).

Tasks — TrackedTask

TrackedTask.record_cost()

def record_cost(
    self,
    service: str,
    cost_usd: Decimal | str,
    *,
    event_type: str = "external_cost",
    cost_confidence: str = "exact",
    pricing_source: str = "manual",
    pricing_version: str | None = None,
    details: dict[str, Any] | None = None,
) -> Event

Record a non-LLM cost event for this task. Same signature as the module-level record_cost().

Raises ValueError if event_type is not "external_cost" or "compute_cost".

TrackedTask.record_usage()

def record_usage(
    self,
    service: str,
    units: int | float = 1,
    *,
    details: dict[str, Any] | None = None,
) -> Event

Compute cost from the rate registry (units × registered_rate) and record the event.

Raises ValueError if no rate is registered for service.

TrackedTask.record_llm_call()

def record_llm_call(
    self,
    provider: str,
    model: str,
    input_tokens: int,
    output_tokens: int,
    cost_usd: Decimal | str | None = None,
    *,
    cost_confidence: str | None = None,
    pricing_source: str | None = None,
    cached_tokens: int = 0,
    latency_ms: int | None = None,
    details: dict[str, Any] | None = None,
    error_type: str | None = None,
) -> Event

Manually record an LLM call event. When cost_usd is None, cost is auto-computed via PricingEngine.

TrackedTask.mark_retry()

def mark_retry(
    self,
    reason: str,
    *,
    cost_usd: Decimal | str = Decimal("0"),
    retry_of: uuid.UUID | None = None,
) -> Event

Explicitly flag the current operation as a retry.

TrackedTask.mark_not_retry()

def mark_not_retry(self, event_id: uuid.UUID | None = None) -> Event | None

Override a false-positive retry detection. When event_id is None, the most recent retry event for this task is un-flagged. Returns the updated Event, or None if no retry event was found.

TrackedTask.link_trace()

def link_trace(self, provider: str, trace_id: str) -> None

Link an external trace to this task. Stored in task.metadata["_trace_links"]. Multiple traces from different providers can be linked.

TrackedTask.get_trace_links()

def get_trace_links(self) -> list[dict[str, str]]

Return all linked traces as a list of {"provider": ..., "trace_id": ...} dicts.

TrackedTask.end()

def end(self, status: str = "success") -> None

Close a manually-started task, compute aggregates, and persist. Must be called exactly once for tasks created via CostTracker.start_task().

Raises RuntimeError if called more than once.

TrackedTask.task_id

uuid.UUID — the unique identifier for this task.

TrackedTask.task

The underlying Task dataclass.

Instrumentation

dexcost.instrument_openai() / dexcost.uninstrument_openai()

def instrument_openai(tracker: CostTracker) -> None
def uninstrument_openai() -> None

Patch / restore openai.resources.chat.completions.Completions.create (sync + async).

dexcost.instrument_anthropic() / dexcost.uninstrument_anthropic()

def instrument_anthropic(tracker: CostTracker) -> None
def uninstrument_anthropic() -> None

Patch / restore anthropic.resources.messages.Messages.create (sync + async).

dexcost.instrument_litellm() / dexcost.uninstrument_litellm()

def instrument_litellm(tracker: CostTracker) -> None
def uninstrument_litellm() -> None

Patch / restore litellm.completion (sync) and litellm.acompletion (async).

dexcost.instrument_gemini() / dexcost.uninstrument_gemini()

def instrument_gemini(tracker: CostTracker) -> None
def uninstrument_gemini() -> None

Patch / restore google.genai.models.Models.generate_content (sync + streaming).

dexcost.instrument_bedrock() / dexcost.uninstrument_bedrock()

def instrument_bedrock(tracker: CostTracker) -> None
def uninstrument_bedrock() -> None

Patch / restore botocore.client.BaseClient._make_api_call — intercepts bedrock-runtime InvokeModel calls only.

dexcost.instrument_cohere() / dexcost.uninstrument_cohere()

def instrument_cohere(tracker: CostTracker) -> None
def uninstrument_cohere() -> None

Patch / restore cohere.Client.chat and cohere.AsyncClient.chat, plus chat_stream when present.

dexcost.instrument_mcp() / dexcost.uninstrument_mcp()

def instrument_mcp(tracker: CostTracker) -> None
def uninstrument_mcp() -> None

Patch / restore mcp.ClientSession.call_tool.

ALL_SUPPORTED_INSTRUMENTS

ALL_SUPPORTED_INSTRUMENTS: list[str] = [
    "openai", "anthropic", "litellm", "gemini", "bedrock", "cohere", "mcp"
]

The full list of SDK names accepted by CostTracker.instrument() and the auto_instrument parameter.

Pricing & rates

PricingEngine

class PricingEngine:
    def __init__(
        self,
        data_path: str | Path | None = None,
        *,
        auto_update: bool = False,
        api_key: str | None = None,
    ) -> None

Calculates LLM costs from model name and token counts. Loads bundled LiteLLM pricing data on construction.

PricingEngine.get_cost()

def get_cost(
    self,
    model: str,
    input_tokens: int,
    output_tokens: int,
    cached_tokens: int = 0,
    cache_creation_tokens: int = 0,
) -> CostResult

Calculate the cost for an LLM call. Returns a CostResult.

PricingEngine.set_custom_pricing()

def set_custom_pricing(
    self,
    model: str,
    input_per_1k: Decimal | str | float,
    output_per_1k: Decimal | str | float,
) -> None

Register custom per-token pricing for a model. Custom pricing takes precedence over the bundled data.

PricingEngine.pricing_version

str — hash of the currently loaded pricing data, for reproducibility.

PricingEngine.close()

Cancel the background update timer, if running.

CostResult

@dataclass(frozen=True)
class CostResult:
    cost_usd: Decimal
    cost_confidence: str
    pricing_source: str
    pricing_version: str

Result of a PricingEngine.get_cost() call.

RateRegistry

Registry of per-service cost rates for non-LLM services.

RateRegistry.register()

def register(self, service: str, per: str, cost_usd: Decimal | str) -> None

Register a per-unit cost rate.

RateRegistry.get()

def get(self, service: str) -> RateEntry | None

Return the RateEntry for service, or None.

RateRegistry.load()

def load(self, path: str | Path) -> None

Load rates from a YAML file. Expected keys: rates.<service>.per and rates.<service>.cost_usd.

RateRegistry.export()

def export(self, path: str | Path) -> None

Export current rates to a YAML file, sorted by service name.

RateRegistry.pricing_version

str — deterministic SHA-256 hash of all registered rates.

RateEntry

@dataclass(frozen=True)
class RateEntry:
    service: str
    per: str
    cost_usd: Decimal

Types

DexcostConfig

@dataclass
class DexcostConfig:
    api_key: str | None = None
    storage: str | None = None
    batch_size: int = 100
    flush_interval_seconds: float = 5.0
    buffer_path: str | None = None
    redact_fields: list[str] = field(default_factory=list)
    hash_customer_id: bool = False
    environment: str | None = None
    track_network: bool = True
    network_event_threshold_bytes: int = 102_400
    network_event_on_error: bool = True
    network_event_latency_ms: int = 0

Global SDK configuration. Created by init() and stored globally.

DexcostContext

@dataclass
class DexcostContext:
    customer_id: str | None = None
    project_id: str | None = None
    metadata: dict[str, Any] = field(default_factory=dict)
    agent: str | None = None

Attribution context stored in a ContextVar by set_context().

Task

The dataclass persisted to SQLite for each tracked business task.

Event

The dataclass persisted for each individual cost event within a task.

EventType

Enum of valid event_type values: llm_call, external_cost, compute_cost, retry_marker, network, gpu_cost, gpu_utilization_signal.

TaskStatus

Enum of valid task status values: pending, success, failed.

CostConfidence

Enum of valid cost_confidence values: exact, computed, estimated, unknown.

PricingSource

Enum of valid pricing_source values: litellm, tokencost, provider_response, manual, custom, rate_registry, service_catalog, unknown.

InvalidAPIKeyError

class InvalidAPIKeyError(ValueError)

Raised by validate_api_key() when an API key does not start with dx_live_ or dx_test_.

validate_api_key()

def validate_api_key(key: str | None) -> str | None

Validate API key format. Returns "live", "test", or None if key is None. Raises InvalidAPIKeyError for invalid formats.

TrackedOpenAI / TrackedAnthropic

Thin wrappers around the OpenAI and Anthropic client classes that auto-record LLM events without monkey-patching. Useful in environments where monkey-patching won't work, or when you want an explicit alternative to auto-instrumentation. Pass a CostTracker via the tracker parameter to connect pricing and storage:

from dexcost import CostTracker
from dexcost.clients import TrackedOpenAI, TrackedAnthropic

tracker = CostTracker(auto_instrument=[])
client = TrackedOpenAI(tracker=tracker)
# client.chat.completions.create(...) auto-records events inside an active task

TrackedOpenAI and TrackedAnthropic are also importable directly from dexcost (they are in __all__).

ServiceCatalog

Manages the bundled 163-service catalog. ServiceCatalog.lookup(url) returns a catalog entry for a URL's hostname, or None. ServiceCatalog.extract_cost(entry, headers, body) returns a CostExtractionResult with amount, confidence, pricing_source, and service_name.

SessionManager / get_session_manager()

SessionManager manages auto-created "session" tasks for LLM calls that occur outside an explicit dexcost.task() context. Call get_session_manager() (exported from dexcost) to retrieve the global instance. Session tasks are created automatically when a tracked LLM call fires with no active task; the session task_type defaults to "agent_session" but respects the agent parameter of set_context().

SyncWorker

Background worker that flushes buffered Event and Task records to the Control Layer. Started automatically by init() when an API key is present and storage != "local". Exposes start(), stop(), and flush().

CostTracker

High-level tracker class. The global instance is managed by dexcost.init(). Use directly when you need multiple independent trackers:

from dexcost import CostTracker

tracker = CostTracker(auto_instrument=["openai"])
tracker.register_rate("maps.googleapis.com", per="request", cost_usd="0.005")

with tracker.task(task_type="route_plan") as t:
    t.record_usage("maps.googleapis.com", units=4)

Key methods: instrument(name), uninstrument(name), task(...), start_task(...), track_task(...) (decorator), register_rate(...), load_rates(path), export_rates(path), get_cost(...), set_custom_pricing(...).

tracker.instrumented returns a frozenset[str] of currently patched SDK names.