CrewAI Tools: Building Custom Agents & Tool Integration
CrewAI is a framework for building multi-agent AI systems. Tools are the actions agents can performβfrom web searches to database queries. This guide explains how to create and use tools in CrewAI.
What is a CrewAI Tool?
A tool is any action an AI agent can take. Without tools, an agent can only chat. With tools, it can:
- π Search the web
- πΎ Access databases
- π§ Send emails
- π Analyze data
- π¨ Generate content
- And anything else you can code
How CrewAI Works
User Task
β
[Agent 1] ββ Uses Tool A ββ Returns result
β
[Agent 2] ββ Uses Tool B ββ Returns result
β
[Agent 3] ββ Uses Tool C ββ Returns result
β
Final Output
Each agent decides which tools to use and when.
Installing CrewAI
pip install crewai
# With additional tools package
pip install crewai-tools
Creating Your First Tool
Method 1: Using the @tool Decorator
from crewai import Agent, Task, Crew, Tool
from crewai_tools import tool
@tool
def search_database(query: str) -> str:
"""
Search the company database for information.
Args:
query: The search term to look up
"""
# Your database search logic here
return f"Found results for {query}"
# Use it in an agent
agent = Agent(
role="Data Analyst",
goal="Find information in our database",
tools=[search_database]
)
Method 2: Using Tool Class
from crewai import Tool
def multiply(a: int, b: int) -> int:
"""Multiply two numbers"""
return a * b
math_tool = Tool(
name="Multiply",
func=multiply,
description="Multiplies two numbers together"
)
agent = Agent(
role="Calculator",
goal="Solve math problems",
tools=[math_tool]
)
Built-in CrewAI Tools
CrewAI comes with pre-built tools you can use immediately:
Web Search
from crewai_tools import tool
from crewai_tools.tools import DuckDuckGoSearchRun
search_tool = DuckDuckGoSearchRun()
agent = Agent(
role="Researcher",
goal="Find information online",
tools=[search_tool]
)
File Operations
from crewai_tools import FileReadTool, FileWriteTool
read_tool = FileReadTool(file_path="data.txt")
write_tool = FileWriteTool(file_path="output.txt")
agent = Agent(
role="File Manager",
goal="Read and write files",
tools=[read_tool, write_tool]
)
Complete Example: Multi-Agent Research System
from crewai import Agent, Task, Crew
from crewai_tools import DuckDuckGoSearchRun, tool
# Define custom tools
@tool
def analyze_sentiment(text: str) -> str:
"""Analyze the sentiment of text"""
# Simple sentiment analysis
positive_words = ['good', 'great', 'excellent', 'amazing']
negative_words = ['bad', 'terrible', 'awful', 'horrible']
text_lower = text.lower()
if any(word in text_lower for word in positive_words):
return "positive"
elif any(word in text_lower for word in negative_words):
return "negative"
return "neutral"
# Create agents
researcher = Agent(
role="Research Analyst",
goal="Find information about a topic",
backstory="You are an expert researcher who finds accurate information",
tools=[DuckDuckGoSearchRun()],
verbose=True
)
analyst = Agent(
role="Content Analyst",
goal="Analyze and summarize information",
backstory="You are expert at analyzing content and finding key insights",
tools=[analyze_sentiment],
verbose=True
)
# Create tasks
research_task = Task(
description="Research the latest developments in AI",
agent=researcher,
expected_output="A summary of recent AI developments"
)
analysis_task = Task(
description="Analyze the sentiment of the research findings",
agent=analyst,
expected_output="Sentiment analysis of the findings"
)
# Create crew
crew = Crew(
agents=[researcher, analyst],
tasks=[research_task, analysis_task],
verbose=True
)
# Execute
result = crew.kickoff()
print(result)
Advanced: Creating Complex Tools
Tool with Multiple Inputs
@tool
def fetch_user_data(user_id: int, include_details: bool = False) -> dict:
"""
Fetch user information from database.
Args:
user_id: The ID of the user to fetch
include_details: Whether to include detailed profile info
Returns:
Dictionary with user data
"""
# Simulate database fetch
user_data = {
'id': user_id,
'name': f'User {user_id}',
'email': f'user{user_id}@example.com'
}
if include_details:
user_data['preferences'] = {'theme': 'dark', 'language': 'en'}
user_data['activity'] = {'last_login': '2024-03-15'}
return user_data
Tool with Error Handling
@tool
def calculate_discount(price: float, discount_percent: float) -> str:
"""Calculate discounted price."""
try:
if not 0 <= discount_percent <= 100:
raise ValueError(f"Discount must be 0-100%, got {discount_percent}")
if price < 0:
raise ValueError("Price cannot be negative")
final_price = price * (1 - discount_percent / 100)
return f"Original: ${price:.2f}, Discount: {discount_percent}%, Final: ${final_price:.2f}"
except ValueError as e:
return f"Error: {str(e)}"
Tool with External API Call
import requests
@tool
def get_weather(city: str) -> str:
"""Get current weather for a city."""
try:
response = requests.get(
f"https://api.openweathermap.org/data/2.5/weather",
params={
'q': city,
'appid': 'YOUR_API_KEY',
'units': 'metric'
}
)
if response.status_code == 200:
data = response.json()
return f"Weather in {city}: {data['weather'][0]['main']}, Temp: {data['main']['temp']}Β°C"
else:
return f"Error fetching weather for {city}"
except Exception as e:
return f"Error: {str(e)}"
Tool Requirements
Effective tools need:
- Clear Name - Describe what it does
- Documentation - Docstring explaining parameters
- Type Hints - So agents know what inputs to provide
- Error Handling - Graceful failures
- Reasonable Output - Return data the agent can use
Good Tool Example
@tool
def convert_currency(amount: float, from_currency: str, to_currency: str) -> str:
"""
Convert between two currencies using real exchange rates.
Args:
amount: The amount to convert (must be positive)
from_currency: Source currency code (e.g., 'USD')
to_currency: Target currency code (e.g., 'EUR')
Returns:
Formatted string with conversion result
"""
# Proper implementation with error handling
if amount <= 0:
return "Error: Amount must be positive"
# Get exchange rates (example)
rates = {'USD': 1.0, 'EUR': 0.92, 'GBP': 0.79}
if from_currency not in rates or to_currency not in rates:
return "Error: Unsupported currency"
converted = amount * (rates[to_currency] / rates[from_currency])
return f"{amount} {from_currency} = {converted:.2f} {to_currency}"
Using Multiple Tools in One Agent
from crewai import Agent, Tool
@tool
def fetch_weather(city: str) -> str:
"""Get weather for a city"""
return f"Weather in {city}: Sunny, 72Β°F"
@tool
def fetch_calendar(date: str) -> str:
"""Get calendar events for a date"""
return f"Events on {date}: Team meeting at 2pm, Lunch at 1pm"
@tool
def send_notification(message: str) -> str:
"""Send a notification"""
return f"Notification sent: {message}"
# One agent with multiple tools
assistant = Agent(
role="Personal Assistant",
goal="Help with daily tasks",
backstory="You are a helpful personal assistant",
tools=[fetch_weather, fetch_calendar, send_notification]
)
# Agent can decide which tool to use for each task
Real-World Use Cases
1. Customer Support Agent
@tool
def search_knowledge_base(query: str) -> str:
"""Search customer support knowledge base"""
# Implementation...
pass
@tool
def get_order_status(order_id: str) -> str:
"""Check customer order status"""
# Implementation...
pass
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""Send email to customer"""
# Implementation...
pass
support_agent = Agent(
role="Support Specialist",
goal="Help customers with their issues",
tools=[search_knowledge_base, get_order_status, send_email]
)
2. Data Analysis Agent
@tool
def query_database(sql: str) -> str:
"""Execute database query"""
# Implementation...
pass
@tool
def create_chart(data: list, chart_type: str) -> str:
"""Generate chart from data"""
# Implementation...
pass
@tool
def generate_report(data: dict) -> str:
"""Generate analysis report"""
# Implementation...
pass
analyst = Agent(
role="Data Analyst",
goal="Analyze data and provide insights",
tools=[query_database, create_chart, generate_report]
)
3. Content Creation Agent
@tool
def fetch_research_data(topic: str) -> str:
"""Research a topic online"""
# Implementation...
pass
@tool
def generate_outline(topic: str) -> str:
"""Generate content outline"""
# Implementation...
pass
@tool
def save_content(filename: str, content: str) -> str:
"""Save content to file"""
# Implementation...
pass
writer = Agent(
role="Content Writer",
goal="Create high-quality content",
tools=[fetch_research_data, generate_outline, save_content]
)
Best Practices
- Keep Tools Focused - One tool = one clear action
- Descriptive Docstrings - Help agents understand what to do
- Error Handling - Return meaningful error messages
- Rate Limiting - If calling external APIs, add delays
- Logging - Track tool usage for debugging
- Testing - Test tools independently before using in agents
- Documentation - Document tool behavior for other developers
Tool Execution Flow
Agent needs to act
β
Looks at available tools
β
Determines best tool to use
β
Prepares tool input
β
Executes tool
β
Analyzes result
β
Decides next step (use another tool or return answer)
Common Patterns
Pattern 1: Tool Chaining
# Task requires multiple tool calls in sequence
# Agent automatically chains tools:
# fetch_data() β process_data() β save_results()
Pattern 2: Fallback Tools
# Provide alternative tools if primary fails
@tool
def get_price_from_api(product_id: str) -> str:
"""Primary: Get price from API"""
pass
@tool
def get_price_from_cache(product_id: str) -> str:
"""Fallback: Get price from cache"""
pass
Pattern 3: Conditional Tools
# Agent decides which tool based on input
# If user_id provided β fetch_user_tool
# If email provided β find_user_by_email_tool
Conclusion
CrewAI tools empower agents to take action:
- @tool decorator - Simplest way to create tools
- Error handling - Essential for robust systems
- Documentation - Helps agents make smart decisions
- Multiple tools - Agents choose the right tool for each task
Start simple, add tools as needed, and let your agents take action!