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

# Stream AgentRail task lifecycle events — Python SDK

> Use client.stream_events with StreamOptions to receive real-time task lifecycle events as Pydantic models, with cursor-based reconnection.

The `stream_events` method opens a server-sent event (SSE) connection and yields `TaskLifecycleEvent` Pydantic models as they arrive. Pass a `StreamOptions` instance to filter by event type, set a heartbeat interval, and resume from a cursor after reconnection.

## Basic usage

```python theme={null}
from agentrail import StreamOptions, TaskEventType

async for event in client.stream_events(
    StreamOptions(
        event_types=[
            TaskEventType.TASK_UPDATED.value,
            TaskEventType.TASK_REVIEWED.value,
            TaskEventType.TASK_SHIPPED.value,
        ],
        heartbeat_seconds=30,
    )
):
    print(event.id, event.type)
```

The `async for` loop runs until the server closes the stream or an error is raised. Each `event` is a validated `TaskLifecycleEvent` with `id`, `type`, and `data` fields.

## Reconnect with a cursor

After a disconnect, pass the last received event ID as `cursor` to resume the stream without missing events:

```python theme={null}
from agentrail import StreamOptions, TaskEventType

last_event_id: str | None = None

async for event in client.stream_events(
    StreamOptions(
        event_types=[
            TaskEventType.TASK_UPDATED.value,
            TaskEventType.TASK_REVIEWED.value,
            TaskEventType.TASK_SHIPPED.value,
        ],
        heartbeat_seconds=30,
        cursor=last_event_id,
    )
):
    last_event_id = event.id
    print(event.id, event.type)
```

The server replays any events that occurred after the cursor position.

## Event types

| `TaskEventType` value         | String value    | When it fires                                  |
| ----------------------------- | --------------- | ---------------------------------------------- |
| `TaskEventType.TASK_UPDATED`  | `task.updated`  | The task's status, branch, or metadata changed |
| `TaskEventType.TASK_REVIEWED` | `task.reviewed` | A review decision was posted                   |
| `TaskEventType.TASK_SHIPPED`  | `task.shipped`  | The task was shipped (merged and deployed)     |

<Tip>
  Set `heartbeat_seconds` to a value lower than your infrastructure's idle connection timeout. A value of `30` works well for most deployments.
</Tip>

## StreamOptions fields

| Field               | Type                | Description                                                    |
| ------------------- | ------------------- | -------------------------------------------------------------- |
| `event_types`       | `list[str] \| None` | Event type strings to subscribe to. `None` receives all types. |
| `task_id`           | `str \| None`       | Scope the stream to a single task.                             |
| `cursor`            | `str \| None`       | Resume from this event ID.                                     |
| `heartbeat_seconds` | `int \| None`       | Interval for server-sent heartbeat pings.                      |

## Required scope

Streaming events requires the `events:read` scope on your API key. A missing scope raises an `InsufficientScopeError` before the stream opens.