Merge branch 'main' into devin/1735586761-fix-hierarchical-delegation

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

src/crewai/agent.py

@@ -406,8 +406,17 @@ class Agent(BaseAgent):
             callbacks=[TokenCalcHandler(self._token_process)],
         )
 
-    def get_delegation_tools(self, agents: List[BaseAgent]):
-        agent_tools = AgentTools(agents=agents)
+    def get_delegation_tools(self, agents: List[BaseAgent], task: Optional[Task] = None) -> List[BaseTool]:
+        """Get the delegation tools for this agent.
+        
+        Args:
+            agents: List of agents that can be delegated to
+            task: Optional task context for delegation
+            
+        Returns:
+            List of delegation tools
+        """
+        agent_tools = AgentTools(agents=agents, task=task, i18n=self.i18n)
         tools = agent_tools.tools()
         return tools
 

src/crewai/agents/agent_builder/base_agent.py

@@ -232,7 +232,7 @@ class BaseAgent(ABC, BaseModel):
         pass
 
     @abstractmethod
-    def get_delegation_tools(self, agents: List["BaseAgent"]) -> List[BaseTool]:
+    def get_delegation_tools(self, agents: List["BaseAgent"], task=None) -> List[BaseTool]:
         """Set the task tools that init BaseAgenTools class."""
         pass
 

src/crewai/crew.py

@@ -798,15 +798,34 @@ class Crew(BaseModel):
         return None
 
     def _prepare_tools(self, agent: BaseAgent, task: Task, tools: List[Tool]) -> List[Tool]:
+        """Prepare tools for an agent, including delegation tools if allowed.
+        
+        Args:
+            agent: Agent that will receive the tools
+            task: Task being executed
+            tools: List of existing tools
+            
+        Returns:
+            List of tools with delegation and other tools added
+        """
         # Add delegation tools if agent allows delegation
         if agent.allow_delegation:
             if self.process == Process.hierarchical:
                 if self.manager_agent:
-                    tools = self._update_manager_tools(task, tools)
+                    # For hierarchical process, handle both manager and regular agent tools
+                    if agent == self.manager_agent:
+                        # Manager can delegate to all regular agents
+                        tools = self._inject_delegation_tools(tools, agent, [a for a in self.agents if a != agent], task=task)
+                    elif task and task.async_execution and task.agent:
+                        # For async tasks in hierarchical mode, only allow delegation to the task's assigned agent
+                        tools = self._inject_delegation_tools(tools, agent, [task.agent], task=task)
+                    else:
+                        # Regular agents can delegate to manager and other agents
+                        delegation_agents = [self.manager_agent] + [a for a in self.agents if a != agent]
+                        tools = self._inject_delegation_tools(tools, agent, delegation_agents, task=task)
                 else:
                     raise ValueError("Manager agent is required for hierarchical process.")
-
-            elif agent and agent.allow_delegation:
+            else:
                 tools = self._add_delegation_tools(task, tools)
 
         # Add code execution tools if agent allows code execution
@@ -839,8 +858,19 @@ class Crew(BaseModel):
 
         return tools
 
-    def _inject_delegation_tools(self, tools: List[Tool], task_agent: BaseAgent, agents: List[BaseAgent]):
-        delegation_tools = task_agent.get_delegation_tools(agents)
+    def _inject_delegation_tools(self, tools: List[Tool], task_agent: BaseAgent, agents: List[BaseAgent], task=None):
+        """Inject delegation tools for the given agent.
+        
+        Args:
+            tools: List of existing tools
+            task_agent: Agent that will receive the delegation tools
+            agents: List of agents that can be delegated to
+            task: Optional task context for delegation
+            
+        Returns:
+            List of tools with delegation tools added
+        """
+        delegation_tools = task_agent.get_delegation_tools(agents, task=task)
         return self._merge_tools(tools, delegation_tools)
 
     def _add_multimodal_tools(self, agent: BaseAgent, tools: List[Tool]):
@@ -852,8 +882,17 @@ class Crew(BaseModel):
         return self._merge_tools(tools, code_tools)
 
     def _add_delegation_tools(self, task: Task, tools: List[Tool]):
-        agents_for_delegation = [agent for agent in self.agents if agent != task.agent]
-        if len(self.agents) > 1 and len(agents_for_delegation) > 0 and task.agent:
+        agents_for_delegation = []
+        if self.process == Process.hierarchical and self.manager_agent:
+            # In hierarchical mode, allow delegation to manager if the task agent isn't the manager
+            if task.agent != self.manager_agent:
+                agents_for_delegation = [self.manager_agent]
+            # Also include regular agents for delegation
+            agents_for_delegation.extend([agent for agent in self.agents if agent != task.agent])
+        else:
+            agents_for_delegation = [agent for agent in self.agents if agent != task.agent]
+        
+        if len(agents_for_delegation) > 0 and task.agent:
             if not tools:
                 tools = []
             tools = self._inject_delegation_tools(tools, task.agent, agents_for_delegation)
@@ -868,9 +907,23 @@ class Crew(BaseModel):
     def _update_manager_tools(self, task: Task, tools: List[Tool]):
         if self.manager_agent:
             if task.agent:
-                tools = self._inject_delegation_tools(tools, task.agent, [task.agent])
+                # In hierarchical mode with async execution, only allow delegation to the task's assigned agent
+                if task.async_execution and self.process == Process.hierarchical:
+                    tools = self._inject_delegation_tools(tools, task.agent, [task.agent] if task.agent else [])
+                else:
+                    # In hierarchical mode, allow bidirectional delegation
+                    if self.process == Process.hierarchical:
+                        # For non-manager agents, allow delegation to manager and other agents
+                        if task.agent != self.manager_agent:
+                            delegation_agents = [self.manager_agent] + self.agents
+                            tools = self._inject_delegation_tools(tools, task.agent, [a for a in delegation_agents if a != task.agent])
+                        else:
+                            # For manager, allow delegation to all agents
+                            tools = self._inject_delegation_tools(tools, task.agent, [a for a in self.agents if a != task.agent])
+                    else:
+                        tools = self._inject_delegation_tools(tools, task.agent, self.agents, task=task)
             else:
-                tools = self._inject_delegation_tools(tools, self.manager_agent, self.agents)
+                tools = self._inject_delegation_tools(tools, self.manager_agent, self.agents, task=task)
         return tools
 
     def _get_context(self, task: Task, task_outputs: List[TaskOutput]):

src/crewai/tools/agent_tools/agent_tools.py

@@ -9,24 +9,36 @@ from .delegate_work_tool import DelegateWorkTool
 class AgentTools:
     """Manager class for agent-related tools"""
 
-    def __init__(self, agents: list[BaseAgent], i18n: I18N = I18N()):
+    def __init__(self, agents: list[BaseAgent], i18n: I18N = I18N(), task=None):
         self.agents = agents
         self.i18n = i18n
+        self.task = task
 
     def tools(self) -> list[BaseTool]:
         """Get all available agent tools"""
-        coworkers = ", ".join([f"{agent.role}" for agent in self.agents])
+        # Format coworkers list based on agents and task context
+        if len(self.agents) == 1:
+            coworkers = self.agents[0].role
+        elif self.task and hasattr(self.task, 'async_execution') and self.task.async_execution and hasattr(self.task, 'agent') and self.task.agent:
+            # For async tasks with a specific agent, only show that agent
+            coworkers = self.task.agent.role
+        else:
+            # Show all agents for non-async tasks or when no specific agent is assigned
+            coworkers = ", ".join([agent.role for agent in self.agents])
+
+        # Ensure coworkers list doesn't have extra spaces or newlines
+        coworkers = coworkers.strip()
 
         delegate_tool = DelegateWorkTool(
             agents=self.agents,
             i18n=self.i18n,
-            description=self.i18n.tools("delegate_work").format(coworkers=coworkers),  # type: ignore
+            description=f"Delegate a specific task to one of the following coworkers: {coworkers}\n",
         )
 
         ask_tool = AskQuestionTool(
             agents=self.agents,
             i18n=self.i18n,
-            description=self.i18n.tools("ask_question").format(coworkers=coworkers),  # type: ignore
+            description=f"Ask a specific question to one of the following coworkers: {coworkers}\n",
         )
 
         return [delegate_tool, ask_tool]

tests/crew_test.py

@@ -43,7 +43,7 @@ writer = Agent(
     role="Senior Writer",
     goal="Write the best content about AI and AI agents.",
     backstory="You're a senior writer, specialized in technology, software engineering, AI and startups. You work as a freelancer and are now working on writing content for a new customer.",
-    allow_delegation=False,
+    allow_delegation=True,
 )
 
 

tests/test_hierarchical_delegation.py

@@ -0,0 +1,123 @@
+from unittest.mock import MagicMock
+
+import pytest
+from langchain_core.language_models.base import BaseLanguageModel
+
+from crewai import Agent, Crew, Process, Task
+
+
+def test_hierarchical_delegation_tool_availability():
+    """Test that all agents are available for delegation in hierarchical mode."""
+    # Mock LLM to avoid actual API calls
+    mock_llm = MagicMock(spec=BaseLanguageModel)
+    
+    # Create agents
+    manager = Agent(
+        role="Manager",
+        goal="Manage the team",
+        backstory="I am a manager",
+        allow_delegation=True,
+        llm=mock_llm
+    )
+    
+    kb_agent = Agent(
+        role="kb_retriever_agent",
+        goal="Retrieve knowledge",
+        backstory="I am a knowledge retrieval specialist",
+        allow_delegation=True,
+        llm=mock_llm
+    )
+    
+    worker = Agent(
+        role="Worker",
+        goal="Do the work",
+        backstory="I am a worker",
+        allow_delegation=True,
+        llm=mock_llm
+    )
+    
+    # Create a task assigned to the manager
+    task = Task(
+        description="Complex task requiring delegation",
+        expected_output="Task completion status",
+        agent=manager
+    )
+    
+    # Create the crew with hierarchical process
+    crew = Crew(
+        agents=[kb_agent, worker],  # Manager should not be in agents list when using hierarchical process
+        tasks=[task],
+        process=Process.hierarchical,
+        manager_agent=manager  # Explicitly set the manager agent for hierarchical process
+    )
+    
+    # Get the manager's tools
+    tools_for_task = task.tools or manager.tools or []
+    manager_tools = crew._prepare_tools(manager, task, tools_for_task)
+    
+    # Find delegation tools
+    delegation_tools = [tool for tool in manager_tools if tool.name == "Delegate work to coworker"]
+    assert len(delegation_tools) > 0, "Delegation tool should be present"
+    
+    # Get the delegation tool description
+    delegate_tool = delegation_tools[0]
+    tool_description = str(delegate_tool.description)
+    
+    # Verify all agents are available for delegation
+    assert "kb_retriever_agent" in tool_description, "kb_retriever_agent should be available for delegation"
+    assert "Worker" in tool_description, "Worker should be available for delegation"
+
+def test_hierarchical_delegation_tool_updates():
+    """Test that delegation tools are properly updated when task agent changes."""
+    mock_llm = MagicMock(spec=BaseLanguageModel)
+    
+    manager = Agent(
+        role="Manager",
+        goal="Manage the team",
+        backstory="I am a manager",
+        allow_delegation=True,
+        llm=mock_llm
+    )
+    
+    kb_agent = Agent(
+        role="kb_retriever_agent",
+        goal="Retrieve knowledge",
+        backstory="I am a knowledge retrieval specialist",
+        allow_delegation=True,
+        llm=mock_llm
+    )
+    
+    # Create tasks for different agents
+    manager_task = Task(
+        description="Manager task",
+        expected_output="Manager task completion status",
+        agent=manager
+    )
+    
+    kb_task = Task(
+        description="KB task",
+        expected_output="KB task completion status",
+        agent=kb_agent
+    )
+    
+    # Create crew
+    crew = Crew(
+        agents=[kb_agent],  # Manager should not be in agents list when using hierarchical process
+        tasks=[manager_task, kb_task],
+        process=Process.hierarchical,
+        manager_agent=manager  # Explicitly set the manager agent for hierarchical process
+    )
+    
+    # Test manager's tools
+    manager_tools_for_task = manager_task.tools or manager.tools or []
+    manager_tools = crew._prepare_tools(manager, manager_task, manager_tools_for_task)
+    manager_delegation = [t for t in manager_tools if t.name == "Delegate work to coworker"]
+    assert len(manager_delegation) > 0, "Manager should have delegation tool"
+    assert "kb_retriever_agent" in str(manager_delegation[0].description), "Manager should see kb_retriever_agent"
+    
+    # Test kb_agent's tools
+    kb_tools_for_task = kb_task.tools or kb_agent.tools or []
+    kb_tools = crew._prepare_tools(kb_agent, kb_task, kb_tools_for_task)
+    kb_delegation = [t for t in kb_tools if t.name == "Delegate work to coworker"]
+    assert len(kb_delegation) > 0, "KB agent should have delegation tool"
+    assert "Manager" in str(kb_delegation[0].description), "KB agent should see manager"