fixing tests for delegations and coding

src/crewai/crew.py

@@ -534,9 +534,6 @@ class Crew(BaseModel):
             if not agent.function_calling_llm:  # type: ignore # "BaseAgent" has no attribute "function_calling_llm"
                 agent.function_calling_llm = self.function_calling_llm  # type: ignore # "BaseAgent" has no attribute "function_calling_llm"
 
-            if agent.allow_code_execution:  # type: ignore # BaseAgent" has no attribute "allow_code_execution"
-                agent.tools += agent.get_code_execution_tools()  # type: ignore # "BaseAgent" has no attribute "get_code_execution_tools"; maybe "get_delegation_tools"?
-
             if not agent.step_callback:  # type: ignore # "BaseAgent" has no attribute "step_callback"
                 agent.step_callback = self.step_callback  # type: ignore # "BaseAgent" has no attribute "step_callback"
 
@@ -684,6 +681,7 @@ class Crew(BaseModel):
                 goal=i18n.retrieve("hierarchical_manager_agent", "goal"),
                 backstory=i18n.retrieve("hierarchical_manager_agent", "backstory"),
                 tools=AgentTools(agents=self.agents).tools(),
+                allow_delegation=True,
                 llm=self.manager_llm,
                 verbose=self.verbose,
             )
@@ -726,9 +724,16 @@ class Crew(BaseModel):
                     f"No agent available for task: {task.description}. Ensure that either the task has an assigned agent or a manager agent is provided."
                 )
 
+            # Determine which tools to use - task tools take precedence over agent tools
+            tools_for_task = task.tools if task.tools else agent_to_use.tools or []
+            # Add delegation tools if agent allows delegation
+            if agent_to_use.allow_delegation:
+                tools_for_task = self._prepare_tools(task, tools_for_task)
+
+            # Add code execution tools if agent allows code execution
+            if agent_to_use.allow_code_execution:
+                tools_for_task += agent_to_use.get_code_execution_tools()
 
-            tools_for_task = agent_to_use.tools or task.tools or []
-            tools_for_task = self._prepare_tools(task, tools_for_task)
             self._log_task_start(task, agent_to_use.role)
 
             if isinstance(task, ConditionalTask):

tests/crew_test.py

@@ -579,6 +579,80 @@ def test_task_tools_override_agent_tools():
     assert len(new_researcher.tools) == 1
     assert isinstance(new_researcher.tools[0], TestTool)
 
+@pytest.mark.vcr(filter_headers=["authorization"])
+def test_task_tools_override_agent_tools_with_allow_delegation():
+    """
+    Test that task tools override agent tools while preserving delegation tools when allow_delegation=True
+    """
+    from typing import Type
+    from crewai.tools import BaseTool
+    from pydantic import BaseModel, Field
+
+    class TestToolInput(BaseModel):
+        query: str = Field(..., description="Query to process")
+
+    class TestTool(BaseTool):
+        name: str = "Test Tool"
+        description: str = "A test tool that just returns the input"
+        args_schema: Type[BaseModel] = TestToolInput
+
+        def _run(self, query: str) -> str:
+            return f"Processed: {query}"
+
+    class AnotherTestTool(BaseTool):
+        name: str = "Another Test Tool"
+        description: str = "Another test tool"
+        args_schema: Type[BaseModel] = TestToolInput
+
+        def _run(self, query: str) -> str:
+            return f"Another processed: {query}"
+
+    # Set up agents with tools and allow_delegation
+    researcher_with_delegation = researcher.model_copy()
+    researcher_with_delegation.allow_delegation = True
+    researcher_with_delegation.tools = [TestTool()]
+
+    writer_for_delegation = writer.model_copy()
+
+    # Create a task with different tools
+    task = Task(
+        description="Write a test task",
+        expected_output="Test output",
+        agent=researcher_with_delegation,
+        tools=[AnotherTestTool()],
+    )
+
+    crew = Crew(
+        agents=[researcher_with_delegation, writer_for_delegation],
+        tasks=[task],
+        process=Process.sequential,
+    )
+
+    mock_task_output = TaskOutput(
+        description="Mock description",
+        raw="mocked output",
+        agent="mocked agent"
+    )
+
+    # We mock execute_sync to verify which tools get used at runtime
+    with patch.object(Task, "execute_sync", return_value=mock_task_output) as mock_execute_sync:
+        crew.kickoff()
+
+        # Inspect the call kwargs to verify the actual tools passed to execution
+        _, kwargs = mock_execute_sync.call_args
+        used_tools = kwargs["tools"]
+
+        # Confirm AnotherTestTool is present but TestTool is not
+        assert any(isinstance(tool, AnotherTestTool) for tool in used_tools), "AnotherTestTool should be present"
+        assert not any(isinstance(tool, TestTool) for tool in used_tools), "TestTool should not be present among used tools"
+
+        # Confirm delegation tool(s) are present
+        assert any("delegate" in tool.name.lower() for tool in used_tools), "Delegation tool should be present"
+
+    # Finally, make sure the agent's original tools remain unchanged
+    assert len(researcher_with_delegation.tools) == 1
+    assert isinstance(researcher_with_delegation.tools[0], TestTool)
+
 @pytest.mark.vcr(filter_headers=["authorization"])
 def test_crew_verbose_output(capsys):
     tasks = [
@@ -1366,11 +1440,22 @@ def test_code_execution_flag_adds_code_tool_upon_kickoff():
 
     crew = Crew(agents=[programmer], tasks=[task])
 
-    with patch.object(Agent, "execute_task") as executor:
+    mock_task_output = TaskOutput(
-        executor.return_value = "ok"
+        description="Mock description",
+        raw="mocked output",
+        agent="mocked agent"
+    )
+
+    with patch.object(Task, "execute_sync", return_value=mock_task_output) as mock_execute_sync:
         crew.kickoff()
-        assert len(programmer.tools) == 1
+
-        assert programmer.tools[0].__class__ == CodeInterpreterTool
+        # Get the tools that were actually used in execution
+        _, kwargs = mock_execute_sync.call_args
+        used_tools = kwargs["tools"]
+
+        # Verify that exactly one tool was used and it was a CodeInterpreterTool
+        assert len(used_tools) == 1, "Should have exactly one tool"
+        assert isinstance(used_tools[0], CodeInterpreterTool), "Tool should be CodeInterpreterTool"
 
 @pytest.mark.vcr(filter_headers=["authorization"])
 def test_delegation_is_not_enabled_if_there_are_only_one_agent():
@@ -2792,3 +2877,78 @@ def test_hierarchical_verbose_false_manager_agent():
 
     assert crew.manager_agent is not None
     assert not crew.manager_agent.verbose
+
+
+def test_task_tools_preserve_code_execution_tools():
+    """
+    Test that task tools don't override code execution tools when allow_code_execution=True
+    """
+    from typing import Type
+    from crewai.tools import BaseTool
+    from crewai_tools import CodeInterpreterTool
+    from pydantic import BaseModel, Field
+
+    class TestToolInput(BaseModel):
+        """Input schema for TestTool."""
+        query: str = Field(..., description="Query to process")
+
+    class TestTool(BaseTool):
+        name: str = "Test Tool"
+        description: str = "A test tool that just returns the input"
+        args_schema: Type[BaseModel] = TestToolInput
+
+        def _run(self, query: str) -> str:
+            return f"Processed: {query}"
+
+    # Create a programmer agent with code execution enabled
+    programmer = Agent(
+        role="Programmer",
+        goal="Write code to solve problems.",
+        backstory="You're a programmer who loves to solve problems with code.",
+        allow_delegation=True,
+        allow_code_execution=True,
+    )
+
+    # Create a code reviewer agent
+    reviewer = Agent(
+        role="Code Reviewer",
+        goal="Review code for bugs and improvements",
+        backstory="You're an experienced code reviewer who ensures code quality and best practices.",
+        allow_delegation=True,
+        allow_code_execution=True,
+    )
+
+    # Create a task with its own tools
+    task = Task(
+        description="Write a program to calculate fibonacci numbers.",
+        expected_output="A working fibonacci calculator.",
+        agent=programmer,
+        tools=[TestTool()]
+    )
+
+    crew = Crew(
+        agents=[programmer, reviewer],
+        tasks=[task],
+        process=Process.sequential,
+    )
+
+    mock_task_output = TaskOutput(
+        description="Mock description",
+        raw="mocked output",
+        agent="mocked agent"
+    )
+
+    with patch.object(Task, "execute_sync", return_value=mock_task_output) as mock_execute_sync:
+        crew.kickoff()
+
+        # Get the tools that were actually used in execution
+        _, kwargs = mock_execute_sync.call_args
+        used_tools = kwargs["tools"]
+
+        # Verify all expected tools are present
+        assert any(isinstance(tool, TestTool) for tool in used_tools), "Task's TestTool should be present"
+        assert any(isinstance(tool, CodeInterpreterTool) for tool in used_tools), "CodeInterpreterTool should be present"
+        assert any("delegate" in tool.name.lower() for tool in used_tools), "Delegation tool should be present"
+
+        # Verify the total number of tools (TestTool + CodeInterpreter + 2 delegation tools)
+        assert len(used_tools) == 4, "Should have TestTool, CodeInterpreter, and 2 delegation tools"