fix: remove unused imports and variables to fix lint issues

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

src/crewai/agent.py

@@ -155,6 +155,10 @@ class Agent(BaseAgent):
         default=None,
         description="The Agent's role to be used from your repository.",
     )
+    target_agents: Optional[List[str]] = Field(
+        default=None,
+        description="List of agent roles that this agent can delegate tasks to. If None, can delegate to all agents.",
+    )
 
     @model_validator(mode="before")
     def validate_from_repository(cls, v):
@@ -549,6 +553,18 @@ class Agent(BaseAgent):
         )
 
     def get_delegation_tools(self, agents: List[BaseAgent]):
+        if hasattr(self, 'target_agents') and self.target_agents is not None:
+            filtered_agents = [
+                agent for agent in agents 
+                if agent.role in self.target_agents
+            ]
+            if len(filtered_agents) < len(self.target_agents):
+                available_roles = [agent.role for agent in agents]
+                missing_roles = set(self.target_agents) - set(available_roles)
+                if hasattr(self, '_logger'):
+                    self._logger.log("warning", f"Some target agents not found for delegation: {missing_roles}")
+            agents = filtered_agents
+        
         agent_tools = AgentTools(agents=agents)
         tools = agent_tools.tools()
         return tools

tests/test_target_agents_delegation.py

@@ -0,0 +1,128 @@
+"""Test target_agents delegation functionality."""
+
+from crewai.agent import Agent
+
+def test_target_agents_filters_delegation_tools():
+    """Test that target_agents properly filters available agents for delegation."""
+    researcher = Agent(
+        role="researcher",
+        goal="Research topics",
+        backstory="Expert researcher",
+        allow_delegation=True,
+        target_agents=["writer"]
+    )
+    
+    writer = Agent(
+        role="writer", 
+        goal="Write content",
+        backstory="Expert writer",
+        allow_delegation=False
+    )
+    
+    analyst = Agent(
+        role="analyst",
+        goal="Analyze data", 
+        backstory="Expert analyst",
+        allow_delegation=False
+    )
+    
+    tools = researcher.get_delegation_tools([writer, analyst])
+    delegate_tool = tools[0]
+    
+    result = delegate_tool.run(
+        coworker="writer",
+        task="Write an article",
+        context="About AI"
+    )
+    assert "Error executing tool" not in result
+    
+    result = delegate_tool.run(
+        coworker="analyst", 
+        task="Analyze data",
+        context="Sales data"
+    )
+    assert "Error executing tool" in result
+    assert "analyst" not in result.lower() or "not found" in result.lower()
+
+def test_target_agents_none_allows_all():
+    """Test that target_agents=None allows delegation to all agents."""
+    researcher = Agent(
+        role="researcher",
+        goal="Research topics", 
+        backstory="Expert researcher",
+        allow_delegation=True,
+        target_agents=None  # Should allow all
+    )
+    
+    writer = Agent(
+        role="writer",
+        goal="Write content",
+        backstory="Expert writer"
+    )
+    
+    analyst = Agent(
+        role="analyst", 
+        goal="Analyze data",
+        backstory="Expert analyst"
+    )
+    
+    tools = researcher.get_delegation_tools([writer, analyst])
+    delegate_tool = tools[0]
+    
+    result1 = delegate_tool.run(coworker="writer", task="Write", context="test")
+    result2 = delegate_tool.run(coworker="analyst", task="Analyze", context="test")
+    
+    assert "Error executing tool" not in result1
+    assert "Error executing tool" not in result2
+
+def test_target_agents_empty_list_blocks_all():
+    """Test that target_agents=[] blocks delegation to all agents."""
+    researcher = Agent(
+        role="researcher",
+        goal="Research topics",
+        backstory="Expert researcher", 
+        allow_delegation=True,
+        target_agents=[]  # Should block all
+    )
+    
+    writer = Agent(
+        role="writer",
+        goal="Write content", 
+        backstory="Expert writer"
+    )
+    
+    tools = researcher.get_delegation_tools([writer])
+    delegate_tool = tools[0]
+    
+    result = delegate_tool.run(
+        coworker="writer",
+        task="Write an article", 
+        context="About AI"
+    )
+    assert "Error executing tool" in result
+
+def test_target_agents_with_invalid_names():
+    """Test behavior when target_agents contains invalid agent names."""
+    researcher = Agent(
+        role="researcher",
+        goal="Research topics",
+        backstory="Expert researcher",
+        allow_delegation=True, 
+        target_agents=["writer", "nonexistent_agent"]
+    )
+    
+    writer = Agent(
+        role="writer",
+        goal="Write content",
+        backstory="Expert writer" 
+    )
+    
+    tools = researcher.get_delegation_tools([writer])
+    delegate_tool = tools[0]
+    
+    result = delegate_tool.run(
+        coworker="writer",
+        task="Write an article",
+        context="About AI"
+    )
+    assert "Error executing tool" not in result

tests/tools/agent_tools/agent_tools_test.py

@@ -131,3 +131,49 @@ 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"
     )
+
+def test_target_agents_delegation_filtering():
+    """Test that target_agents properly filters delegation targets."""
+    # that an agent with target_agents=["writer"] would only delegate to writer
+    Agent(
+        role="researcher",
+        goal="make the best research and analysis on content about AI and AI agents",
+        backstory="You're an expert researcher, specialized in technology",
+        allow_delegation=True,
+        target_agents=["writer"]  # Can only delegate to writer
+    )
+    
+    writer = Agent(
+        role="writer", 
+        goal="Write engaging content",
+        backstory="You're an expert writer",
+        allow_delegation=False
+    )
+    
+    analyst = Agent(
+        role="analyst",
+        goal="Analyze data trends", 
+        backstory="You're an expert analyst",
+        allow_delegation=False
+    )
+    
+    tools = AgentTools(agents=[writer]).tools()  # Only writer available
+    delegate_tool = tools[0]
+    
+    result = delegate_tool.run(
+        coworker="writer",
+        task="Write about AI trends", 
+        context="Latest developments"
+    )
+    assert "Error executing tool" not in result
+    
+    tools_with_analyst = AgentTools(agents=[analyst]).tools()
+    delegate_tool_analyst = tools_with_analyst[0]
+    
+    result = delegate_tool_analyst.run(
+        coworker="analyst",
+        task="Analyze trends",
+        context="Data analysis"
+    )
+    # This should work since we're passing analyst directly to AgentTools
+    assert "Error executing tool" not in result