Python SDK

pip install "xpander-sdk[agno]"

from xpander_sdk import Agents, Tasks, ToolsRepository, KnowledgeBases

# SDK automatically uses environment variables
# Set XPANDER_API_KEY. XPANDER_ORGANIZATION_ID is optional on the cloud
# edition, and required when pointing at a self-hosted Agent Controller.

# Initialize modules directly
agents = Agents()
tasks = Tasks()
tools = ToolsRepository()
knowledge = KnowledgeBases()

from xpander_sdk import Agents

# Initialize modules (automatically uses environment variables)
agents = Agents()

# List all agents
agent_list = await agents.alist()

# Load a specific agent
agent = await agents.aget("agent-id")

# Create and execute a task
task = await agent.acreate_task(
    prompt="Analyze this data",
    file_urls=["https://example.com/data.csv"],
    events_streaming=True
)

# Stream task events
async for event in task.aevents():
    print(f"Event: {event.type} at {event.time}")
    if event.type == "task_finished":
        print(f"Result: {event.data.result}")
        break

import time
from xpander_sdk import Agents

# Initialize modules
agents = Agents()

# List all agents
agent_list = agents.list()

# Load a specific agent
agent = agents.get("agent-id")

# Create a task. It returns at once, with status pending
task = agent.create_task(prompt="Analyze this data")

# Wait for the result
while task.status.value in ("pending", "executing"):
    time.sleep(2)
    task.reload()

print(f"Task Status: {task.status}")
print(f"Task Result: {task.result}")

from xpander_sdk import on_task
from xpander_sdk.modules.events.events_module import Events

# Define task handler using decorator
@on_task
async def handle_task(task):
    print(f"Processing task: {task.id}")
    
    # Your custom logic here
    task.result = "Task processed successfully"
    return task

# Start event listener
events = Events()
await events.start(on_execution_request=handle_task)

from xpander_sdk import on_task

# Define task handler using decorator
@on_task
def handle_task(task):
    print(f"Processing task: {task.id}")
    
    # Your custom logic here
    task.result = "Task processed successfully"
    return task

from xpander_sdk import on_boot, on_shutdown, on_task

# Initialize resources before task processing
@on_boot
async def initialize_resources():
    print("Setting up database and cache...")
    # Initialization logic here

# Handle tasks with initialized resources
@on_task
async def process_task(task):
    print(f"Processing task: {task.id}")
    task.result = "Task completed"
    return task

# Clean up during shutdown
@on_shutdown
async def cleanup_resources():
    print("Cleaning up resources...")
    # Cleanup logic here

from xpander_sdk.exceptions.module_exception import ModuleException

try:
    agent = await agents.aget("invalid-agent-id")
except ModuleException as e:
    print(f"Error {e.status_code}: {e.description}")