Task Methods

The Tasks API provides methods for creating, monitoring, and managing tasks assigned to xpander agents.

add_task() / addTask()

Creates a new task for the agent with the specified input.

task = agent.add_task(
    input="Research quantum computing",
    thread_id="thread_abc123",  # Optional
    files=[{"name": "data.csv", "content": "base64-encoded-content"}]  # Optional
)
ParameterTypeRequiredDescription
inputstringYesTask description or user message
thread_idstringNoThread ID for connecting related tasks
filesList[Dict]NoFiles to attach to the task
use_workerbooleanNoWhether to use a worker for execution (default: true)

Returns: Task object

After calling add_task(), the agent’s execution property is initialized, and agent.messages is populated with system instructions and the user input.

is_finished() / isFinished()

Checks if the current task has finished.

is_done = agent.is_finished()

Returns: boolean - True if task has finished, False otherwise.

A task is considered finished when the AI model calls the xpfinish-agent-execution-finished tool or if the task has failed.

get_execution_status() / getExecutionStatus()

Fetches the latest status of the current task.

status = agent.get_execution_status()

Returns: Task status object with the following properties:

interface TaskStatus {
  id: string;
  status: "RUNNING" | "COMPLETED" | "FAILED" | "PENDING";
  createdAt: string;
  updatedAt: string;
  input: string;
  memoryThreadId: string;
}

retrieve_execution_result() / retrieveExecutionResult()

Retrieves the result of a completed task, including the thread ID which can be used to connect related tasks.

result = agent.retrieve_execution_result()

Returns: Result object with the following properties:

interface ExecutionResult {
  status: "COMPLETED" | "FAILED";
  result?: string;      // Present when status is "COMPLETED"
  error?: string;       // Present when status is "FAILED"
  executionId: string;
  threadId: string;     // Thread ID for connecting related tasks
}

list_executions() / listExecutions()

Lists tasks for the current agent.

tasks = agent.list_executions()

Returns: Array of task objects.

get_execution() / getExecution()

Gets a task by ID.

task = agent.get_execution(execution_id="execution-1234")
ParameterTypeRequiredDescription
execution_idstringYesID of the task

Returns: Task object

Task Execution Properties

agent.execution

The current execution task object, available after calling add_task().

execution = agent.execution
execution_id = execution.id

To retrieve the thread ID, you should use the result from retrieve_execution_result() rather than the execution object.

Type Definitions

Execution

Represents a task assigned to an agent.

interface Execution {
  id: string;
  status: "RUNNING" | "COMPLETED" | "FAILED" | "PENDING";
  createdAt: string;
  updatedAt: string;
  input: string;
  result?: string;
  error?: string;
  memoryThreadId: string;
}

ExecutionStatus

Represents the current status of a task.

interface ExecutionStatus {
  id: string;
  status: "RUNNING" | "COMPLETED" | "FAILED" | "PENDING";
  createdAt: string;
  updatedAt: string;
  input: string;
  memoryThreadId: string;
}

ExecutionResult

Represents the result of a completed task.

interface ExecutionResult {
  status: "COMPLETED" | "FAILED";
  result?: string;      // Present when status is "COMPLETED"
  error?: string;       // Present when status is "FAILED"
  executionId: string;
  threadId: string;     // Thread ID for connecting related tasks
}

Usage Examples

Basic Task Execution

# Create a task
agent.add_task(input="Summarize the latest news on AI")

# Run until completion
while not agent.is_finished():
    # Run LLM to process messages and generate tool calls
    response = openai_client.chat.completions.create(
        model="gpt-4o",
        messages=agent.messages,
        tools=agent.get_tools(llm_provider=LLMProvider.OPEN_AI),
        temperature=0
    )
    
    # Add response to messages
    agent.add_messages(response.model_dump())
    
    # Process tool calls
    tool_calls = XpanderClient.extract_tool_calls(
        llm_response=response.model_dump(),
        llm_provider=LLMProvider.OPEN_AI
    )
    
    if tool_calls:
        agent.run_tools(tool_calls=tool_calls)

# Get result
result = agent.retrieve_execution_result()
print(result.result)

Multi-Agent Workflow

# Start with first agent
agent_a = client.agents.get(agent_id="agent_a_id")
agent_a.add_task(input="Collect data for analysis")

# Process with first agent
# ... (code for processing the task) ...

# Get the execution result which contains the thread ID
result = agent_a.retrieve_execution_result()
thread_id = result.thread_id

# Transfer to second agent using the thread ID
agent_b = client.agents.get(agent_id="agent_b_id")
agent_b.add_task(
    input="Analyze the collected data",
    thread_id=thread_id  # Continue the same conversation
)

# Process with second agent
# ... (code for processing the task) ...

Retrieving Past Tasks

# Get all tasks for the agent
tasks = agent.list_executions()

# Get a specific task by ID
task = agent.get_execution(execution_id="execution-1234")

# Get the result of a completed task
if task.status == "COMPLETED":
    result = agent.retrieve_execution_result()
    print(result.result)