Skip to content

Observability API Reference

Plug-and-play observability system for Promptise agents. Controls what gets observed, how much detail is captured, and where events are sent. Includes configurable transporters and a callback handler that bridges LangChain events into the observability collector.

Core Collector

ObservabilityCollector

promptise.observability.ObservabilityCollector

Thread-safe collector for structured timeline events.

Records events from agents, tools, prompts, and the runtime. Events are dispatched to registered :class:BaseTransporter instances in real time. Supports ring-buffer eviction via max_entries.

Parameters:

Name Type Description Default
session_name str

Human-readable label for this session.

 'promptise'
max_entries int

Maximum number of entries to retain. Oldest entries are evicted when the buffer is full. Defaults to 100,000.

 100000

Example::

collector = ObservabilityCollector("my-agent-session")
collector.add_transporter(ConsoleTransporter())

entry = collector.record(
    TimelineEventType.TOOL_CALL,
    agent_id="my-agent",
    details="Calling search_files",
    metadata={"tool_name": "search_files"},
)
add_transporter(transporter)

Register a transporter to receive timeline events.

Parameters:

Name Type Description Default
transporter Any

Any object with an on_event(entry) method.

 required
clear()

Clear all timeline entries and reset the session start time.

get_agents()

Return sorted unique agent IDs seen in the timeline.

Returns:

Type Description
list[str]

Sorted list of agent ID strings (None entries excluded).
get_phases()

Return sorted unique phase labels seen in the timeline.

Returns:

Type Description
list[str]

Sorted list of phase strings (None entries excluded).
get_stats()

Return aggregated statistics for the current session.

Returns:

Type Description
dict[str, Any]

Dict with token counts, latency percentiles, error counts,
dict[str, Any]

and per-agent/per-type/per-category breakdowns.
get_timeline()

Return all entries sorted by timestamp.

Returns:

Type Description
list[TimelineEntry]

Sorted list of :class:TimelineEntry objects.
query(*, event_types=None, agent_ids=None, limit=None, offset=None, metadata_key=None, metadata_value=None)

Query the timeline with optional filters.

Parameters:

Name Type Description Default
event_types list[TimelineEventType | str] | None

Restrict to these event types (enum or string values).

 None
agent_ids list[str] | None

Restrict to these agent IDs.

 None
limit int | None

Maximum number of results to return.

 None
offset int | None

Number of results to skip from the start.

 None
metadata_key str | None

Restrict to entries whose metadata contains this key.

 None
metadata_value Any

If set, also match the metadata value for metadata_key.

 None

Returns:

Type Description
list[TimelineEntry]

Filtered and sliced list of :class:TimelineEntry objects.
record(event_type, *, agent_id=None, phase=None, details=None, duration=None, parent_id=None, metadata=None)

Record a single timeline event.

Parameters:

Name Type Description Default
event_type TimelineEventType

The event type.

 required
agent_id str | None

Agent that produced the event.

 None
phase str | None

Execution phase label.

 None
details str | None

Human-readable event description.

 None
duration float | None

Duration in seconds (for span events).

 None
parent_id str | None

Parent entry for hierarchical tracing.

 None
metadata dict[str, Any] | None

Arbitrary structured data.

 None

Returns:

Type Description
TimelineEntry

The created :class:TimelineEntry.
record_event(event_type, agent_id=None, data=None)

Record an event using the data keyword (prompt bridge API).

This is a convenience wrapper around :meth:record for callers that pass structured data as a data dict rather than keyword args.

Parameters:

Name Type Description Default
event_type TimelineEventType

The event type.

 required
agent_id str | None

Agent that produced the event.

 None
data dict[str, Any] | None

Structured event data, stored as metadata.

 None

Returns:

Type Description
TimelineEntry

The created :class:TimelineEntry.
span(event_type, **kwargs)

Context manager that records an event and measures its duration.

The entry is recorded immediately on entry; its duration field is set to the elapsed seconds when the context exits (even if an exception is raised).

Parameters:

Name Type Description Default
event_type TimelineEventType

The event type.

 required
**kwargs Any

Passed through to :meth:record.

 {}

Yields:

Name Type Description
The TimelineEntry

class:TimelineEntry being tracked.

Example::

with collector.span(TimelineEventType.PHASE_START, phase="pipeline") as e:
    await do_work()
print(e.duration)  # seconds
to_dict()

Serialise the full session to a dictionary.

Returns:

Type Description
dict[str, Any]

Dict with session metadata, entry list, and stats.
to_json(indent=None)

Serialise the full session to a JSON string.

Parameters:

Name Type Description Default
indent int | None

Optional JSON indentation level.

 None

Returns:

Type Description
str

JSON-encoded session data.
to_ndjson()

Serialise entries as newline-delimited JSON.

Each line is an independent JSON object containing the entry fields plus a session_name key.

Returns:

Type Description
str

NDJSON string (empty string if no entries).

TimelineEntry

promptise.observability.TimelineEntry dataclass

A single event recorded on the observability timeline.

Parameters:

Name Type Description Default
entry_id str

Unique identifier for this entry.

 required
timestamp float

Unix epoch seconds when the event was recorded.

 required
event_type TimelineEventType

The type of event.

 required
category TimelineEventCategory

High-level category derived from the event type.

 required
agent_id str | None

Identifier of the agent that produced the event.

 None
phase str | None

Optional orchestration or execution phase label.

 None
details str | None

Human-readable description of the event.

 None
duration float | None

Duration in seconds (for span events).

 None
parent_id str | None

Parent entry ID for hierarchical tracing.

 None
metadata dict[str, Any]

Arbitrary structured data for the event.

 dict()
to_dict()

Serialise the entry to a plain dictionary.

Returns:

Type Description
dict[str, Any]

Dict with all fields; event_type and category are
dict[str, Any]

their string values.

TimelineEventType

promptise.observability.TimelineEventType

Bases: str, Enum

All timeline event types.

Values are dot-notation strings that appear in serialized output.

TimelineEventCategory

promptise.observability.TimelineEventCategory

Bases: str, Enum

High-level category groupings for timeline events.

Configuration

ObservabilityConfig

promptise.observability_config.ObservabilityConfig dataclass

Configuration for the observability system.

Pass as observe=config to :func:build_agent or use the shorthand observe=True for sensible defaults.

Examples::

# Defaults: STANDARD level, HTML transporter
ObservabilityConfig()

# Enterprise: full detail, multiple transporters
ObservabilityConfig(
    level=ObserveLevel.FULL,
    record_prompts=True,
    transporters=[
        TransporterType.HTML,
        TransporterType.STRUCTURED_LOG,
        TransporterType.CONSOLE,
        TransporterType.PROMETHEUS,
    ],
    output_dir="./observability",
    log_file="./logs/events.jsonl",
    console_live=True,
    correlation_id="req-abc-123",
)

console_live = False class-attribute instance-attribute

When True with CONSOLE transporter, start a background thread that prints events in real-time.

correlation_id = None class-attribute instance-attribute

Optional correlation ID that ties all events to an external request or trace.

level = ObserveLevel.STANDARD class-attribute instance-attribute

How much detail to capture.

log_file = None class-attribute instance-attribute

File path for the STRUCTURED_LOG transporter.

max_entries = 100000 class-attribute instance-attribute

Maximum timeline entries before oldest are evicted (ring buffer).

on_event = None class-attribute instance-attribute

User callback for the CALLBACK transporter. Receives a single :class:TimelineEntry argument.

otlp_endpoint = 'http://localhost:4317' class-attribute instance-attribute

gRPC endpoint for the OTLP transporter.

output_dir = None class-attribute instance-attribute

Directory for HTML and JSON output files.

prometheus_port = 9090 class-attribute instance-attribute

Port for the Prometheus metrics endpoint.

record_prompts = False class-attribute instance-attribute

When True, store full prompt/response text in metadata. Off by default for privacy.

session_name = 'promptise' class-attribute instance-attribute

Human-readable session identifier embedded in reports/logs.

transporters = field(default_factory=(lambda: [TransporterType.HTML])) class-attribute instance-attribute

Which transporter backends receive events.

webhook_headers = field(default_factory=dict) class-attribute instance-attribute

Custom HTTP headers for the WEBHOOK transporter (e.g. auth tokens).

webhook_url = None class-attribute instance-attribute

Target URL for the WEBHOOK transporter.

ObserveLevel

promptise.observability_config.ObserveLevel

Bases: str, Enum

Controls how much detail the observability system captures.

BASIC = 'basic' class-attribute instance-attribute

Tool calls + agent I/O + errors only.

FULL = 'full' class-attribute instance-attribute

Everything in STANDARD plus prompt/response content and streaming tokens.

OFF = 'off' class-attribute instance-attribute

Observability disabled.

STANDARD = 'standard' class-attribute instance-attribute

Everything in BASIC plus every LLM turn with token usage and latency.

TransporterType

promptise.observability_config.TransporterType

Bases: str, Enum

Available backends for receiving observability events.

CALLBACK = 'callback' class-attribute instance-attribute

Invoke a user-provided Python callable for each event.

CONSOLE = 'console' class-attribute instance-attribute

Real-time Rich console output with color-coded events.

HTML = 'html' class-attribute instance-attribute

Self-contained interactive HTML report (default).

JSON = 'json' class-attribute instance-attribute

JSON file export (full session dump + NDJSON streaming).

OTLP = 'otlp' class-attribute instance-attribute

OpenTelemetry span export via OTLP gRPC. Requires the [all] extra: pip install "promptise[all]".

PROMETHEUS = 'prometheus' class-attribute instance-attribute

Prometheus metrics (counters, histograms) for Grafana dashboards.

STRUCTURED_LOG = 'log' class-attribute instance-attribute

JSON log lines, one per event. Compatible with ELK, Datadog, Splunk, CloudWatch, and other enterprise logging pipelines.

WEBHOOK = 'webhook' class-attribute instance-attribute

HTTP POST each event (or batch) to a configurable URL.

ExportFormat

Backward-compatible alias for TransporterType.

Transporters

create_transporters

promptise.observability_transporters.create_transporters(config, collector)

Create and register transporter instances from an ObservabilityConfig.

This is called automatically by build_agent() when the observe parameter is enabled.

Parameters:

Name Type Description Default
config Any

An :class:ObservabilityConfig instance.

 required
collector Any

The :class:ObservabilityCollector to register with.

 required

Returns:

Type Description
list[BaseTransporter]

List of created transporter instances.

BaseTransporter

promptise.observability_transporters.BaseTransporter

Bases: ABC

Abstract base class for all observability transporters.

Subclasses must implement: - on_event(entry) — called for each timeline event in real time. - flush() — finalize / export pending data (may be async).

Optionally override close() for cleanup.

close()

Release resources. Called on shutdown.

flush() abstractmethod

Finalize and export any buffered data.

on_event(entry) abstractmethod

Process a single :class:TimelineEntry event.

HTMLReportTransporter

promptise.observability_transporters.HTMLReportTransporter

Bases: BaseTransporter

Generates a self-contained interactive HTML report on flush.

Events are buffered in memory. When :meth:flush is called (typically at the end of a session), the full report is generated via the existing generate_report() infrastructure.

Parameters:

Name Type Description Default
output_dir str

Directory for the report file. Defaults to "./reports".

 './reports'
session_name str

Embedded in the filename.

 'promptise'
flush()

Write a self-contained HTML report of collected events.

JSONFileTransporter

promptise.observability_transporters.JSONFileTransporter

Bases: BaseTransporter

Writes NDJSON lines (one per event) and a full session dump on flush.

Supports two output modes: - Streaming: Each event is appended as a single JSON line to the file. - Dump: On flush(), a full session JSON is written to a separate file.

Parameters:

Name Type Description Default
output_dir str

Directory for output files. Defaults to "./reports".

 './reports'
session_name str

Base name for output files.

 'promptise'
stream bool

If True, write NDJSON lines in real time. Default: True.

 True
flush()

Write full session JSON dump.

StructuredLogTransporter

promptise.observability_transporters.StructuredLogTransporter

Bases: BaseTransporter

Enterprise-grade structured logging — one JSON line per event.

Each event becomes a log line like::

{"timestamp":"2026-02-20T19:47:31Z", "level":"INFO", "service":"promptise",
 "session":"my-run", "agent_id":"code-reviewer", "event_type":"llm.end",
 "duration_ms":1250, "tokens":450, "message":"LLM call completed", ...}

Compatible with ELK stack, Datadog, Splunk, CloudWatch, and any JSON log ingestion pipeline.

Parameters:

Name Type Description Default
log_file str | None

File path for structured logs. If None, writes to stdout.

 None
session_name str

Embedded in each log line.

 'promptise'
service_name str

Service identifier for log aggregation.

 'promptise'
correlation_id str | None

Optional trace/request correlation ID.

 None
on_event(entry)

Format entry as structured JSON log line.

ConsoleTransporter

promptise.observability_transporters.ConsoleTransporter

Bases: BaseTransporter

Real-time color-coded console output.

Uses Rich if available for beautiful formatting, otherwise falls back to plain print().

Color coding: - Green: LLM events (start, end, stream) - Yellow: Tool events (call, result) - Red: Errors (LLM error, tool error, task failed) - Blue: Agent events (input, output) - Magenta: Phase / orchestration events - Cyan: Session events

Parameters:

Name Type Description Default
live bool

If True, enable real-time console output for each event. Default: True.

 True
verbose bool

If True, include metadata in output. Default: False.

 False
flush()

Print session summary.

PrometheusTransporter

promptise.observability_transporters.PrometheusTransporter

Bases: BaseTransporter

Bridges timeline events to Prometheus metrics.

Auto-creates and increments counters and histograms:

  • promptise_llm_calls_total (counter: agent_id, model)
  • promptise_llm_tokens_total (counter: agent_id, token_type)
  • promptise_llm_duration_seconds (histogram: agent_id, model)
  • promptise_tool_calls_total (counter: agent_id, tool_name)
  • promptise_tool_duration_seconds (histogram: agent_id, tool_name)
  • promptise_tool_errors_total (counter: agent_id, tool_name)
  • promptise_events_total (counter: event_type, category)

If the prometheus_client package is not installed, metrics are tracked internally as plain Python dicts (accessible via :attr:metrics).

Parameters:

Name Type Description Default
port int

Port for the Prometheus /metrics HTTP endpoint. If set to 0, no HTTP server is started. Default: 0 (no server).

 0

OTLPTransporter

promptise.observability_transporters.OTLPTransporter

Bases: BaseTransporter

Converts timeline entries to OpenTelemetry spans and exports via OTLP.

Each agent invocation is a parent span; each LLM call and tool call are child spans with proper parent-child relationships.

Requires the [all] extra::

pip install "promptise[all]"

Compatible with Jaeger, Datadog, Honeycomb, Grafana Tempo, and any OTLP-compatible backend.

Parameters:

Name Type Description Default
endpoint str

OTLP gRPC endpoint. Default: "http://localhost:4317".

 'http://localhost:4317'
service_name str

OpenTelemetry service name. Default: "promptise".

 'promptise'
flush()

Force-flush the span processor.

WebhookTransporter

promptise.observability_transporters.WebhookTransporter

Bases: BaseTransporter

HTTP POST each event (or batch) to a configurable URL.

Supports: - Single mode (POST per event) - Batch mode (buffer N events, flush periodically) - Custom HTTP headers (for auth tokens) - Retry with exponential backoff - Async non-blocking delivery via background thread

Useful for Slack, Discord, PagerDuty, or custom webhook integrations.

Parameters:

Name Type Description Default
url str

Target URL for the webhook.

 required
headers dict[str, str] | None

Custom HTTP headers (e.g. {"Authorization": "Bearer ..."})

 None
batch_size int

Buffer events and send in batches. 0 = send per event.

 0
max_retries int

Maximum retry attempts for failed deliveries.

 3
timeout float

HTTP request timeout in seconds.

 10.0
flush()

Send any buffered events.

CallbackTransporter

promptise.observability_transporters.CallbackTransporter

Bases: BaseTransporter

Invoke a user-provided Python callable for each event.

The simplest extensibility point — users can do anything they want with each event.

Parameters:

Name Type Description Default
callback Callable[..., Any]

A callable that receives a single :class:TimelineEntry.

 required

Callback Handler

PromptiseCallbackHandler

promptise.callback_handler.PromptiseCallbackHandler

Bases: BaseCallbackHandler

LangChain callback handler → ObservabilityCollector bridge.

Captures every LLM turn, tool call, token count, latency, retry, and error. Designed to be the only integration point needed between LangChain's event system and Promptise's observability.

get_summary()

Return a summary of all tracked metrics.

on_llm_new_token(token, *, run_id, **kwargs)

Capture streaming tokens (FULL level only).