Writing clear CrewAI task descriptions that agents follow

Writing clear CrewAI task descriptions is the difference between agents that deliver useful output and agents that hallucinate, loop, or ignore constraints. The framework gives you powerful primitives — roles, goals, backstories, tools — but the task description is where intent becomes instruction. This guide walks through crewai clear task description writing patterns that survive contact with real models, with runnable code you can drop into a project today.

Step 1: Define the task contract before you write the description

Every task needs an explicit contract: what inputs it consumes, what output it produces, and what “done” looks like. Write this contract as a docstring or comment before touching the Task constructor. It forces you to separate the what from the how and catches ambiguity early.

# tasks/research_task.py
"""
Task contract: ResearchCompetitorPricing

Inputs:
  - competitor_names: list[str] (validated non-empty, max 10)
  - product_category: str (one of: "saas", "hardware", "consumer_goods")
  - target_market: str (ISO 3166-1 alpha-2 country code)

Output:
  - JSON object matching CompetitorPricingReport schema (see schemas/pricing.py)

Success criteria:
  - At least 3 pricing data points per competitor
  - All prices in USD with conversion source cited
  - Confidence score >= 0.7 for each data point
  - Completes within 120 seconds

Failure modes:
  - Returns partial data with `incomplete: true` and `missing: list[str]`
  - Raises ResearchError if zero competitors return data
"""

Verify: Read the contract aloud. If you cannot explain it to a colleague in two sentences, rewrite it.

Step 2: Structure the description with explicit sections

CrewAI task descriptions accept free text, but models follow structure. Use a consistent template with labeled sections. This is the single highest-leverage change you can make to crewai clear task description writing.

from crewai import Task
from agents import research_analyst
from schemas.pricing import CompetitorPricingReport

research_task = Task(
    description=(
        "## Objective\n"
        "Research current pricing for each competitor in `competitor_names` "
        "for the `product_category` in `target_market`.\n\n"
        "## Inputs (provided in context)\n"
        "- competitor_names: {competitor_names}\n"
        "- product_category: {product_category}\n"
        - target_market: {target_market}\n\n"
        "## Required output format\n"
        "Return a JSON object that validates against the `CompetitorPricingReport` "
        "Pydantic model. Do not include markdown fences or explanatory text.\n\n"
        "## Constraints\n"
        "- Use only the `web_search` and `fetch_url` tools\n"
        "- Cite sources with URLs and access timestamps\n"
        "- Convert all prices to USD using the exchange rate from "
        "`https://api.exchangerate.host/latest` at time of fetch\n"
        "- Minimum 3 data points per competitor\n"
        "- Confidence score per data point based on source recency and authority\n\n"
        "## Quality gates (self-check before returning)\n"
        "1. Does every competitor have >= 3 priced plans?\n"
        "2. Are all prices in USD with conversion source?\n"
        "3. Is confidence >= 0.7 for each point?\n"
        "4. Is the JSON valid against the schema?\n\n"
        "## Failure handling\n"
        "If any competitor yields zero data points, set `incomplete: true` "
        "and list missing competitors in `missing` array. Do not fabricate data."
    ),
    expected_output=(
        "A JSON object matching the CompetitorPricingReport schema with "
        "pricing data for all requested competitors, or an incomplete report "
        "with missing competitors listed."
    ),
    agent=research_analyst,
    output_json=CompetitorPricingReport,
    tools=["web_search", "fetch_url"],
)

Verify: Run the task with crewai run --task research_task --inputs '{"competitor_names": ["Notion", "Obsidian"], "product_category": "saas", "target_market": "US"}'. Check that the output validates against CompetitorPricingReport and that incomplete is false.

Step 3: Bind inputs to the task context explicitly

CrewAI interpolates {variable} placeholders in the description from the kickoff inputs or prior task outputs. Make this binding visible in the description so the agent (and you) knows exactly what data is available.

# Good: explicit binding in description
description = (
    "Analyze the quarterly report for `{company_ticker}` "
    "covering fiscal year `{fiscal_year}` Q`{quarter}`.\n\n"
    "The full report text is available in the context variable `report_text`.\n"
    "Previous analysis from the sentiment task is in `sentiment_summary`."
)

# Bad: implicit, ambiguous
description = "Analyze the quarterly report for the company."

When inputs come from upstream tasks, document the expected shape:

description = (
    "## Upstream context\n"
    "The `competitor_analysis` task produced a `CompetitorLandscape` object:\n"
    "```json\n"
    "{\n"
    "  \"competitors\": [\n"
    "    {\"name\": \"string\", \"market_share_pct\": \"float\", \"tier\": \"string\"}\n"
    "  ],\n"
    "  \"market_size_usd\": \"float\"\n"
    "}\n"
    "```\n"
    "Use `competitors` filtered to `tier == 'primary'` for this task."
)

Verify: Add a debug task that prints the rendered description before the real task runs:

debug_task = Task(
    description="Print the rendered description for research_task to verify interpolation.",
    agent=research_analyst,
    expected_output="The full rendered description text.",
    callback=lambda output: print(f"RENDERED:\n{output.raw}")
)

Step 4: Constrain output with schemas, not prose

expected_output is documentation. output_json (or output_pydantic) is enforcement. Always use a Pydantic model for structured output. The model becomes the ground truth for validation and for the agent’s self-correction loop.

# schemas/pricing.py
from pydantic import BaseModel, Field, field_validator
from typing import Literal
from datetime import datetime

class PricingDataPoint(BaseModel):
    competitor: str
    plan_name: str
    price_usd: float = Field(gt=0)
    billing_period: Literal["monthly", "annual", "one_time"]
    features: list[str] = Field(min_length=1)
    source_url: str
    source_accessed_at: datetime
    confidence: float = Field(ge=0.0, le=1.0)

    @field_validator("confidence")
    @classmethod
    def confidence_threshold(cls, v: float) -> float:
        if v < 0.7:
            raise ValueError("Confidence must be >= 0.7")
        return v

class CompetitorPricingReport(BaseModel):
    competitors: list[PricingDataPoint]
    incomplete: bool = False
    missing: list[str] = Field(default_factory=list)
    generated_at: datetime = Field(default_factory=datetime.utcnow)
    schema_version: str = "1.0"

Then reference it in the task:

research_task = Task(
    description=...,
    expected_output="JSON matching CompetitorPricingReport",
    agent=research_analyst,
    output_json=CompetitorPricingReport,  # Enforces schema at runtime
)

Verify: Intentionally break the schema — remove a required field from the agent’s output — and confirm the task fails validation with a clear error. Then fix and confirm it passes.

Step 5: Write tool-use instructions the model can actually follow

Models struggle with “use the web search tool” unless you specify when, what query, and how to parse results. Give the agent a mini-algorithm.

description = (
    "## Tool usage protocol\n"
    "For EACH competitor in `competitor_names`:\n"
    "1. **Search**: `web_search(query=\"{competitor} pricing {product_category} {target_market} 2024\")`\n"
    "2. **Filter**: From results, select URLs matching:\n"
    "   - Official pricing page (contains /pricing, /plans, /pricing/)\n"
    "   - Published within last 12 months\n"
    "   - Not a third-party comparison site\n"
    "3. **Fetch**: `fetch_url(url=selected_url)` for each candidate\n"
    "4. **Extract**: Parse pricing tables for plan names, prices, billing periods, features\n"
    "5. **Validate**: Confirm at least 3 distinct plans extracted\n"
    "6. **Convert**: Fetch USD exchange rate if prices not in USD\n"
    "7. **Score**: Assign confidence:\n"
    "   - 0.9: Official pricing page, current year\n"
    "   - 0.8: Official page, last year\n"
    "   - 0.7: Press release or blog announcement\n"
    "   - <0.7: Discard\n\n"
    "If step 2 yields zero URLs for a competitor, add to `missing` and continue."
)

Verify: Run with a competitor known to have no public pricing (e.g., enterprise-only). Confirm the agent adds it to missing rather than hallucinating prices.

Step 6: Add self-check checkpoints inside the description

Agents drift. Embed explicit verification steps the agent must perform before returning. Frame them as a checklist the agent “ticks off” in its reasoning.

description = (
    "...\n\n"
    "## Pre-return verification (reason through each)\n"
    "☐ Every requested competitor appears in `competitors` array OR in `missing`\n"
    "☐ Each data point has all required fields: competitor, plan_name, price_usd, "
    "billing_period, features, source_url, source_accessed_at, confidence\n"
    "☐ All prices > 0 and in USD\n"
    "☐ All confidence scores >= 0.7\n"
    "☐ All source_urls are valid HTTP/HTTPS URLs\n"
    "☐ All source_accessed_timestamps are within the last 5 minutes\n"
    "☐ `incomplete` is true iff `missing` is non-empty\n"
    "☐ JSON validates against CompetitorPricingReport schema\n\n"
    "If any check fails, revise and re-check. Do not return until all pass."
)

Verify: Add logging to the agent’s LLM calls (set verbose=True on the crew) and confirm the model’s reasoning includes the checklist items.

Step 7: Handle iteration with context-aware retry guidance

When a task fails validation, CrewAI can retry. Make the retry productive by telling the agent what went wrong and how to fix it. Use the callback or a custom retry handler to inject failure context.

from crewai import Task
from schemas.pricing import CompetitorPricingReport, ValidationError

def pricing_retry_handler(task_output, attempt: int, max_attempts: int):
    """Inject validation errors into the next attempt's context."""
    if task_output.pydantic_error:
        error_detail = task_output.pydantic_error.errors()
        return {
            "retry_context": (
                f"Attempt {attempt} of {max_attempts} failed validation:\n"
                f"{error_detail}\n\n"
                "Fix the specific fields mentioned. Common issues:\n"
                "- Missing `source_accessed_at`: add datetime.utcnow() for each point\n"
                "- Confidence < 0.7: re-evaluate source authority or discard\n"
                "- Price not in USD: fetch exchange rate and convert\n"
                "- Extra fields not in schema: remove them\n"
            )
        }
    return {}

research_task = Task(
    description=...,
    expected_output=...,
    agent=research_analyst,
    output_json=CompetitorPricingReport,
    max_retries=3,
    callback=pricing_retry_handler,
)

Verify: Force a validation error (e.g., return confidence 0.5) and watch the retry log. Confirm the agent corrects the specific field on the next attempt.

Step 8: Test with adversarial inputs

Happy-path testing is not enough. Build a test harness that feeds the task edge cases and asserts on behavior.

# tests/test_research_task.py
import pytest
from crewai import Crew
from tasks.research_task import research_task
from agents import research_analyst

@pytest.mark.parametrize("inputs,expect_incomplete", [
    ({"competitor_names": ["Notion", "Obsidian"], "product_category": "saas", "target_market": "US"}, False),
    ({"competitor_names": ["InternalToolXYZ"], "product_category": "saas", "target_market": "US"}, True),
    ({"competitor_names": [], "product_category": "saas", "target_market": "US"}, True),  # invalid, but test handling
    ({"competitor_names": ["Notion"], "product_category": "invalid_category", "target_market": "US"}, False),  # agent should handle gracefully
])
def test_research_task_handles_inputs(inputs, expect_incomplete):
    crew = Crew(agents=[research_analyst], tasks=[research_task], verbose=False)
    result = crew.kickoff(inputs=inputs)
    report = result.pydantic
    assert isinstance(report.incomplete, bool)
    if expect_incomplete:
        assert report.incomplete is True
        assert len(report.missing) > 0
    else:
        assert report.incomplete is False
        assert len(report.competitors) >= 3  # at least 3 data points total
        for point in report.competitors:
            assert point.confidence >= 0.7
            assert point.price_usd > 0

Verify: Run pytest tests/test_research_task.py -v. All cases should pass. If the empty list case crashes, add input validation to the task description or a pre-task guard.

Step 9: Document the task interface for downstream consumers

Other tasks (and humans) consume this task’s output. Write a machine-readable interface description that lives with the task.

# tasks/research_task.py (continued)
TASK_INTERFACE = {
    "name": "research_competitor_pricing",
    "version": "1.2.0",
    "description": "Researches current pricing for competitors in a product category and market.",
    "inputs": {
        "competitor_names": {"type": "array", "items": {"type": "string"}, "minItems": 1, "maxItems": 10},
        "product_category": {"type": "string", "enum": ["saas", "hardware", "consumer_goods"]},
        "target_market": {"type": "string", "pattern": "^[A-Z]{2}$"},
    },
    "outputs": {
        "schema": "schemas.pricing.CompetitorPricingReport",
        "example": {
            "competitors": [
                {
                    "competitor": "Notion",
                    "plan_name": "Plus",
                    "price_usd": 10.0,
                    "billing_period": "monthly",
                    "features": ["Unlimited blocks", "Unlimited file uploads", "30-day page history"],
                    "source_url": "https://www.notion.so/pricing",
                    "source_accessed_at": "2024-01-15T10:30:00Z",
                    "confidence": 0.9
                }
            ],
            "incomplete": False,
            "missing": [],
            "generated_at": "2024-01-15T10:30:05Z",
            "schema_version": "1.0"
        }
    },
    "side_effects": ["web_search API calls", "fetch_url HTTP requests"],
    "timeout_seconds": 120,
    "retries": 3,
}

Verify: Generate OpenAPI docs from TASK_INTERFACE and confirm downstream tasks can import and validate against it.

Step 10: Version and migrate task descriptions like code

Task descriptions are prompts. They drift, they break, they improve. Treat them as versioned artifacts.

# tasks/__init__.py
from .research_task import research_task, TASK_INTERFACE

TASK_REGISTRY = {
    "research_competitor_pricing": {
        "task": research_task,
        "interface": TASK_INTERFACE,
        "changelog": [
            {"version": "1.2.0", "date": "2024-01-15", "changes": "Added confidence threshold validation; fixed USD conversion for JPY"},
            {"version": "1.1.0", "date": "2023-11-03", "changes": "Added missing array for incomplete results"},
            {"version": "1.0.0", "date": "2023-09-12", "changes": "Initial release"},
        ],
    },
}

When you change a description, bump the version, update the changelog, and re-run the test suite. If you’re routing tasks through a gateway that meters per-token usage — n4n.ai surfaces this automatically — you can correlate version changes with cost and latency shifts.

Verify: After any description change, run the full test suite and compare token usage against the previous version. A 20% token increase without quality improvement is a regression.

Quick reference checklist

• Contract written before description
• Description uses labeled sections (Objective, Inputs, Output format, Constraints, Quality gates, Failure handling)
• All inputs explicitly bound with {variable} syntax
• Output constrained by Pydantic model via output_json
• Tool-use protocol specifies query construction, filtering, extraction, validation
• Pre-return verification checklist embedded
• Retry handler injects specific validation errors
• Adversarial test cases cover missing data, invalid inputs, schema violations
• Machine-readable interface documented with example
• Versioned with changelog

Follow these steps and your CrewAI tasks become deterministic components — debuggable, testable, and reliable enough to put in a production pipeline.