CrewAI Tutorial (Python): implementing guardrails for intermediate developers

This tutorial shows you how to add guardrails to a CrewAI workflow so agents only accept valid inputs, return structured outputs, and fail fast when something is off. You need this when you’re moving from demos to production, where bad model output can break downstream tools, corrupt data, or trigger unsafe actions.

What You'll Need

• •Python 3.10+
• •crewai
• •pydantic
• •OpenAI API key set as OPENAI_API_KEY
• •A working virtual environment
• •Basic familiarity with Agent, Task, and Crew

Install the packages:

pip install crewai pydantic

Set your API key:

export OPENAI_API_KEY="your-key-here"

Step-by-Step

1. •Start with a simple agent and task.
   We’ll build a small research workflow first, then add guardrails around the task output so the crew cannot continue with malformed data.

from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role="Research Analyst",
    goal="Summarize customer complaints clearly",
    backstory="You analyze support tickets for patterns.",
    verbose=True,
)

task = Task(
    description="Summarize the top complaint in one short paragraph.",
    expected_output="A concise summary of the complaint.",
    agent=researcher,
)

crew = Crew(
    agents=[researcher],
    tasks=[task],
    process=Process.sequential,
)

result = crew.kickoff()
print(result)

2. •Define a strict schema for what “good” output looks like.
   Guardrails work best when you make validity explicit. Here we use Pydantic to enforce a structured response that downstream code can trust.

from pydantic import BaseModel, Field

class ComplaintSummary(BaseModel):
    category: str = Field(..., min_length=3)
    severity: int = Field(..., ge=1, le=5)
    summary: str = Field(..., min_length=20)

3. •Add a guardrail function that validates raw agent output.
   In CrewAI, guardrails are just Python functions that inspect the model’s output and either accept it or reject it with an error message.

import json

def validate_complaint_summary(output):
    try:
        data = json.loads(output) if isinstance(output, str) else output
        parsed = ComplaintSummary(**data)
        return True, parsed.model_dump()
    except Exception as e:
        return False, f"Invalid output format: {e}"

4. •Attach the guardrail to your task and force JSON output.
   The model must now return data that matches your schema. If it doesn’t, CrewAI will retry based on your configuration instead of passing broken data forward.

guarded_task = Task(
    description=(
        "Analyze the top customer complaint and return JSON with keys "
        "`category`, `severity`, and `summary`."
    ),
    expected_output="Valid JSON matching the ComplaintSummary schema.",
    agent=researcher,
    guardrail=validate_complaint_summary,
    guardrail_max_retries=2,
)

guarded_crew = Crew(
    agents=[researcher],
    tasks=[guarded_task],
    process=Process.sequential,
)

guarded_result = guarded_crew.kickoff()
print(guarded_result)

5. •Add a second layer of protection for business rules.
   Schema validation catches shape errors, but not all bad outputs are structurally invalid. For example, you may want severity capped at 3 for non-critical issues or reject vague summaries.

def business_rule_guardrail(output):
    try:
        data = json.loads(output) if isinstance(output, str) else output
        parsed = ComplaintSummary(**data)

        if parsed.severity >= 4 and "urgent" not in parsed.summary.lower():
            return False, "High severity complaints must mention urgency."

        return True, parsed.model_dump()
    except Exception as e:
        return False, f"Business rule validation failed: {e}"

6. •Chain validated output into the next task safely.
   Once your first task is guarded, you can pass trusted structured data into downstream tasks without re-parsing free-form text.

triage_agent = Agent(
    role="Triage Specialist",
    goal="Recommend next action based on validated complaint data",
    backstory="You route issues to the correct support queue.",
)

triage_task = Task(
    description="Recommend a next action using the validated complaint summary.",
    expected_output="A routing recommendation.",
    agent=triage_agent,
)

pipeline_crew = Crew(
    agents=[researcher, triage_agent],
    tasks=[guarded_task, triage_task],
    process=Process.sequential,
)

pipeline_result = pipeline_crew.kickoff()
print(pipeline_result)

Testing It

Run the script with a few different prompts that should produce both valid and invalid structures. A good test is to intentionally ask for plain English instead of JSON and confirm the guardrail rejects it or retries until it gets a valid response.

Then test edge cases in your validation logic by tightening the schema fields or adding business rules that should fail on purpose. If your guardrail is working correctly, bad outputs never reach the next task and the final result stays predictable.

Also inspect logs with verbose=True so you can see whether retries are happening and why validation failed. In production, those failure messages are what you’ll use to tune prompts and reduce retry rates.

Next Steps

• •Add custom retry messaging so the model knows exactly how to fix invalid outputs.
• •Move from simple schema checks to tool-output validation before calling external systems.
• •Build a shared guardrail module for all crews in your codebase so every team uses the same validation rules.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.