Become an Agentic Architect

Building Custom Tools in CrewAI

In CrewAI, custom tools let your agents call your own Python logic or APIs as first‑class capabilities. You can build them in two main ways: by subclassing BaseTool or by using the @tool decorator, with optional caching and async support.


1. When and Why to Build Custom Tools

Use custom tools when agents need to:

  • Call internal APIs (CRM, ticketing, billing, logs).
  • Run business logic (pricing, routing, scoring, validation).
  • Do specialized I/O (custom DBs, files, internal services).

You define the tool once, then attach it to any agent via tools=[my_tool, ...].


2. Pattern 1 – Subclassing BaseTool

Best when you want strong typing, validation, and full control.

2.1 Basic sync tool with input schema

# python
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"

Key ideas:

  • name + description are what the agent sees when deciding to use the tool.
  • args_schema (Pydantic) validates inputs and documents parameters.
  • _run(...) contains your sync business logic.

You then instantiate and pass it to an agent:

my_tool = MyCustomTool()
agent = Agent(..., tools=[my_tool])

3. Pattern 2 – @tool Decorator (Quick & Concise)

Great for simple tools and rapid prototyping.

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"
  • The decorator wraps your function as a Tool object.
  • The docstring explains when the agent should call it.
  • You attach my_simple_tool directly in tools=[...] on an agent.

4. Adding Caching to Tools

CrewAI lets tools decide when to cache their results via cache_function.

from crewai.tools import tool

@tool("Tool with Caching")
def cached_tool(argument: str) -> str:
    """Tool functionality description."""
    # Expensive computation or API call
    return "Cacheable result"

def my_cache_strategy(arguments: dict, result: str) -> bool:
    # Custom rule: return True to cache, False to skip
    return True  # or some_condition_based_on(arguments, result)

cached_tool.cache_function = my_cache_strategy

Use this to avoid repeated expensive calls (e.g., slow APIs, large queries).


5. Building Async Tools

Async tools prevent blocking when you do HTTP requests, DB calls, or other I/O.

5.1 Async function with @tool

Simplest async option:

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()

The framework will await this tool in async workflows.

5.2 Async BaseTool subclass (_run + _arun)

Use this when you want both sync and async entry points.

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()
  • _run is used in regular (sync) crews.
  • _arun is used in async flows / async crew execution.

6. Your approach as  “Agentic Architect”

Recap:

  1. Start with a decorator tool for a toy task (e.g., calculator, date formatter).
  2. Introduce BaseTool + Pydantic schema for more serious, typed tools (e.g., CRM fetch).
  3. Add caching for expensive operations and let them tune cache_function.
  4. Finish with an async web/API tool to show non‑blocking patterns.

The core mindset: you’re not just prompting; you are designing a toolbelt that your agents can intelligently pick from as they work.