> ## Documentation Index
> Fetch the complete documentation index at: https://agent-observability-docs.splunk.com/llms.txt
> Use this file to discover all available pages before exploring further.

# agent_stream

## 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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
def project(self) -> Project | None
```

Get the project this log stream belongs to.

### query

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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()
)