feat: implement hierarchical agent delegation with allowed_agents parameter

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"
     )
@@ -136,6 +140,17 @@ class BaseAgent(ABC, BaseModel):
     def process_model_config(cls, values):
         return process_config(values, cls)
 
+    @field_validator("allowed_agents")
+    @classmethod
+    def validate_allowed_agents(cls, allowed_agents: Optional[List[str]]) -> Optional[List[str]]:
+        """Validate the allowed_agents parameter."""
+        if allowed_agents is not None:
+            if not isinstance(allowed_agents, list):
+                raise ValueError("allowed_agents must be a list of strings")
+            if not all(isinstance(agent, str) for agent in allowed_agents):
+                raise ValueError("all entries in allowed_agents must be strings")
+        return allowed_agents
+
     @field_validator("tools")
     @classmethod
     def validate_tools(cls, tools: List[Any]) -> List[BaseTool]:

src/crewai/tools/agent_tools/base_agent_tools.py

@@ -31,38 +31,43 @@ class BaseAgentTool(BaseTool):
             if agent_name is None:
                 agent_name = ""
 
-            # It is important to remove the quotes from the agent name.
-            # The reason we have to do this is because less-powerful LLM's
-            # have difficulty producing valid JSON.
-            # As a result, we end up with invalid JSON that is truncated like this:
-            # {"task": "....", "coworker": "....
-            # when it should look like this:
-            # {"task": "....", "coworker": "...."}
             agent_name = agent_name.casefold().replace('"', "").replace("\n", "")
-            agent = [  # type: ignore # Incompatible types in assignment (expression has type "list[BaseAgent]", variable has type "str | None")
+            available_agents = [
                 available_agent
                 for available_agent in self.agents
                 if available_agent.role.casefold().replace("\n", "") == agent_name
             ]
+
+            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 target_agent.role not in delegating_agent.allowed_agents:
+                    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)
+                    )
+
+            task_with_assigned_agent = Task(
+                description=task,
+                agent=target_agent,
+                expected_output=target_agent.i18n.slice("manager_request"),
+                i18n=target_agent.i18n,
+            )
+            return target_agent.execute_task(task_with_assigned_agent, context)
         except Exception as _:
             return self.i18n.errors("agent_tool_unexisting_coworker").format(
                 coworkers="\n".join(
                     [f"- {agent.role.casefold()}" for agent in self.agents]
                 )
             )
-
-        if not agent:
-            return self.i18n.errors("agent_tool_unexisting_coworker").format(
-                coworkers="\n".join(
-                    [f"- {agent.role.casefold()}" for agent in self.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,
-            agent=agent,
-            expected_output=agent.i18n.slice("manager_request"),
-            i18n=agent.i18n,
-        )
-        return agent.execute_task(task_with_assigned_agent, context)

src/crewai/translations/en.json

@@ -29,6 +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",
     "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.",

tests/tools/agent_tools/agent_tools_test.py

@@ -124,3 +124,60 @@ def test_ask_question_to_wrong_agent():
         result
         == "\nError executing tool. coworker mentioned not found, it must be one of the following options:\n- researcher\n"
     )
+
+
+@pytest.mark.vcr(filter_headers=["authorization"])
+def test_delegate_work_with_allowed_agents():
+    executive = Agent(
+        role="Executive Director",
+        goal="Lead the team effectively",
+        backstory="You're an experienced executive",
+        allow_delegation=True,
+        allowed_agents=["Communications Manager"]
+    )
+    
+    comms_manager = Agent(
+        role="Communications Manager",
+        goal="Manage communications",
+        backstory="You're a communications expert",
+        allow_delegation=False
+    )
+    
+    tools = AgentTools(agents=[executive, comms_manager]).tools()
+    delegate_tool = tools[0]
+    
+    result = delegate_tool.run(
+        coworker="Communications Manager",
+        task="Handle PR",
+        context="Important announcement"
+    )
+    assert "Error" not in result
+
+
+@pytest.mark.vcr(filter_headers=["authorization"])
+def test_delegate_work_with_unauthorized_agent():
+    executive = Agent(
+        role="Executive Director",
+        goal="Lead the team effectively",
+        backstory="You're an experienced executive",
+        allow_delegation=True,
+        allowed_agents=["Communications Manager"]
+    )
+    
+    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 "not authorized to delegate" in result