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
sanitizer Callable[[dict[str, Any]], dict[str, Any]] | None

Optional callable applied to every event before it is stored or sent to a transporter: it gets the event's metadata, and {"details": <text>} for its details, and returns the cleaned dict. :func:redact_sensitive is the one :func:~promptise.agent.build_agent uses. None (default) records events unchanged.

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

for_user(user_id)

Return all entries owned by user_id (sorted by timestamp).

Convenience wrapper over :meth:query for the common multi-tenant case: "show me everything user X did in this session." Returns an empty list when the user has no events.

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.

get_users()

Return sorted unique user_id values seen in the timeline.

Events without a user_id (e.g. system events recorded outside of an invocation) are excluded.

purge_user(user_id)

Delete every timeline entry owned by user_id from the buffer.

Does not propagate to already-flushed transporters (e.g. a writable JSON file on disk) — those are external sinks and must be purged by their owners. Returns the number of in-memory entries removed.

Use it to delete one user's data on request.

query(*, event_types=None, agent_ids=None, user_ids=None, session_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
user_ids list[str] | None

Restrict to events owned by these users (matches :attr:TimelineEntry.user_id).

None
session_ids list[str] | None

Restrict to events tied to these session IDs (matches :attr:TimelineEntry.session_id).

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, user_id=None, session_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. When omitted and an agent run is being observed (see :class:AgentRun), the run's agent.input entry.

None
user_id str | None

Explicit owning user for this event. When omitted, the collector reads :func:promptise.agent.get_current_caller and attaches caller.user_id automatically.

None
session_id str | None

Conversation session associated with this event. Falls back to caller.metadata['session_id'].

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
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
user_id str | None

Owning user for this event (from :class:~promptise.agent.CallerContext). None means the event was produced outside of a per-caller invocation.

None
session_id str | None

Conversation session associated with the event, when available. Populated from caller.metadata['session_id'] or an explicit session_id kwarg to :meth:record.

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.

AgentRun

promptise.observability.AgentRun dataclass

One observed agent invocation, from agent.input to agent.output.

:class:~promptise.agent.PromptiseAgent opens a run for every invocation it observes and makes it current for the duration of the call. While a run is current, :meth:ObservabilityCollector.record sets each event's parent_id to :attr:entry_id (unless one is passed explicitly), which is what links LLM turns, tool calls and cache events to the invocation that caused them — the OpenTelemetry transporter builds its span tree from it. The callback handler also adds the run's token usage and call counts here, so agent.output can report per-invocation totals.

Attributes:

Name Type Description
entry_id str

entry_id of the run's agent.input event.

started float

Epoch seconds when the run started.

prompt_tokens int

Input tokens used by this run's LLM calls.

completion_tokens int

Output tokens used by this run's LLM calls.

llm_calls int

LLM calls made during this run.

tool_calls int

Tool calls made during this run.

errors int

LLM and tool errors during this run.

cache_hit bool

Whether the run was answered from the semantic cache.

total_tokens property

Input plus output tokens used by this run.

get_current_run

promptise.observability.get_current_run()

Return the :class:AgentRun being observed in this context, if any.

redact_sensitive

promptise.observability.redact_sensitive(metadata)

Replace credentials and common PII in event metadata with placeholders.

The sanitizer :func:~promptise.agent.build_agent installs on the collector it creates (ObservabilityConfig.redact_sensitive, on by default), so tool arguments, tool results, error messages and recorded prompt text cannot carry secrets into reports, log files or exported traces. It uses the same patterns as the event notifier: API keys (sk-…), AWS access keys, GitHub tokens, Bearer tokens, credentials in URLs (://user:password@), card numbers, US social security numbers and email addresses.

String values are scrubbed at any depth; other values (numbers, flags) are kept as they are. Pattern matching cannot recognise every kind of personal data (names, addresses), so set record_tool_io=False and leave record_prompts off when traces leave your trust boundary.

Parameters:

Name Type Description Default
metadata dict[str, Any]

Event metadata (or any JSON-like dict).

required

Returns:

Type Description
dict[str, Any]

A new dict with sensitive substrings replaced.

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 = None class-attribute instance-attribute

Whether to store prompt, response, agent input and output text (truncated to 2,000 characters) in event metadata.

None (default) follows :attr:level: on at FULL, off below it. True records text at any level; False never records it, even at FULL.

record_tool_io = True class-attribute instance-attribute

Whether to store tool arguments (tool.call) and result previews (tool.result) in event metadata, truncated to 2,000 characters.

On by default — arguments and results are what make a trace debuggable. Tool arguments often carry user data (names, emails, account numbers), so set False when traces leave your trust boundary: events then keep the tool name, latency and error status, plus arguments_length and result_length. Independent of :attr:record_prompts.

redact_sensitive = True class-attribute instance-attribute

Replace credentials and common PII in every recorded event with placeholders before it is stored or exported (see :func:promptise.observability.redact_sensitive): API keys, AWS and GitHub tokens, Bearer tokens, passwords in URLs, card numbers, US social security numbers and email addresses become [API_KEY], Bearer [REDACTED], [EMAIL] and so on. Applies to the collector :func:~promptise.agent.build_agent creates; a collector passed as observer= keeps its own sanitizer. On by default; set False only when traces stay inside your trust boundary.

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.

  • OFF: nothing. The agent still has a (permanently empty) collector, so agent.get_stats() and timeline queries keep working, but no events are recorded and no transporters are created.
  • BASIC: agent.input / agent.output / agent.error per invocation (agent.output carries the run's token totals), tool calls and results, LLM and tool errors, cache events.
  • STANDARD: BASIC plus llm.start / llm.end for every LLM turn, with token usage, latency and the tools the model asked for.
  • FULL: STANDARD plus prompt, response, agent input and output text (unless record_prompts=False) and streamed-token counts.

Tool arguments and results are controlled separately by :attr:ObservabilityConfig.record_tool_io.

BASIC = 'basic' class-attribute instance-attribute

Agent input/output, tool calls, errors and cache events; no per-LLM-turn events.

FULL = 'full' class-attribute instance-attribute

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

OFF = 'off' class-attribute instance-attribute

Observability disabled: nothing is recorded.

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.

Events are read from the collector when the report is written, so nothing is buffered here. :meth:flush (called on agent shutdown) writes <output_dir>/<session_name>-report-<timestamp>.html; :meth:write writes to an exact path, which is what :meth:PromptiseAgent.generate_report uses.

The page shows the session's stats (taken from :meth:ObservabilityCollector.get_stats, so they match agent.get_stats()), a filterable timeline, and each event's metadata on click. All data is embedded; the page loads nothing.

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'
title str

Page title and heading.

'Promptise Agent Report'
flush()

Write a timestamped report into :attr:output_dir.

write(path)

Write the report to exactly path and return it.

Creates missing parent directories and overwrites an existing file.

Raises:

Type Description
RuntimeError

When no collector is attached.

OSError

When the file cannot be written.

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 <output_dir>/<session_name>-events.ndjson. - Dump: On flush(), a full session JSON is written to a separate, timestamped file.

The NDJSON file is opened in append mode, so events from every run with the same session_name accumulate in it.

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

Exports agent runs to OpenTelemetry as traces, via OTLP gRPC.

Each agent invocation becomes one trace: an invoke_agent span from agent.input to agent.output (or agent.error), with a chat <model> child span per LLM call (llm.start → llm.end) and an execute_tool <name> child span per tool call (tool.call → tool.result / tool.error). Spans carry the real start and end times of the work, so their duration is the latency. Other events recorded during the run (cache hits, retries, approvals …) become span events on the run span; events outside a run become standalone spans.

If an OpenTelemetry span is active when the agent is invoked (for example the server span of an instrumented web framework), the run span becomes its child, so the agent run joins the request's trace.

Attributes follow the OpenTelemetry GenAI semantic conventions where one exists (gen_ai.operation.name, gen_ai.agent.name, gen_ai.request.model, gen_ai.usage.input_tokens / output_tokens, gen_ai.tool.name) plus enduser.id and session.id from the :class:~promptise.agent.CallerContext, promptise.correlation_id, and each event's scalar metadata as promptise.<key>. Failed LLM calls, tool calls and runs get an ERROR status.

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'
correlation_id str | None

Optional ID added to every span as promptise.correlation_id.

None
tracer_provider Any | None

An existing TracerProvider to emit spans through (for example your application's). When given, endpoint and service_name are not used and the provider is flushed but not shut down by :meth:close. When omitted, a private provider with an OTLP exporter is created; the global tracer provider is never replaced.

None
close()

End spans still open, export them, and shut down the private provider.

flush()

Export finished spans now.

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.

What is recorded depends on level (see :class:~promptise.observability_config.ObserveLevel): OFF records nothing, BASIC skips the per-turn llm.start / llm.end events (token usage is still added to the current agent run), STANDARD records every turn, and FULL also records prompt and response text unless record_prompts=False.

Parameters:

Name Type Description Default
collector Any

The :class:~promptise.observability.ObservabilityCollector.

required
agent_id str | None

Agent identifier stamped on every event.

None
record_prompts bool | None

Record prompt and response text. None follows level (on at FULL only); True / False force it.

None
level ObserveLevel

Detail level.

STANDARD
record_tool_io bool

Record tool arguments and result previews. When False, only their lengths are recorded.

True
get_summary()

Return a summary of all tracked metrics.

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

Capture streaming tokens (FULL level only).

token is typed to match LangChain-core's LLMManagerMixin base, which widened it from str to str | list[str | dict] in the 1.x line. The wider annotation satisfies the override contract on both the 0.3.x and 1.x major versions.