Guardrails: checking a task's answer
A guardrail is a function that checks a task's answer, and on failure sends the reason back to the agent to try the task again, up to a limit.
Last updated: 28 Sep, 2026 · CrewAI 1.15
A masking hook edits a bad reply after the fact. A guardrail rejects it instead and asks the agent to write it again, the way an editor sends a draft back.
A guardrail function
A guardrail takes the task's output and returns a pair: accept with a value, or reject with a reason the agent will read.
def no_card_numbers(output):
if re.search(r"\d{4} ?\d{4} ?\d{4} ?\d{4}", output.raw):
return (False, "Remove the card number.") # reject; the agent retries
return (True, output.raw) # accept this valueWriting the check
This one looks for a sixteen-digit card number in the reply.
import re
def no_card_numbers(output):
if re.search(r"\d{4} ?\d{4} ?\d{4} ?\d{4}", output.raw):
return (False, "Remove the card number. Never repeat one to a customer.")
return (True, output.raw)
Passing it when the task is made
A Task reads its guardrail when it is created. Assigning one to an existing task does nothing in 1.15.22, and no error tells you so, so pass it to Task.
look = Task(description="Find the order in this message: {question}",
expected_output="The order's status.", agent=clerk)
reply = Task(description="Write the customer a reply.",
expected_output="A short, friendly email.", agent=writer,
guardrail=no_card_numbers)
crew = Crew(agents=[clerk, writer], tasks=[look, reply], verbose=False)
View the code here
from crewai.tools import tool
ORDERS = {"A17": "shipped on 3 March", "C40": "waiting for stock"}
@tool
def lookup_order(order_id: str) -> str:
"""Look up an order's shipping status by its id, such as A17."""
status = ORDERS.get(order_id)
return f"{order_id} {status}." if status else f"{order_id} is not an order we have."
import json
import os
import re
from crewai import BaseLLM
os.environ["OTEL_SDK_DISABLED"] = "true"
os.environ["CREWAI_DISABLE_TELEMETRY"] = "true"
os.environ["CREWAI_TRACING_ENABLED"] = "false"
os.environ["CREWAI_DISABLE_VERSION_CHECK"] = "true"
class ShopLLM(BaseLLM):
script: list = []
def supports_function_calling(self):
return True
def call(self, messages, tools=None, **kwargs):
if isinstance(messages, str):
messages = [{"role": "user", "content": messages}]
if self.script:
return self.script.pop(0)
names = [t["function"]["name"] for t in tools or []]
return self.decide(messages, names)
def decide(self, messages, tools):
last = messages[-1]
if last["role"] == "tool":
return last["content"]
text = last["content"]
orders = re.findall(r"\b[A-Z]\d+\b", text)
wanted = "refund_order" if "refund" in text.lower() else "lookup_order"
matches = [name for name in tools if name.endswith(wanted)]
if orders and matches:
args = json.dumps({"order_id": orders[0]})
return [{"id": f"call_{orders[0]}", "type": "function",
"function": {"name": matches[0], "arguments": args}}]
if "working with:" in text:
context = text.split("working with:")[1].strip().split("\n\n")[0]
return f"Dear customer, {context}"
if orders:
return f"I have no way to look up {orders[0]} yet."
return "Hello. Which order is this about?"
A reply sent back and rewritten
The writer is scripted with a bad reply then a clean one, so you can watch the retry.
import os
os.environ["OTEL_SDK_DISABLED"] = "true"
os.environ["CREWAI_DISABLE_TELEMETRY"] = "true"
from crewai import Agent, Crew, Task
from shop_llm import ShopLLM
from tools import lookup_order
clerk = Agent(role="Order clerk", goal="Find the status of customers' orders",
backstory="You can look up any order in the shop's system.",
llm=ShopLLM(model="shop"), tools=[lookup_order])
writer = Agent(role="Reply writer", goal="Write replies to customers",
backstory="You write short, friendly emails.",
llm=ShopLLM(model="shop"))
import re
def no_card_numbers(output):
if re.search(r"\d{4} ?\d{4} ?\d{4} ?\d{4}", output.raw):
return (False, "Remove the card number. Never repeat one to a customer.")
return (True, output.raw)
look = Task(description="Find the order in this message: {question}",
expected_output="The order's status.", agent=clerk)
reply = Task(description="Write the customer a reply.",
expected_output="A short, friendly email.", agent=writer,
guardrail=no_card_numbers)
crew = Crew(agents=[clerk, writer], tasks=[look, reply], verbose=False)
writer.llm.script = ["Refunded to card 4111 1111 1111 1234.", "Refunded to your card."]
print(crew.kickoff(inputs={"question": "Where is my order A17?"}).raw)Refunded to your card.
What the retry did
- The first reply failed the check, so its card number never left the crew.
- The task ran again with the failed answer and your reason added to its prompt under
Previous attempt failed validation. - The second reply passed, and that clean answer is the task's result.
When it never passes
Give the writer only bad replies and cap the retries, and the task fails with the last reason instead of leaking anything.
import os
os.environ["OTEL_SDK_DISABLED"] = "true"
os.environ["CREWAI_DISABLE_TELEMETRY"] = "true"
from crewai import Agent, Crew, Task
from shop_llm import ShopLLM
from tools import lookup_order
clerk = Agent(role="Order clerk", goal="Find the status of customers' orders",
backstory="You can look up any order in the shop's system.",
llm=ShopLLM(model="shop"), tools=[lookup_order])
writer = Agent(role="Reply writer", goal="Write replies to customers",
backstory="You write short, friendly emails.",
llm=ShopLLM(model="shop"))
import re
def no_card_numbers(output):
if re.search(r"\d{4} ?\d{4} ?\d{4} ?\d{4}", output.raw):
return (False, "Remove the card number. Never repeat one to a customer.")
return (True, output.raw)
look = Task(description="Find the order in this message: {question}",
expected_output="The order's status.", agent=clerk)
reply = Task(description="Write the customer a reply.",
expected_output="A short, friendly email.", agent=writer,
guardrail=no_card_numbers)
crew = Crew(agents=[clerk, writer], tasks=[look, reply], verbose=False)
reply.guardrail_max_retries = 1
writer.llm.script = ["Card 4111 1111 1111 1234."] * 2
try:
crew.kickoff(inputs={"question": "Where is my order A17?"})
except Exception as error:
print(error)Task failed guardrail validation after 1 retries. Last error: Remove the card number. Never repeat one to a customer.
guardrail_max_retries defaults to a small number; this task allowed one retry. Set it yourself when you need a known cap.
Guardrail vs a masking hook
| Masking hook | Guardrail | |
|---|---|---|
| What it does | Edits the reply in place | Rejects it and asks again |
| The agent learns | Nothing; the edit is silent | The reason, on its next attempt |
| On repeated failure | Keeps editing | Fails the task after the cap |
When to guard a task
- An answer must never contain something: a card number, a promise you cannot keep.
- An answer must have a shape you can check in Python before it leaves the crew.
Task(...), not assigned afterward, and nothing warns you. A task can take a list, guardrails=[...], run in order.Related
- Previous: Structured output with output_pydantic
- Next: Conditional tasks that run only sometimes
- Reference: Tasks
- Return
(True, output.raw.upper())on success and print the result. - Add a second guardrail that rejects replies longer than 60 characters.
- Set
guardrail_max_retriesto 0 with the first script.
Little by little, you're building something great.