spareparts-sdk

Drop-in OpenTelemetry tracing for OpenAI, Anthropic, and Gemini LLM calls, captured locally and optionally exported to Spare Parts Core.

import spareparts_sdk as spareparts

spareparts.init(service_name="mechanic-assistant")

with spareparts.workflow("diagnose"):
    response = client.messages.create(...)

Install with the providers you use:

pip install 'spareparts-sdk[anthropic]'   # or [openai], [gemini], [all]

Tracing

After init(), any instrumented provider call in this process is traced and written to a local SQLite file (./.spareparts/traces.db by default). Each span row carries the model, provider, prompt/completion content, input/output/cache token counts, cost, latency, status, and any exception type and message.

init() takes:

project_id, environment, service_name, and local_dir resolve the same way: explicit kwarg > env var (SPAREPARTS_PROJECT, SPAREPARTS_ENV, SPAREPARTS_SERVICE_NAME, SPAREPARTS_LOCAL_DIR) > a repo-root spareparts.toml's [capture] table > the default above.

# spareparts.toml
[capture]
project = "sparepartslabs/spareparts"
service_name = "spareparts-api"
environment = "production"
local_dir = ".spareparts"

Remote export

Pass endpoint and api_key together to send the same canonical spans to Core. Failed batches stay pending in memory; call flush() before completing a short-lived job so it can retry or fail the job instead of losing telemetry. Span IDs make retries idempotent.

spareparts.init(endpoint="https://api.example", api_key=token, capture_content=False)
# run model work
if not spareparts.flush():
    raise RuntimeError("telemetry was not accepted")

Workflow tracking

workflow(name) opens a named parent span. Every LLM call made inside becomes a child of it, so a trace's root span name identifies the feature the calls belong to, which is what makes per-feature cost and latency grouping possible. It works as a context manager or as a decorator on sync and async functions:

@spareparts.workflow("mechanic-assistant")
async def _ai_reply(message: str) -> str:
    response = await client.messages.create(...)
    return response.content[0].text

An exception raised inside a workflow propagates unchanged; the span is marked error and keeps the exception type and message.

Reading traces

traces.db is plain SQLite with a single spans table:

sqlite3 .spareparts/traces.db \
  "SELECT started_at, name, model, input_tokens, output_tokens, cost, latency_ms
     FROM spans ORDER BY started_at DESC LIMIT 20;"

Group a run's cost by feature via the root span:

SELECT root.name, COUNT(*) AS calls, ROUND(SUM(child.cost), 4) AS cost_usd
  FROM spans child
  JOIN spans root ON root.span_id = child.parent_span_id
 GROUP BY root.name
 ORDER BY cost_usd DESC;

The schema is the contract with anything that reads the file, so it changes only as a public interface would.

Tests

pip install -e '.[dev]'
pytest