Introduction

The Events Module in xpander.ai SDK provides functionality for handling background tasks and event streaming. It integrates seamlessly with the platform for real-time task execution and monitoring.

Overview

The SDK supports comprehensive event handling systems:
  • Agent Event Listeners
    • Set up using the @on_task decorators
    • Handle incoming task execution requests
    • Integrate with Server Sent Events (SSE) for real-time communication
  • Task Event Streaming
    • Monitor updates in real-time during task execution
    • Track task progress, tool calls, and lifecycle events

Examples

Agent Event Listeners

Handle incoming task execution requests for agent deployment.

Asynchronous Example

Using @on_task decorator to create an event handler:
from xpander_sdk import on_task

@on_task
async def handle_task(task):
    print(f"Processing task: {task.id}")
    # Custom task logic
    task.result = "Task processed successfully"
    return task

Synchronous Example

from xpander_sdk import on_task

@on_task
def handle_task(task):
    print(f"Processing task: {task.id}")
    # Custom task logic
    task.result = "Task processed successfully"
    return task

Task Event Streaming

Monitor task execution in real-time.

Asynchronous Example

from xpander_sdk import Tasks, Agent
from xpander_sdk.models.events import TaskUpdateEventType

# Load agent
agent = Agent.load("agent-id")

# Create task with event streaming enabled
task = await agent.acreate_task(
    prompt="Analyze this dataset",
    file_urls=["https://example.com/data.csv"],
    events_streaming=True
)

# Stream events
async for event in task.aevents():
    print(f"Event Type: {event.type}")
    if event.type == TaskUpdateEventType.TaskFinished:
        break

Synchronous Example

# Monitor task events
for event in task.events():
    print(f"Event: {event.type}")
    if event.type == "task_finished":
        break

Continue to the Events API Reference for detailed class and method documentation.

Support

For additional help:

Contact

For further assistance, contact support: dev@xpander.ai