AgentStream

Object-centric interface for Splunk AO log streams. This class provides an intuitive way to work with Splunk AO log streams, offering methods for managing log streams and their associated metrics. Examples

Create a new log stream and persist it

agent_stream = AgentStream(name=“Production Logs”, project_name=“My AI Project”).create()

Get an existing log stream

agent_stream = AgentStream.get(name=“Production Logs”, project_name=“My AI Project”)

AgentStreams can also be created through Project instances

from splunk_ao.project import Project project = Project.get(name=“My AI Project”) agent_stream = project.create_agent_stream(name=“Production Logs”)

Enable evaluators on the agent stream

from splunk_ao.schema.metrics import SplunkAOEvaluators local_evaluators = agent_stream.set_metrics([ SplunkAOEvaluators.correctness, SplunkAOEvaluators.completeness, “context_relevance” ])

Refresh agent stream state from API

agent_stream.refresh()

context

def context(self) -> Any

Get a Splunk AO context manager for this log stream. This is a convenient method that returns a pre-configured splunk_ao_context for this log stream, eliminating the need to specify project and log stream names. Examples agent_stream = AgentStream.get( name=“Production Logs”, project_name=“My AI Project” ) with agent_stream.context():

Your logging code here

response = openai_client.chat.completions.create(…)

create

def create(self) -> AgentStream

Persist this log stream to the API. Examples agent_stream = AgentStream(name=“Production Logs”, project_name=“My AI Project”).create() assert agent_stream.is_synced()

export_records

def export_records(self,
                   record_type: RecordType=RecordType.TRACE,
                   filters: builtins.list[FilterType] | None=None,
                   sort: LogRecordsSortClause=LogRecordsSortClause(column_id='created_at', ascending=False),
                   export_format: LLMExportFormat=LLMExportFormat.JSONL,
                   column_ids: builtins.list[str] | None=None,
                   redact: bool=True) -> Iterator[dict[str, Any]]

Export records from this log stream. This method provides a convenient way to export records without needing to specify project_id or log_stream_id. Arguments

record_type: The type of records to export (SPAN, TRACE, or SESSION).
filters: A list of filters to apply to the export.
sort: A sort clause to order the exported records.
export_format: The desired format for the exported data.
column_ids: A list of column IDs to include in the export.
redact: Redact sensitive data from the response.

get

def get(cls,
        *,
        name: str,
        project_id: str | None=None,
        project_name: str | None=None) -> AgentStream | None

Get an existing log stream by name. Arguments

name (str): The log stream name.
project_id (Optional[str]): The project ID. If neither project_id nor project_name is provided, falls back to SPLUNK_AO_PROJECT_ID or SPLUNK_AO_PROJECT environment variables.
project_name (Optional[str]): The project name. If neither project_id nor project_name is provided, falls back to SPLUNK_AO_PROJECT environment variable.

get_metrics

def get_metrics(self) -> builtins.list[str]

Get the list of evaluators currently enabled on this agent stream. Examples agent_stream = AgentStream.get(name=“Production Logs”, project_name=“My Project”) current_evaluators = agent_stream.get_metrics() print(f”Currently enabled: {current_evaluators}“)

get_sessions

def get_sessions(self,
                 filters: builtins.list[FilterType] | None=None,
                 sort: LogRecordsSortClause | None=None,
                 limit: int=100,
                 starting_token: int=0) -> QueryResult

Query sessions in this log stream. This is a convenience method that queries for sessions specifically. Arguments

filters: A list of filters to apply to the query.
sort: A sort clause to order the query results.
limit: The maximum number of records to return.
starting_token: The token for the next page of results.

get_spans

def get_spans(self,
              filters: builtins.list[FilterType] | None=None,
              sort: LogRecordsSortClause | None=None,
              limit: int=100,
              starting_token: int=0) -> QueryResult

Query spans in this log stream. This is a convenience method that queries for spans specifically. Arguments

filters: A list of filters to apply to the query.
sort: A sort clause to order the query results.
limit: The maximum number of records to return.
starting_token: The token for the next page of results.

get_traces

def get_traces(self,
               filters: builtins.list[FilterType] | None=None,
               sort: LogRecordsSortClause | None=None,
               limit: int=100,
               starting_token: int=0) -> QueryResult

Query traces in this log stream. This is a convenience method that queries for traces specifically. Arguments

filters: A list of filters to apply to the query.
sort: A sort clause to order the query results.
limit: The maximum number of records to return.
starting_token: The token for the next page of results.

list

def list(cls,
         *,
         project_id: str | None=None,
         project_name: str | None=None,
         limit: Unset | int=100,
         starting_token: Unset | int=0) -> list[AgentStream]

List log streams for a project. Returns a single page of results. Use starting_token (from next_starting_token on a prior response) to fetch subsequent pages. Arguments

project_id (Optional[str]): The project ID. If neither project_id nor project_name is provided, falls back to SPLUNK_AO_PROJECT_ID or SPLUNK_AO_PROJECT environment variables.
project_name (Optional[str]): The project name. If neither project_id nor project_name is provided, falls back to SPLUNK_AO_PROJECT environment variable.
limit (Union[Unset, int]): Maximum number of log streams to return per page. Defaults to 100.
starting_token (Union[Unset, int]): Pagination token to start from. Defaults to 0 (first page).

project

def project(self) -> Project | None

Get the project this log stream belongs to.

query

def query(self,
          record_type: RecordType,
          filters: builtins.list[FilterType] | None=None,
          sort: LogRecordsSortClause | None=None,
          limit: int=100,
          starting_token: int=0) -> QueryResult

Query records in this log stream. This method provides a convenient way to search spans, traces, or sessions within the current log stream without needing to specify project_id or log_stream_id. Arguments

record_type: The type of records to query (SPAN, TRACE, or SESSION).
filters: A list of filters to apply to the query.
sort: A sort clause to order the query results.
limit: The maximum number of records to return.
starting_token: The token for the next page of results.

refresh

def refresh(self) -> None

Refresh this log stream’s state from the API. Updates all attributes with the latest values from the remote API and sets the state to SYNCED. Examples agent_stream.refresh() assert agent_stream.is_synced()

session_columns

def session_columns(self) -> ColumnCollection

Get available columns for sessions in this log stream. Examples agent_stream = AgentStream.get(name=“Production Logs”, project_name=“My AI Project”) columns = agent_stream.session_columns

Access a specific column

model_column = columns[“model”]

Filter using columns

sessions = agent_stream.get_sessions( filters=[columns[“model”].equals(“gpt-4o-mini”)], sort=columns[“created_at”].descending() )

set_metrics

def set_metrics(self,
                metrics: builtins.list[SplunkAOEvaluators | Metric | LocalMetricConfig | str]) -> builtins.list[LocalMetricConfig]

Set (replace) the evaluators on this agent stream. This replaces any existing evaluators with the new list. The metrics parameter name is retained for API compatibility. Arguments

metrics: List of evaluators to set. Supports:
- SplunkAOEvaluators enum values (e.g., SplunkAOEvaluators.correctness)
- Metric objects (including from Metric.get(id=”…”))
- LocalMetricConfig objects for custom scoring functions
- String names of built-in evaluators

span_columns

def span_columns(self) -> ColumnCollection

Get available columns for spans in this log stream. Examples agent_stream = AgentStream.get(name=“Production Logs”, project_name=“My AI Project”) columns = agent_stream.span_columns

Access a specific column

input_column = columns[“input”]

Filter using columns

spans = agent_stream.get_spans( filters=[columns[“input”].contains(“world”)], sort=columns[“created_at”].descending() )

trace_columns

def trace_columns(self) -> ColumnCollection

Get available columns for traces in this log stream. Examples agent_stream = AgentStream.get(name=“Production Logs”, project_name=“My AI Project”) columns = agent_stream.trace_columns

Access a specific column

input_column = columns[“input”]

Filter using columns

traces = agent_stream.get_traces( filters=[columns[“input”].contains(“largest”)], sort=columns[“created_at”].descending() )

Overview

Python SDK

API

agent_stream

AgentStream

Create a new log stream and persist it

Get an existing log stream

AgentStreams can also be created through Project instances

Enable evaluators on the agent stream

Refresh agent stream state from API

context

Your logging code here

create

export_records

get

get_metrics

get_sessions

get_spans

get_traces

list

project

query

refresh

session_columns

Access a specific column

Filter using columns

set_metrics

span_columns

Access a specific column

Filter using columns

trace_columns

Access a specific column

Filter using columns

​AgentStream

​Create a new log stream and persist it

​Get an existing log stream

​AgentStreams can also be created through Project instances

​Enable evaluators on the agent stream

​Refresh agent stream state from API

​context

​Your logging code here

​create

​export_records

​get

​get_metrics

​get_sessions

​get_spans

​get_traces

​list

​project

​query

​refresh

​session_columns

​Access a specific column

​Filter using columns

​set_metrics

​span_columns

​Access a specific column

​Filter using columns

​trace_columns

​Access a specific column

​Filter using columns

AgentStream

Create a new log stream and persist it

Get an existing log stream

AgentStreams can also be created through Project instances

Enable evaluators on the agent stream

Refresh agent stream state from API

context

Your logging code here

create

export_records

get

get_metrics

get_sessions

get_spans

get_traces

list

project

query

refresh

session_columns

Access a specific column

Filter using columns

set_metrics

span_columns

Access a specific column

Filter using columns

trace_columns

Access a specific column

Filter using columns