Observability API Reference

All classes are importable from anchor.observability. For usage examples see the Observability Guide.

SpanKind

String enum representing the category of a traced operation.

from anchor.observability import SpanKind

Span

Immutable Pydantic model representing a single traced operation. Spans form a tree via parent_span_id.

from anchor.observability import Span

TraceRecord

Immutable Pydantic model grouping all spans from a single pipeline execution.

from anchor.observability import TraceRecord

MetricPoint

Immutable Pydantic model for a single metric measurement.

from anchor.observability import MetricPoint

Tracer

Creates and manages spans within traces.

from anchor.observability import Tracer

Constructor: Tracer() -- no parameters.

[!NOTE] Tracer is not thread-safe. Synchronise externally or use one instance per thread.start_trace(name, attributes=None) -> TraceRecord

end_trace(trace) -> TraceRecord -- finalises a trace, computes end_time and total_duration_ms, removes it from the active set.

start_span(trace_id, name, kind, parent_span_id=None, attributes=None) -> Span

end_span(span, status="ok", attributes=None) -> Span -- finalises a span with end_time, duration_ms, and merged attributes.

get_trace(trace_id) -> TraceRecord | None -- look up an active trace by ID.

TracingCallback

Pipeline callback that automatically traces execution via spans.

from anchor.observability import TracingCallback

Constructor:

TracingCallback(
    tracer: Tracer | None = None,
    exporters: list[SpanExporter] | None = None,
    metrics_collector: MetricsCollector | None = None,
)

Properties:

Callback methods (called by the pipeline automatically):

• on_pipeline_start(query: QueryBundle) -> None
• on_step_start(step_name: str, items: list[ContextItem]) -> None
• on_step_end(step_name: str, items: list[ContextItem], time_ms: float) -> None
• on_step_error(step_name: str, error: Exception) -> None
• on_pipeline_end(result: ContextResult) -> None

ConsoleSpanExporter

Exports spans as structured JSON to Python's logging system.

from anchor.observability import ConsoleSpanExporter

Constructor: ConsoleSpanExporter(log_level: int = logging.INFO)

export(spans: list[Span]) -> None -- logs each span as a JSON object.

InMemorySpanExporter

Stores spans in memory for testing and debugging.

from anchor.observability import InMemorySpanExporter

Constructor: InMemorySpanExporter()

export(spans: list[Span]) -> None -- appends spans to internal buffer.

get_spans() -> list[Span] -- returns a copy of all exported spans.

clear() -> None -- removes all stored spans.

FileSpanExporter

Writes spans as JSON-Lines to a file in append mode.

from anchor.observability import FileSpanExporter

Constructor: FileSpanExporter(path: str | Path)

export(spans: list[Span]) -> None -- appends one JSON object per span.

OTLPSpanExporter

Exports spans to an OpenTelemetry collector via OTLP/HTTP.

from anchor.observability import OTLPSpanExporter

[!CAUTION] Requires opentelemetry-exporter-otlp-proto-http and opentelemetry-sdk. Install with pip install astro-anchor[otlp].Constructor:

OTLPSpanExporter(
    endpoint: str = "http://localhost:4318",
    service_name: str = "anchor",
    headers: dict[str, str] | None = None,
)

export(spans: list[Span]) -> None -- converts and exports spans.

export_record(record: TraceRecord) -> None -- exports all spans from a TraceRecord.

shutdown() -> None -- flushes pending spans and shuts down.

InMemoryMetricsCollector

Stores metric points in memory with summary statistics.

from anchor.observability import InMemoryMetricsCollector

Constructor: InMemoryMetricsCollector()

record(metric: MetricPoint) -> None -- stores a metric point.

flush() -> None -- no-op; metrics remain in memory.

get_metrics(name: str | None = None) -> list[MetricPoint] -- returns stored metrics, optionally filtered by name.

get_summary(name: str) -> dict[str, Any] -- returns {"min", "max", "avg", "count", "p50", "p95"} for the named metric. Empty dict if no matches.

clear() -> None -- removes all stored metrics.

LoggingMetricsCollector

Logs each metric as structured JSON via Python's logging module.

from anchor.observability import LoggingMetricsCollector

Constructor: LoggingMetricsCollector(log_level: int = logging.INFO)

record(metric: MetricPoint) -> None -- logs the metric immediately.

flush() -> None -- no-op.

OTLPMetricsExporter

Exports metrics to an OpenTelemetry collector via OTLP/HTTP.

from anchor.observability import OTLPMetricsExporter

[!CAUTION] Requires opentelemetry-exporter-otlp-proto-http and opentelemetry-sdk. Install with pip install astro-anchor[otlp].Constructor:

OTLPMetricsExporter(
    endpoint: str = "http://localhost:4318",
    service_name: str = "anchor",
    headers: dict[str, str] | None = None,
)

record(metric: MetricPoint) -> None -- records as an OTel gauge. Raises TypeError if not a MetricPoint.

flush() -> None -- flushes buffered metrics.

shutdown() -> None -- shuts down the exporter.

CostEntry

Immutable Pydantic model for a single cost record.

from anchor.observability import CostEntry

CostSummary

Immutable Pydantic model for aggregated cost.

from anchor.observability import CostSummary

CostTracker

Thread-safe tracker for accumulating cost entries.

from anchor.observability import CostTracker

Constructor: CostTracker() -- no parameters.

Property: entries -> list[CostEntry] -- copy of all recorded entries.

record(...) -> CostEntry

def record(
    self,
    operation: str,
    model: str,
    input_tokens: int = 0,
    output_tokens: int = 0,
    cost_per_input_token: float = 0.0,
    cost_per_output_token: float = 0.0,
    metadata: dict[str, Any] | None = None,
) -> CostEntry

Computes cost_usd = input_tokens * cost_per_input_token + output_tokens * cost_per_output_token.

summary() -> CostSummary -- aggregates all entries with per-model and per-operation breakdowns.

reset() -> None -- clears all recorded entries.

CostTrackingCallback

Pipeline callback that records cost entries from step execution metadata.

from anchor.observability import CostTrackingCallback

Constructor: CostTrackingCallback(tracker: CostTracker)

Records a cost entry on on_step_end when items contain cost_model in their metadata. Other callback methods are no-ops.

Expected metadata keys on ContextItem.metadata: