CrewAI custom tools are the extension point that turns a language model into an agent that can act on your systems. When the built-in connectors don’t cover your internal APIs, you need to define your own. This guide walks through the exact code to build, register, and debug a custom tool against a real CrewAI agent, from environment setup to verification in a production-style loop.
Step 1: Install CrewAI and configure the LLM
Install the core package and the tools helper library:
pip install crewai crewai-toolsCrewAI expects an OpenAI-compatible chat model by default. Point it at your provider of choice. If you’re routing through n4n.ai, the OpenAI-compatible endpoint covers 240+ models and handles fallback when a provider is degraded, so your agent’s reasoning loop stays up without custom retry code.
from crewai import LLM
llm = LLM(
model="gpt-4o-mini",
base_url="https://api.n4n.ai/v1", # OpenAI-compatible
api_key="your-key",
)Set the environment variable OPENAI_API_KEY only if you use OpenAI directly; otherwise pass credentials per-LLM instance. Keep the model identifier aligned with what your gateway supports.
Step 2: Write a minimal custom tool with the @tool decorator
The fastest path to CrewAI custom tools is the @tool decorator. The decorator infers the schema from type hints and the docstring. Keep the docstring precise—the LLM reads it to decide when to call the tool.
from crewai.tools import tool
@tool("lookup_inventory")
def lookup_inventory(sku: str) -> str:
"""Look up current stock count for a product SKU.
Args:
sku: The product identifier, e.g. 'A-123'.
Returns:
A string with the available quantity and warehouse location.
"""
# Mock external call
inventory = {"A-123": (42, "DC-East"), "B-456": (0, "DC-West")}
if sku not in inventory:
return f"SKU {sku} not found"
qty, loc = inventory[sku]
return f"{qty} units at {loc}"The function name becomes the tool name unless you pass a string to the decorator. The function must return a string. CrewAI serializes the output back to the model as an observation. Avoid returning Python objects; if you need structure, dump to JSON.
Step 3: Define a structured tool with BaseTool for validation
When inputs need constraints or you want explicit Pydantic schemas, subclass BaseTool from crewai_tools. This gives you control over argument validation and error surfacing.
from crewai_tools import BaseTool
from pydantic import BaseModel, Field
class InventoryInput(BaseModel):
sku: str = Field(..., description="Product SKU, uppercase letters and digits")
class InventoryTool(BaseTool):
name: str = "lookup_inventory_struct"
args_schema: type[BaseModel] = InventoryInput
def _run(self, sku: str) -> str:
if not sku.isupper():
return "Error: SKU must be uppercase"
# real request would go here
return f"Stock for {sku}: 12"
async def _arun(self, sku: str) -> str:
# async variant for non-blocking IO
return self._run(sku)Use _run for sync, _arun for async. Raising exceptions inside _run will break the agent loop; return error strings instead. The args_schema forces the LLM to produce conforming JSON, which reduces malformed calls.
Step 4: Attach the tool to an agent
Create an agent and pass your CrewAI custom tools in the tools list. The agent’s role and goal determine how aggressively it calls them.
from crewai import Agent, Task, Crew
inventory_tool = lookup_inventory
agent = Agent(
role="Supply Chain Analyst",
goal="Answer stock questions accurately using the inventory tool",
backstory="You track warehouse levels and report shortages.",
tools=[inventory_tool],
llm=llm,
verbose=True,
)
task = Task(
description="How many A-123 units are available?",
expected_output="A clear statement of quantity and location.",
agent=agent,
)
crew = Crew(agents=[agent], tasks=[task])
result = crew.kickoff()
print(result)You can pass multiple tools. The agent will pick based on descriptions. If two tools overlap in description, the model may guess wrong—make each docstring distinct.
Step 5: Execute and inspect the tool call
Run the script. With verbose=True, CrewAI prints the LLM’s thought process, including the tool call and its result. You should see a line similar to:
Action: lookup_inventory
Action Input: {"sku": "A-123"}
Observation: 42 units at DC-EastIf the agent ignores the tool, the docstring is likely too vague. Rewrite it to state exactly what the tool does and when to use it. Also confirm the model you selected supports tool calling; some small models do not emit the required function-call format.
To capture the call programmatically, wrap the tool with a logging closure:
def logged_tool(func):
def wrapper(*args, **kwargs):
print("TOOL CALLED", func.name, args, kwargs)
return func(*args, **kwargs)
return wrapper
agent.tools = [logged_tool(t) for t in agent.tools]Step 6: Verify the tool works in isolation
Before trusting the agent, unit-test the underlying function. CrewAI custom tools are just Python callables; test them directly.
def test_lookup():
out = lookup_inventory.run({"sku": "A-123"})
assert "42" in out
out2 = lookup_inventory.run({"sku": "ZZZ"})
assert "not found" in out2
test_lookup()For the BaseTool variant, call InventoryTool().run({"sku": "A-123"}). Verification succeeds when the assertions pass and the crew’s final answer incorporates the observation. Add this test to CI so regressions surface before deploy.
Step 7: Handle real network calls and failures
Production tools hit APIs. Wrap requests with timeouts and return structured error strings.
import requests
from tenacity import retry, stop_after_attempt, wait_fixed
@tool("fetch_order_status")
def fetch_order_status(order_id: str) -> str:
"""Fetch fulfillment status for an order ID from the internal API."""
@retry(stop=stop_after_attempt(3), wait=wait_fixed(1))
def _get():
return requests.get(
f"https://internal.api/orders/{order_id}",
timeout=5,
)
try:
resp = _get()
resp.raise_for_status()
data = resp.json()
return f"Order {order_id}: {data['status']}"
except requests.RequestException as e:
return f"Failed to fetch order {order_id}: {e}"Never let a traceback escape _run or the decorated function. The agent cannot recover from an unhandled exception, and the crew will halt. Use retries at the I/O layer, but cap them so the agent can fall back to reporting failure.
Step 8: Load tools dynamically from configuration
If you manage many CrewAI custom tools, register them from a config dict to avoid hardcoding.
TOOL_REGISTRY = {
"lookup_inventory": lookup_inventory,
"fetch_order_status": fetch_order_status,
}
def build_agent(tool_names: list[str], llm) -> Agent:
tools = [TOOL_REGISTRY[n] for n in tool_names]
return Agent(
role="Ops",
goal="Execute configured tools",
backstory="Config-driven operator",
tools=tools,
llm=llm,
)This pattern lets you swap tools per environment without touching agent logic. Pair it with a YAML file that lists tool names per agent role, and load it at startup.
Common pitfalls with CrewAI custom tools
- Weak docstrings: The LLM uses the description to decide invocation. “Does stuff” fails; “Returns stock count for SKU” works.
- Non-string returns: Returning a dict causes serialization issues. Convert to JSON or a formatted string.
- Mutable global state: Tools should be stateless or explicitly scoped. Side effects complicate debugging.
- Missing input validation: Use Pydantic schemas for anything beyond a single scalar.
- Silent failures: Returning
Noneconfuses the model. Always return a human-readable string.
Verify success in production
A custom tool is successful when: (1) the unit test passes, (2) the agent’s verbose log shows the correct Action/Observation pair, and (3) the final crew output references the tool’s data. Add a CI check that runs the unit test and a sandbox crew kickoff with a fixed prompt to catch regressions.
If you route LLM traffic through a gateway, confirm provider cache-control hints are forwarded so repeated tool-adjacent calls hit cache. That’s orthogonal to tool code but reduces latency and cost per agent step.
Write the tools as small, testable surfaces. The agent’s intelligence comes from the model; your job is to give it reliable, well-described levers.