Agent Tasks

Understanding Tasks in PraisonAI

Tasks are units of work that agents execute. Each task has a description, expected output, and optional configuration for routing, retries, and advanced features.

Quick Start

Basic Task

from praisonaiagents import Agent, Task, AgentTeam

agent = Agent(name="Researcher", instructions="Research topics")

task = Task(
    action="Research AI trends in 2024",
    expected_output="A summary report",
    agent=agent
)

team = AgentTeam(agents=[agent], tasks=[task])
team.start()

With Handler

# Custom function instead of agent
def process_data(context):
    return {"result": "Processed"}

task = Task(
    action="Process the data",
    handler=process_data
)

---

Core Parameters

Parameter	Type	Description
action	str	What the task should accomplish
expected_output	str	What the output should look like
agent	Agent	Agent to execute the task
name	str	Optional identifier
tools	list	Tools available to the task

description is deprecated. Use action instead.

---

Task Dependencies

Use context to chain tasks:

task1 = Task(
    name="research",
    action="Research AI trends",
    agent=researcher
)

task2 = Task(
    name="write",
    action="Write a report based on research",
    agent=writer,
    context=[task1]  # or depends_on=[task1]
)

---

Conditional Execution

Simple Condition (should_run)

task = Task(
    action="Optional step",
    should_run=lambda ctx: ctx.get("enabled", False)
)

When/Then/Else Routing

task = Task(
    action="Evaluate score",
    when="{{score}} > 80",      # Condition expression
    then_task="celebrate",       # If true
    else_task="retry"           # If false
)

---

Loop Support

Iterate over a list:

task = Task(
    action="Process {{item}}",
    loop_over="items",    # Variable containing the list
    loop_var="item"       # Name for each item
)

---

Callbacks

Use on_task_complete for task completion notifications:

def on_complete(output):
    print(f"Task done: {output.raw}")

task = Task(
    action="Do something",
    agent=agent,
    on_task_complete=on_complete
)

The callback parameter is deprecated. Use on_task_complete instead.

---

Guardrails

Validate task output before accepting:

def validate_output(output):
    if len(output.raw) < 100:
        return (False, "Output too short")
    return (True, output)

task = Task(
    action="Write detailed report",
    agent=writer,
    guardrails=validate_output,
    max_retries=3
)

---

Robustness Features

Parameter	Type	Description
max_retries	int	Maximum retry attempts (default: 3)
retry_delay	float	Seconds between retries
skip_on_failure	bool	Continue workflow if task fails

task = Task(
    action="Optional enrichment",
    skip_on_failure=True,   # Workflow continues if this fails
    retry_delay=1.0,        # Wait between retries
    max_retries=3
)

---

Output Configuration

Parameter	Type	Description
output_file	str	Save output to file
output_json	BaseModel	Parse output as JSON
output_pydantic	BaseModel	Validate with Pydantic
output_variable	str	Store output in workflow variable

from pydantic import BaseModel

class Report(BaseModel):
    title: str
    summary: str

task = Task(
    action="Generate report",
    output_pydantic=Report,
    output_file="report.json"
)

---

Feature Configs

Enable advanced features per-task:

task = Task(
    action="Complex task",
    autonomy=True,         # Agent autonomy level
    knowledge=["docs/"],   # Knowledge base paths
    web=True,              # Web search enabled
    reflection=True,       # Self-reflection
    planning=True,         # Task planning
    caching=True           # Response caching
)

---

Task Types

Type	Description
task	Standard task (default)
decision	Branching logic with conditions
loop	Iterate over items

# Decision task
task = Task(
    action="Decide next step",
    task_type="decision",
    condition={
        "approve": ["next_task"],
        "reject": ["error_task"]
    }
)

---

Async Execution

task = Task(
    action="Long running task",
    agent=agent,
    async_execution=True
)

# Run async
result = await team.astart()

---

Full Parameter Reference

All Task Parameters

Parameter	Type	Default	Description
action	str	-	What the task should do (required)
expected_output	str	”Complete successfully”	Expected output format
agent	Agent	None	Agent to execute
name	str	None	Task identifier
tools	list	[]	Available tools
context	list	[]	Dependent tasks
depends_on	list	[]	Alias for context
handler	Callable	None	Custom function
should_run	Callable	None	Condition to run
loop_over	str	None	Variable containing list
loop_var	str	”item”	Loop variable name
when	str	None	Condition expression
then_task	str	None	Task if condition true
else_task	str	None	Task if condition false
on_task_complete	Callable	None	Completion callback
guardrails	Callable/str	None	Output validation
max_retries	int	3	Max retry attempts
retry_delay	float	0.0	Retry delay seconds
skip_on_failure	bool	False	Continue on failure
async_execution	bool	False	Run async
output_file	str	None	Save to file
output_json	BaseModel	None	JSON output model
output_pydantic	BaseModel	None	Pydantic validation
output_variable	str	None	Workflow variable name
task_type	str	”task”	task/decision/loop
routing	dict	Next task routing
images	list	[]	Image inputs
input_file	str	None	Input file path
memory	Memory	None	Memory instance
variables	dict	Task variables
description	str	-	Deprecated - use action

---

Related

Agents

Create and configure agents

AgentTeam

Orchestrate multiple agents

Callbacks

Task completion callbacks

Guardrails

Output validation