CrewAI Tutorial (Python): implementing guardrails for beginners

This tutorial shows you how to add guardrails to a CrewAI workflow in Python so your agents reject bad inputs, sanitize outputs, and stop unsafe actions before they happen. You need this when you want an agent to stay inside a narrow business policy, especially in banking or insurance where malformed or risky outputs are not acceptable.

What You'll Need

• •Python 3.10+
• •crewai
• •pydantic
• •An OpenAI API key
• •A .env file for secrets
• •Basic familiarity with CrewAI agents, tasks, and crews

Install the packages:

pip install crewai pydantic python-dotenv

Set your API key:

export OPENAI_API_KEY="your-key-here"

Step-by-Step

1. •Start by defining a strict schema for the output you want. Guardrails work best when the model has a clear contract to satisfy, so we’ll use Pydantic to validate the response shape.

from pydantic import BaseModel, Field

class PolicySummary(BaseModel):
    customer_name: str = Field(min_length=2)
    policy_type: str
    risk_level: str
    summary: str = Field(min_length=20)

def validate_policy_summary(data: dict) -> PolicySummary:
    return PolicySummary(**data)

2. •Next, create a guardrail function that checks the raw task output before it is accepted. If validation fails, CrewAI can retry the task instead of passing bad data downstream.

from typing import Tuple

def policy_summary_guardrail(result) -> Tuple[bool, str]:
    try:
        if hasattr(result, "raw"):
            payload = result.raw
        else:
            payload = result

        if isinstance(payload, str):
            import json
            payload = json.loads(payload)

        validate_policy_summary(payload)
        return True, "Output is valid."
    except Exception as e:
        return False, f"Invalid output: {e}"

3. •Now define an agent and a task that produces structured policy analysis. The important part is attaching the guardrail to the task and telling the model to return JSON only.

from crewai import Agent, Task, Crew, Process
from crewai.llm import LLM

llm = LLM(model="gpt-4o-mini")

policy_analyst = Agent(
    role="Policy Analyst",
    goal="Summarize insurance policies accurately",
    backstory="You analyze policy details for internal review.",
    llm=llm,
    verbose=True,
)

task = Task(
    description=(
        "Summarize this insurance policy into JSON with keys "
        "customer_name, policy_type, risk_level, summary."
    ),
    expected_output="Valid JSON matching the PolicySummary schema.",
    agent=policy_analyst,
    guardrail=policy_summary_guardrail,
    guardrail_max_retries=2,
)

4. •Build and run the crew. If the output fails validation, CrewAI will retry up to the limit you set; if it still fails, you get a clear error instead of a broken payload.

crew = Crew(
    agents=[policy_analyst],
    tasks=[task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff(inputs={
    "policy_text": (
        "Customer: Jane Doe. Policy type: Auto. "
        "Risk level: Medium. Summary: Covers collision and liability."
    )
})

print(result)

5. •Add a second guardrail for content filtering if you need business rules beyond schema validation. This is where you block disallowed phrases like payment instructions, legal advice, or anything outside scope.

DISALLOWED_TERMS = ["wire transfer", "guaranteed approval", "legal advice"]

def content_guardrail(result) -> tuple[bool, str]:
    text = getattr(result, "raw", str(result)).lower()
    for term in DISALLOWED_TERMS:
        if term in text:
            return False, f"Disallowed term found: {term}"
    return True, "Content passes policy."

review_task = Task(
    description="Review the previous answer for compliance with internal policy.",
    expected_output="A compliant review result.",
    agent=policy_analyst,
    guardrail=content_guardrail,
)

Testing It

Run the script with a valid example first and confirm that the crew returns structured JSON matching your schema. Then deliberately break the output by asking for missing fields or adding extra text and verify that the guardrail triggers retries instead of accepting invalid data.

If you want to test failure handling directly, temporarily lower guardrail_max_retries to 1 and pass an input that makes validation fail. You should see CrewAI surface an error after retry exhaustion rather than returning malformed output.

For production-style testing, add unit tests around your guardrail functions so they can be exercised without calling the LLM every time.

Next Steps

• •Add custom validators for regulated fields like claim amounts or underwriting decisions.
• •Chain multiple tasks so one agent generates content and another agent validates it.
• •Move from simple string checks to policy-based rules using JSON Schema or Pydantic model validators.

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.