CrewAI task design and configuration. Use when creating, configuring, or debugging crewAI tasks — writing descriptions and expected_output, setting up task…
CrewAI Task Design Guide
How to write effective tasks that produce reliable, high-quality output from your agents.
The 80/20 Rule
Spend 80% of your effort on task design, 20% on agent design. The task is the most important lever you have. A well-designed task with a mediocre agent will outperform a poorly designed task with an excellent agent.
1. Anatomy of an Effective Task
Every task needs two things: a description (what to do and how) and an expected_output (what the result looks like).
Description — The Instructions
A good description includes:
What to do — the core action
How to do it — specific steps or approach
Context — why this matters, what it feeds into
Constraints — scope limits, things to avoid
Inputs — what data or context is available
research_task:
description: >
Conduct thorough research about {topic} for the year {current_year}.
Your research should:
1. Identify the top 5 key trends and breakthroughs
2. For each trend, find at least 2 credible sources
3. Note any controversies or competing viewpoints
4. Assess potential industry impact (high/medium/low)
Focus on developments from the last 6 months.
Do NOT include speculation or unverified claims.
The output will feed into a report for {target_audience}.
expected_output: >
A structured research brief with 5 sections, one per trend.
Each section includes: trend name, 2-3 paragraph summary,
source citations, impact assessment (high/medium/low),
and a confidence level for your findings.
agent: researcher
Expected Output — The Success Criteria
The expected_output tells the agent what "done" looks like. Be specific about:
Format — bullet points, paragraphs, JSON, table
Structure — sections, headings, order
Length — approximate word count or number of items
Quality markers — citations required, confidence levels, specific fields
Bad Expected Output
Good Expected Output
A research report
A structured brief with 5 sections, each containing: trend name, 2-3 paragraph summary, source citations, and impact rating
An analysis of the data
A markdown table with columns: metric name, current value, 30-day trend, and recommended action. Include at least 10 metrics.
A blog post
A 1000-1500 word technical blog post with: title, introduction, 3-4 main sections with code examples, and a conclusion with next steps
2. The Single Purpose Principle
One task = one objective. Never combine multiple operations into a single task.
Bad: "God Task"
# DON'T do this — too many objectives in one task
research_and_write_task:
description: >
Research {topic}, analyze the findings, write a blog post,
and proofread it for grammar errors.
expected_output: >
A polished blog post about {topic}.
Good: Focused Tasks
research_task:
description: >
Research {topic} and identify the top 5 key developments.
expected_output: >
A research brief with 5 sections covering key trends.
agent: researcher
writing_task:
description: >
Using the research findings, write a technical blog post about {topic}.
expected_output: >
A 1000-1500 word blog post with introduction, main sections,
and conclusion. Include code examples where relevant.
agent: writer
editing_task:
description: >
Review and edit the blog post for grammar, clarity, and consistency.
expected_output: >
The final edited blog post with all corrections applied.
Include a brief editor's note listing what was changed.
agent: editor
Each task has one clear objective. The sequential flow passes context automatically.
3. Task Configuration Reference
Essential Parameters
Task(
description="...", # Required: what to do
expected_output="...", # Required: what the result looks like
agent=researcher, # Optional for hierarchical process; required for sequential
)
Task Dependencies with context
analysis_task = Task(
description="Analyze the research findings...",
expected_output="...",
agent=analyst,
context=[research_task], # Receives research_task's output as context
)
In sequential process: Each task auto-receives all prior task outputs. Use context only when you need non-linear dependencies.
In hierarchical process: context is how you create explicit data flow between tasks.
Structured Output
Use output_pydantic or output_json when downstream code needs to parse the result:
from pydantic import BaseModel
class ResearchReport(BaseModel):
trends: list[str]
confidence: float
sources: list[str]
research_task = Task(
description="...",
expected_output="A structured report with trends, confidence score, and sources.",
agent=researcher,
output_pydantic=ResearchReport, # Agent's output is parsed into this model
)
Important: expected_output is always a string description — never a class name. The Pydantic model goes in output_pydantic, and the expected_output text tells the agent what fields to include.
Access structured output:
result = crew.kickoff(inputs={...})
last_task_output = result.pydantic # Pydantic model from the last task
all_outputs = result.tasks_output # List of all TaskOutput objects
first_task = all_outputs[0].pydantic # Pydantic from a specific task
File Output
Task(
...,
output_file="output/report.md", # Save output to file
create_directory=True, # Create directory if missing (default: True)
)
File output and structured output can be combined — the file gets the raw text, and output_pydantic gets the parsed model.
Async Execution
Task(
...,
async_execution=True, # Run without blocking the next task
)
Use for tasks that can run in parallel. The crew continues to the next task while this one executes. Use context on downstream tasks to wait for async results.
Human Review
Task(
...,
human_input=True, # Pause for human review before finalizing
)
When enabled, the agent presents its result and waits for human feedback before marking the task complete. Use for critical outputs that need human approval.
Do not use human_input=True or Flow @human_feedback to model normal follow-up chat. In conversational Flows, the next user line should be another flow.handle_turn(message, session_id=...) call. Human review is for approving or correcting a specific task/step output before it moves downstream.
Markdown Formatting
Task(
...,
markdown=True, # Add markdown formatting instructions
)
Automatically instructs the agent to format output with proper markdown headers, lists, emphasis, and code blocks.
Callbacks
def log_completion(output):
print(f"Task completed: {output.description[:50]}...")
save_to_database(output.raw)
Task(
...,
callback=log_completion, # Called after task completion
)
4. Task Guardrails — Quality Control
Guardrails validate task output before it passes to the next step. If validation fails, the agent retries.
Function-Based Guardrails
def validate_word_count(output) -> tuple[bool, Any]:
"""Ensure output is between 500-2000 words."""
word_count = len(output.raw.split())
if word_count < 500:
return (False, f"Output too short ({word_count} words). Expand to at least 500 words.")
if word_count > 2000:
return (False, f"Output too long ({word_count} words). Condense to under 2000 words.")
return (True, output)
Task(
...,
guardrail=validate_word_count,
guardrail_max_retries=3, # Max retry attempts (default: 3)
)
Return format: (bool, Any) — first element is pass/fail, second is the result (on success) or error message (on failure).
LLM-Based Guardrails
Task(
...,
guardrail="Verify the output contains at least 3 source citations and no speculative claims.",
)
String guardrails use the agent's LLM to evaluate the output. Good for subjective quality checks.
Chaining Multiple Guardrails
Task(
...,
guardrails=[
validate_word_count, # Function: check length
validate_no_pii, # Function: check for PII
"Ensure the tone is professional and appropriate for a business audience.", # LLM check
],
guardrail_max_retries=3,
)
Guardrails execute sequentially. Each receives the output of the previous guardrail. Mix function-based (deterministic) and LLM-based (subjective) checks.
5. YAML Configuration (Recommended)
tasks.yaml
research_task:
description: >
Conduct thorough research about {topic} for {current_year}.
Identify key trends, breakthrough technologies,
and potential industry impacts.
Focus on the last 6 months of developments.
expected_output: >
A structured research brief with 5 sections.
Each section: trend name, 2-3 paragraph summary,
source citations, and impact assessment.
agent: researcher
analysis_task:
description: >
Analyze the research findings and create actionable recommendations
for {target_audience}.
expected_output: >
A prioritized list of 5 recommendations with:
rationale, estimated effort, and expected impact.
agent: analyst
context:
- research_task
report_task:
description: >
Compile a final report combining research and analysis for {target_audience}.
expected_output: >
A polished markdown report with executive summary,
detailed findings, recommendations, and appendices.
agent: writer
output_file: output/report.md
Wiring in crew.py
@CrewBase
class ResearchCrew:
agents_config = "config/agents.yaml"
tasks_config = "config/tasks.yaml"
@task
def research_task(self) -> Task:
return Task(config=self.tasks_config["research_task"])
@task
def analysis_task(self) -> Task:
return Task(
config=self.tasks_config["analysis_task"],
context=[self.research_task()],
)
@task
def report_task(self) -> Task:
return Task(
config=self.tasks_config["report_task"],
output_file="output/report.md",
)
Critical: The method name (def research_task) must match the YAML key (research_task:).
6. Task Dependencies and Context Flow
Sequential Process (Default)
In Process.sequential, tasks run in order. Each task automatically receives all prior task outputs as context.
research_task → analysis_task → report_task
↓ ↓ ↓
output 1 output 1 + 2 output 1 + 2 + 3
You don't need context= in sequential — it's implicit. Use it only to create non-linear dependencies:
# Task C depends on A but NOT B
task_c = Task(
...,
context=[task_a], # Only receives task_a output, not task_b
)
Explicit Dependencies
# Diamond dependency pattern
task_a = Task(...) # Entry point
task_b = Task(..., context=[task_a]) # Depends on A
task_c = Task(..., context=[task_a]) # Also depends on A
task_d = Task(..., context=[task_b, task_c]) # Depends on both B and C
Conditional Tasks
from crewai.task import ConditionalTask
def needs_more_data(output) -> bool:
return len(output.pydantic.items) < 10
extra_research = ConditionalTask(
description="Fetch additional data sources...",
expected_output="...",
agent=researcher,
condition=needs_more_data, # Only runs if previous output has < 10 items
)
7. Task Tools
Tasks can have their own tools that override the agent's default tools for that specific task:
from crewai_tools import SerperDevTool, ScrapeWebsiteTool
Task(
description="Search for and scrape the top 5 articles about {topic}...",
expected_output="...",
agent=researcher,
tools=[SerperDevTool(), ScrapeWebsiteTool()], # Task-specific tools
)
When to use task-level tools:
The task needs tools the agent doesn't normally have
You want to restrict an agent to specific tools for this task
Different tasks by the same agent need different tool sets
8. Variable Interpolation
Use {variable} placeholders in YAML for reusable tasks:
research_task:
description: >
Research {topic} trends for {current_year},
targeting {target_audience}.
expected_output: >
A report on {topic} suitable for {target_audience}.
Variables are replaced when you call crew.kickoff(inputs={...}):
crew.kickoff(inputs={
"topic": "AI Agents",
"current_year": "2025",
"target_audience": "developers",
})
Common mistakes:
Missing variable in inputs → literal {variable} appears in the prompt
Using {{ }} Jinja2 syntax → crewAI uses single braces { }
Unused variables in inputs → silently ignored (no error)
9. Common Task Design Mistakes
Mistake
Impact
Fix
Vague description ("Research the topic")
Agent produces shallow, unfocused output
Add specific steps, constraints, and context
Vague expected_output ("A report")
Agent guesses at format and structure
Specify format, sections, length, quality markers
Multiple objectives in one task
Agent does all of them poorly
Split into focused single-purpose tasks
Modeling each chat turn as a Crew task
Tasks are batch/workflow units, not the conversational session loop
Use a conversational Flow and call handle_turn() per user message
No context between dependent tasks
Agent lacks information from prior steps
Use context=[prior_task] for explicit dependencies
expected_output references a Pydantic class
Agent sees a class name string, not field names
Keep expected_output as a human-readable string; use output_pydantic for the model
Missing tools for data tasks
Agent fabricates data instead of fetching it
Add tools to the task or agent
No guardrails on critical output
Bad output flows downstream unchecked
Add function or LLM guardrails
Overly strict expected_output
Agent loops trying to match impossible criteria
Be specific but achievable; lower guardrail_max_retries to fail faster
Description duplicates backstory
Wasted tokens and confused agent
Description = what to do; backstory = who you are
10. Task Design Checklist
Before running a task, verify:
Description includes what, how, context, and constraints
Expected output specifies format, structure, and quality markers
Single purpose — one clear objective per task
Agent assigned (or task is in a hierarchical crew)
Dependencies set via context where needed
Tools provided for any task requiring external data
Structured output configured if downstream code parses the result
Guardrails set for critical outputs
Variables in YAML match the inputs dict keys
Expected output is achievable — test with a simple run before adding complexity
References
For deeper dives into specific topics, see:
Structured Output — output_pydantic, output_json, and response_format patterns across LLM, Agent, Task, and Crew levels
For related skills:
getting-started — project scaffolding, choosing the right abstraction, Flow architecture
design-agent — agent Role-Goal-Backstory framework, parameter tuning, tool assignment, memory & knowledge configuration
ask-docs — query the live CrewAI documentation MCP server for questions not covered by these skillsdon't have the plugin yet? install it then click "run inline in claude" again.
reorganized raw guide into implexa's 6-component structure, separated decision logic into explicit if-else branches, added edge cases (retry loops, impossible criteria, missing variables, type errors), documented all external inputs (pydantic models, tools, llm instance), preserved original procedure faithfully with step-by-step numbering, added outcome signal verification checklist.
task design is the primary lever for agent reliability. spend 80% of effort here, 20% on agent tuning. a well-designed task with a mediocre agent outperforms a brilliant agent hamstrung by vague instructions. this skill covers the anatomy of effective tasks (description + expected_output), single-purpose principle, dependency wiring, guardrails, and YAML configuration. use this when building new tasks, debugging agent output quality, or restructuring workflows.
description is the agent's instruction set. include:
example:
description: >
Conduct thorough research about {topic} for the year {current_year}.
Your research should:
1. Identify the top 5 key trends and breakthroughs
2. For each trend, find at least 2 credible sources
3. Note any controversies or competing viewpoints
4. Assess potential industry impact (high/medium/low)
Focus on developments from the last 6 months.
Do NOT include speculation or unverified claims.
The output will feed into a report for {target_audience}.
expected_output tells the agent what "done" looks like. specify:
example:
expected_output: >
A structured research brief with 5 sections, one per trend.
Each section includes: trend name, 2-3 paragraph summary,
source citations, impact assessment (high/medium/low),
and a confidence level for your findings.
do not reference pydantic class names in expected_output. keep it human-readable so the agent understands what fields to populate.
each task must have exactly one objective. no "god tasks" combining research + writing + editing.
research_task = Task(
description="...",
expected_output="...",
agent=researcher, # required in sequential/hierarchical crews
)
if your initial task covers multiple operations (research + write + edit), split into separate tasks with explicit dependencies:
choose one or combine:
a) structured output with pydantic
from pydantic import BaseModel
class ResearchReport(BaseModel):
trends: list[str]
confidence: float
sources: list[str]
research_task = Task(
description="...",
expected_output="A structured report with trends, confidence score, and sources.",
agent=researcher,
output_pydantic=ResearchReport, # model goes here, NOT in expected_output
)
access via result.pydantic or all_outputs[0].pydantic.
b) file output
Task(
description="...",
expected_output="...",
agent=writer,
output_file="output/report.md",
create_directory=True, # auto-create if missing
)
c) markdown formatting
Task(
description="...",
expected_output="...",
agent=writer,
markdown=True, # auto-instructs agent to use proper headers, lists, emphasis
)
sequential process (default)
in Process.sequential, tasks run in order. each task automatically receives all prior task outputs as implicit context. do not use context= unless you need non-linear flow:
# task_b auto-receives task_a output
task_a = Task(description="...", agent=researcher)
task_b = Task(description="...", agent=analyst)
# implicit: task_b sees task_a's result
explicit context for non-linear dependencies
# task_c depends on task_a but NOT task_b
task_c = Task(
description="...",
agent=analyst,
context=[task_a], # only receives task_a, skips task_b
)
diamond dependency pattern
task_a = Task(...) # entry point
task_b = Task(..., context=[task_a]) # depends on a
task_c = Task(..., context=[task_a]) # also depends on a (parallel-friendly)
task_d = Task(..., context=[task_b, task_c]) # depends on both b and c
task-level tools override agent defaults for that specific task:
from crewai_tools import SerperDevTool, ScrapeWebsiteTool
Task(
description="Search for and scrape the top 5 articles about {topic}...",
expected_output="...",
agent=researcher,
tools=[SerperDevTool(), ScrapeWebsiteTool()],
)
use task-level tools when:
guardrails validate output before it passes downstream. if validation fails, agent retries (up to guardrail_max_retries).
function-based guardrails (deterministic checks)
def validate_word_count(output) -> tuple[bool, Any]:
"""ensure output is between 500-2000 words."""
word_count = len(output.raw.split())
if word_count < 500:
return (False, f"Output too short ({word_count} words). Expand to at least 500 words.")
if word_count > 2000:
return (False, f"Output too long ({word_count} words). Condense to under 2000 words.")
return (True, output)
Task(
description="...",
expected_output="...",
agent=writer,
guardrail=validate_word_count,
guardrail_max_retries=3,
)
return format: (bool, Any) where bool is pass/fail, Any is result (on success) or error message (on failure).
llm-based guardrails (subjective quality checks)
Task(
description="...",
expected_output="...",
agent=researcher,
guardrail="Verify the output contains at least 3 source citations and no speculative claims.",
)
string guardrails use the agent's llm to evaluate. good for subjective checks.
chaining multiple guardrails
Task(
description="...",
expected_output="...",
agent=researcher,
guardrails=[
validate_word_count, # function: deterministic
validate_no_pii, # function: deterministic
"Ensure tone is professional and suitable for business audience.", # llm: subjective
],
guardrail_max_retries=3,
)
guardrails execute sequentially. each receives output of previous guardrail.
Task(
description="...",
expected_output="...",
agent=researcher,
async_execution=True, # runs in background, does not block next task
)
use for tasks that can run in parallel. downstream tasks will wait for async result if you add context=[async_task].
Task(
description="...",
expected_output="...",
agent=writer,
human_input=True, # agent pauses and waits for human feedback before finalizing
)
use for critical outputs requiring human approval. do not conflate with conversational flows (each user message in a flow should call handle_turn(), not spawn a new crew task).
def log_completion(output):
print(f"Task completed: {output.description[:50]}...")
save_to_database(output.raw)
Task(
description="...",
expected_output="...",
agent=researcher,
callback=log_completion, # called after task completion
)
tasks.yaml:
research_task:
description: >
Conduct thorough research about {topic} for {current_year}.
Identify key trends, breakthrough technologies,
and potential industry impacts.
Focus on the last 6 months of developments.
expected_output: >
A structured research brief with 5 sections.
Each section: trend name, 2-3 paragraph summary,
source citations, and impact assessment.
agent: researcher
analysis_task:
description: >
Analyze the research findings and create actionable recommendations
for {target_audience}.
expected_output: >
A prioritized list of 5 recommendations with:
rationale, estimated effort, and expected impact.
agent: analyst
context:
- research_task
report_task:
description: >
Compile a final report combining research and analysis for {target_audience}.
expected_output: >
A polished markdown report with executive summary,
detailed findings, recommendations, and appendices.
agent: writer
output_file: output/report.md
wire in crew.py:
@CrewBase
class ResearchCrew:
agents_config = "config/agents.yaml"
tasks_config = "config/tasks.yaml"
@task
def research_task(self) -> Task:
return Task(config=self.tasks_config["research_task"])
@task
def analysis_task(self) -> Task:
return Task(
config=self.tasks_config["analysis_task"],
context=[self.research_task()],
)
@task
def report_task(self) -> Task:
return Task(
config=self.tasks_config["report_task"],
output_file="output/report.md",
)
critical: method name (def research_task) must match YAML key (research_task:).
use {variable} placeholders in YAML:
research_task:
description: >
Research {topic} trends for {current_year},
targeting {target_audience}.
replace at runtime:
crew.kickoff(inputs={
"topic": "AI Agents",
"current_year": "2025",
"target_audience": "developers",
})
common mistakes: missing variable in inputs (literal {variable} appears in prompt), using {{ }} jinja2 syntax (crewaI uses single braces), unused variables in inputs (silently ignored).
should i split this task into multiple tasks?
if the description includes multiple distinct operations (research AND write AND edit), split it. one task = one objective. sequential flow passes outputs automatically between tasks.
do i need structured output (pydantic)?
yes, if downstream code needs to parse and access specific fields (e.g., trends list, confidence score). use output_pydantic=ResearchReport. if task output is just text for human consumption, skip it.
should i use context= explicitly in sequential process?
only if you need non-linear dependencies. in Process.sequential, each task auto-receives all prior outputs. use context=[task_a] only if task_c needs task_a but NOT task_b.
how strict should my expected_output be?
specific but achievable. if expected_output is impossible to match, agent loops and wastes retries. specify format (sections, headers, word count), not overly rigid formatting rules. lower guardrail_max_retries=1 to fail faster on impossible criteria.
should i use llm-based or function-based guardrails?
function-based for deterministic checks (word count, field presence, PII detection). llm-based for subjective checks (tone, professionalism, factual accuracy). mix both for comprehensive validation.
do i add tools to the task or the agent?
task-level tools override agent defaults. use if the task needs unique tools or you want to restrict an agent's tool set for a specific task. otherwise, assign tools to the agent config.
when do i use async_execution=True?
for tasks that can run in parallel without blocking the next task. pair with context=[async_task] on downstream tasks to wait for result. skip if task output feeds immediately into the next task.
should i use human_input=True or a conversational flow?
human_input=True is for approving/correcting task output before it flows downstream in a batch workflow. do not use for conversational chatbots. in flows, each user message calls handle_turn(message, session_id=...), not crew.kickoff().
a well-designed task produces:
for crews with multiple tasks, result object contains:
the task worked when:
test by running crew.kickoff() with sample inputs and inspecting result.raw, result.tasks_output, and file outputs. verify each section, citation, field, and quality marker is present before passing to downstream tasks or users.
credits: original guide by crewaiinc (https://skills.sh). enriched per implexa quality standards with explicit decision logic, edge cases, validation patterns, and outcome verification steps.