Skip to content

Core API

The core API provides the fundamental building blocks for tracing AI agent applications.

prela.init()

prela.init(service_name=None, exporter=None, auto_instrument=True, sample_rate=None, capture_for_replay=False, project_id=None, n8n_webhook_port=None, n8n_webhook_host='0.0.0.0', **kwargs)

Initialize Prela tracing with one line of code.

This is the primary entry point for the Prela SDK. It: 1. Creates a tracer with the specified configuration 2. Sets it as the global tracer 3. Auto-instruments detected LLM SDKs (Anthropic, OpenAI, etc.) 4. Optionally starts n8n webhook receiver for zero-code workflow tracing 5. Returns the tracer for manual span creation

After calling init(), all LLM SDK calls are automatically traced!

Parameters:

Name Type Description Default
service_name str | None

Name of your service (default: $PRELA_SERVICE_NAME or "default")

None
exporter str | BaseExporter | None

Where to send traces: - "console": Pretty-print to console (default) - "file": Write to JSONL file - "http": Send to HTTP endpoint (Railway, cloud backend) - BaseExporter instance: Custom exporter - Default: $PRELA_EXPORTER or "console"

None
auto_instrument bool

Whether to auto-instrument detected libraries (default: True, disable with $PRELA_AUTO_INSTRUMENT=false)

True
sample_rate float | None

Sampling rate 0.0-1.0 (default: $PRELA_SAMPLE_RATE or 1.0)

None
capture_for_replay bool

Enable full replay data capture (default: False) When enabled, captures complete request/response data including: - LLM: Full prompts, responses, streaming chunks, model info - Tools: Input args, output, side effects flag - Retrieval: Queries, documents, scores, metadata - Agents: System prompts, available tools, memory, config Use for debugging, testing, and auditing. Increases storage costs.

False
project_id str | None

Project ID for multi-tenant deployments (default: $PRELA_PROJECT_ID or None) Used for: - Organizing traces in multi-project deployments - Filtering dashboards by project - n8n webhook routing (?project={project_id})

None
n8n_webhook_port int | None

Port for n8n webhook receiver (optional, default: None) Set to enable n8n webhook-based tracing (e.g., 8787) Example: n8n_webhook_port=8787

None
n8n_webhook_host str

Host for n8n webhook receiver (default: "0.0.0.0") Usually "0.0.0.0" for accepting external connections

'0.0.0.0'
**kwargs Any

Additional arguments passed to exporter For ConsoleExporter: verbosity, color, show_timestamps For FileExporter: directory, format, max_file_size_mb, rotate For HTTPExporter: endpoint, api_key, bearer_token, compress, headers

{}

Returns:

Type Description
Tracer

Configured Tracer instance (also set as global tracer)

Environment Variables

PRELA_SERVICE_NAME: Default service name PRELA_PROJECT_ID: Default project ID for multi-tenant setups PRELA_EXPORTER: Default exporter ("console", "file", or "http") PRELA_SAMPLE_RATE: Default sampling rate (0.0-1.0) PRELA_CAPTURE_REPLAY: Enable replay capture ("true", "1", or "yes") PRELA_AUTO_INSTRUMENT: Enable auto-instrumentation ("true" or "false") PRELA_DEBUG: Enable debug logging ("true" or "false") PRELA_TRACE_DIR: Directory for file exporter (default: ./traces) PRELA_HTTP_ENDPOINT: HTTP endpoint for http exporter PRELA_API_KEY: API key for http exporter PRELA_N8N_WEBHOOK_PORT: Port for n8n webhook receiver (optional)

Example
import prela

# Simple initialization
prela.init(service_name="my-agent")

# All Anthropic/OpenAI calls now auto-traced!
from anthropic import Anthropic
client = Anthropic()
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}]
)
# Trace is automatically captured and exported

# Manual span creation
with prela.get_tracer().span("custom_operation") as span:
    span.set_attribute("key", "value")
    # Do work...

Example with console exporter (verbose mode):

import prela

prela.init(
    service_name="my-agent",
    exporter="console",
    verbosity="verbose",  # "minimal", "normal", or "verbose"
    color=True,
    show_timestamps=True
)

Example with file exporter
import prela

prela.init(
    service_name="my-agent",
    exporter="file",
    directory="./traces",
    max_file_size_mb=100,  # 100 MB per file
    rotate=True
)

Example with HTTP exporter (Railway deployment):

import prela

prela.init(
    service_name="my-agent",
    exporter="http",
    endpoint="https://prela-ingest-gateway-xxx.railway.app/v1/traces",
    api_key="your-api-key",  # Optional
    compress=True  # Enable gzip compression
)

Example with n8n webhook receiver
import prela

# Start webhook receiver on port 8787
prela.init(
    service_name="n8n-workflows",
    exporter="http",
    endpoint="https://prela-ingest-gateway-xxx.railway.app/v1/traces",
    n8n_webhook_port=8787
)

# Now configure n8n HTTP Request node to POST to:
# http://your-server:8787/webhook
# Body: {"workflow": "{{ $workflow }}", "execution": "{{ $execution }}", ...}
Example with custom exporter
from prela import init, BaseExporter, ExportResult

class MyExporter(BaseExporter):
    def export(self, spans):
        # Send to your backend
        return ExportResult.SUCCESS

init(service_name="my-agent", exporter=MyExporter())

Tracer

prela.core.tracer.Tracer

Main tracer for creating and managing spans.

The Tracer is responsible for: - Creating spans with proper trace/span IDs - Managing trace context and span hierarchies - Applying sampling decisions - Exporting completed spans

Example
from prela.core.tracer import Tracer
from prela.exporters.console import ConsoleExporter

tracer = Tracer(
    service_name="my-agent",
    exporter=ConsoleExporter()
)

# Create spans using context manager
with tracer.span("operation") as span:
    span.set_attribute("key", "value")
    # Nested spans inherit trace context
    with tracer.span("sub-operation") as child:
        child.set_attribute("nested", True)

Functions

__init__(service_name='default', exporter=None, sampler=None, capture_for_replay=False)

Initialize a tracer.

Parameters:

Name Type Description Default
service_name str

Name of the service (added to all spans as service.name)

'default'
exporter BaseExporter | None

Exporter for sending spans to backend (None = no export)

None
sampler BaseSampler | None

Sampler for controlling trace volume (default: AlwaysOnSampler)

None
capture_for_replay bool

If True, capture full replay data (default: False)

False

span(name, span_type=SpanType.CUSTOM, attributes=None)

Create a new span as a context manager.

The span is automatically: - Started when entering the context - Ended when exiting the context - Exported if it's a root span and sampling decision is True - Linked to parent span if one exists in current context

Exceptions are automatically captured and recorded on the span.

Parameters:

Name Type Description Default
name str

Name of the span (e.g., "process_request", "llm_call")

required
span_type SpanType

Type of operation (LLM, TOOL, AGENT, etc.)

CUSTOM
attributes dict[str, Any] | None

Initial attributes to set on the span

None

Yields:

Name Type Description
Span Span

The created span (can be used to add attributes/events)

Example
with tracer.span("database_query", SpanType.CUSTOM) as span:
    span.set_attribute("query", "SELECT * FROM users")
    result = execute_query()
    span.set_attribute("row_count", len(result))

set_global()

Set this tracer as the global default.

After calling this, get_tracer() will return this tracer instance. This is useful for auto-instrumentation where instrumentors need access to a tracer without explicit passing.

Example
tracer = Tracer(service_name="my-app")
tracer.set_global()

# Later, from anywhere in the code
from prela.core.tracer import get_tracer
tracer = get_tracer()

Span

prela.core.span.Span

A span represents a unit of work in a distributed trace.

Spans are immutable after being ended. Any attempt to modify an ended span will raise a RuntimeError.

Functions

__init__(span_id=None, trace_id=None, parent_span_id=None, name='', span_type=SpanType.CUSTOM, started_at=None, ended_at=None, status=SpanStatus.PENDING, status_message=None, attributes=None, events=None, _ended=False, replay_snapshot=None)

Initialize a new span.

Parameters:

Name Type Description Default
span_id str | None

Unique identifier for this span (generates UUID if not provided)

None
trace_id str | None

Trace ID this span belongs to (generates UUID if not provided)

None
parent_span_id str | None

Parent span ID if this is a child span

None
name str

Human-readable name for this span

''
span_type SpanType

Type of operation this span represents

CUSTOM
started_at datetime | None

When the span started (uses current time if not provided)

None
ended_at datetime | None

When the span ended (None if still running)

None
status SpanStatus

Current status of the span

PENDING
status_message str | None

Optional message describing the status

None
attributes dict[str, Any] | None

Key-value pairs of metadata

None
events list[SpanEvent] | None

List of events that occurred during span execution

None
_ended bool

Internal flag for immutability (do not set manually)

False
replay_snapshot Any

Optional replay data for deterministic re-execution

None

end(end_time=None)

End the span.

Parameters:

Name Type Description Default
end_time datetime | None

When the span ended (uses current time if not provided)

None

Raises:

Type Description
RuntimeError

If the span has already ended

set_attribute(key, value)

Set an attribute on the span.

Parameters:

Name Type Description Default
key str

Attribute key

required
value Any

Attribute value

required

Raises:

Type Description
RuntimeError

If the span has already ended

add_event(name, attributes=None)

Add an event to the span.

Parameters:

Name Type Description Default
name str

Event name

required
attributes dict[str, Any] | None

Optional event attributes

None

Raises:

Type Description
RuntimeError

If the span has already ended

to_dict()

Convert span to dictionary representation.

Returns:

Type Description
dict[str, Any]

Dictionary containing all span data

from_dict(data) classmethod

Create span from dictionary representation.

Parameters:

Name Type Description Default
data dict[str, Any]

Dictionary containing span data

required

Returns:

Type Description
Span

Reconstructed Span instance

SpanEvent

prela.core.span.SpanEvent dataclass

An event that occurred during a span's execution.

Functions

to_dict()

Convert event to dictionary representation.

from_dict(data) classmethod

Create event from dictionary representation.

SpanType

prela.core.span.SpanType

Bases: Enum

Type of span in the trace.

SpanStatus

prela.core.span.SpanStatus

Bases: Enum

Status of a span.

Context Management

prela.core.context.get_current_context()

Get the current trace context.

Returns:

Type Description
TraceContext | None

The active trace context, or None if no context is active

prela.core.context.get_current_span()

Get the currently active span.

Returns:

Type Description
Span | None

The active span, or None if no context or no active span

prela.core.context.new_trace_context(trace_id=None, sampled=True, baggage=None)

Create a new trace context for the duration of the context manager.

This context manager creates a new trace context, sets it as active, yields it for use, and automatically resets it on exit.

Parameters:

Name Type Description Default
trace_id str | None

Unique identifier for this trace (generates UUID if not provided)

None
sampled bool

Whether this trace should be sampled/recorded

True
baggage dict[str, str] | None

Initial baggage metadata to propagate

None

Yields:

Type Description
TraceContext

The newly created trace context

Example

with new_trace_context() as ctx: ... span = Span(name="operation", trace_id=ctx.trace_id) ... ctx.push_span(span) ... # Do work ... ctx.pop_span()

prela.core.context.copy_context_to_thread(func)

Create a wrapper that copies the current context to a new thread.

This function captures the current contextvars context at call time and creates a wrapper that will run the function in that context. This is essential for maintaining trace continuity when using thread pools.

IMPORTANT: Call this function INSIDE the context you want to propagate, BEFORE submitting to the thread pool.

Parameters:

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

The function to wrap

required

Returns:

Type Description
Callable[..., Any]

Wrapped function that will run in the captured context

Example

def background_task(): ... span = get_current_span() ... print(f"Span: {span}")

with new_trace_context() as ctx: ... # Capture context NOW, before submitting to pool ... wrapped = copy_context_to_thread(background_task) ... with ThreadPoolExecutor() as executor: ... future = executor.submit(wrapped) ... future.result()

prela.core.context.get_current_trace_id()

Get the current trace ID.

Returns:

Type Description
str | None

The active trace ID, or None if no context is active

prela.core.context.set_context(ctx)

Set the current trace context.

Parameters:

Name Type Description Default
ctx TraceContext

The trace context to set as active

required

Returns:

Type Description
Token[TraceContext | None]

A token that can be used to reset the context

prela.core.context.reset_context(token)

Reset the context to its previous value.

Parameters:

Name Type Description Default
token Token[TraceContext | None]

The token returned by set_context

required

TraceContext

prela.core.context.TraceContext

A trace context manages the current trace and span stack.

This class maintains the active trace ID, a stack of active spans, and baggage (inherited metadata) that propagates through the trace.

Functions

__init__(trace_id=None, sampled=True, baggage=None)

Initialize a new trace context.

Parameters:

Name Type Description Default
trace_id str | None

Unique identifier for this trace (generates UUID if not provided)

None
sampled bool

Whether this trace should be sampled/recorded

True
baggage dict[str, str] | None

Initial baggage metadata to propagate

None

push_span(span)

Push a span onto the stack.

Parameters:

Name Type Description Default
span Span

The span to make active

required

pop_span()

Pop the current span from the stack.

Returns:

Type Description
Span | None

The popped span, or None if stack was empty

current_span()

Get the currently active span.

Returns:

Type Description
Span | None

The span at the top of the stack, or None if stack is empty

Sampling

prela.core.sampler.BaseSampler

Bases: ABC

Abstract base class for trace samplers.

Samplers determine whether a trace should be collected based on the trace ID and potentially other factors.

Functions

should_sample(trace_id) abstractmethod

Determine if a trace should be sampled.

Parameters:

Name Type Description Default
trace_id str

The trace ID to make a sampling decision for

required

Returns:

Type Description
bool

True if the trace should be sampled, False otherwise

prela.core.sampler.AlwaysOnSampler

Bases: BaseSampler

Sampler that always samples every trace.

Use this in development or when you need complete trace coverage. Be aware this may generate high data volumes in production.

Functions

should_sample(trace_id)

Always return True.

Parameters:

Name Type Description Default
trace_id str

The trace ID (unused)

required

Returns:

Type Description
bool

Always True

prela.core.sampler.AlwaysOffSampler

Bases: BaseSampler

Sampler that never samples any traces.

Use this to completely disable tracing, for example during maintenance windows or in testing environments.

Functions

should_sample(trace_id)

Always return False.

Parameters:

Name Type Description Default
trace_id str

The trace ID (unused)

required

Returns:

Type Description
bool

Always False

prela.core.sampler.ProbabilitySampler

Bases: BaseSampler

Sampler that samples traces with a fixed probability.

This sampler uses a deterministic hash-based approach to ensure consistent sampling decisions for the same trace ID across different services and processes.

Functions

__init__(rate)

Initialize the probability sampler.

Parameters:

Name Type Description Default
rate float

Sampling rate between 0.0 and 1.0 (inclusive)

required

Raises:

Type Description
ValueError

If rate is not between 0.0 and 1.0

should_sample(trace_id)

Sample based on trace ID hash.

Uses MD5 hash of trace_id to make a deterministic sampling decision. This ensures the same trace_id always gets the same sampling decision across different processes and services.

Parameters:

Name Type Description Default
trace_id str

The trace ID to make a sampling decision for

required

Returns:

Type Description
bool

True if the trace should be sampled, False otherwise

prela.core.sampler.RateLimitingSampler

Bases: BaseSampler

Sampler that limits the number of traces sampled per second.

This sampler uses a token bucket algorithm to enforce a maximum rate of sampled traces per second. Useful for controlling costs and backend load.

Functions

__init__(traces_per_second)

Initialize the rate limiting sampler.

Parameters:

Name Type Description Default
traces_per_second float

Maximum number of traces to sample per second

required

Raises:

Type Description
ValueError

If traces_per_second is negative

should_sample(trace_id)

Sample if tokens are available.

Uses a token bucket algorithm: tokens regenerate at the configured rate, and each sampling decision consumes one token.

Parameters:

Name Type Description Default
trace_id str

The trace ID (unused)

required

Returns:

Type Description
bool

True if a token is available, False otherwise

Clock Utilities

prela.core.clock.now()

Get the current UTC time with microsecond precision.

Returns:

Type Description
datetime

Current datetime in UTC with microsecond precision

Example

timestamp = now() timestamp.tzinfo == timezone.utc True

prela.core.clock.monotonic_ns()

Get monotonic time in nanoseconds.

This is useful for measuring durations as it's not affected by system clock adjustments.

Returns:

Type Description
int

Monotonic time in nanoseconds

Example

start = monotonic_ns()

... do work ...

end = monotonic_ns() elapsed_ms = duration_ms(start, end)

prela.core.clock.duration_ms(start_ns, end_ns)

Calculate duration in milliseconds from nanosecond timestamps.

Parameters:

Name Type Description Default
start_ns int

Start time in nanoseconds (from monotonic_ns)

required
end_ns int

End time in nanoseconds (from monotonic_ns)

required

Returns:

Type Description
float

Duration in milliseconds

Example

start = monotonic_ns() end = start + 1_500_000 # 1.5ms later duration_ms(start, end) 1.5

prela.core.clock.format_timestamp(dt)

Format a datetime as ISO 8601 string.

Parameters:

Name Type Description Default
dt datetime

Datetime to format

required

Returns:

Type Description
str

ISO 8601 formatted string

Example

dt = datetime(2024, 1, 15, 12, 30, 45, 123456, tzinfo=timezone.utc) format_timestamp(dt) '2024-01-15T12:30:45.123456+00:00'

prela.core.clock.parse_timestamp(s)

Parse an ISO 8601 timestamp string.

Parameters:

Name Type Description Default
s str

ISO 8601 formatted timestamp string

required

Returns:

Type Description
datetime

Parsed datetime object

Raises:

Type Description
ValueError

If the string is not a valid ISO 8601 timestamp

Example

dt = parse_timestamp('2024-01-15T12:30:45.123456+00:00') dt.year 2024