Intermediate Crewai 2 min read

CrewAI Custom Tools: Connect Agents to Any API or Service

#crewai #tools #integrations #api #python #langchain
📚

Read these first:

Tools Are What Make Agents Useful

A CrewAI agent without tools can only generate text. With tools, agents can search the web, query databases, call APIs, read files, and interact with any external system. Tools are the bridge between your AI agents and the real world.

CrewAI integrates with LangChain’s tool ecosystem and adds its own crewai-tools package with ready-to-use integrations. This guide covers building custom tools and using built-in ones.

The @tool Decorator: Fastest Way to Create a Tool

The simplest way to give an agent a capability is the @tool decorator from LangChain:

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

@tool
def search_database(query: str) -> str:
    """Search the product database for items matching the query.
    Input should be a search term like 'laptop' or 'wireless headphones'."""
    # Simulate a database lookup
    products = {
        "laptop": "Dell XPS 15 — $1,299 — In stock",
        "headphones": "Sony WH-1000XM5 — $349 — In stock",
        "monitor": "LG 27UK850-W — $449 — Low stock",
    }
    results = [v for k, v in products.items() if query.lower() in k]
    return "\n".join(results) if results else "No products found."

@tool
def check_inventory(product_id: str) -> str:
    """Check the current inventory level for a product by its ID.
    Input should be a product ID like 'DELL-XPS-15'."""
    inventory = {"DELL-XPS-15": 24, "SONY-WH1000XM5": 8, "LG-27UK850": 2}
    count = inventory.get(product_id, 0)
    status = "In stock" if count > 5 else ("Low stock" if count > 0 else "Out of stock")
    return f"Product {product_id}: {count} units — {status}"

# Assign tools to an agent
researcher = Agent(
    role="Product Researcher",
    goal="Find accurate product information and inventory data.",
    backstory="Expert at navigating product databases and providing accurate information.",
    tools=[search_database, check_inventory],
    verbose=True,
)

The docstring is critical — CrewAI agents read it to decide when and how to use the tool.

Built-in crewai-tools

The crewai-tools package provides ready-to-use tools:

pip install crewai-tools
from crewai_tools import (
    SerperDevTool,        # Google search via Serper API
    WebsiteSearchTool,    # RAG over a website
    FileReadTool,         # Read files from disk
    DirectoryReadTool,    # List directory contents
    PDFSearchTool,        # RAG over PDFs
    CSVSearchTool,        # Search CSV files
    YoutubeVideoSearchTool, # Search YouTube videos
    GithubSearchTool,     # Search GitHub repos
)

# Web search (requires SERPER_API_KEY env var)
web_search = SerperDevTool()

# Read any file
file_reader = FileReadTool()

# Search inside a website
website_rag = WebsiteSearchTool(website="https://docs.crewai.com")

agent = Agent(
    role="Research Analyst",
    goal="Research topics using the web and files.",
    backstory="Expert at finding information from multiple sources.",
    tools=[web_search, file_reader, website_rag],
)

Custom Tool with Pydantic Validation

For tools with multiple parameters, use a Pydantic schema:

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

class WeatherInput(BaseModel):
    city: str = Field(description="City name, e.g. 'Seoul' or 'New York'")
    unit: str = Field(default="celsius", description="Temperature unit: 'celsius' or 'fahrenheit'")

class WeatherTool(BaseTool):
    name: str = "get_weather"
    description: str = (
        "Get current weather for a city. "
        "Use this when the task requires current weather information."
    )
    args_schema: Type[BaseModel] = WeatherInput

    def _run(self, city: str, unit: str = "celsius") -> str:
        # In production, call a real weather API here
        temp = 22 if unit == "celsius" else 72
        return f"Weather in {city}: {temp}°{'C' if unit == 'celsius' else 'F'}, partly cloudy, humidity 65%"

weather_tool = WeatherTool()

BaseTool gives you type safety, automatic docstring generation from name/description, and compatibility with CrewAI’s tool validation.

Database Query Tool

A real-world example — a tool that queries SQLite:

from crewai.tools import BaseTool
from pydantic import BaseModel, Field
import sqlite3
from pathlib import Path

class SQLQueryInput(BaseModel):
    query: str = Field(description="A read-only SQL SELECT query")

class DatabaseQueryTool(BaseTool):
    name: str = "query_database"
    description: str = (
        "Execute a read-only SQL query on the customer database. "
        "Use for questions about customers, orders, or products. "
        "Only SELECT queries are allowed."
    )
    args_schema: type = SQLQueryInput
    db_path: str = "data/customers.db"

    def _run(self, query: str) -> str:
        # Safety: only allow SELECT
        if not query.strip().upper().startswith("SELECT"):
            return "Error: only SELECT queries are allowed"
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.execute(query)
            rows = cursor.fetchmany(20)  # limit results
            cols = [d[0] for d in cursor.description]
            conn.close()
            if not rows:
                return "No results found."
            result = [", ".join(cols)]
            result += [", ".join(str(v) for v in row) for row in rows]
            return "\n".join(result)
        except Exception as e:
            return f"Query error: {e}"

db_tool = DatabaseQueryTool(db_path="data/customers.db")

HTTP API Tool

For calling external REST APIs:

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

class APICallInput(BaseModel):
    endpoint: str = Field(description="API endpoint path, e.g. '/users/123'")
    method: str = Field(default="GET", description="HTTP method: GET or POST")

class InternalAPITool(BaseTool):
    name: str = "call_internal_api"
    description: str = "Call the internal company API to get business data."
    args_schema: type = APICallInput
    base_url: str = "https://api.internal.example.com"
    api_key: str = ""

    def _run(self, endpoint: str, method: str = "GET") -> str:
        url = f"{self.base_url}{endpoint}"
        headers = {"Authorization": f"Bearer {self.api_key}"}
        try:
            if method == "GET":
                resp = requests.get(url, headers=headers, timeout=10)
            else:
                return "Only GET is supported"
            resp.raise_for_status()
            return resp.text[:2000]  # limit response size
        except Exception as e:
            return f"API error: {e}"

Caching Tool Results

For expensive tools (web searches, database queries), cache results to avoid repeated calls:

from crewai.tools import BaseTool
import hashlib
import json

class CachedSearchTool(BaseTool):
    name: str = "cached_web_search"
    description: str = "Search the web with caching to avoid duplicate requests."
    _cache: dict = {}

    def _run(self, query: str) -> str:
        cache_key = hashlib.md5(query.encode()).hexdigest()
        if cache_key in self._cache:
            return f"[cached] {self._cache[cache_key]}"

        # Perform actual search
        result = self._do_search(query)
        self._cache[cache_key] = result
        return result

    def _do_search(self, query: str) -> str:
        # Real search implementation here
        return f"Search results for '{query}': ..."

Full Example: Research Crew with Custom Tools

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool
from langchain.tools import tool

@tool
def save_report(content: str) -> str:
    """Save a report to a markdown file. Input is the full report text."""
    from pathlib import Path
    Path("report.md").write_text(content)
    return "Report saved to report.md"

web_search = SerperDevTool()

researcher = Agent(
    role="Market Researcher",
    goal="Research AI agent frameworks thoroughly.",
    backstory="Expert researcher with access to the web.",
    tools=[web_search],
)

writer = Agent(
    role="Technical Writer",
    goal="Write comprehensive reports from research.",
    backstory="Technical writer who produces clear, structured reports.",
    tools=[save_report],
)

research_task = Task(
    description="Research the top 5 AI agent frameworks in 2025. Find key features, GitHub stars, and use cases.",
    expected_output="A structured research summary with framework comparison.",
    agent=researcher,
)

report_task = Task(
    description="Write a 500-word comparison report and save it using the save_report tool.",
    expected_output="Report saved to report.md",
    agent=writer,
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, report_task],
    process=Process.sequential,
)

result = crew.kickoff()
print(result.raw)

Frequently Asked Questions

Can I use the same tool in multiple agents?

Yes — tool instances are stateless (or you can make them stateful with care). Pass the same tool object to multiple agents. If the tool has state (like the caching example), be aware that agents share that state.

What’s the difference between @tool and BaseTool?

@tool is a quick decorator for simple functions with one string parameter. BaseTool is a class-based approach for complex tools with multiple typed parameters, custom validation, and initialization logic. Use @tool for quick tools, BaseTool for production tools.

How do I handle tool errors gracefully?

Return an error string instead of raising an exception. The agent reads the error as tool output and can adapt its approach:

def _run(self, query: str) -> str:
    try:
        return self._execute(query)
    except Exception as e:
        return f"Tool error: {e}. Try a different approach."

Can CrewAI tools call other AI models?

Yes. A tool can call OpenAI, Anthropic, or any API internally. This enables patterns like: a summarizer tool that calls a cheap model (gpt-4o-mini) to compress long text before the main agent processes it.

Next Steps

Related Articles

langchain LangChain Agents and Tools: Build Agents That Take Action crewai CrewAI Flows: Event-Driven Pipelines for Complex AI Workflows crewai CrewAI Memory and Knowledge: Agents That Learn and Remember