Tool Sharing and Delegation

1from crewai_tools import (
2    # Search Tools
3    SerperDevTool,         # Google search via Serper API
4    DuckDuckGoSearchTool,  # DuckDuckGo search (no API key needed)
5
6    # Web Tools
7    WebsiteSearchTool,     # Search within a specific website
8    ScrapeWebsiteTool,     # Scrape website content
9
10    # File Tools
11    FileReadTool,          # Read files
12    DirectoryReadTool,     # List directory contents
13    DirectorySearchTool,   # Search in directories
14
15    # Code Tools
16    CodeInterpreterTool,   # Execute Python code
17    GithubSearchTool,      # Search GitHub
18
19    # Document Tools
20    PDFSearchTool,         # Search in PDFs
21    DOCXSearchTool,        # Search in Word docs
22    CSVSearchTool,         # Search in CSV files
23
24    # API Tools
25    JSONSearchTool,        # Search in JSON files
26    XMLSearchTool,         # Search in XML files
27)
28
29# Example: Agent with multiple tools
30from crewai import Agent
31
32research_agent = Agent(
33    role="Research Analyst",
34    goal="Find and analyze information from various sources",
35    backstory="Expert researcher with access to multiple data sources",
36    tools=[
37        SerperDevTool(),
38        WebsiteSearchTool(),
39        PDFSearchTool()
40    ]
41)

1from crewai_tools import SerperDevTool, WebsiteSearchTool
2import os
3
4# Configure tool with API key
5os.environ["SERPER_API_KEY"] = "your-key"
6search_tool = SerperDevTool()
7
8# Configure tool with specific settings
9website_tool = WebsiteSearchTool(
10    website="https://docs.python.org",
11    config={
12        "llm": {
13            "provider": "openai",
14            "config": {
15                "model": "gpt-4o-mini",  # Cheaper model for search
16            }
17        },
18        "embedder": {
19            "provider": "openai",
20            "config": {
21                "model": "text-embedding-3-small"
22            }
23        }
24    }
25)
26
27# Tools with callbacks
28def on_tool_use(tool_name, input_data, output):
29    print(f"Tool: {tool_name}")
30    print(f"Input: {input_data}")
31    print(f"Output: {output[:100]}...")
32
33# Monitor tool usage in agent
34agent = Agent(
35    role="Researcher",
36    goal="Research topics",
37    backstory="Expert researcher",
38    tools=[search_tool, website_tool],
39    step_callback=lambda step: print(f"Step: {step}")
40)

1"""
2Pattern: Each agent has exclusive tools
3
4Benefits:
5- Clear ownership of capabilities
6- No tool conflicts
7- Specialized expertise
8"""
9
10from crewai import Agent
11from crewai_tools import SerperDevTool, CodeInterpreterTool, FileReadTool
12
13# Web researcher - only web tools
14web_researcher = Agent(
15    role="Web Researcher",
16    goal="Find information online",
17    backstory="Expert at web research",
18    tools=[SerperDevTool()]  # Only web search
19)
20
21# Code analyst - only code tools
22code_analyst = Agent(
23    role="Code Analyst",
24    goal="Analyze and execute code",
25    backstory="Expert programmer",
26    tools=[CodeInterpreterTool()]  # Only code execution
27)
28
29# Document analyst - only file tools
30doc_analyst = Agent(
31    role="Document Analyst",
32    goal="Analyze documents and files",
33    backstory="Expert at document analysis",
34    tools=[FileReadTool()]  # Only file reading
35)
36
37# Each agent is the specialist for their domain

1"""
2Pattern: Multiple agents share common tools
3
4Benefits:
5- Flexibility in task assignment
6- Redundancy for critical capabilities
7- Efficient resource usage
8"""
9
10# Shared tool instances
11shared_search = SerperDevTool()
12shared_file_reader = FileReadTool()
13
14# Multiple agents share the same tools
15researcher_a = Agent(
16    role="Market Researcher",
17    goal="Research market trends",
18    backstory="Market analysis expert",
19    tools=[shared_search, shared_file_reader]
20)
21
22researcher_b = Agent(
23    role="Technical Researcher",
24    goal="Research technology developments",
25    backstory="Tech research expert",
26    tools=[shared_search, shared_file_reader]  # Same tools
27)
28
29# Both can search and read files
30# Manager in hierarchical can assign to either based on availability

1"""
2Pattern: Central tool pool with role-based access
3
4Benefits:
5- Centralized tool management
6- Role-based permissions
7- Easy to update tools
8"""
9
10from dataclasses import dataclass
11from typing import List, Dict
12
13@dataclass
14class ToolPool:
15    """Centralized tool management."""
16
17    tools: Dict[str, list] = None
18
19    def __post_init__(self):
20        self.tools = {
21            "research": [
22                SerperDevTool(),
23                WebsiteSearchTool()
24            ],
25            "analysis": [
26                CodeInterpreterTool(),
27                FileReadTool()
28            ],
29            "writing": []  # Writers don't need tools
30        }
31
32    def get_tools(self, roles: List[str]) -> list:
33        """Get tools for given roles."""
34        all_tools = []
35        for role in roles:
36            all_tools.extend(self.tools.get(role, []))
37        return list(set(all_tools))  # Deduplicate
38
39
40# Usage
41tool_pool = ToolPool()
42
43research_agent = Agent(
44    role="Researcher",
45    goal="Research topics",
46    backstory="Expert researcher",
47    tools=tool_pool.get_tools(["research"])
48)
49
50analyst_agent = Agent(
51    role="Analyst",
52    goal="Analyze data",
53    backstory="Expert analyst",
54    tools=tool_pool.get_tools(["research", "analysis"])  # Multiple roles
55)

1"""
2Delegation allows agents to ask other agents for help.
3
4When an agent with allow_delegation=True encounters a task
5outside their expertise, they can delegate to a more suitable agent.
6"""
7
8from crewai import Agent, Task, Crew, Process
9
10# Lead agent can delegate
11lead_researcher = Agent(
12    role="Lead Research Director",
13    goal="Oversee comprehensive research projects",
14    backstory="""You lead research teams and know when to leverage
15    specialist expertise. You delegate detailed work to specialists.""",
16    allow_delegation=True  # Can ask others for help
17)
18
19# Specialist cannot delegate (focuses on their work)
20data_specialist = Agent(
21    role="Data Specialist",
22    goal="Perform deep data analysis",
23    backstory="Expert in statistical analysis and data processing",
24    allow_delegation=False  # Focuses on own expertise
25)
26
27web_specialist = Agent(
28    role="Web Research Specialist",
29    goal="Find and verify online information",
30    backstory="Expert at web research and source verification",
31    allow_delegation=False
32)
33
34# When lead_researcher needs specialized work,
35# they can delegate to the appropriate specialist

1"""
2Delegation Patterns:
3
41. Expertise-Based: Delegate to domain experts
52. Capacity-Based: Delegate when overloaded
63. Quality-Based: Delegate for higher quality
74. Hierarchical: Manager delegates to workers
8"""
9
10# Pattern 1: Expertise-based delegation
11manager = Agent(
12    role="Research Manager",
13    goal="Coordinate research and ensure quality",
14    backstory="""You coordinate research teams. When you encounter
15    specialized needs, you delegate to the appropriate expert.""",
16    allow_delegation=True
17)
18
19# Manager will automatically delegate:
20# - Code questions → Code specialist
21# - Data analysis → Data specialist
22# - Web research → Web specialist
23
24# Pattern 2: Hierarchical delegation in crews
25from crewai import Crew, Process
26
27hierarchical_crew = Crew(
28    agents=[manager, data_specialist, web_specialist],
29    tasks=tasks,
30    process=Process.hierarchical,
31    manager_agent=manager  # Manager handles all delegation
32)

1from crewai import Agent
2
3# Delegation with constraints
4controlled_agent = Agent(
5    role="Team Lead",
6    goal="Lead the team effectively",
7    backstory="""You lead a specialized team. You delegate thoughtfully,
8    ensuring tasks go to the right team members. You never delegate
9    critical decisions - only execution tasks.""",
10    allow_delegation=True,
11    max_iter=10  # Limits back-and-forth delegation
12)
13
14# Agent that explicitly avoids delegation
15focused_agent = Agent(
16    role="Specialist",
17    goal="Complete specialized tasks independently",
18    backstory="""You are an independent specialist who takes pride in
19    completing tasks yourself. You prefer to work through challenges
20    rather than passing them to others.""",
21    allow_delegation=False
22)
23
24# Delegation-only agent (coordinator)
25coordinator = Agent(
26    role="Project Coordinator",
27    goal="Coordinate work across the team",
28    backstory="""You coordinate work but don't execute tasks yourself.
29    Your job is to understand requirements and assign work to the
30    right specialists.""",
31    allow_delegation=True
32    # Note: This agent should have no tools - their job is to delegate
33)

1from crewai.tools import BaseTool
2from pydantic import BaseModel, Field
3from typing import Type
4
5
6class CalculatorInput(BaseModel):
7    """Input schema for calculator tool."""
8    expression: str = Field(
9        description="Mathematical expression to evaluate"
10    )
11
12
13class CalculatorTool(BaseTool):
14    """Tool for evaluating mathematical expressions."""
15
16    name: str = "Calculator"
17    description: str = "Evaluates mathematical expressions safely"
18    args_schema: Type[BaseModel] = CalculatorInput
19
20    def _run(self, expression: str) -> str:
21        """Execute the calculation."""
22        try:
23            # Safe evaluation (limited operations)
24            allowed_chars = set("0123456789+-*/().  ")
25            if not all(c in allowed_chars for c in expression):
26                return "Error: Invalid characters in expression"
27
28            result = eval(expression)
29            return f"Result: {result}"
30        except Exception as e:
31            return f"Error: {str(e)}"
32
33
34# Use the custom tool
35calculator = CalculatorTool()
36math_agent = Agent(
37    role="Math Assistant",
38    goal="Help with mathematical calculations",
39    backstory="Expert mathematician",
40    tools=[calculator]
41)

1from crewai.tools import BaseTool
2from pydantic import BaseModel, Field
3import httpx
4from typing import Type
5
6
7class WeatherInput(BaseModel):
8    """Input for weather tool."""
9    city: str = Field(description="City name for weather lookup")
10
11
12class WeatherTool(BaseTool):
13    """Tool for getting weather information."""
14
15    name: str = "Weather Lookup"
16    description: str = "Get current weather for a city"
17    args_schema: Type[BaseModel] = WeatherInput
18
19    def __init__(self, api_key: str):
20        super().__init__()
21        self.api_key = api_key
22
23    def _run(self, city: str) -> str:
24        """Fetch weather data."""
25        try:
26            url = f"https://api.weatherapi.com/v1/current.json"
27            params = {"key": self.api_key, "q": city}
28
29            response = httpx.get(url, params=params)
30            data = response.json()
31
32            if "error" in data:
33                return f"Error: {data['error']['message']}"
34
35            current = data["current"]
36            return f"""Weather in {city}:
37            Temperature: {current['temp_c']}°C
38            Condition: {current['condition']['text']}
39            Humidity: {current['humidity']}%"""
40
41        except Exception as e:
42            return f"Error fetching weather: {str(e)}"
43
44
45# Usage
46weather_tool = WeatherTool(api_key="your-key")
47assistant = Agent(
48    role="Weather Assistant",
49    goal="Provide weather information",
50    backstory="Helpful weather assistant",
51    tools=[weather_tool]
52)

1from crewai.tools import BaseTool
2from pydantic import BaseModel, Field
3from typing import Type, Optional
4import sqlite3
5
6
7class QueryInput(BaseModel):
8    """Input for database queries."""
9    query: str = Field(description="SQL SELECT query to execute")
10
11
12class DatabaseTool(BaseTool):
13    """Tool for querying a SQLite database."""
14
15    name: str = "Database Query"
16    description: str = "Execute SELECT queries on the database"
17    args_schema: Type[BaseModel] = QueryInput
18
19    def __init__(self, db_path: str):
20        super().__init__()
21        self.db_path = db_path
22
23    def _run(self, query: str) -> str:
24        """Execute the query safely."""
25        # Security: Only allow SELECT queries
26        if not query.strip().upper().startswith("SELECT"):
27            return "Error: Only SELECT queries are allowed"
28
29        try:
30            conn = sqlite3.connect(self.db_path)
31            cursor = conn.cursor()
32            cursor.execute(query)
33            results = cursor.fetchall()
34            columns = [desc[0] for desc in cursor.description]
35            conn.close()
36
37            if not results:
38                return "No results found"
39
40            # Format results
41            output = f"Columns: {columns}\n"
42            output += f"Results ({len(results)} rows):\n"
43            for row in results[:10]:  # Limit to 10 rows
44                output += f"  {row}\n"
45
46            return output
47
48        except Exception as e:
49            return f"Query error: {str(e)}"
50
51
52# Usage
53db_tool = DatabaseTool(db_path="./data.db")
54data_agent = Agent(
55    role="Data Analyst",
56    goal="Analyze database data",
57    backstory="Expert SQL analyst",
58    tools=[db_tool]
59)