feat: add target_agents parameter to allow constrained delegation

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

tests/test_target_agents_delegation.py

@@ -0,0 +1,132 @@
+"""Test target_agents delegation functionality."""
+
+import pytest
+from crewai.agent import Agent
+from crewai.crew import Crew
+from crewai.task import Task
+from crewai.tools.agent_tools.agent_tools import AgentTools
+
+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,48 @@ 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."""
+    researcher = 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