feat: implement hierarchical agent delegation with allowed_agents parameter

- Add allowed_agents parameter to BaseAgent
- Add validation for allowed_agents list
- Update delegation tools to respect restrictions
- Add error messages for unauthorized delegation
- Add tests for hierarchical delegation

Co-Authored-By: Joe Moura <joao@crewai.com>

src/crewai/agents/agent_builder/base_agent.py

@@ -107,6 +107,10 @@ class BaseAgent(ABC, BaseModel):
         default=False,
         description="Enable agent to delegate and ask questions among each other.",
     )
+    allowed_agents: Optional[List[str]] = Field(
+        default=None,
+        description="List of agent roles that this agent is allowed to delegate tasks to.",
+    )
     tools: Optional[List[Any]] = Field(
         default_factory=list, description="Tools at agents' disposal"
     )
@@ -174,6 +178,13 @@ class BaseAgent(ABC, BaseModel):
                     f"{field} must be provided either directly or through config"
                 )
 
+        # Validate allowed_agents if delegation is enabled
+        if self.allow_delegation and self.allowed_agents is not None:
+            if not isinstance(self.allowed_agents, list):
+                raise ValueError("allowed_agents must be a list of strings")
+            if not all(isinstance(agent, str) for agent in self.allowed_agents):
+                raise ValueError("all entries in allowed_agents must be strings")
+
         # Set private attributes
         self._logger = Logger(verbose=self.verbose)
         if self.max_rpm and not self._rpm_controller:

src/crewai/tools/agent_tools/base_agent_tools.py

@@ -1,6 +1,5 @@
 from typing import Optional, Union
-
-from pydantic import Field
+from pydantic import UUID4, Field
 
 from crewai.agents.agent_builder.base_agent import BaseAgent
 from crewai.task import Task
@@ -12,6 +11,7 @@ class BaseAgentTool(BaseTool):
     """Base class for agent-related tools"""
 
     agents: list[BaseAgent] = Field(description="List of available agents")
+    agent_id: UUID4 = Field(description="ID of the agent using this tool")
     i18n: I18N = Field(
         default_factory=I18N, description="Internationalization settings"
     )
@@ -58,6 +58,15 @@ class BaseAgentTool(BaseTool):
                 )
             )
 
+        # Check if delegation is allowed based on allowed_agents list
+        delegating_agent = [a for a in self.agents if a.id == self.agent_id][0]
+        if (delegating_agent.allowed_agents is not None and 
+            agent[0].role not in delegating_agent.allowed_agents):
+            return self.i18n.errors("agent_tool_unauthorized_delegation").format(
+                coworker=agent[0].role,
+                allowed_agents="\n".join([f"- {role}" for role in delegating_agent.allowed_agents])
+            )
+
         agent = agent[0]
         task_with_assigned_agent = Task(  # type: ignore # Incompatible types in assignment (expression has type "Task", variable has type "str")
             description=task,

src/crewai/translations/en.json

@@ -33,7 +33,8 @@
     "tool_usage_error": "I encountered an error: {error}",
     "tool_arguments_error": "Error: the Action Input is not a valid key, value dictionary.",
     "wrong_tool_name": "You tried to use the tool {tool}, but it doesn't exist. You must use one of the following tools, use one at time: {tools}.",
-    "tool_usage_exception": "I encountered an error while trying to use the tool. This was the error: {error}.\n Tool {tool} accepts these inputs: {tool_inputs}"
+    "tool_usage_exception": "I encountered an error while trying to use the tool. This was the error: {error}.\n Tool {tool} accepts these inputs: {tool_inputs}",
+    "agent_tool_unauthorized_delegation": "I cannot delegate this task to {coworker} as I am only allowed to delegate to: \n{allowed_agents}"
   },
   "tools": {
     "delegate_work": "Delegate a specific task to one of the following coworkers: {coworkers}\nThe input to this tool should be the coworker, the task you want them to do, and ALL necessary context to execute the task, they know nothing about the task, so share absolute everything you know, don't reference things but instead explain them.",

tests/agents/test_delegation.py

@@ -0,0 +1,166 @@
+import pytest
+from unittest.mock import MagicMock
+
+from crewai.agent import Agent
+from crewai.crew import Crew
+from crewai.task import Task
+from crewai.tools.agent_tools.delegate_work_tool import DelegateWorkTool
+
+def test_delegate_work_with_allowed_agents():
+    """Test successful delegation to allowed agent."""
+    # Create agents
+    executive = Agent(
+        role="Executive Director",
+        goal="Manage the team",
+        backstory="An experienced manager",
+        allow_delegation=True,
+        allowed_agents=["Communications Manager"]
+    )
+    comms_manager = Agent(
+        role="Communications Manager",
+        goal="Handle communications",
+        backstory="A skilled communicator",
+        allow_delegation=False
+    )
+    
+    # Mock LLM to avoid actual API calls
+    mock_content = "Thought: I will handle this task\nFinal Answer: Task completed successfully"
+    mock_response = {
+        "choices": [{
+            "message": {
+                "content": mock_content
+            }
+        }]
+    }
+    executive.llm = MagicMock()
+    executive.llm.invoke = MagicMock(return_value=mock_response)
+    executive.llm.call = MagicMock(return_value=mock_content)
+    comms_manager.llm = MagicMock()
+    comms_manager.llm.invoke = MagicMock(return_value=mock_response)
+    comms_manager.llm.call = MagicMock(return_value=mock_content)
+    
+    # Create crew and tool
+    crew = Crew(agents=[executive, comms_manager])
+    tool = DelegateWorkTool(
+        name="Delegate work to coworker",
+        description="Tool for delegating work to coworkers",
+        agents=[executive, comms_manager],
+        agent_id=executive.id
+    )
+    
+    # Test delegation
+    result = tool._execute(
+        agent_name="Communications Manager",
+        task="Write a press release",
+        context="Important company announcement"
+    )
+    
+    # Verify delegation was allowed
+    assert "error" not in result.lower()
+    assert "unauthorized" not in result.lower()
+
+def test_delegate_work_with_unauthorized_agent():
+    """Test failed delegation to unauthorized agent."""
+    # Create agents
+    executive = Agent(
+        role="Executive Director",
+        goal="Manage the team",
+        backstory="An experienced manager",
+        allow_delegation=True,
+        allowed_agents=["Communications Manager"]
+    )
+    tech_manager = Agent(
+        role="Tech Manager",
+        goal="Manage technology",
+        backstory="A tech expert",
+        allow_delegation=False
+    )
+    
+    # Mock LLM to avoid actual API calls
+    mock_content = "Thought: I will handle this task\nFinal Answer: Task completed successfully"
+    mock_response = {
+        "choices": [{
+            "message": {
+                "content": mock_content
+            }
+        }]
+    }
+    executive.llm = MagicMock()
+    executive.llm.invoke = MagicMock(return_value=mock_response)
+    executive.llm.call = MagicMock(return_value=mock_content)
+    tech_manager.llm = MagicMock()
+    tech_manager.llm.invoke = MagicMock(return_value=mock_response)
+    tech_manager.llm.call = MagicMock(return_value=mock_content)
+    
+    # Create crew and tool
+    crew = Crew(agents=[executive, tech_manager])
+    tool = DelegateWorkTool(
+        name="Delegate work to coworker",
+        description="Tool for delegating work to coworkers",
+        agents=[executive, tech_manager],
+        agent_id=executive.id
+    )
+    
+    # Test delegation
+    result = tool._execute(
+        agent_name="Tech Manager",
+        task="Update servers",
+        context="Server maintenance needed"
+    )
+    
+    # Verify delegation was blocked with proper error message
+    assert "cannot delegate this task" in result.lower()
+    assert "tech manager" in result.lower()
+    assert "communications manager" in result.lower()
+
+def test_delegate_work_without_allowed_agents():
+    """Test delegation works normally when no allowed_agents is specified."""
+    # Create agents
+    manager = Agent(
+        role="Manager",
+        goal="Manage the team",
+        backstory="An experienced manager",
+        allow_delegation=True  # No allowed_agents specified
+    )
+    worker = Agent(
+        role="Worker",
+        goal="Do the work",
+        backstory="A skilled worker",
+        allow_delegation=False
+    )
+    
+    # Mock LLM to avoid actual API calls
+    mock_content = "Thought: I will handle this task\nFinal Answer: Task completed successfully"
+    mock_response = {
+        "choices": [{
+            "message": {
+                "content": mock_content
+            }
+        }]
+    }
+    manager.llm = MagicMock()
+    manager.llm.invoke = MagicMock(return_value=mock_response)
+    manager.llm.call = MagicMock(return_value=mock_content)
+    worker.llm = MagicMock()
+    worker.llm.invoke = MagicMock(return_value=mock_response)
+    worker.llm.call = MagicMock(return_value=mock_content)
+    
+    # Create crew and tool
+    crew = Crew(agents=[manager, worker])
+    tool = DelegateWorkTool(
+        name="Delegate work to coworker",
+        description="Tool for delegating work to coworkers",
+        agents=[manager, worker],
+        agent_id=manager.id
+    )
+    
+    # Test delegation
+    result = tool._execute(
+        agent_name="Worker",
+        task="Complete task",
+        context="Important task"
+    )
+    
+    # Verify delegation was allowed
+    assert "error" not in result.lower()
+    assert "unauthorized" not in result.lower()