diff --git a/src/crewai/agents/agent_builder/base_agent.py b/src/crewai/agents/agent_builder/base_agent.py index f3aae7f83..b096483f7 100644 --- a/src/crewai/agents/agent_builder/base_agent.py +++ b/src/crewai/agents/agent_builder/base_agent.py @@ -109,7 +109,9 @@ class BaseAgent(ABC, BaseModel): ) allowed_agents: Optional[List[str]] = Field( default=None, - description="List of agent roles that this agent is allowed to delegate tasks to.", + description="List of agent roles that this agent is allowed to delegate tasks to. " + "If None, delegation is unrestricted when allow_delegation is True.", + examples=["Researcher", "Analyst"] ) tools: Optional[List[Any]] = Field( default_factory=list, description="Tools at agents' disposal" diff --git a/src/crewai/exceptions.py b/src/crewai/exceptions.py new file mode 100644 index 000000000..69b34eb1a --- /dev/null +++ b/src/crewai/exceptions.py @@ -0,0 +1,10 @@ +"""Exceptions for CrewAI.""" + +class AgentLookupError(Exception): + """Exception raised when an agent cannot be found.""" + pass + + +class UnauthorizedDelegationError(Exception): + """Exception raised when an agent attempts unauthorized delegation.""" + pass diff --git a/src/crewai/tools/agent_tools/base_agent_tools.py b/src/crewai/tools/agent_tools/base_agent_tools.py index ca6ff956a..6ed24c371 100644 --- a/src/crewai/tools/agent_tools/base_agent_tools.py +++ b/src/crewai/tools/agent_tools/base_agent_tools.py @@ -3,12 +3,14 @@ from typing import Optional, Union from pydantic import Field from crewai.agents.agent_builder.base_agent import BaseAgent +from crewai.exceptions import AgentLookupError, UnauthorizedDelegationError from crewai.task import Task from crewai.tools.base_tool import BaseTool from crewai.utilities import I18N +from crewai.utilities.agent_lookup import AgentLookupMixin -class BaseAgentTool(BaseTool): +class BaseAgentTool(BaseTool, AgentLookupMixin): """Base class for agent-related tools""" agents: list[BaseAgent] = Field(description="List of available agents") @@ -24,6 +26,20 @@ class BaseAgentTool(BaseTool): coworker = coworker[1:-1].split(",")[0] return coworker + def can_delegate_to(self, delegating_agent: BaseAgent, target_agent: BaseAgent) -> bool: + """Check if an agent can delegate to another agent. + + Args: + delegating_agent: The agent attempting to delegate + target_agent: The agent being delegated to + + Returns: + bool: True if delegation is allowed, False otherwise + """ + return (delegating_agent.allow_delegation and + (not delegating_agent.allowed_agents or + target_agent.role in delegating_agent.allowed_agents)) + def _execute( self, agent_name: Union[str, None], task: str, context: Union[str, None] ) -> str: @@ -31,32 +47,21 @@ class BaseAgentTool(BaseTool): if agent_name is None: agent_name = "" - agent_name = agent_name.casefold().replace('"', "").replace("\n", "") - 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 = self.get_agent_by_role(agent_name, self.agents) + if not target_agent: + raise AgentLookupError( + f"Agent with role '{agent_name}' not found. Available agents: " + f"{', '.join(agent.role 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) - ) + if delegating_agent and not self.can_delegate_to(delegating_agent, target_agent): + raise UnauthorizedDelegationError( + f"Agent '{delegating_agent.role}' cannot delegate to '{target_agent.role}'. " + f"Allowed targets: {', '.join(delegating_agent.allowed_agents or [])}" + ) task_with_assigned_agent = Task( description=task, @@ -65,9 +70,15 @@ class BaseAgentTool(BaseTool): i18n=target_agent.i18n, ) return target_agent.execute_task(task_with_assigned_agent, context) - except Exception as _: + except AgentLookupError as e: return self.i18n.errors("agent_tool_unexisting_coworker").format( - coworkers="\n".join( - [f"- {agent.role.casefold()}" for agent in self.agents] - ) + coworkers="\n".join(f"- {agent.role}" for agent in self.agents) ) + except UnauthorizedDelegationError as e: + 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 or [])) + ) + except Exception as e: + return self.i18n.errors("tool_usage_error").format(error=str(e)) diff --git a/src/crewai/translations/en.json b/src/crewai/translations/en.json index 280764642..76b681ef2 100644 --- a/src/crewai/translations/en.json +++ b/src/crewai/translations/en.json @@ -29,7 +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", + "agent_tool_unauthorized_delegation": "\nAuthorization Error: Agent '{agent}' cannot delegate to '{target}'.\nAllowed delegation targets:\n{allowed}\nPlease check agent configuration or contact administrator.\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.", diff --git a/src/crewai/utilities/agent_lookup.py b/src/crewai/utilities/agent_lookup.py new file mode 100644 index 000000000..02c7e9ef1 --- /dev/null +++ b/src/crewai/utilities/agent_lookup.py @@ -0,0 +1,26 @@ +"""Agent lookup utilities for CrewAI.""" + +from typing import List, Optional, Union +from ..agents.agent_builder.base_agent import BaseAgent +from ..exceptions import AgentLookupError + + +class AgentLookupMixin: + """Mixin class for agent lookup functionality.""" + + def get_agent_by_role(self, role: str, agents: List[BaseAgent]) -> Union[BaseAgent, None]: + """Find an agent by role, case-insensitive. + + Args: + role: The role to search for + agents: List of agents to search through + + Returns: + The found agent or None + """ + normalized_role = role.casefold().replace('"', "").replace("\n", "") + return next( + (agent for agent in agents + if agent.role.casefold().replace("\n", "") == normalized_role), + None + ) diff --git a/tests/tools/agent_tools/agent_tools_test.py b/tests/tools/agent_tools/agent_tools_test.py index 9ec426d08..379655995 100644 --- a/tests/tools/agent_tools/agent_tools_test.py +++ b/tests/tools/agent_tools/agent_tools_test.py @@ -126,6 +126,65 @@ def test_ask_question_to_wrong_agent(): ) +@pytest.mark.vcr(filter_headers=["authorization"]) +def test_delegate_work_with_null_allowed_agents(): + """Test delegation when allowed_agents is None (unrestricted delegation).""" + executive = Agent( + role="Executive Director", + goal="Lead the team effectively", + backstory="You're an experienced executive", + allow_delegation=True, + allowed_agents=None + ) + + 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" not in result + + +@pytest.mark.vcr(filter_headers=["authorization"]) +def test_delegate_work_with_empty_allowed_agents(): + """Test delegation when allowed_agents is an empty list (no delegation allowed).""" + executive = Agent( + role="Executive Director", + goal="Lead the team effectively", + backstory="You're an experienced executive", + allow_delegation=True, + allowed_agents=[] + ) + + 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 "Authorization Error" in result + + @pytest.mark.vcr(filter_headers=["authorization"]) def test_delegate_work_with_allowed_agents(): executive = Agent(