← Back to DevBytes

CrewAI Tools and Toolkits: Extending Agent Capabilities

Understanding CrewAI Tools and Toolkits

In CrewAI, tools are the atomic units of capability that transform agents from simple text generators into action-oriented entities. A tool is essentially a function or class that an agent can invoke to perform a specific task—searching the web, reading files, executing code, calling an API, or querying a database. Toolkits, on the other hand, are curated collections of related tools bundled together around a common domain or integration, such as the GitHubToolkit for repository management or the SerperToolkit for web search operations.

When you assign a tool to an agent, you are giving that agent the ability to reach outside its language model context and interact with the real world. The agent does not execute the tool directly; instead, the CrewAI framework orchestrates the tool call, passes the result back to the agent's reasoning loop, and allows the agent to incorporate that information into its next decision or final response.

What Exactly Are CrewAI Tools?

At a technical level, a CrewAI tool is a Python class that inherits from BaseTool (or is decorated with @tool) and exposes a set of properties and methods that define how the agent should invoke it. Each tool must declare:

Here is a minimal custom tool definition:

from crewai_tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type

class WeatherInput(BaseModel):
    city: str = Field(description="The city name to get weather for")
    country: str = Field(default="US", description="Two-letter country code")

class WeatherTool(BaseTool):
    name: str = "Weather Fetcher"
    description: str = "Fetches current weather data for a given city and country."
    args_schema: Type[BaseModel] = WeatherInput

    def _run(self, city: str, country: str = "US") -> str:
        """Simulates fetching weather data. Replace with a real API call."""
        # In production, call an actual weather API here
        weather_data = {
            "New York": "72°F, Partly Cloudy",
            "London": "60°F, Rainy",
            "Tokyo": "68°F, Sunny"
        }
        key = f"{city}, {country}"
        return weather_data.get(city, f"Weather data for {key} is not available.")

# Using the custom tool with an agent
from crewai import Agent

weather_agent = Agent(
    role="Weather Reporter",
    goal="Provide accurate weather information for requested cities.",
    backstory="You are a meteorologist with access to real-time weather data.",
    tools=[WeatherTool()],
    verbose=True
)

Notice how args_schema uses a Pydantic BaseModel. This is critical because the LLM uses the schema's field names and descriptions to correctly structure arguments when invoking the tool. Without a well-defined schema, the agent may pass malformed or incomplete arguments.

The @tool Decorator Approach

For simpler use cases, CrewAI provides a @tool decorator that converts a plain Python function into a tool without requiring you to manually subclass BaseTool. This is ideal for quick integrations:

from crewai_tools import tool
from pydantic import BaseModel, Field

class MultiplyInput(BaseModel):
    x: float = Field(description="First number")
    y: float = Field(description="Second number")

@tool("Multiplication Tool")
def multiply_tool(x: float, y: float) -> str:
    """Multiplies two numbers and returns the result as a string."""
    result = x * y
    return f"The product of {x} and {y} is {result}"

# The decorated function is now a BaseTool instance you can assign to agents
math_agent = Agent(
    role="Calculator",
    goal="Perform arithmetic operations accurately.",
    backstory="You are a precise mathematical computation engine.",
    tools=[multiply_tool],
    verbose=True
)

The decorator infers the args_schema from the function's type hints and the Pydantic model you pass. The first argument to @tool() is the human-readable name, and the docstring becomes the tool description that the LLM reads.

Why Tools and Toolkits Matter

Without tools, a CrewAI agent is limited to its training data and the context provided in prompts. Tools bridge the gap between reasoning and action. They matter because:

Built-in Toolkits Overview

CrewAI ships with a rich set of pre-built toolkits maintained by the community and core team. These toolkits wrap popular APIs and services into ready-to-use tool collections. Some of the most commonly used toolkits include:

Here is how you would equip a research agent with the Serper toolkit for web search capabilities:

import os
from crewai import Agent
from crewai_tools import SerperToolkit

# Set your Serper API key as an environment variable
os.environ["SERPER_API_KEY"] = "your-serper-api-key-here"

# Instantiate the toolkit
serper_toolkit = SerperToolkit()

# The toolkit contains multiple tools; you can assign them individually
research_agent = Agent(
    role="Senior Researcher",
    goal="Conduct thorough internet research and synthesize findings.",
    backstory="You are an expert researcher with 15 years of experience in investigative journalism.",
    tools=serper_toolkit.get_tools(),  # Gets all tools from the toolkit
    verbose=True
)

# Alternatively, pick specific tools from the toolkit
# tools=[serper_toolkit.get_tool("search"), serper_toolkit.get_tool("news")]

Composing a Multi-Tool Agent Crew

The true power of tools emerges when multiple agents with different tool sets collaborate. Consider a data analysis crew where one agent scrapes web data, another analyzes it with code execution, and a third writes a report:

import os
from crewai import Agent, Task, Crew, Process
from crewai_tools import (
    SerperToolkit,
    ScrapeWebsiteTool,
    CodeInterpreterToolkit,
    FileWriteTool
)

os.environ["SERPER_API_KEY"] = "your-key"
os.environ["OPENAI_API_KEY"] = "your-openai-key"

# Agent 1: Web researcher with search and scraping tools
serper = SerperToolkit()
scraper = ScrapeWebsiteTool()

researcher = Agent(
    role="Data Researcher",
    goal="Find and extract relevant data from the web about a given topic.",
    backstory="You excel at discovering hard-to-find information online.",
    tools=[*serper.get_tools(), scraper],
    verbose=True
)

# Agent 2: Data analyst with code execution capabilities
code_interpreter = CodeInterpreterToolkit()

analyst = Agent(
    role="Data Analyst",
    goal="Analyze raw data and produce statistical summaries and visualizations.",
    backstory="You are a quantitative analyst who writes Python code to uncover insights.",
    tools=code_interpreter.get_tools(),
    verbose=True,
    allow_code_execution=True
)

# Agent 3: Report writer with file writing capability
writer = Agent(
    role="Report Writer",
    goal="Compose a polished final report based on the analyst's findings.",
    backstory="You transform analytical results into clear, actionable reports.",
    tools=[FileWriteTool()],
    verbose=True
)

# Define tasks that flow from research → analysis → writing
research_task = Task(
    description="Research the latest trends in renewable energy adoption across Europe. "
                "Gather data on solar, wind, and hydro capacity additions for 2024.",
    expected_output="A structured dataset with country-level renewable energy statistics.",
    agent=researcher,
    output_file="raw_data.md"
)

analysis_task = Task(
    description="Using the raw data collected, compute growth percentages, "
                "identify the top 5 countries by capacity addition, and create a summary table.",
    expected_output="A markdown table with analysis results and key insights.",
    agent=analyst,
    output_file="analysis.md"
)

write_task = Task(
    description="Write a comprehensive executive summary report based on the analysis. "
                "Include an introduction, key findings, data table, and recommendations.",
    expected_output="A final report in markdown format, saved to 'final_report.md'.",
    agent=writer,
    output_file="final_report.md"
)

# Assemble the crew
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, write_task],
    process=Process.sequential,
    verbose=True
)

result = crew.kickoff()
print(result)

This example demonstrates the essential pattern: each agent receives only the tools relevant to its role, creating a secure and focused workflow. The researcher cannot execute arbitrary code, the analyst cannot write files arbitrarily, and the writer cannot scrape websites—each agent's capabilities are precisely scoped.

Creating Custom Toolkits

When your project requires a cohesive set of custom tools—perhaps wrapping an internal API or a proprietary data source—you can build your own toolkit by subclassing BaseToolkit:

from crewai_tools import BaseTool, BaseToolkit, tool
from pydantic import BaseModel, Field
from typing import Type, List

# Define input schemas
class InvoiceLookupInput(BaseModel):
    invoice_id: str = Field(description="The invoice number to look up")

class CreateInvoiceInput(BaseModel):
    customer_name: str = Field(description="Full name of the customer")
    amount: float = Field(description="Invoice amount in USD")
    description: str = Field(description="Line item description")

# Custom tools for an internal billing system
class LookupInvoiceTool(BaseTool):
    name: str = "Invoice Lookup"
    description: str = "Retrieves details of an invoice by its ID from the billing system."
    args_schema: Type[BaseModel] = InvoiceLookupInput

    def _run(self, invoice_id: str) -> str:
        # Simulate a database lookup
        mock_db = {
            "INV-001": "Customer: Acme Corp, Amount: $1,250.00, Status: Paid",
            "INV-002": "Customer: Beta LLC, Amount: $890.00, Status: Pending"
        }
        return mock_db.get(invoice_id, f"Invoice {invoice_id} not found.")

class CreateInvoiceTool(BaseTool):
    name: str = "Create Invoice"
    description: str = "Creates a new invoice in the billing system for a customer."
    args_schema: Type[BaseModel] = CreateInvoiceInput

    def _run(self, customer_name: str, amount: float, description: str) -> str:
        # In production, this would call your billing API
        new_id = f"INV-{hash(customer_name + str(amount)) % 10000:04d}"
        return f"Invoice created successfully. ID: {new_id}, Customer: {customer_name}, Amount: ${amount:.2f}"

# Bundle them into a custom toolkit
class BillingToolkit(BaseToolkit):
    def get_tools(self) -> List[BaseTool]:
        return [LookupInvoiceTool(), CreateInvoiceTool()]

# Use the custom toolkit
billing_toolkit = BillingToolkit()

billing_agent = Agent(
    role="Billing Specialist",
    goal="Handle customer invoice inquiries and create new invoices accurately.",
    backstory="You are a billing department expert with access to the internal billing system.",
    tools=billing_toolkit.get_tools(),
    verbose=True
)

The BaseToolkit interface requires only that you implement get_tools() returning a list of BaseTool instances. This pattern keeps your tool collections organized, reusable, and testable.

Best Practices for Tools and Toolkits

Over time, the community and core maintainers have identified several practices that lead to more reliable, maintainable, and secure tool usage:

Advanced: Tool Call Observation and Error Recovery

When an agent invokes a tool, CrewAI captures the interaction in the agent's execution context. You can observe these calls programmatically by enabling verbose mode or by hooking into the callback system. Here is an example of wrapping a tool with logging and error recovery:

import logging
import time
from crewai_tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class CurrencyConversionInput(BaseModel):
    amount: float = Field(description="Amount to convert")
    from_currency: str = Field(description="ISO 4217 currency code, e.g., USD")
    to_currency: str = Field(description="ISO 4217 currency code, e.g., EUR")

class RobustCurrencyTool(BaseTool):
    name: str = "Currency Converter"
    description: str = (
        "Converts an amount from one currency to another using real-time exchange rates. "
        "Use this tool when you need accurate currency conversion. "
        "Provide the amount, source currency code (3 letters), and target currency code (3 letters)."
    )
    args_schema: Type[BaseModel] = CurrencyConversionInput
    max_retries: int = 3
    base_delay: float = 1.0

    def _run(self, amount: float, from_currency: str, to_currency: str) -> str:
        logger.info(f"Tool called: convert {amount} {from_currency} → {to_currency}")

        # Input validation
        if amount < 0:
            return "Error: Amount must be non-negative. Please provide a positive number."
        if len(from_currency) != 3 or len(to_currency) != 3:
            return "Error: Currency codes must be exactly 3 letters (ISO 4217 format)."

        for attempt in range(1, self.max_retries + 1):
            try:
                # Simulate an API call that might fail transiently
                result = self._fetch_exchange_rate(amount, from_currency, to_currency)
                logger.info(f"Tool succeeded on attempt {attempt}: {result}")
                return result
            except Exception as e:
                logger.warning(f"Attempt {attempt} failed: {e}")
                if attempt < self.max_retries:
                    sleep_time = self.base_delay * (2 ** (attempt - 1))
                    time.sleep(sleep_time)
                else:
                    return (
                        f"Error: Failed to convert after {self.max_retries} attempts. "
                        f"Last error: {str(e)}. Please try again later or verify the currency codes."
                    )

    def _fetch_exchange_rate(self, amount: float, from_cur: str, to_cur: str) -> str:
        # Mock exchange rates — replace with an actual API like exchangerate-api.com
        rates = {"USD": 1.0, "EUR": 0.92, "GBP": 0.79, "JPY": 149.5, "CAD": 1.36}
        if from_cur not in rates or to_cur not in rates:
            raise ValueError(f"Unsupported currency: {from_cur} or {to_cur}")
        
        # Simulate occasional transient failure
        import random
        if random.random() < 0.2:
            raise ConnectionError("Temporary network issue.")
        
        converted = amount * (rates[to_cur] / rates[from_cur])
        return f"{amount:.2f} {from_cur} = {converted:.2f} {to_cur} (rate: {rates[to_cur]/rates[from_cur]:.4f})"

This robust tool design demonstrates several best practices in action: input validation, retry logic with exponential backoff, detailed logging, and error messages that guide the agent toward corrective action. When the agent receives an error string, it can interpret the message and either retry with corrected arguments or inform the user appropriately.

Tool Result Context Management

One common pitfall is overflowing the LLM context window with excessively large tool results. If a web scraping tool returns 50KB of HTML, the agent may lose the ability to reason effectively. Mitigate this by:

Here is a pattern for a search tool that manages result size:

class SearchResultInput(BaseModel):
    query: str = Field(description="Search query string")
    max_results: int = Field(default=5, description="Maximum number of results to return, between 1 and 20")

class BoundedSearchTool(BaseTool):
    name: str = "Web Search"
    description: str = (
        "Searches the web and returns a condensed list of results. "
        "Use max_results to control output size. Defaults to 5 results."
    )
    args_schema: Type[BaseModel] = SearchResultInput

    def _run(self, query: str, max_results: int = 5) -> str:
        # Clamp max_results to prevent context overflow
        max_results = max(1, min(max_results, 20))
        
        # Simulated search results
        all_results = [
            {"title": "Result A", "snippet": "This is the first result about " + query},
            {"title": "Result B", "snippet": "Second relevant finding for " + query},
            {"title": "Result C", "snippet": "Additional information regarding " + query},
            # ... many more results
        ]
        
        # Truncate and format
        selected = all_results[:max_results]
        formatted = "\n".join(
            f"{i+1}. {r['title']}: {r['snippet'][:200]}" 
            for i, r in enumerate(selected)
        )
        
        if len(all_results) > max_results:
            formatted += f"\n\n(Showing {max_results} of {len(all_results)} total results. "
            formatted += "Increase max_results to see more.)"
        
        return formatted

Security Considerations for Tools

Because tools grant agents the ability to perform real-world actions, security must be a first-class concern. Consider these guidelines:

Conclusion

Tools and toolkits are the mechanism through which CrewAI agents transcend the boundaries of language models and become capable actors in the digital world. By carefully designing tools with clear descriptions, validated input schemas, robust error handling, and appropriate scoping to agent roles, you can build crews that perform complex, multi-step workflows reliably and securely. Start with built-in toolkits to accelerate development, then graduate to custom tools and toolkits as your application's needs grow. Remember that every tool you grant an agent is both a capability and a responsibility—design them thoughtfully, test them thoroughly, and always keep the agent's context window and security posture in mind. With these practices in place, your CrewAI agents will be equipped to tackle increasingly ambitious real-world tasks.

— Ad —

Google AdSense will appear here after approval

← Back to all articles