refactor: optimize agent lookup and improve error handling

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

src/crewai/agents/agent_builder/base_agent.py

@@ -109,7 +109,9 @@ class BaseAgent(ABC, BaseModel):
     )
     allowed_agents: Optional[List[str]] = Field(
         default=None,
-        description="List of agent roles that this agent is allowed to delegate tasks to.",
+        description="List of agent roles that this agent is allowed to delegate tasks to. "
+                   "If None, delegation is unrestricted when allow_delegation is True.",
+        examples=["Researcher", "Analyst"]
     )
     tools: Optional[List[Any]] = Field(
         default_factory=list, description="Tools at agents' disposal"

src/crewai/exceptions.py

@@ -0,0 +1,10 @@
+"""Exceptions for CrewAI."""
+
+class AgentLookupError(Exception):
+    """Exception raised when an agent cannot be found."""
+    pass
+
+
+class UnauthorizedDelegationError(Exception):
+    """Exception raised when an agent attempts unauthorized delegation."""
+    pass

src/crewai/tools/agent_tools/base_agent_tools.py

@@ -3,12 +3,14 @@ from typing import Optional, Union
 from pydantic import Field
 
 from crewai.agents.agent_builder.base_agent import BaseAgent
+from crewai.exceptions import AgentLookupError, UnauthorizedDelegationError
 from crewai.task import Task
 from crewai.tools.base_tool import BaseTool
 from crewai.utilities import I18N
+from crewai.utilities.agent_lookup import AgentLookupMixin
 
 
-class BaseAgentTool(BaseTool):
+class BaseAgentTool(BaseTool, AgentLookupMixin):
     """Base class for agent-related tools"""
 
     agents: list[BaseAgent] = Field(description="List of available agents")
@@ -24,6 +26,20 @@ class BaseAgentTool(BaseTool):
                 coworker = coworker[1:-1].split(",")[0]
         return coworker
 
+    def can_delegate_to(self, delegating_agent: BaseAgent, target_agent: BaseAgent) -> bool:
+        """Check if an agent can delegate to another agent.
+        
+        Args:
+            delegating_agent: The agent attempting to delegate
+            target_agent: The agent being delegated to
+            
+        Returns:
+            bool: True if delegation is allowed, False otherwise
+        """
+        return (delegating_agent.allow_delegation and 
+                (not delegating_agent.allowed_agents or 
+                 target_agent.role in delegating_agent.allowed_agents))
+
     def _execute(
         self, agent_name: Union[str, None], task: str, context: Union[str, None]
     ) -> str:
@@ -31,32 +47,21 @@ class BaseAgentTool(BaseTool):
             if agent_name is None:
                 agent_name = ""
 
-            agent_name = agent_name.casefold().replace('"', "").replace("\n", "")
+            target_agent = self.get_agent_by_role(agent_name, self.agents)
-            available_agents = [
+            if not target_agent:
-                available_agent
+                raise AgentLookupError(
-                for available_agent in self.agents
+                    f"Agent with role '{agent_name}' not found. Available agents: "
-                if available_agent.role.casefold().replace("\n", "") == agent_name
+                    f"{', '.join(agent.role for agent in self.agents)}"
-            ]
-
-            if not available_agents:
-                return self.i18n.errors("agent_tool_unexisting_coworker").format(
-                    coworkers="\n".join(
-                        [f"- {agent.role.casefold()}" for agent in self.agents]
-                    )
                 )
-
-            target_agent = available_agents[0]
             delegating_agent = next(
                 (agent for agent in self.agents if agent.allow_delegation), None
             )
 
-            if delegating_agent and delegating_agent.allowed_agents:
+            if delegating_agent and not self.can_delegate_to(delegating_agent, target_agent):
-                if target_agent.role not in delegating_agent.allowed_agents:
+                raise UnauthorizedDelegationError(
-                    return self.i18n.errors("agent_tool_unauthorized_delegation").format(
+                    f"Agent '{delegating_agent.role}' cannot delegate to '{target_agent.role}'. "
-                        agent=delegating_agent.role,
+                    f"Allowed targets: {', '.join(delegating_agent.allowed_agents or [])}"
-                        target=target_agent.role,
+                )
-                        allowed="\n".join(f"- {role}" for role in delegating_agent.allowed_agents)
-                    )
 
             task_with_assigned_agent = Task(
                 description=task,
@@ -65,9 +70,15 @@ class BaseAgentTool(BaseTool):
                 i18n=target_agent.i18n,
             )
             return target_agent.execute_task(task_with_assigned_agent, context)
-        except Exception as _:
+        except AgentLookupError as e:
             return self.i18n.errors("agent_tool_unexisting_coworker").format(
-                coworkers="\n".join(
+                coworkers="\n".join(f"- {agent.role}" for agent in self.agents)
-                    [f"- {agent.role.casefold()}" for agent in self.agents]
-                )
             )
+        except UnauthorizedDelegationError as e:
+            return self.i18n.errors("agent_tool_unauthorized_delegation").format(
+                agent=delegating_agent.role,
+                target=target_agent.role,
+                allowed="\n".join(f"- {role}" for role in (delegating_agent.allowed_agents or []))
+            )
+        except Exception as e:
+            return self.i18n.errors("tool_usage_error").format(error=str(e))

src/crewai/translations/en.json

@@ -29,7 +29,7 @@
     "force_final_answer_error": "You can't keep going, this was the best you could do.\n {formatted_answer.text}",
     "force_final_answer": "Now it's time you MUST give your absolute best final answer. You'll ignore all previous instructions, stop using any tools, and just return your absolute BEST Final answer.",
     "agent_tool_unexisting_coworker": "\nError executing tool. coworker mentioned not found, it must be one of the following options:\n{coworkers}\n",
-    "agent_tool_unauthorized_delegation": "\nError executing tool. Agent '{agent}' is not authorized to delegate to '{target}'. Allowed agents are:\n{allowed}\n",
+    "agent_tool_unauthorized_delegation": "\nAuthorization Error: Agent '{agent}' cannot delegate to '{target}'.\nAllowed delegation targets:\n{allowed}\nPlease check agent configuration or contact administrator.\n",
     "task_repeated_usage": "I tried reusing the same input, I must stop using this action input. I'll try something else instead.\n\n",
     "tool_usage_error": "I encountered an error: {error}",
     "tool_arguments_error": "Error: the Action Input is not a valid key, value dictionary.",

src/crewai/utilities/agent_lookup.py

@@ -0,0 +1,26 @@
+"""Agent lookup utilities for CrewAI."""
+
+from typing import List, Optional, Union
+from ..agents.agent_builder.base_agent import BaseAgent
+from ..exceptions import AgentLookupError
+
+
+class AgentLookupMixin:
+    """Mixin class for agent lookup functionality."""
+    
+    def get_agent_by_role(self, role: str, agents: List[BaseAgent]) -> Union[BaseAgent, None]:
+        """Find an agent by role, case-insensitive.
+        
+        Args:
+            role: The role to search for
+            agents: List of agents to search through
+            
+        Returns:
+            The found agent or None
+        """
+        normalized_role = role.casefold().replace('"', "").replace("\n", "")
+        return next(
+            (agent for agent in agents 
+             if agent.role.casefold().replace("\n", "") == normalized_role),
+            None
+        )

tests/tools/agent_tools/agent_tools_test.py

@@ -126,6 +126,65 @@ def test_ask_question_to_wrong_agent():
     )
 
 
+@pytest.mark.vcr(filter_headers=["authorization"])
+def test_delegate_work_with_null_allowed_agents():
+    """Test delegation when allowed_agents is None (unrestricted delegation)."""
+    executive = Agent(
+        role="Executive Director",
+        goal="Lead the team effectively",
+        backstory="You're an experienced executive",
+        allow_delegation=True,
+        allowed_agents=None
+    )
+    
+    research_manager = Agent(
+        role="Research Manager",
+        goal="Manage research",
+        backstory="You're a research expert",
+        allow_delegation=False
+    )
+    
+    tools = AgentTools(agents=[executive, research_manager]).tools()
+    delegate_tool = tools[0]
+    
+    result = delegate_tool.run(
+        coworker="Research Manager",
+        task="Handle research",
+        context="Important research"
+    )
+    assert "Error" not in result
+
+
+@pytest.mark.vcr(filter_headers=["authorization"])
+def test_delegate_work_with_empty_allowed_agents():
+    """Test delegation when allowed_agents is an empty list (no delegation allowed)."""
+    executive = Agent(
+        role="Executive Director",
+        goal="Lead the team effectively",
+        backstory="You're an experienced executive",
+        allow_delegation=True,
+        allowed_agents=[]
+    )
+    
+    research_manager = Agent(
+        role="Research Manager",
+        goal="Manage research",
+        backstory="You're a research expert",
+        allow_delegation=False
+    )
+    
+    tools = AgentTools(agents=[executive, research_manager]).tools()
+    delegate_tool = tools[0]
+    
+    result = delegate_tool.run(
+        coworker="Research Manager",
+        task="Handle research",
+        context="Important research"
+    )
+    assert "Error" in result
+    assert "Authorization Error" in result
+
+
 @pytest.mark.vcr(filter_headers=["authorization"])
 def test_delegate_work_with_allowed_agents():
     executive = Agent(