CrewAI Custom Tools: Connect Agents to External APIs

Build CrewAI custom tools that call external APIs. Add auth, error handling, and schema validation to give your agents real-world data access.

Problem: CrewAI Agents Are Blind Without Custom Tools

CrewAI ships with built-in tools for web search and file I/O. But the moment you need live stock prices, your internal database, a weather API, or any proprietary endpoint — you're on your own. Without a custom tool, agents hallucinate data or flatly fail.

You'll learn:

  • How to build a typed CrewAI tool that calls any REST API
  • How to handle auth headers, rate limits, and API errors cleanly
  • How to wire the tool into an agent and verify it runs end-to-end

Time: 20 min | Difficulty: Intermediate


Why Custom Tools Work the Way They Do

CrewAI tools are Python classes that inherit from BaseTool. The agent decides when to call a tool based on its name and description — these strings are injected directly into the LLM prompt. Poor descriptions mean the agent ignores the tool or misuses it.

Each tool exposes a single _run() method. CrewAI handles the invocation, argument parsing, and result passing back to the agent automatically.

Key rule: the tool's input schema (via Pydantic) controls what arguments the LLM can pass in. Define it tightly and you prevent malformed API calls before they happen.


Solution

Step 1: Install Dependencies

# CrewAI 0.80+ required for typed tool schemas
pip install crewai crewai-tools pydantic requests python-dotenv

Verify:

python -c "import crewai; print(crewai.__version__)"

Expected: 0.80.x or higher.


Step 2: Create the Tool File

Create tools/weather_tool.py. We'll use the Open-Meteo API (free, no auth) as a working example, then show how to add auth.

# tools/weather_tool.py
import requests
from pydantic import BaseModel, Field
from crewai.tools import BaseTool


class WeatherInput(BaseModel):
    """Input schema — the LLM must match this exactly when calling the tool."""
    city: str = Field(description="City name, e.g. 'Berlin' or 'Tokyo'")
    days: int = Field(
        default=1,
        ge=1,
        le=7,
        description="Forecast days (1–7). Default is 1 for today only."
    )


class WeatherTool(BaseTool):
    name: str = "get_weather_forecast"
    description: str = (
        "Fetches a real-time weather forecast for a given city. "
        "Use this whenever the task requires current or upcoming weather data. "
        "Returns temperature, precipitation, and wind speed."
    )
    args_schema: type[BaseModel] = WeatherInput

    def _run(self, city: str, days: int = 1) -> str:
        # Step 1: resolve city name to lat/lon via geocoding API
        geo_url = "https://geocoding-api.open-meteo.com/v1/search"
        geo_resp = requests.get(geo_url, params={"name": city, "count": 1}, timeout=5)
        geo_resp.raise_for_status()
        results = geo_resp.json().get("results")

        if not results:
            return f"City '{city}' not found. Try a different spelling."

        lat = results[0]["latitude"]
        lon = results[0]["longitude"]

        # Step 2: fetch forecast
        forecast_url = "https://api.open-meteo.com/v1/forecast"
        params = {
            "latitude": lat,
            "longitude": lon,
            "daily": "temperature_2m_max,precipitation_sum,windspeed_10m_max",
            "forecast_days": days,
            "timezone": "auto",
        }
        forecast_resp = requests.get(forecast_url, params=params, timeout=5)
        forecast_resp.raise_for_status()
        data = forecast_resp.json()["daily"]

        # Step 3: format output for the LLM to parse easily
        lines = [f"Weather forecast for {city} ({days} day(s)):"]
        for i, date in enumerate(data["time"]):
            lines.append(
                f"  {date}: max {data['temperature_2m_max'][i]}°C, "
                f"rain {data['precipitation_sum'][i]}mm, "
                f"wind {data['windspeed_10m_max'][i]}km/h"
            )
        return "\n".join(lines)

Why the description matters: The LLM reads name + description verbatim when deciding whether to call this tool. Be explicit about when to use it and what it returns.


Step 3: Add Auth for Private APIs

Most production APIs require a key. Here's the pattern using environment variables:

# tools/github_tool.py
import os
import requests
from pydantic import BaseModel, Field
from crewai.tools import BaseTool


class GitHubInput(BaseModel):
    repo: str = Field(description="GitHub repo in 'owner/repo' format, e.g. 'openai/openai-python'")


class GitHubStarsTool(BaseTool):
    name: str = "get_github_repo_stats"
    description: str = (
        "Returns star count, open issues, and last push date for a GitHub repository. "
        "Use when the task involves evaluating a GitHub project's popularity or activity."
    )
    args_schema: type[BaseModel] = GitHubInput

    def _run(self, repo: str) -> str:
        token = os.getenv("GITHUB_TOKEN")
        headers = {"Accept": "application/vnd.github+json"}

        # Auth is optional for public repos but prevents rate-limiting (60 → 5000 req/hr)
        if token:
            headers["Authorization"] = f"Bearer {token}"

        url = f"https://api.github.com/repos/{repo}"
        resp = requests.get(url, headers=headers, timeout=10)

        if resp.status_code == 404:
            return f"Repository '{repo}' not found."
        if resp.status_code == 403:
            return "GitHub rate limit hit. Set GITHUB_TOKEN in your environment."

        resp.raise_for_status()
        d = resp.json()

        return (
            f"{repo}: {d['stargazers_count']} stars, "
            f"{d['open_issues_count']} open issues, "
            f"last pushed {d['pushed_at'][:10]}"
        )

Store your token in .env:

# .env
GITHUB_TOKEN=ghp_your_token_here

Step 4: Handle Errors Gracefully

Agents break badly when a tool raises an unhandled exception — the whole crew halts. Return error strings instead of raising:

def _run(self, city: str, days: int = 1) -> str:
    try:
        # ... API calls ...
    except requests.exceptions.Timeout:
        # Tell the agent what happened so it can decide what to do next
        return f"Weather API timed out for '{city}'. Try again or use a fallback source."
    except requests.exceptions.HTTPError as e:
        return f"Weather API error {e.response.status_code}: {e.response.text[:200]}"
    except Exception as e:
        return f"Unexpected error fetching weather: {str(e)}"

The agent receives the error string, can report it to the user, or retry with different arguments.


Step 5: Wire Tools Into an Agent and Crew

# main.py
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process
from tools.weather_tool import WeatherTool
from tools.github_tool import GitHubStarsTool

load_dotenv()

weather_tool = WeatherTool()
github_tool = GitHubStarsTool()

researcher = Agent(
    role="Research Analyst",
    goal="Gather accurate real-world data to answer user questions",
    backstory="You retrieve live data using tools rather than relying on memory.",
    tools=[weather_tool, github_tool],
    verbose=True,
    # gpt-4o is recommended for reliable tool selection; 3.5-turbo misses calls
    llm="gpt-4o",
)

task = Task(
    description=(
        "Find the current weather in Tokyo for the next 3 days. "
        "Also check the star count for 'langchain-ai/langchain' on GitHub."
    ),
    expected_output="A brief report with Tokyo's 3-day forecast and LangChain's GitHub stats.",
    agent=researcher,
)

crew = Crew(
    agents=[researcher],
    tasks=[task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff()
print(result)

Verification

Run the crew and look for tool invocations in the verbose output:

python main.py

You should see:

[Agent: Research Analyst] Using tool: get_weather_forecast
[Agent: Research Analyst] Tool result: Weather forecast for Tokyo (3 day(s)):
  2026-03-09: max 14°C, rain 0.0mm, wind 18km/h
  ...
[Agent: Research Analyst] Using tool: get_github_repo_stats
[Agent: Research Analyst] Tool result: langchain-ai/langchain: 94200 stars, ...

If a tool is never called, the agent is ignoring it — usually because the description doesn't match the task wording. Make the description more specific about when to trigger it.

Debug tool selection:

# Print which tools the agent sees
for tool in researcher.tools:
    print(f"{tool.name}: {tool.description[:80]}")

What You Learned

  • BaseTool + Pydantic schema is the correct pattern — avoid the @tool decorator for anything beyond trivial use cases, as it offers no type safety
  • name and description are prompt engineering — vague descriptions cause tool misuse
  • Return error strings, never raise exceptions inside _run(), so the agent can recover
  • GITHUB_TOKEN bumps rate limits from 60 to 5,000 req/hr — always set it in production

When NOT to use custom tools: if the data changes less than once a day (e.g., documentation, model specs), inject it into the agent's backstory instead. Tool calls cost latency and tokens every run.

Tested on CrewAI 0.80.4, Python 3.12, gpt-4o-2024-11-20, Ubuntu 24.04 and macOS 15