​
Introduction

The Tasks Module provides comprehensive functionality for managing task execution, monitoring progress, and streaming real-time events within the xpander.ai platform.

​
Overview

In this module, you can:
  • Create and execute tasks with detailed configuration options
  • Monitor task execution status and progress in real-time
  • Stream task events for responsive application development
  • Manage task lifecycle including stopping and updating tasks
  • Handle both synchronous and asynchronous task operations

​
Examples

​
Creating and Executing Tasks

This example demonstrates how to create and execute tasks with both asynchronous and synchronous approaches.

​
Asynchronous Example

Using acreate() to create tasks asynchronously: 
from xpander_sdk import Tasks

tasks = Tasks()
task = await tasks.acreate(
    agent_id="agent-123",
    prompt="Analyze the sales data",
    file_urls=["https://example.com/sales_data.csv"],
    events_streaming=True
)
print(f"Created task: {task.id}")

​
Synchronous Example

Using create() to create tasks synchronously: 
from xpander_sdk import Tasks

tasks = Tasks()
task = tasks.create(
    agent_id="agent-123",
    prompt="Analyze the sales data",
    file_urls=["https://example.com/sales_data.csv"]
)
print(f"Created task: {task.id}")

​
Retrieving Task Status

Learn how to retrieve and monitor task execution status.

​
Asynchronous Example

task = await tasks.aget("task-456")
print(f"Task status: {task.status}")
print(f"Task result: {task.result}")

​
Synchronous Example

task = tasks.get("task-456")
print(f"Task status: {task.status}")
print(f"Task result: {task.result}")

​
Streaming Task Events

Monitor real-time task execution with event streaming.

​
Asynchronous Example

async for event in task.aevents():
    print(f"Event: {event.type} at {event.time}")
    
    if event.type == "task_finished":
        print("Task completed!")
        break

​
Synchronous Example

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

​
Managing Task Lifecycle

Control task execution with update and stop operations.

​
Asynchronous Example

# Stop a running task
stopped_task = await tasks.astop("task-456")
print(f"Task {stopped_task.id} has been stopped")

# Update task status
updated_task = await tasks.aupdate(
    task_id="task-456",
    status=AgentExecutionStatus.Completed,
    result="Analysis completed successfully"
)

​
Synchronous Example

# Stop a running task
stopped_task = tasks.stop("task-456")
print(f"Task {stopped_task.id} has been stopped")

# Update task status
updated_task = tasks.update(
    task_id="task-456",
    status=AgentExecutionStatus.Completed,
    result="Analysis completed successfully"
)
Continue to the Tasks API Reference for detailed documentation on classes and methods.

Agents Module

Agent management and task creation

Events Module

Event-driven programming and real-time streaming

Tools Repository

Tool integration and invocation

Configuration

SDK configuration and authentication

​
Support

For additional help: