Chapter 8 · Custom Tool Development

Chapter goals: Master two custom tool writing patterns — BaseTool subclass and @tool decorator — use Pydantic for parameter validation, control caching with cache_function, and safely wrap internal REST APIs as agent-accessible tools.

8.1 Two Custom Tool Writing Patterns

CrewAI offers two paths, both documented on the Tools concept page:

• Inherit BaseTool: suits complex tools — need state, multiple methods, typed output;
• @tool decorator: suits "one function = one thing" simple tools with minimal code.

from crewai.tools import BaseTool, tool


# Pattern 1: @tool decorator (recommended for getting started)
# First arg is the tool name; the function docstring becomes the tool description — the model uses it to decide when to call
@tool("exchange_rate_converter")
def exchange_rate(amount: float, currency: str) -> str:
    """Convert an amount to CNY at a fixed demo exchange rate. Use when currency conversion is needed."""
    rates = {"USD": 7.2, "EUR": 7.8, "JPY": 0.048}
    if currency not in rates:
        return f"Error: currency {currency} is not supported. Supported: {list(rates)}"
    return f"{amount} {currency} ≈ {amount * rates[currency]:.2f} CNY"


# Pattern 2: inherit BaseTool (when you need state or finer control)
from typing import Type

from pydantic import BaseModel, Field


class WeatherInput(BaseModel):
    """Input schema for the weather query tool."""
    city: str = Field(..., description="City name, e.g. 'Beijing'")
    days: int = Field(3, ge=1, le=7, description="Forecast days, 1-7")


class WeatherTool(BaseTool):
    name: str = "weather_forecast"
    description: str = "Query the weather forecast for a city over the next few days. Must be called before answering weather-related questions."
    args_schema: Type[BaseModel] = WeatherInput   # Declare and validate params with a Pydantic model

    def _run(self, city: str, days: int = 3) -> str:
        # In production, call a real weather API here; demo returns fixed data
        return f"{city} next {days} days: sunny turning cloudy, 18-26°C."


weather_tool = WeatherTool()

description determines call quality

The model relies entirely on description and each parameter's Field(description=...) to judge "when to use this tool and what params to pass." A vaguely described tool = a tool that won't be called correctly. Descriptions should include: functionality + when to use it + key constraints.

8.2 Parameter Validation and Failure Design

Tool parameters are validated by Pydantic before entering _run. Two more key conventions:

1. Return value must be a string (or a typed output model): this is the text fed to the model;
2. Recoverable errors return an error explanation, not raise exceptions: giving the agent a chance to self-correct.

from typing import Type

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


class OrderQueryInput(BaseModel):
    order_id: str = Field(..., description="Order ID, format: ORD-2024-0001")

    @field_validator("order_id")
    @classmethod
    def check_format(cls, v: str) -> str:
        if not v.startswith("ORD-"):
            raise ValueError("Order ID must start with ORD-")
        return v


class OrderQueryTool(BaseTool):
    name: str = "query_order"
    description: str = "Query order status by order ID. Only use when the user provides a valid order ID."
    args_schema: Type[BaseModel] = OrderQueryInput

    def _run(self, order_id: str) -> str:
        db = {"ORD-2024-0001": "Shipped"}
        status = db.get(order_id)
        if status is None:
            # Return an actionable hint rather than raising — the agent can ask the user to re-verify
            return f"Order {order_id} not found. Please verify the order ID with the user and retry."
        return f"Order {order_id} current status: {status}"

The newer CrewAI also provides ToolFailure (imported from crewai.tools.tool_failure): when a tool "didn't raise an exception but truly failed", returning ToolFailure(message=..., retryable=...) lets the framework explicitly record the failure, working with the Crew's tool_failure_policy (warn/raise) for unified governance — far more reliable than stuffing "error" into a string and letting the model guess.

8.3 Caching: cache_function

All tools have result caching enabled by default (identical params reuse the previous result), which is critical for expensive external calls. Use cache_function to finely control "which results are worth caching":

from crewai.tools import tool


@tool("real_time_gold_price")
def gold_price(city: str = "shanghai") -> str:
    """Query the current gold price. Do not rely on caching in price-sensitive scenarios."""
    return "768.50 CNY/gram"   # Demo value


def cache_func(args, result) -> bool:
    # Return True means "this result is allowed to be cached"
    # Example: do not cache query failures to avoid reusing wrong answers
    return "error" not in result.lower()


gold_price.cache_function = cache_func

Tools that don't need caching: set gold_price.cache_function = lambda args, result: False to force re-execution every time.

8.4 Practice: Wrapping an Internal REST API as a Tool

The most common enterprise requirement: let the agent safely query internal systems. The core principles are whitelist operations + parameter收敛 + no implementation detail leaks:

import os
from typing import Type

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


class CrmCustomerInput(BaseModel):
    """CRM customer query input."""
    customer_name: str = Field(..., min_length=2, description="Customer full name or unique ID")


class CrmLookupTool(BaseTool):
    name: str = "crm_customer_lookup"
    description: str = (
        "Query a customer's tier, assigned sales rep, and last order time from the company CRM. "
        "Exact name/ID queries only — no fuzzy search; one customer per call."
    )
    args_schema: Type[BaseModel] = CrmCustomerInput

    def _run(self, customer_name: str) -> str:
        try:
            resp = requests.get(
                "https://crm.internal.example.com/api/v1/customers",
                params={"q": customer_name},
                headers={
                    # Service account token from env var — never hardcode
                    "Authorization": f"Bearer {os.environ['CRM_TOKEN']}",
                },
                timeout=10,
            )
            resp.raise_for_status()
            data = resp.json()

        except requests.Timeout:
            return "Error: CRM service timed out; please retry later."
        except requests.HTTPError as e:
            return f"Error: CRM returned {e.response.status_code}; check your query conditions."

        items = data.get("results", [])
        if not items:
            return f"No customer found matching '{customer_name}' in CRM."
        c = items[0]
        return (
            f"Customer: {c['name']} | Tier: {c['tier']} | "
            f"Sales rep: {c['owner']} | Last order: {c['last_order_at']}"
        )


crm_tool = CrmLookupTool()

Design review:

• Timeout is mandatory (timeout=10); without it, one hang stalls the entire crew;
• Translate exceptions into human-readable text for the model — never let it see a Python traceback and guess;
• Expose only necessary fields: don't pass through all dozens of internal API fields; pick what the task needs, saving tokens and preventing info leaks.

8.5 Chapter Summary

• Simple tools use the @tool decorator; complex tools inherit BaseTool and implement _run;
• args_schema uses a Pydantic BaseModel to declare and validate parameters in one step;
• The tool description (including each parameter's description) is the model's sole basis for choosing and passing params;
• Recoverable errors should return explanatory text or ToolFailure, enabling agent self-correction;
• cache_function controls which results get cached; external API tools must set timeouts.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. In a BaseTool subclass, which method should contain the tool's actual execution logic?

2. Regarding args_schema, which statement is correct?

3. What does cache_function returning False mean?

4. When wrapping an internal REST API as a tool, which practice is NOT recommended?

🛠️ Hands-On Exercises

1. Write a @tool-decorated tool that queries a mock database of products; validate the category parameter with an enum; cache successful results but not empty-result calls.
2. Wrap a real external API (e.g., a free weather or news API) as a BaseTool subclass; add timeout, error translation, and whitelist input validation.
3. Create a tool with cache_function = lambda args, result: False; call it twice in a crew and confirm the underlying function runs both times.

Now you can build your own tools. Next, hierarchical process and the manager role.