Create Custom Tools

Comprehensive guide on crafting, using, and managing custom tools within the CrewAI framework, including new functionalities and error handling.

Creating and Utilizing Tools in CrewAI

This guide provides detailed instructions on creating custom tools for the CrewAI framework and how to efficiently manage and utilize these tools, incorporating the latest functionalities such as tool delegation, error handling, and dynamic tool calling. It also highlights the importance of collaboration tools, enabling agents to perform a wide range of actions.

Want to publish your tool for the community? If you’re building a tool that others could benefit from, check out the Publish Custom Tools guide to learn how to package and distribute your tool on PyPI.

Subclassing BaseTool

To create a personalized tool, inherit from BaseTool and define the necessary attributes, including the args_schema for input validation, and the _run method.

Code

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

class MyToolInput(BaseModel):
    """Input schema for MyCustomTool."""
    argument: str = Field(..., description="Description of the argument.")

class MyCustomTool(BaseTool):
    name: str = "Name of my tool"
    description: str = "What this tool does. It's vital for effective utilization."
    args_schema: Type[BaseModel] = MyToolInput

    def _run(self, argument: str) -> str:
        # Your tool's logic here
        return "Tool's result"

Using the tool Decorator

Alternatively, you can use the tool decorator @tool. This approach allows you to define the tool’s attributes and functionality directly within a function, offering a concise and efficient way to create specialized tools tailored to your needs.

Code

from crewai.tools import tool

@tool("Tool Name")
def my_simple_tool(question: str) -> str:
    """Tool description for clarity."""
    # Tool logic here
    return "Tool output"

Best Practice: Define Typed Outputs

When a tool returns structured data, define a Pydantic output model. This helps the agent read the result as clear fields instead of guessing from plain text.

Typed outputs are useful for results with stable fields, such as IDs, status values, scores, prices, or lists. Plain strings are still fine for short prose results.

Direct Python calls still receive the value your tool returns. When an agent uses a typed tool, CrewAI sends the agent JSON based on the output model.

Return a Pydantic Model

CrewAI infers the output schema when your BaseTool has a Pydantic return annotation.

Code

from crewai.tools import BaseTool
from pydantic import BaseModel, Field

class InventoryResult(BaseModel):
    sku: str = Field(description="The product SKU.")
    quantity: int = Field(description="Units available.")
    needs_reorder: bool = Field(description="Whether the item should be reordered.")

class InventoryTool(BaseTool):
    name: str = "Inventory Check"
    description: str = "Check current stock for a product SKU."

    def _run(self, sku: str) -> InventoryResult:
        quantity = {"SKU-123": 14, "SKU-456": 0}.get(sku, 0)
        return InventoryResult(sku=sku, quantity=quantity, needs_reorder=quantity < 5)

tool = InventoryTool()
result = tool.run(sku="SKU-123")

# Direct Python calls receive the raw Pydantic object.
print(result.quantity)

When an agent calls InventoryTool, it receives JSON like this:

{"sku":"SKU-123","quantity":14,"needs_reorder":false}

Use result_schema with Dictionary Results

If your tool returns a dictionary, set result_schema explicitly. You can do this on a BaseTool subclass or with the @tool decorator:

Code

from crewai.tools import tool
from pydantic import BaseModel, Field

class ProductResult(BaseModel):
    sku: str = Field(description="The product SKU.")
    name: str = Field(description="The product name.")
    in_stock: bool = Field(description="Whether the product is available.")

@tool("Product Lookup", result_schema=ProductResult)
def product_lookup(sku: str) -> dict[str, object]:
    """Look up product availability by SKU."""
    catalog = {
        "SKU-123": ("Noise-canceling headset", True),
        "SKU-456": ("USB-C dock", False),
    }
    name, in_stock = catalog.get(sku, ("Unknown product", False))
    return {
        "sku": sku,
        "name": name,
        "in_stock": in_stock,
    }

Customize the Text Sent to the Agent

By default, typed tool outputs are sent to the agent as JSON. If the agent should receive a short summary instead, subclass BaseTool and override format_output_for_agent.

Code

from crewai.tools import BaseTool
from pydantic import BaseModel, Field

class InventoryResult(BaseModel):
    sku: str = Field(description="The product SKU.")
    quantity: int = Field(description="Units available.")
    needs_reorder: bool = Field(description="Whether the item should be reordered.")

class InventoryTool(BaseTool):
    name: str = "Inventory Check"
    description: str = "Check current stock for a product SKU."

    def _run(self, sku: str) -> InventoryResult:
        quantity = {"SKU-123": 14, "SKU-456": 0}.get(sku, 0)
        return InventoryResult(sku=sku, quantity=quantity, needs_reorder=quantity < 5)

    def format_output_for_agent(self, raw_result: object) -> str:
        result = InventoryResult.model_validate(raw_result)
        status = "reorder needed" if result.needs_reorder else "stock is healthy"
        return f"{result.sku}: {result.quantity} units. {status}."

tool = InventoryTool()
result = tool.run(sku="SKU-123")

# Direct Python calls receive the raw Pydantic object.
print(result.quantity)

The override only changes what the agent sees. Direct calls to tool.run(...) still return the normal Python value.

Defining a Cache Function for the Tool

To optimize tool performance with caching, define custom caching strategies using the cache_function attribute.

Code

@tool("Tool with Caching")
def cached_tool(argument: str) -> str:
    """Tool functionality description."""
    return "Cacheable result"

def my_cache_strategy(arguments: dict, result: str) -> bool:
    # Define custom caching logic
    return True if some_condition else False

cached_tool.cache_function = my_cache_strategy

Creating Async Tools

CrewAI supports async tools for non-blocking I/O operations. This is useful when your tool needs to make HTTP requests, database queries, or other I/O-bound operations.

Using the @tool Decorator with Async Functions

The simplest way to create an async tool is using the @tool decorator with an async function:

Code

import aiohttp
from crewai.tools import tool

@tool("Async Web Fetcher")
async def fetch_webpage(url: str) -> str:
    """Fetch content from a webpage asynchronously."""
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

Subclassing BaseTool with Async Support

For more control, subclass BaseTool and implement both _run (sync) and _arun (async) methods:

Code

import requests
import aiohttp
from crewai.tools import BaseTool
from pydantic import BaseModel, Field

class WebFetcherInput(BaseModel):
    """Input schema for WebFetcher."""
    url: str = Field(..., description="The URL to fetch")

class WebFetcherTool(BaseTool):
    name: str = "Web Fetcher"
    description: str = "Fetches content from a URL"
    args_schema: type[BaseModel] = WebFetcherInput

    def _run(self, url: str) -> str:
        """Synchronous implementation."""
        return requests.get(url).text

    async def _arun(self, url: str) -> str:
        """Asynchronous implementation for non-blocking I/O."""
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                return await response.text()

By adhering to these guidelines and incorporating new functionalities and collaboration tools into your tool creation and management processes, you can leverage the full capabilities of the CrewAI framework, enhancing both the development experience and the efficiency of your AI agents.