Skip to main content

​
List Events

List events for a task. 
events = client.events.list(
    task_id="task-uuid",
    agent_id="agent-uuid",
    limit=50,
)
Parameters:
  • task_id (str): Task UUID
  • agent_id (str, optional): Filter by agent
  • limit (int, optional): Max results
  • page_number (int, optional): Page number
  • last_processed_event_id (str, optional): Get events after this ID
  • order_by (str, optional): Sort field
  • order_direction (str, optional): “asc” or “desc”
Returns: EventListResponse

​
Retrieve Event

Get an event by ID. 
event = client.events.retrieve("event-uuid")
Parameters:
  • event_id (str): Event UUID
Returns: Event

​
Event Model

event.id          # Event UUID
event.task_id     # Parent task
event.content     # Event content
event.sequence_id # Order sequence
event.created_at  # Creation timestamp

​
Event Content

Events contain content objects:

​
TextContent

if isinstance(event.content, TextContent):
    text = event.content.content
    author = event.content.author  # "user" or "agent"

​
DataContent

if isinstance(event.content, DataContent):
    data = event.content.data  # dict

​
Polling Pattern

For agents that poll for events: 
last_seen = None

while task_running:
    events = await client.events.list(
        task_id=task_id,
        agent_id=agent_id,
        last_processed_event_id=last_seen,
        limit=10,
    )

    for event in events:
        await process_event(event)
        last_seen = event.id

​
Usage in Agent Handlers

Events are delivered to handlers automatically: 
@server.on_event
async def handle_event(ctx: TaskContext, event: Event):
    # event is already the current event
    print(f"Received: {event.id}")
    print(f"Content: {event.content}")
To list all events in a handler: 
@server.on_event
async def handle_event(ctx: TaskContext, event: Event):
    all_events = await ctx.events.list()
    print(f"Total events: {len(all_events)}")