CrewAI Task Guardrails

Small, focused examples of task guardrails in CrewAI: reusable validator factories, a stand-alone re-implementation of the retry loop so the contract can be seen without an LLM, and a one-agent demo crew whose guardrail is guaranteed to trip.

Docs: Task guardrails in the CrewAI documentation.

The guardrail contract

A guardrail is a function that receives the task output and returns a tuple:

def guardrail(output) -> tuple[bool, Any]:
    ...
    return (True, validated_output)     # accepted; a str result replaces task_output.raw
    return (False, "what went wrong")   # rejected; the message goes back to the agent

• output is a TaskOutput (output.raw is the text; output.pydantic / output.json_dict when the task requested structured output). The validators in this project also accept a plain str, which is convenient for tests and demos.
• Attach it with Task(guardrail=fn) or chain several with Task(guardrails=[fn1, fn2, ...]) - each receives the validated output of the previous one.
• When it fails, CrewAI appends "Previous attempt failed validation: <message>" plus the previous result to the agent's context and runs the task again, up to guardrail_max_retries times (default 3). After that it raises Exception("Task failed guardrail validation after N retries. Last error: ..."). max_retries= still works as an alias but is deprecated.
• CrewAI checks the callable when the Task is built: exactly one parameter without a default, and, if you annotate the return type, it must be tuple[bool, Any] (or Tuple[bool, Any]). Because the annotation is inspected at runtime, guardrails/validators.py does not use from __future__ import annotations.
• Returning None as the validated output is an error; return the text (or a cleaned version of it).

Write the error message for the agent, not for a log: say what was measured, what the limit is and what to change ("Summary has 212 words; the limit is 150. Cut at least 62 words...").

Code-based or LLM-based guardrails?

CrewAI also accepts a string: Task(guardrail="The summary must be under 150 words and contain no jargon") builds an LLMGuardrail that asks the agent's LLM to judge the output. Pick by the kind of rule:

| Use a code guardrail when | Use an LLM guardrail when | |---|---| | the rule is measurable: word counts, JSON validity, schema, required headings, forbidden strings, regexes | the rule is subjective: tone, clarity, "suitable for a general audience" | | you need determinism and zero extra cost per retry | the rule is easier to describe than to code | | the failure message must be precise (the agent fixes what you name) | you accept some variance in verdicts and an extra model call per check | | the output feeds another program (parse it, do not eyeball it) | the output is prose for a human |

Mix them in one list when both apply: guardrails=[validate_json_schema(schema), "The description field must be written for non-experts"]. Put the code checks first so the model is only consulted when the mechanical checks already pass.

Code map

crewai-task-guardrails/
├── guardrails/
│   ├── __init__.py       re-exports the factories
│   ├── validators.py     validate_summary_length, validate_json_schema, validate_no_placeholders,
│   │                     validate_contains_sections, all_of; output_text() handles str/TaskOutput;
│   │                     a JSON-Schema checker (jsonschema if installed, minimal fallback otherwise)
│   ├── retry.py          run_with_guardrail(agent_fn, guardrail, max_retries): the retry loop
│   │                     CrewAI runs, reproduced with a plain callable in place of the agent
│   ├── crew.py           demo crew: Technical Writer agent + summarize task with
│   │                     guardrail=validate_summary_length(150), guardrail_max_retries=3;
│   │                     install_guardrail_logger() prints guardrail events
│   └── samples.py        source text, canned outputs, a small JSON schema
├── run.py                --demo (offline) or the live crew
├── tests/
│   ├── test_validators.py       every validator with str and TaskOutput; CrewAI accepts the
│   │                            callables; CrewAI's own process_guardrail() with our tuples
│   └── test_retry_and_crew.py   retry loop, crew wiring, event logger
├── .env.example
├── requirements.txt
└── pytest.ini

How the pieces fit: a factory such as validate_summary_length(150) returns a closure with the contract above (and a readable __name__, which CrewAI reports in its guardrail events). crew.build_summary_task passes that closure as guardrail= and sets guardrail_max_retries. In live mode CrewAI drives the loop; in demo mode retry.run_with_guardrail drives the same loop with a fake agent that shortens its answer whenever it receives feedback.

flowchart LR
    A[Agent produces output] --> G{guardrail(output)}
    G -->|"(True, validated)"| N[validated text replaces task_output.raw<br/>next task / crew result]
    G -->|"(False, message)"| R{retries left?}
    R -->|yes| F[feedback added to context:<br/>previous attempt failed validation: message] --> A
    R -->|no| E[Exception: failed after N retries]

Run

python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt

python run.py --demo                     # offline, no keys
pytest                                   # no network

cp .env.example .env                     # add OPENAI_API_KEY
python run.py                            # live crew: asks for ~200 words, accepts 150
python run.py --target-words 300 --max-words 120 --max-retries 2 --verbose

GUARDRAIL_DEMO_MODEL (default openai/gpt-4o-mini) and OPENAI_API_BASE select the model / endpoint for the live run.

Sample output

python run.py --demo:

1. Guardrails on canned outputs

  PASS validate_summary_length(max_words=150) <- summary_120_words
       validated output: 'Retrieval-augmented generation grounds a language model in retrieved p'
  FAIL validate_summary_length(max_words=150) <- summary_212_words
       Summary has 212 words; the limit is 150. Cut at least 62 words and keep only the essential points.
  PASS validate_json_schema       <- json_fenced
       validated output: '{"title": "RAG in one page", "word_count": 120, "tags": ["rag"]}'
  FAIL validate_json_schema       <- json_missing_field
       JSON does not match the schema: $: 'word_count' is a required property.
  FAIL validate_no_placeholders   <- placeholders
       Output still contains placeholder text: '[Insert customer name]', '{{agent_name}}', 'TODO'. Replace every placeholder with real content.
  FAIL validate_contains_sections <- report_missing_section
       Missing section heading(s): Recommendations. Use exactly these headings: Summary, Findings, Recommendations.

2. Retry loop with a fake agent (limit 150 words, max_retries=3)

  agent produced 240 words (first attempt)
  agent produced 190 words (after feedback: ### Previous attempt failed validation: Summary has 240 word)
  agent produced 140 words (after feedback: ### Previous attempt failed validation: Summary has 190 word)
  attempt 1: FAIL - Summary has 240 words; the limit is 150. Cut at least 90 words and keep only the essential points.
  attempt 2: FAIL - Summary has 190 words; the limit is 150. Cut at least 40 words and keep only the essential points.
  attempt 3: PASS
  accepted after 2 retries: 140 words

3. Same loop with an agent that never complies (max_retries=1)

  raised GuardrailRetryError: Task failed guardrail validation after 1 retries. Last error: Summary has 300 words; ...

Live run (python run.py), illustrative - word counts vary by model. The prompt asks for 200 words so the first attempt normally fails and CrewAI retries with the message in context:

asking for ~200 words, guardrail accepts 150, max_retries=3

[guardrail] attempt 1: running validate_summary_length(max_words=150) (function)
[guardrail] attempt 1: FAIL - Summary has 203 words; the limit is 150. Cut at least 53 words ...
[guardrail] attempt 2: running validate_summary_length(max_words=150) (function)
[guardrail] attempt 2: PASS

final output (138 words):
Retrieval-augmented generation pairs a language model with a search step ...

Notes

• validate_json_schema returns the canonical JSON text with code fences removed, so a task that also sets output_json= gets clean input on the retry that passes.
• validate_contains_sections accepts Markdown headings (## Findings), bold lines (**Findings**) and Findings: lines; matching is case-insensitive unless case_sensitive=True.
• validate_no_placeholders matches [Insert ...], {{ var }}, <insert ...>, upper-case TODO / TBD / XXX, lorem ipsum and [...]; pass your own list of regexes to replace the defaults.
• Guardrail events (LLMGuardrailStartedEvent / LLMGuardrailCompletedEvent) are emitted for function guardrails too; install_guardrail_logger subscribes to them through crewai_event_bus.on(...).