fix: update test error message expectations to match custom validator messages

Co-Authored-By: João <joao@crewai.com>

src/crewai/agent.py

@@ -565,7 +565,7 @@ class Agent(BaseAgent):
 
     def get_delegation_tools(self, agents: List[BaseAgent]):
         agent_tools = AgentTools(agents=agents)
-        tools = agent_tools.tools()
+        tools = agent_tools.tools(delegating_agent=self)
         return tools
 
     def get_multimodal_tools(self) -> Sequence[BaseTool]:

src/crewai/agents/agent_builder/base_agent.py

@@ -1,8 +1,11 @@
+from __future__ import annotations
+
 import uuid
 from abc import ABC, abstractmethod
 from copy import copy as shallow_copy
 from hashlib import md5
-from typing import Any, Callable, Dict, List, Optional, TypeVar
+from typing import Any, Callable, Dict, List, Optional, TypeVar, Union
+from collections.abc import Sequence
 
 from pydantic import (
     UUID4,
@@ -25,7 +28,6 @@ from crewai.security.security_config import SecurityConfig
 from crewai.tools.base_tool import BaseTool, Tool
 from crewai.utilities import I18N, Logger, RPMController
 from crewai.utilities.config import process_config
-from crewai.utilities.converter import Converter
 from crewai.utilities.string_utils import interpolate_only
 
 T = TypeVar("T", bound="BaseAgent")
@@ -108,6 +110,10 @@ class BaseAgent(ABC, BaseModel):
         default=False,
         description="Enable agent to delegate and ask questions among each other.",
     )
+    allowed_agents: Optional[List[Union[str, BaseAgent]]] = Field(
+        default=None,
+        description="List of agent roles or agent instances that this agent can delegate to. If None, can delegate to all agents when allow_delegation=True. Empty list prevents all delegation.",
+    )
     tools: Optional[List[BaseTool]] = Field(
         default_factory=list, description="Tools at agents' disposal"
     )
@@ -195,6 +201,22 @@ class BaseAgent(ABC, BaseModel):
                 )
         return processed_tools
 
+    @field_validator("allowed_agents", mode="before")
+    @classmethod
+    def validate_allowed_agents(cls, allowed_agents: Optional[Sequence[Union[str, BaseAgent]]]) -> Optional[List[Union[str, BaseAgent]]]:
+        """Validate the allowed_agents list."""
+        if allowed_agents is None:
+            return None
+        
+        if not isinstance(allowed_agents, Sequence) or isinstance(allowed_agents, str):
+            raise ValueError("allowed_agents must be a list or tuple of agent roles (strings) or agent instances")
+        
+        for agent in allowed_agents:
+            if not isinstance(agent, (str, BaseAgent)):
+                raise ValueError("Each item in allowed_agents must be either a string (agent role) or a BaseAgent instance")
+        
+        return list(allowed_agents)
+
     @model_validator(mode="after")
     def validate_and_set_attributes(self):
         # Validate required fields

src/crewai/tools/agent_tools/agent_tools.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from crewai.agents.agent_builder.base_agent import BaseAgent
 from crewai.tools.base_tool import BaseTool
 from crewai.utilities import I18N
@@ -13,20 +15,50 @@ class AgentTools:
         self.agents = agents
         self.i18n = i18n
 
-    def tools(self) -> list[BaseTool]:
+    def tools(self, delegating_agent: Optional[BaseAgent] = None) -> list[BaseTool]:
-        """Get all available agent tools"""
+        """Get all available agent tools, filtered by delegating agent's allowed_agents if specified"""
-        coworkers = ", ".join([f"{agent.role}" for agent in self.agents])
+        available_agents = self._filter_allowed_agents(delegating_agent)
+        
+        if not available_agents:
+            return []
+        
+        coworkers = ", ".join([f"{agent.role}" for agent in available_agents])
 
         delegate_tool = DelegateWorkTool(
-            agents=self.agents,
+            agents=available_agents,
             i18n=self.i18n,
             description=self.i18n.tools("delegate_work").format(coworkers=coworkers),  # type: ignore
         )
 
         ask_tool = AskQuestionTool(
-            agents=self.agents,
+            agents=available_agents,
             i18n=self.i18n,
             description=self.i18n.tools("ask_question").format(coworkers=coworkers),  # type: ignore
         )
 
         return [delegate_tool, ask_tool]
+
+    def _filter_allowed_agents(self, delegating_agent: Optional[BaseAgent]) -> list[BaseAgent]:
+        """Filter agents based on the delegating agent's allowed_agents list"""
+        if delegating_agent is None:
+            return self.agents
+        
+        if not hasattr(delegating_agent, 'allowed_agents') or delegating_agent.allowed_agents is None:
+            return self.agents
+        
+        if not delegating_agent.allowed_agents:
+            return []
+        
+        filtered_agents = []
+        for agent in self.agents:
+            for allowed in delegating_agent.allowed_agents:
+                if isinstance(allowed, str):
+                    if agent.role.strip().lower() == allowed.strip().lower():
+                        filtered_agents.append(agent)
+                        break
+                elif isinstance(allowed, BaseAgent):
+                    if agent is allowed:
+                        filtered_agents.append(agent)
+                        break
+        
+        return filtered_agents

tests/test_allowed_agents.py

@@ -0,0 +1,263 @@
+"""Test allowed_agents functionality for hierarchical delegation."""
+
+import pytest
+from crewai.agent import Agent
+from crewai.crew import Crew
+from crewai.task import Task
+from crewai.tools.agent_tools.agent_tools import AgentTools
+
+
+@pytest.fixture
+def agents():
+    """Create test agents for delegation testing."""
+    manager = Agent(
+        role="Manager",
+        goal="Manage the team",
+        backstory="You are a team manager",
+        allow_delegation=True,
+    )
+    
+    researcher = Agent(
+        role="Researcher", 
+        goal="Research topics",
+        backstory="You are a researcher",
+        allow_delegation=False,
+    )
+    
+    writer = Agent(
+        role="Writer",
+        goal="Write content", 
+        backstory="You are a writer",
+        allow_delegation=False,
+    )
+    
+    analyst = Agent(
+        role="Analyst",
+        goal="Analyze data",
+        backstory="You are an analyst", 
+        allow_delegation=False,
+    )
+    
+    return manager, researcher, writer, analyst
+
+
+def test_allowed_agents_with_role_strings(agents):
+    """Test allowed_agents with role strings."""
+    manager, researcher, writer, analyst = agents
+    
+    manager.allowed_agents = ["Researcher", "Writer"]
+    
+    agent_tools = AgentTools(agents=[researcher, writer, analyst])
+    tools = agent_tools.tools(delegating_agent=manager)
+    
+    assert len(tools) == 2
+    delegate_tool = tools[0]
+    
+    assert len(delegate_tool.agents) == 2
+    agent_roles = [agent.role for agent in delegate_tool.agents]
+    assert "Researcher" in agent_roles
+    assert "Writer" in agent_roles
+    assert "Analyst" not in agent_roles
+
+
+def test_allowed_agents_with_agent_instances(agents):
+    """Test allowed_agents with agent instances."""
+    manager, researcher, writer, analyst = agents
+    
+    manager.allowed_agents = [researcher, analyst]
+    
+    agent_tools = AgentTools(agents=[researcher, writer, analyst])
+    tools = agent_tools.tools(delegating_agent=manager)
+    
+    assert len(tools) == 2
+    delegate_tool = tools[0]
+    
+    assert len(delegate_tool.agents) == 2
+    assert researcher in delegate_tool.agents
+    assert analyst in delegate_tool.agents
+    assert writer not in delegate_tool.agents
+
+
+def test_allowed_agents_mixed_types(agents):
+    """Test allowed_agents with mixed role strings and agent instances."""
+    manager, researcher, writer, analyst = agents
+    
+    manager.allowed_agents = ["Researcher", writer]
+    
+    agent_tools = AgentTools(agents=[researcher, writer, analyst])
+    tools = agent_tools.tools(delegating_agent=manager)
+    
+    delegate_tool = tools[0]
+    assert len(delegate_tool.agents) == 2
+    assert researcher in delegate_tool.agents
+    assert writer in delegate_tool.agents
+    assert analyst not in delegate_tool.agents
+
+
+def test_allowed_agents_empty_list(agents):
+    """Test allowed_agents with empty list (no delegation allowed)."""
+    manager, researcher, writer, analyst = agents
+    
+    manager.allowed_agents = []
+    
+    agent_tools = AgentTools(agents=[researcher, writer, analyst])
+    tools = agent_tools.tools(delegating_agent=manager)
+    
+    assert len(tools) == 0
+
+
+def test_allowed_agents_none(agents):
+    """Test allowed_agents with None (delegate to all agents)."""
+    manager, researcher, writer, analyst = agents
+    
+    manager.allowed_agents = None
+    
+    agent_tools = AgentTools(agents=[researcher, writer, analyst])
+    tools = agent_tools.tools(delegating_agent=manager)
+    
+    delegate_tool = tools[0]
+    assert len(delegate_tool.agents) == 3
+    assert researcher in delegate_tool.agents
+    assert writer in delegate_tool.agents
+    assert analyst in delegate_tool.agents
+
+
+def test_allowed_agents_case_insensitive_matching(agents):
+    """Test that role matching is case-insensitive."""
+    manager, researcher, writer, analyst = agents
+    
+    manager.allowed_agents = ["researcher", "WRITER"]
+    
+    agent_tools = AgentTools(agents=[researcher, writer, analyst])
+    tools = agent_tools.tools(delegating_agent=manager)
+    
+    delegate_tool = tools[0]
+    assert len(delegate_tool.agents) == 2
+    assert researcher in delegate_tool.agents
+    assert writer in delegate_tool.agents
+
+
+def test_allowed_agents_validation():
+    """Test validation of allowed_agents field."""
+    agent = Agent(
+        role="Test",
+        goal="Test",
+        backstory="Test",
+        allowed_agents=["Role1", "Role2"]
+    )
+    assert agent.allowed_agents == ["Role1", "Role2"]
+    
+    agent = Agent(
+        role="Test",
+        goal="Test", 
+        backstory="Test",
+        allowed_agents=None
+    )
+    assert agent.allowed_agents is None
+    
+    with pytest.raises(ValueError, match="allowed_agents must be a list or tuple of agent roles"):
+        Agent(
+            role="Test",
+            goal="Test",
+            backstory="Test",
+            allowed_agents="invalid"
+        )
+    
+    with pytest.raises(ValueError, match="Each item in allowed_agents must be either a string"):
+        Agent(
+            role="Test",
+            goal="Test",
+            backstory="Test",
+            allowed_agents=[123]
+        )
+
+
+def test_crew_integration_with_allowed_agents(agents):
+    """Test integration with Crew class."""
+    manager, researcher, writer, analyst = agents
+    
+    manager.allowed_agents = ["Researcher"]
+    
+    task = Task(
+        description="Research AI trends",
+        expected_output="Research report",
+        agent=manager
+    )
+    
+    crew = Crew(
+        agents=[manager, researcher, writer, analyst],
+        tasks=[task]
+    )
+    
+    tools = crew._prepare_tools(manager, task, [])
+    
+    delegation_tools = [tool for tool in tools if "Delegate work" in tool.name or "Ask question" in tool.name]
+    
+    if delegation_tools:
+        delegate_tool = next(tool for tool in delegation_tools if "Delegate work" in tool.name)
+        assert len(delegate_tool.agents) == 1
+        assert delegate_tool.agents[0].role == "Researcher"
+
+
+def test_backward_compatibility_no_allowed_agents(agents):
+    """Test that agents without allowed_agents work as before."""
+    manager, researcher, writer, analyst = agents
+    
+    assert manager.allowed_agents is None
+    
+    agent_tools = AgentTools(agents=[researcher, writer, analyst])
+    tools = agent_tools.tools(delegating_agent=manager)
+    
+    assert len(tools) == 2
+    delegate_tool = tools[0]
+    assert len(delegate_tool.agents) == 3
+    assert researcher in delegate_tool.agents
+    assert writer in delegate_tool.agents
+    assert analyst in delegate_tool.agents
+
+
+def test_no_delegating_agent_parameter(agents):
+    """Test AgentTools.tools() without delegating_agent parameter."""
+    manager, researcher, writer, analyst = agents
+    
+    agent_tools = AgentTools(agents=[researcher, writer, analyst])
+    tools = agent_tools.tools()  # No delegating_agent parameter
+    
+    assert len(tools) == 2
+    delegate_tool = tools[0]
+    assert len(delegate_tool.agents) == 3
+
+
+def test_allowed_agents_with_nonexistent_role(agents):
+    """Test allowed_agents with role that doesn't exist in available agents."""
+    manager, researcher, writer, analyst = agents
+    
+    manager.allowed_agents = ["Researcher", "NonExistentRole"]
+    
+    agent_tools = AgentTools(agents=[researcher, writer, analyst])
+    tools = agent_tools.tools(delegating_agent=manager)
+    
+    delegate_tool = tools[0]
+    assert len(delegate_tool.agents) == 1
+    assert researcher in delegate_tool.agents
+
+
+def test_allowed_agents_with_nonexistent_instance(agents):
+    """Test allowed_agents with agent instance that doesn't exist in available agents."""
+    manager, researcher, writer, analyst = agents
+    
+    other_agent = Agent(
+        role="Other",
+        goal="Other goal",
+        backstory="Other backstory"
+    )
+    
+    manager.allowed_agents = [researcher, other_agent]
+    
+    agent_tools = AgentTools(agents=[researcher, writer, analyst])
+    tools = agent_tools.tools(delegating_agent=manager)
+    
+    delegate_tool = tools[0]
+    assert len(delegate_tool.agents) == 1
+    assert researcher in delegate_tool.agents
+    assert other_agent not in delegate_tool.agents