CrewAI SDK Track Part 6: Tools & Custom Tools

1. Tool Fundamentals

Tools are the bridge between your CrewAI agents and the external world. They allow agents to perform actions — searching the web, reading files, querying databases, calling APIs, or executing code. Without tools, agents can only reason; with tools, they can act.

1.1 Attaching Tools to Agents vs Tasks

Tools can be assigned at two levels: the agent level (available for all tasks that agent performs) or the task level (available only during that specific task). Task-level tools override agent-level tools when both are specified.

from crewai import Agent, Task, Crew
from crewai_tools import SerperDevTool, ScrapeWebsiteTool

# Agent-level tools: available for ALL tasks this agent performs
researcher = Agent(
    role="Senior Research Analyst",
    goal="Find comprehensive information on any topic",
    backstory="Expert researcher with deep web investigation skills",
    tools=[SerperDevTool(), ScrapeWebsiteTool()],
    verbose=True
)

# Task-level tools: available ONLY during this specific task
research_task = Task(
    description="Research the latest advances in quantum computing",
    expected_output="A detailed report with sources",
    agent=researcher,
    tools=[SerperDevTool()]  # Only search, no scraping for this task
)

crew = Crew(agents=[researcher], tasks=[research_task])
result = crew.kickoff()
print(result.raw)

1.2 Tool Caching & Async Execution

CrewAI supports tool-level caching to avoid redundant calls and async execution for I/O-bound tools:

from crewai import Agent, Task, Crew
from crewai_tools import SerperDevTool

# Tool with caching enabled (default behavior)
search_tool = SerperDevTool()

# Agent with cache enabled — repeated identical queries return cached results
researcher = Agent(
    role="Research Analyst",
    goal="Research efficiently without redundant API calls",
    backstory="Expert at finding information quickly",
    tools=[search_tool],
    cache=True,  # Enable caching for this agent's tool calls
    verbose=True
)

task = Task(
    description="Find the current CEO of OpenAI and their background",
    expected_output="Name and brief bio of the current OpenAI CEO",
    agent=researcher
)

crew = Crew(agents=[researcher], tasks=[task])
result = crew.kickoff()
print(result.raw)

2. Creating Custom Tools

CrewAI provides two approaches for creating custom tools: the BaseTool class for complex tools with state, and the @tool decorator for simple function-based tools.

2.1 BaseTool Class Inheritance

For tools that need internal state, complex initialization, or multiple methods, extend BaseTool:

from crewai.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type
import json
import urllib.request

class WeatherInput(BaseModel):
    """Input schema for the weather tool."""
    city: str = Field(..., description="The city name to get weather for")
    units: str = Field(default="metric", description="Temperature units: metric or imperial")

class WeatherTool(BaseTool):
    name: str = "weather_lookup"
    description: str = "Get current weather data for any city worldwide"
    args_schema: Type[BaseModel] = WeatherInput

    def _run(self, city: str, units: str = "metric") -> str:
        """Fetch weather data from a public API."""
        url = f"https://wttr.in/{city}?format=j1"
        try:
            with urllib.request.urlopen(url) as response:
                data = json.loads(response.read().decode())
                current = data["current_condition"][0]
                temp = current["temp_C"] if units == "metric" else current["temp_F"]
                unit_label = "°C" if units == "metric" else "°F"
                return (
                    f"Weather in {city}: {current['weatherDesc'][0]['value']}, "
                    f"Temperature: {temp}{unit_label}, "
                    f"Humidity: {current['humidity']}%"
                )
        except Exception as e:
            return f"Error fetching weather for {city}: {str(e)}"

# Usage with an agent
from crewai import Agent, Task, Crew

weather_agent = Agent(
    role="Weather Reporter",
    goal="Provide accurate weather information",
    backstory="Meteorology expert with access to global weather data",
    tools=[WeatherTool()],
    verbose=True
)

task = Task(
    description="What is the current weather in Tokyo and London?",
    expected_output="Weather report for both cities with temperature and conditions",
    agent=weather_agent
)

crew = Crew(agents=[weather_agent], tasks=[task])
result = crew.kickoff()
print(result.raw)

2.2 @tool Decorator for Function-Based Tools

For simpler tools without internal state, the @tool decorator provides a concise alternative:

from crewai.tools import tool
from crewai import Agent, Task, Crew

@tool("calculate_bmi")
def calculate_bmi(weight_kg: float, height_m: float) -> str:
    """Calculate Body Mass Index given weight in kg and height in meters.

    Args:
        weight_kg: Weight in kilograms
        height_m: Height in meters
    """
    bmi = weight_kg / (height_m ** 2)
    if bmi < 18.5:
        category = "underweight"
    elif bmi < 25:
        category = "normal weight"
    elif bmi < 30:
        category = "overweight"
    else:
        category = "obese"
    return f"BMI: {bmi:.1f} ({category})"

@tool("unit_converter")
def unit_converter(value: float, from_unit: str, to_unit: str) -> str:
    """Convert between common units of measurement.

    Args:
        value: The numeric value to convert
        from_unit: Source unit (kg, lb, km, mi, C, F)
        to_unit: Target unit (kg, lb, km, mi, C, F)
    """
    conversions = {
        ("kg", "lb"): lambda v: v * 2.20462,
        ("lb", "kg"): lambda v: v / 2.20462,
        ("km", "mi"): lambda v: v * 0.621371,
        ("mi", "km"): lambda v: v / 0.621371,
        ("C", "F"): lambda v: (v * 9/5) + 32,
        ("F", "C"): lambda v: (v - 32) * 5/9,
    }
    key = (from_unit, to_unit)
    if key not in conversions:
        return f"Conversion from {from_unit} to {to_unit} not supported"
    result = conversions[key](value)
    return f"{value} {from_unit} = {result:.2f} {to_unit}"

# Use both tools with an agent
health_agent = Agent(
    role="Health Calculator",
    goal="Help users with health-related calculations",
    backstory="Medical assistant specializing in health metrics",
    tools=[calculate_bmi, unit_converter],
    verbose=True
)

task = Task(
    description="Calculate BMI for someone who weighs 180 lbs and is 5 feet 10 inches tall. Convert units as needed.",
    expected_output="BMI calculation with category classification",
    agent=health_agent
)

crew = Crew(agents=[health_agent], tasks=[task])
result = crew.kickoff()
print(result.raw)

2.3 Tool Parameters with Pydantic Models

Define rich input schemas using Pydantic for validation and documentation:

from crewai.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type, Optional, List

class DatabaseQueryInput(BaseModel):
    """Input schema for database query tool."""
    table: str = Field(..., description="The database table to query")
    columns: List[str] = Field(default=["*"], description="Columns to select")
    where: Optional[str] = Field(default=None, description="WHERE clause condition")
    limit: int = Field(default=10, description="Maximum rows to return", ge=1, le=1000)

class DatabaseQueryTool(BaseTool):
    name: str = "database_query"
    description: str = "Execute read-only SQL queries against the application database"
    args_schema: Type[BaseModel] = DatabaseQueryInput

    def _run(self, table: str, columns: List[str] = None, where: str = None, limit: int = 10) -> str:
        """Build and execute a safe SELECT query."""
        cols = ", ".join(columns) if columns else "*"
        query = f"SELECT {cols} FROM {table}"
        if where:
            query += f" WHERE {where}"
        query += f" LIMIT {limit}"
        # In production, execute against actual DB
        return f"Query executed: {query}\n[Results would appear here]"

# Demonstrate schema introspection
tool_instance = DatabaseQueryTool()
print(f"Tool: {tool_instance.name}")
print(f"Description: {tool_instance.description}")
print(f"Schema: {DatabaseQueryInput.model_json_schema()}")

2.4 Error Handling in Tools

Robust tools handle failures gracefully and return informative error messages:

from crewai.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type
import json

class APICallInput(BaseModel):
    endpoint: str = Field(..., description="API endpoint path")
    method: str = Field(default="GET", description="HTTP method")

class RobustAPITool(BaseTool):
    name: str = "api_caller"
    description: str = "Make API calls with comprehensive error handling"
    args_schema: Type[BaseModel] = APICallInput

    def _run(self, endpoint: str, method: str = "GET") -> str:
        """Execute API call with error handling."""
        try:
            # Validate endpoint
            if not endpoint.startswith("/"):
                return "Error: Endpoint must start with '/'"

            # Simulate API call (replace with actual HTTP client)
            import urllib.request
            url = f"https://jsonplaceholder.typicode.com{endpoint}"
            req = urllib.request.Request(url, method=method)
            with urllib.request.urlopen(req, timeout=10) as response:
                data = json.loads(response.read().decode())
                return json.dumps(data, indent=2)[:2000]  # Truncate large responses

        except urllib.error.HTTPError as e:
            return f"HTTP Error {e.code}: {e.reason}"
        except urllib.error.URLError as e:
            return f"Connection Error: {str(e.reason)}"
        except TimeoutError:
            return "Error: Request timed out after 10 seconds"
        except Exception as e:
            return f"Unexpected error: {type(e).__name__}: {str(e)}"

# Test the tool directly
tool = RobustAPITool()
result = tool._run(endpoint="/posts/1")
print(result)

3. Publishing Custom Tools

CrewAI supports a community tool registry where you can share tools for others to use. Package your tools as Python packages following the standard structure.

3.1 Tool Package Structure

# Create a new tool package
crewai tool create my-custom-tool

# Package structure generated:
# my_custom_tool/
# ├── __init__.py
# ├── tool.py          # Your tool implementation
# ├── README.md        # Documentation
# ├── pyproject.toml   # Package metadata
# └── tests/
#     └── test_tool.py # Tests

# my_custom_tool/tool.py
from crewai.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type

class SentimentInput(BaseModel):
    """Input for sentiment analysis tool."""
    text: str = Field(..., description="Text to analyze for sentiment")

class SentimentAnalysisTool(BaseTool):
    name: str = "sentiment_analyzer"
    description: str = "Analyze the sentiment of text and return positive/negative/neutral classification"
    args_schema: Type[BaseModel] = SentimentInput

    def _run(self, text: str) -> str:
        """Analyze sentiment using keyword heuristics."""
        positive_words = {"good", "great", "excellent", "amazing", "love", "happy", "wonderful"}
        negative_words = {"bad", "terrible", "awful", "hate", "sad", "poor", "horrible"}

        words = set(text.lower().split())
        pos_count = len(words & positive_words)
        neg_count = len(words & negative_words)

        if pos_count > neg_count:
            sentiment = "positive"
        elif neg_count > pos_count:
            sentiment = "negative"
        else:
            sentiment = "neutral"

        return f"Sentiment: {sentiment} (positive_signals={pos_count}, negative_signals={neg_count})"

4. Coding Tools

CrewAI includes specialized tools for code generation, execution, and analysis. These enable agents to write, test, and iterate on code autonomously.

4.1 Code Execution Agent

from crewai import Agent, Task, Crew

# Coding agent with built-in code capabilities
coding_agent = Agent(
    role="Senior Python Developer",
    goal="Write clean, tested, production-ready Python code",
    backstory="""Expert Python developer with 15 years of experience.
    You write code that is well-documented, follows PEP 8, includes
    error handling, and has comprehensive docstrings.""",
    allow_code_execution=True,
    verbose=True
)

coding_task = Task(
    description="""Write a Python function that:
    1. Takes a list of dictionaries with 'name' and 'score' keys
    2. Returns the top N entries sorted by score (descending)
    3. Handles edge cases (empty list, N > list length)
    4. Includes type hints and docstring
    5. Include a test demonstrating the function works""",
    expected_output="Complete Python function with tests and example output",
    agent=coding_agent
)

crew = Crew(agents=[coding_agent], tasks=[coding_task])
result = crew.kickoff()
print(result.raw)

5. Advanced Tool Patterns

5.1 Force Tool Output as Result

Sometimes you want a tool’s output to become the task’s final result directly, bypassing further agent reasoning. Use result_as_answer=True:

from crewai import Agent, Task, Crew
from crewai.tools import tool

@tool("generate_report")
def generate_report(topic: str, data_points: str) -> str:
    """Generate a formatted report from data points.

    Args:
        topic: The report topic/title
        data_points: Comma-separated data points to include
    """
    points = [p.strip() for p in data_points.split(",")]
    report = f"# Report: {topic}\n\n"
    report += f"Generated: 2026-05-24\n\n"
    report += "## Key Findings\n\n"
    for i, point in enumerate(points, 1):
        report += f"{i}. {point}\n"
    report += f"\n## Summary\n\nAnalysis covers {len(points)} data points on {topic}."
    return report

# The tool's output becomes the task's final answer directly
report_agent = Agent(
    role="Report Generator",
    goal="Create formatted reports from raw data",
    backstory="Expert technical writer who produces clear, structured reports",
    tools=[generate_report],
    verbose=True
)

report_task = Task(
    description="Generate a report on 'Q2 Sales Performance' with these data points: Revenue up 15%, New customers 230, Churn rate 2.1%, NPS score 72",
    expected_output="A formatted markdown report",
    agent=report_agent,
    output_file="report.md"  # Also save to file
)

crew = Crew(agents=[report_agent], tasks=[report_task])
result = crew.kickoff()
print(result.raw)

5.2 Caching Strategies

Implement custom caching logic to optimize expensive tool calls:

from crewai.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type
import hashlib
import json

class CachedSearchInput(BaseModel):
    query: str = Field(..., description="Search query")

class CachedSearchTool(BaseTool):
    name: str = "cached_search"
    description: str = "Search with built-in response caching to minimize API calls"
    args_schema: Type[BaseModel] = CachedSearchInput
    cache_function: callable = None  # CrewAI's built-in cache hook

    _cache: dict = {}  # Simple in-memory cache

    def _run(self, query: str) -> str:
        """Search with caching layer."""
        # Generate cache key from query
        cache_key = hashlib.md5(query.lower().strip().encode()).hexdigest()

        # Check cache
        if cache_key in self._cache:
            return f"[CACHED] {self._cache[cache_key]}"

        # Simulate expensive search operation
        result = f"Search results for '{query}': Found 5 relevant articles about {query}."

        # Store in cache
        self._cache[cache_key] = result
        return result

# Demonstrate caching behavior
tool = CachedSearchTool()
print(tool._run("machine learning trends"))  # First call - fresh
print(tool._run("machine learning trends"))  # Second call - cached
print(tool._run("deep learning advances"))   # Different query - fresh