Tasks & Workflows

Tasks are the heart of every CrewAI application. A well-defined task tells the agent exactly what to do, what format is expected, and what context is available.

Task Definition in Detail

from crewai import Task

analysis_task = Task(
    description=(
        "Analyze the provided sales data for Q4 2025. "
        "Identify the top 3 products by revenue, "
        "calculate growth rates compared to Q3, "
        "and identify anomalies or notable trends."
    ),
    expected_output=(
        "A structured analysis report with: "
        "1) Top 3 products with revenue figures, "
        "2) Quarter-over-quarter growth rates, "
        "3) Identified anomalies with explanations, "
        "4) Three concrete recommendations"
    ),
    agent=data_analyst,
    output_file="reports/q4_analysis.md"
)

Important Task Parameters

Context Passing

Tasks can use results from previous tasks as context:

research_task = Task(
    description="Research current market trends for {topic}",
    expected_output="Summary of the top 5 trends",
    agent=researcher
)

analysis_task = Task(
    description="Analyze the researched trends for relevance to our company",
    expected_output="Prioritized trend analysis with impact assessment",
    agent=analyst,
    context=[research_task]  # Receives output from research_task
)

writing_task = Task(
    description="Create a management report based on the analysis",
    expected_output="Professional 2-page report",
    agent=writer,
    context=[research_task, analysis_task]  # Receives both outputs
)

Task Dependencies

In sequential process, the order of tasks in the list determines execution order. Context references make dependencies explicit:

research_task (no dependency)
    ↓ Output as context
analysis_task (depends on research_task)
    ↓ Output as context
writing_task (depends on research + analysis)

Callback Functions

Execute a function after each task — for logging, notifications, or data processing:

def on_task_complete(output):
    # Save result to database
    db.save_result(output.raw)
    # Send notification
    slack.send(f"Task completed: {output.description}")
    # Track metrics
    metrics.track("task_completed", {"tokens": output.token_usage})

analysis_task = Task(
    description="Analyze the data...",
    expected_output="Structured report",
    agent=analyst,
    callback=on_task_complete
)

Human Input

For critical tasks, you can enforce human review:

approval_task = Task(
    description="Create the final customer report",
    expected_output="Approved report in PDF format",
    agent=writer,
    human_input=True  # Pauses and waits for human input
)

Workflow Example: Content Pipeline

crew = Crew(
    agents=[researcher, seo_expert, writer, editor],
    tasks=[
        Task(description="Keyword research...", agent=seo_expert),
        Task(description="Topic research...", agent=researcher, context=[task_0]),
        Task(description="Write article...", agent=writer, context=[task_0, task_1]),
        Task(description="Edit article...", agent=editor, context=[task_2], human_input=True),
    ],
    process=Process.sequential
)

Practical tip: Define expected_output as precisely as possible. "A report" is too vague. "A structured report with executive summary, 3 key points, and concrete recommendations" gives the agent clear guardrails.