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
)
| Parameter | Type | Required | Description |
|---|
input | string | Yes | Task description or user message |
thread_id | string | No | Thread ID for connecting related tasks |
files | List[Dict] | No | Files to attach to the task |
use_worker | boolean | No | Whether 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")
| Parameter | Type | Required | Description |
|---|
execution_id | string | Yes | ID 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)
