cendor-core — the foundation

The shared vocabulary every other tool rides: one set of types, one tokenizer, one price table, one instrumentation seam, one event bus. It’s small and stable on purpose — it’s the dependency of the whole stack. You rarely install it directly; it arrives transitively.

pip install cendor-core                # exact token counts by default — tiktoken ships with it

npm i @cendor/core                     # or any tool that depends on it
# token counting via js-tiktoken is bundled — exact counts match Python

Quickstart

Everything below runs offline — token counting and pricing ship bundled, no key, no network:

from cendor.core import tokens, prices

n = tokens.count([{"role": "user", "content": "Summarize this in 3 bullets."}], model="claude-opus-4-8")
cost = prices.estimate("claude-opus-4-8", input_tokens=n, output_tokens=200)
print(n, cost, tokens.method("claude-opus-4-8"))   # e.g. 13  0.005065 USD  bpe-estimate

import { tokens, prices } from '@cendor/core';

const n = tokens.count([{ role: 'user', content: 'Summarize this in 3 bullets.' }], 'claude-opus-4-8');
const cost = prices.estimate('claude-opus-4-8', n, { outputTokens: 200 });
console.log(n, cost.toString(), tokens.method('claude-opus-4-8'));  // e.g. 13  0.005065 USD  bpe-estimate

See it in the stack. The one-wrap-then-every-tool-subscribes flow is walked end to end in Architecture and the Cookbook.

Core concepts

The instrument() seam

Wrap a provider client once, at startup. From then on instrument() intercepts each call, normalizes it into an LLMCall, fills in usage/cost/latency, and publishes it on the event bus — so every sibling tool observes the call by subscribing, never by patching the client. It’s idempotent (re-wrapping is a no-op) and additive (coexists with OpenLLMetry / OpenInference).

The event bus

subscribe / emit, synchronous and in-process. emit runs every subscriber even if one raises — a logging subscriber’s bug can’t starve tokenguard’s enforcement. The first Exception is re-raised after all have run (so intentional control flow like BudgetExceeded still reaches the caller); KeyboardInterrupt/SystemExit propagate immediately. It’s thread-safe: the subscriber list is snapshotted under a lock, then the lock is released before subscribers run, so a callback may safely (un)subscribe from any thread.

Token counting, three tiers

tokens.count() is accurate-first and always offline-capable; tokens.method(model) tells you which path is active:

Exact counting is the default (tiktoken is a required dependency), because truthful token counts are the product. register() a precise counter to override a family.

Prices: offline-first, refreshable

A dated prices.json ships in the wheel, so estimate() works with no network — the offline default. Money is always Decimal (rates are parsed with parse_float=Decimal, so they never round-trip through float). What a snapshot can’t know is whether its rates are still current, so refresh() pulls live rates and age_days()/is_stale() surface staleness.

Because cached_tokens ⊆ input_tokens (normalized across providers), estimate() bills the cached portion once — input_rate*(input − cached) + cached_rate*cached — never at both rates. A model with no published cached rate falls back to the input rate for cache reads (no discount, no double charge).

Cost provenance: reported vs estimated

When a response carries a real billed cost (e.g. a gateway’s usage.cost), instrument() prefers it and tags metadata["cost_reported"] = True; otherwise it prices from the snapshot and tags metadata["cost_estimated"] = True. So a downstream tool or audit can always tell a real figure from an estimate (an unknown model with no reported cost leaves cost = None).

Interceptors (replay / reroute)

add_interceptor(fn) registers a pre-call hook that can return a response to short-circuit the call (replay — used by cassette), a Reroute(…) to rewrite the request before it runs, or MISS to proceed untouched. One seam powers all three, so there’s never a second patch point.

Reroute(**updates) applies its keyword updates to the outgoing call’s kwargs before it executes. Two keys are special-cased so the emitted LLMCall stays consistent with what is actually sent, and so a rewrite works across providers:

• Reroute(model=…) also updates call.model — the downgrade tokenguard uses for on_exceed="downgrade".
• Reroute(messages=…) rewrites the outbound messages — mapped to the provider’s own kwarg (messages for Chat Completions / Anthropic / Bedrock / Ollama, input for the OpenAI Responses API, contents for Gemini) — and updates call.messages. This is how acttrace’s guard() does redact-before-send: the scrubbed messages, not the originals, reach the provider. Applies uniformly to sync, async, and streaming calls (the rewrite happens before the real call runs).

OpenTelemetry (optional)

otel.span(...) emits a GenAI gen_ai.* span when OpenTelemetry is installed, else it’s a no-op. otel.ingest(attrs) turns a managed runtime’s gen_ai.* span attributes into a bus event — with no OpenTelemetry dependency required — so tokenguard/acttrace work even when a runtime owns the call loop.

Functions & classes

instrument() / instrument_tool()

client = instrument(openai_client)     # OpenAI · Anthropic · Bedrock · Gemini · Ollama

@instrument_tool("search")             # wrap a tool so ToolCall events join the stream
def search(q): ...

import { instrument, instrumentTool } from '@cendor/core';

const client = instrument(new OpenAI());       // OpenAI (Chat + Responses) · Anthropic — more landing

const search = instrumentTool('search')(       // wrap a tool so ToolCall events join the stream
  (q) => { /* ... */ });

Detects the client by shape (not model name, so new models work the day they ship), wraps its call entrypoint(s), runs the real call (sync, async, and streaming), fills usage/cost/latency, and emits an LLMCall. Idempotent and additive; an unrecognized client is returned untouched. See Providers for the exact entrypoints wrapped per provider and the streaming note below.

tokens

prices

prices.estimate("gpt-4o", input_tokens=1000, output_tokens=300, cached_tokens=200)  # -> Money
prices.refresh(source="litellm")       # or "openrouter" | "azure" | a static-JSON URL

prices.estimate('gpt-4o', 1000, { outputTokens: 300, cachedTokens: 200 });  // -> Money
await prices.refresh(undefined, { source: 'litellm' });  // or 'openrouter' | 'azure' | a URL

refresh() fetches a static resource over http(s) only (it rejects file:// and other schemes), maps it to our schema in memory (nothing persisted), and normalizes source ids to bare keys (openai/gpt-4o → gpt-4o). See Providers → Live pricing for which sources expose rates.

bus

bus.subscribe(fn)     # idempotent; fn receives each emitted LLMCall / ToolCall
bus.unsubscribe(fn)   # no error if absent
bus.emit(event)       # synchronous dispatch to all subscribers

import { bus } from '@cendor/core';

bus.subscribe(fn);    // idempotent; fn receives each emitted LLMCall / ToolCall
bus.unsubscribe(fn);  // no error if absent
bus.emit(event);      // synchronous dispatch to all subscribers

otel

with otel.span("gpt-4o", provider="openai"):   # gen_ai.* span if OTel installed, else no-op
    ...
otel.ingest({"gen_ai.system": "azure_ai_foundry", "gen_ai.request.model": "gpt-4o",
             "gen_ai.usage.input_tokens": 1000, "gen_ai.usage.output_tokens": 500})  # -> bus event

import { otel } from '@cendor/core';

otel.span('gpt-4o', { provider: 'openai' }, (span) => {   // gen_ai.* span if OTel installed, else no-op
  // ... your model call; `span` is null when @opentelemetry/api isn't installed
});
otel.ingest({ 'gen_ai.system': 'azure_ai_foundry', 'gen_ai.request.model': 'gpt-4o',
              'gen_ai.usage.input_tokens': 1000, 'gen_ai.usage.output_tokens': 500 });  // -> bus event

Types

from cendor.core.types import LLMCall, ToolCall, Usage, Money

Usage(input_tokens=1200, output_tokens=300, cached_tokens=0, reasoning_tokens=0, cache_write=0)  # frozen
Money(0.0135)   # Decimal-backed; +, -, *, comparisons; Money.zero()

import { LLMCall, ToolCall, Usage, Money } from '@cendor/core';

new Usage({ inputTokens: 1200, outputTokens: 300, cachedTokens: 0, reasoningTokens: 0, cacheWrite: 0 });
new Money(0.0135);   // decimal.js-backed; value-equal with Python's Decimal; Money.zero()

• LLMCall (id, provider, model, messages, usage, cost, latency_ms, trace_id, ts, metadata) is the normalized record emitted for every call.
• In Usage, cached_tokens ⊆ input_tokens and reasoning_tokens ⊆ output_tokens (breakdowns, not added to the total). cache_write (Anthropic cache_creation) is a separate billed category (~1.25× input), not in the total.

Modules & protocols

The protocols are typing.Protocols — a library satisfies one by shape, no import or base class. That’s how squeeze is a Compressor for contextkit without either importing the other.

Sink lifecycle (optional). write(entry) is the only required method — isinstance(obj, Sink) matches any write-only sink. A sink may also implement two optional lifecycle methods, which callers invoke via hasattr/getattr guards: flush() (block until buffered records are durably written) and close() (flush, then release resources). These are additive — write-only sinks stay valid. tokenguard.sinks.QueueSink implements both to move durable I/O off the model call’s hot path: the bus runs subscribers inline, so wrapping a SQLite/OTel/file sink in a QueueSink keeps its I/O latency out of every call. QueueSink(SQLiteSink(path)) — enqueue-and-return, drain on a background thread in order, flush()/close() for durability at shutdown.

How it works

%%{init: {"flowchart": {"htmlLabels": false}} }%%
graph LR
    APP["your app /<br/>agent loop"]
    WRAP["instrument client<br/>one wrap, at startup"]
    SEAM["normalize the call"]
    INT["interceptors<br/>replay · reroute"]
    BUS["event bus<br/>LLMCall / ToolCall"]
    SUBS["subscribers<br/>tokenguard · cassette<br/>acttrace · contextkit"]
    OT["OpenTelemetry<br/>gen_ai span"]

    APP -->|"create / stream"| WRAP --> SEAM
    SEAM -->|"pre-call"| INT
    INT -.->|"short-circuit / rewrite"| SEAM
    SEAM -->|"emit on completion"| BUS --> SUBS
    SEAM -.->|"optional"| OT

    classDef seam fill:#2563EB,color:#ffffff,stroke:#1E40AF;
    classDef co fill:#94A3BB,color:#0F172A,stroke:#64748B;
    class WRAP seam;
    class BUS co;

Streaming. With stream=True, the streamed value is passed through to your caller unchanged while instrument() accumulates usage in the background, emitting the LLMCall once, when the stream completes (or is closed early) — with true end-to-end latency. Usage is read from the provider’s own stream reporting where present; when a provider streams no usage, it falls back to an offline estimate flagged metadata["usage_estimated"] = True. Streamed calls carry metadata["streamed"] = True, and the collected chunks are attached at metadata["response"] so cassette can record them.

The streamed value is both an iterator and a context manager — exactly like the provider SDK’s own stream object — so both usage forms work and finalize (emit) exactly once:

stream = client.chat.completions.create(model="gpt-4o", messages=msgs, stream=True)
for chunk in stream:            # iterate…
    ...

with client.chat.completions.create(model="gpt-4o", messages=msgs, stream=True) as stream:
    for chunk in stream:        # …or use it as a context manager (what LangChain does)
        ...
# async: `async for chunk in stream` and `async with … as stream:` likewise

const stream = await client.chat.completions.create({ model: 'gpt-4o', messages: msgs, stream: true });
for await (const chunk of stream) {
  // ... usage accumulates in the background;
}    // the LLMCall is emitted once the stream completes (or is closed early)

This context-manager surface is required by frameworks such as langchain_openai, which consume a streamed completion via with client…create(stream=True) as response:. Unknown attributes (.response, .close(), …) are forwarded to the underlying SDK stream, and closing early (stream.close() / block exit) finalizes the LLMCall once. Replayed streams (via cassette) carry the same iterator + context-manager surface.

Run correlation (trace()). Every LLMCall/ToolCall carries a trace_id (default ""). Set an ambient one with with core.trace("run-id"): to group a unit of work — a direct-SDK agent loop, a request — so its calls share an id downstream (acttrace, your own subscribers). It’s a contextvars binding (nests, works across sync/async); core.current_trace_id() reads it.

from cendor.core import trace
with trace("session-42"):
    client.chat.completions.create(...)      # emitted LLMCall.trace_id == "session-42"

import { trace } from '@cendor/core';
await trace('session-42', () =>
  client.chat.completions.create({ /* ... */ }));  // emitted LLMCall.traceId === 'session-42'

This is a hook, not an orchestrator (see architecture.md): cendor stamps the id you set, it never invents a run graph. The LangChain/LangGraph callback path (providers.md) derives the same trace_id automatically from the framework’s run tree.

Frameworks (LangChain / LangGraph). For frameworks, the SDK-aligned integration point is the framework’s callback system, not client wrapping. cendor.core.langchain.CendorCallbackHandler (optional extra cendor-core[langchain]) records usage + reasoning + tools + run-correlated trace_id with no client touch — recording-only. See providers.md → Frameworks.

Plugs into the stack

core is the seam. Every other tool cooperates through it and nothing else: tools subscribe to the bus, satisfy a protocols type by shape, or register an interceptor — they never import one another. That’s what keeps core the whole stack’s small, stable blast radius.

Honest limits

• Token counts are exact for OpenAI by default — tiktoken is a required dependency, so a normal install counts exactly (no opt-in). Claude/Gemini use tiktoken’s o200k BPE as a close cross-tokenizer proxy (not their native tokenizer); register() a precise counter to override a family. Money is always exact (Decimal).
• Capture is best-effort, not a billing guarantee. A call that raises before returning emits no usage/cost; a streamed response whose provider reports no usage is priced from an offline estimate (flagged usage_estimated). Bedrock’s separate converse_stream entrypoint isn’t wrapped — use converse.
• refresh() never reaches a running service or needs an account — it fetches static JSON over http(s), maps it in memory, and falls back to the bundled snapshot. AWS/GCP catalogs need credentials/SDKs and are intentionally out of core (bring your own mapper=).
• Provider SDKs and OpenTelemetry are optional extras ([openai], [anthropic], [otel]) — never hard dependencies. tiktoken, by contrast, is a required dependency: exact token counts are not optional. (It is fully offline — no network or account — so this keeps the local-first guarantee.)