Stream AgentRail task lifecycle events — Python SDK

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

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:

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)

Set heartbeat_seconds to a value lower than your infrastructure’s idle connection timeout. A value of 30 works well for most deployments.

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.

Overview

TypeScript

Python

Stream AgentRail task lifecycle events — Python SDK

Basic usage

Reconnect with a cursor

Event types

StreamOptions fields

Required scope

​Basic usage

​Reconnect with a cursor

​Event types

​StreamOptions fields

​Required scope

Basic usage

Reconnect with a cursor

Event types

StreamOptions fields

Required scope