Building reliable agents means accepting that your custom tools will fail — network partitions, rate limits, malformed responses, and upstream deprecations are inevitable. This guide covers crewai custom tool error handling patterns that keep your crews operational when things go wrong, moving beyond basic try/except into strategies that preserve agent reasoning and enable debugging.
Design your exception hierarchy first
Don’t let every tool raise generic Exception. A typed hierarchy lets agents (and your monitoring) distinguish between retryable failures, permanent errors, and cases requiring human intervention.
# exceptions.py
class ToolError(Exception):
"""Base exception for all custom tool failures."""
def __init__(self, message: str, *, retryable: bool = False, context: dict | None = None):
super().__init__(message)
self.retryable = retryable
self.context = context or {}
class TransientError(ToolError):
"""Upstream timeout, rate limit, 5xx — safe to retry with backoff."""
def __init__(self, message: str, *, retry_after: float | None = None, **kwargs):
super().__init__(message, retryable=True, **kwargs)
self.retry_after = retry_after
class PermanentError(ToolError):
"""Bad input, auth failure, 4xx — retrying won't help."""
def __init__(self, message: str, **kwargs):
super().__init__(message, retryable=False, **kwargs)
class ValidationError(PermanentError):
"""Input failed schema validation before the tool executed."""
passAgents can now catch TransientError and decide to retry, while PermanentError bubbles up for replanning. This mirrors how HTTP clients distinguish 4xx from 5xx — same principle, applied at the tool layer.
Wrap external calls with a retry policy
Raw requests or httpx calls inside a tool are a liability. Centralize retry logic so every tool gets consistent behavior without boilerplate.
# retry.py
from tenacity import (
retry,
stop_after_attempt,
wait_exponential_jitter,
retry_if_exception_type,
before_sleep_log,
)
import logging
log = logging.getLogger(__name__)
def transient_retry_policy(max_attempts: int = 3, base_wait: float = 1.0):
"""Retry TransientError with exponential backoff and jitter."""
return retry(
reraise=True,
stop=stop_after_attempt(max_attempts),
wait=wait_exponential_jitter(initial=base_wait, max=30.0),
retry=retry_if_exception_type(TransientError),
before_sleep=before_sleep_log(log, logging.WARNING),
)Apply it as a decorator on the actual I/O function, not the tool’s _run method — this keeps the tool class clean and the retry logic testable in isolation.
# tools/web_search.py
from crewai.tools import BaseTool
from pydantic import BaseModel, Field
from .retry import transient_retry_policy
from .exceptions import TransientError, PermanentError
import httpx
class WebSearchInput(BaseModel):
query: str = Field(..., min_length=1, max_length=500)
max_results: int = Field(default=5, ge=1, le=20)
class WebSearchTool(BaseTool):
name: str = "web_search"
args_schema: type[BaseModel] = WebSearchInput
def __init__(self, api_key: str, timeout: float = 10.0):
super().__init__()
self._client = httpx.AsyncClient(timeout=timeout)
self._api_key = api_key
@transient_retry_policy(max_attempts=3)
async def _fetch_results(self, query: str, max_results: int) -> list[dict]:
resp = await self._client.post(
"https://api.search.example/v1/search",
headers={"Authorization": f"Bearer {self._api_key}"},
json={"query": query, "limit": max_results},
)
if resp.status_code == 429:
retry_after = float(resp.headers.get("Retry-After", "2"))
raise TransientError("Rate limited", retry_after=retry_after)
if 500 <= resp.status_code < 600:
raise TransientError(f"Upstream error: {resp.status_code}")
if resp.status_code >= 400:
raise PermanentError(f"API error: {resp.status_code} - {resp.text}")
return resp.json().get("results", [])
def _run(self, query: str, max_results: int = 5) -> str:
# Validation happens via args_schema, but guard anyway
if not query.strip():
raise ValidationError("Query cannot be empty")
try:
results = self._fetch_results(query, max_results)
except TransientError as e:
# Agent sees this and can decide to retry or pivot
raise
except PermanentError as e:
# Include context for debugging
raise PermanentError(str(e), context={"query": query}) from e
return self._format_results(results)
def _format_results(self, results: list[dict]) -> str:
if not results:
return "No results found."
return "\n".join(f"- {r['title']}: {r['snippet']} ({r['url']})" for r in results)Notice the _run method stays thin — it validates, delegates to the retry-wrapped fetcher, and formats output. The async _fetch_results handles the actual I/O with proper error classification.
Return structured error payloads, not strings
When a tool fails, the agent receives whatever _run returns or raises. Returning a plain string like "Error: timeout" forces the agent to parse natural language. Return a structured dict (or JSON string) that the agent can reason over programmatically.
# tools/base.py
from crewai.tools import BaseTool
from pydantic import BaseModel
from typing import Any
import json
class ToolResult(BaseModel):
success: bool
data: Any = None
error: str | None = None
error_code: str | None = None
retryable: bool = False
metadata: dict = {}
class RobustBaseTool(BaseTool):
"""Base tool that returns structured results on both success and failure."""
def _run(self, *args, **kwargs) -> str:
try:
result = self._execute(*args, **kwargs)
return ToolResult(success=True, data=result).model_dump_json()
except ToolError as e:
return ToolResult(
success=False,
error=str(e),
error_code=e.__class__.__name__,
retryable=e.retryable,
metadata=e.context,
).model_dump_json()
except Exception as e:
# Catch-all for bugs — log and return structured failure
log.exception("Unexpected tool error")
return ToolResult(
success=False,
error=f"Internal error: {e}",
error_code="InternalError",
retryable=False,
).model_dump_json()
def _execute(self, *args, **kwargs) -> Any:
"""Subclasses implement this instead of _run."""
raise NotImplementedErrorNow the agent gets consistent JSON whether the tool succeeds or fails:
{"success": false, "error": "Rate limited", "error_code": "TransientError", "retryable": true, "metadata": {"retry_after": 2.0}}The agent’s system prompt can instruct it to check success and retryable before deciding next steps — far more reliable than hoping it interprets “Error: rate limited” correctly.
Implement circuit breakers for degraded upstreams
If an upstream API is consistently failing, hammering it wastes tokens and latency. A circuit breaker trips after a threshold of failures, failing fast for a cooldown period.
# circuit.py
import time
from threading import Lock
from dataclasses import dataclass, field
from enum import Enum
class CircuitState(Enum):
CLOSED = "closed" # Normal operation
OPEN = "open" # Failing fast
HALF_OPEN = "half_open" # Testing recovery
@dataclass
class CircuitBreaker:
failure_threshold: int = 5
success_threshold: int = 2
timeout: float = 30.0 # seconds before half-open
_state: CircuitState = field(default=CircuitState.CLOSED, init=False)
_failure_count: int = field(default=0, init=False)
_success_count: int = field(default=0, init=False)
_last_failure_time: float = field(default=0, init=False)
_lock: Lock = field(default_factory=Lock, init=False)
def call(self, func, *args, **kwargs):
with self._lock:
if self._state == CircuitState.OPEN:
if time.time() - self._last_failure_time >= self.timeout:
self._state = CircuitState.HALF_OPEN
self._success_count = 0
else:
raise TransientError("Circuit breaker open", context={"service": func.__name__})
try:
result = func(*args, **kwargs)
self._on_success()
return result
except TransientError:
self._on_failure()
raise
def _on_success(self):
with self._lock:
if self._state == CircuitState.HALF_OPEN:
self._success_count += 1
if self._success_count >= self.success_threshold:
self._state = CircuitState.CLOSED
self._failure_count = 0
elif self._state == CircuitState.CLOSED:
self._failure_count = 0
def _on_failure(self):
with self._lock:
self._failure_count += 1
self._last_failure_time = time.time()
if self._state == CircuitState.HALF_OPEN:
self._state = CircuitState.OPEN
elif self._failure_count >= self.failure_threshold:
self._state = CircuitState.OPENAttach one breaker per upstream service (not per tool instance) so all tools hitting the same API share state:
# tools/search_with_breaker.py
from .web_search import WebSearchTool
from .circuit import CircuitBreaker, CircuitState
_SEARCH_BREAKER = CircuitBreaker(failure_threshold=5, timeout=60.0)
class ResilientWebSearchTool(WebSearchTool):
def _run(self, query: str, max_results: int = 5) -> str:
def _do_search():
return super()._run(query, max_results)
try:
return _SEARCH_BREAKER.call(_do_search)
except TransientError as e:
if "Circuit breaker open" in str(e):
# Add context so agent knows not to retry immediately
raise TransientError(
"Search service temporarily unavailable",
context={"circuit_state": "open", "retry_after": 60}
) from e
raiseAdd structured logging for observability
You can’t debug production tool failures from agent transcripts alone. Emit structured logs at key points — especially on retry, circuit breaker transitions, and permanent failures.
# logging_config.py
import structlog
import logging
structlog.configure(
processors=[
structlog.processors.add_log_level,
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.JSONRenderer(),
],
wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
context_class=dict,
logger_factory=structlog.PrintLoggerFactory(),
cache_logger_on_first_use=True,
)
log = structlog.get_logger("crewai.tools")Then instrument your tools:
# In _fetch_results
@transient_retry_policy(max_attempts=3)
async def _fetch_results(self, query: str, max_results: int) -> list[dict]:
log.info("search_request", query=query, max_results=max_results)
start = time.perf_counter()
try:
resp = await self._client.post(...)
duration = time.perf_counter() - start
log.info("search_response", status=resp.status_code, duration_ms=duration*1000)
...
except TransientError as e:
log.warning("search_transient_error", error=str(e), retryable=True, **e.context)
raiseThis gives you queryable logs: error_code:TransientError duration_ms:>5000 finds slow failures, circuit_state:open shows breaker trips.
Handle agent-level retries with max_iterations
Even with tool-level retries, agents sometimes loop — calling the same failing tool repeatedly. Configure the crew to limit iterations and expose the failure to your orchestration layer.
from crewai import Crew, Process
from crewai.tasks import Task
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, write_task],
process=Process.sequential,
max_iter=15, # Hard cap on agent loop iterations
verbose=True,
)
result = crew.kickoff(inputs={"topic": "quantum computing"})
if not result.success:
# Your orchestration decides: alert, fallback, human review
handle_crew_failure(result)Pair this with a custom Task callback that logs each tool invocation:
def tool_callback(tool_name: str, args: dict, result: str, error: Exception | None):
log.info("tool_invocation", tool=tool_name, args=args,
success=error is None, error_type=type(error).__name__ if error else None)Common pitfalls and tradeoffs
Pitfall: Swallowing exceptions in _run to return error strings.
This breaks the agent’s ability to distinguish failure modes. Always raise typed exceptions or return structured ToolResult objects.
Pitfall: Retrying non-idempotent operations.
POST requests that mutate state (sending email, charging a card) must not be retried automatically. Mark those tools with retryable=False at the exception level, or use idempotency keys.
Pitfall: Infinite retry loops from agent + tool both retrying.
If your tool retries 3x and the agent retries the task 5x, that’s 15 attempts. Coordinate: either let the tool handle all retries (simpler) or let the agent handle retries with tool returning fast failures. Don’t do both.
Tradeoff: Circuit breaker adds latency on recovery.
The half-open state tests with real requests. For high-stakes upstreams, consider a synthetic health check endpoint instead of live traffic.
Tradeoff: Structured results increase token usage.
JSON error payloads cost more tokens than “Error: timeout”. Worth it — the agent makes better decisions, reducing total turns.
Testing your error paths
Unit test the exception hierarchy and retry policy in isolation. Integration test the full tool with a mock server that returns various failure modes.
# tests/test_web_search_tool.py
import pytest
import respx
import httpx
from tools.web_search import WebSearchTool
from tools.exceptions import TransientError, PermanentError, ValidationError
@respx.mock
@pytest.mark.asyncio
async def test_rate_limit_triggers_transient_error():
tool = WebSearchTool(api_key="test-key")
route = respx.post("https://api.search.example/v1/search").mock(
return_value=httpx.Response(429, headers={"Retry-After": "2"})
)
with pytest.raises(TransientError) as exc:
await tool._fetch_results("test query", 5)
assert exc.value.retryable is True
assert exc.value.retry_after == 2.0
assert route.call_count == 3 # Initial + 2 retries
@respx.mock
@pytest.mark.asyncio
async def test_400_raises_permanent_error():
tool = WebSearchTool(api_key="test-key")
respx.post("https://api.search.example/v1/search").mock(
return_value=httpx.Response(400, text="Invalid query")
)
with pytest.raises(PermanentError):
await tool._fetch_results("test query", 5)
def test_empty_query_raises_validation_error():
tool = WebSearchTool(api_key="test-key")
with pytest.raises(ValidationError):
tool._run("") # Bypasses args_schema validation intentionallyRun these in CI. They catch regressions when upstream APIs change error formats.
Summary checklist
- Define a typed exception hierarchy (
TransientError,PermanentError,ValidationError) - Centralize retry policy with exponential backoff and jitter (tenacity)
- Return structured
ToolResultJSON from every tool, not raw strings - Add circuit breakers per upstream service, not per tool
- Emit structured logs at request/response/retry/breaker boundaries
- Set
max_iteron Crew to bound agent loops - Test each error path with mock servers in CI
These patterns scale from a single-agent prototype to multi-crew production systems. The upfront investment in error architecture pays off every time an upstream API degrades and your agents keep working instead of spinning.