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,
) -> DexcostConfig

Initialize the SDK. Call once at process startup. Returns the DexcostConfig created.

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]

Context manager for grouping costs into a single business task. Supports both with and async with. 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.

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

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.

TaskStatus

Enum of valid task status values: success, failed.

CostConfidence

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

PricingSource

Enum of valid pricing_source values: litellm, custom, manual, 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 ensure a CostTracker is wired before the first call. Useful when you instantiate a provider client before calling dexcost.init().

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

Manages auto-created "session" tasks for LLM calls that occur outside an explicit dexcost.task() context. get_session_manager() returns the global instance.

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.