Implement comprehensive A2A protocol improvements

- Add tool deduplication logic in agent_tools.py
- Re-implement recursion depth limits and validation in base_agent_tools.py
- Add proper type hinting for _agent_tools attributes
- Add consistent tool retrieval method across delegation tools
- Enhance logging for debugging recursive agent invocations

Addresses all remaining PR review comments for full A2A protocol support.

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

src/crewai/tools/agent_tools/agent_tools.py

@@ -1,3 +1,5 @@
+from typing import List, Set
+
 from crewai.agents.agent_builder.base_agent import BaseAgent
 from crewai.tools.base_tool import BaseTool
 from crewai.utilities import I18N
@@ -22,21 +24,33 @@ class AgentTools:
             i18n=self.i18n,
             description=self.i18n.tools("delegate_work").format(coworkers=coworkers),  # type: ignore
         )
-        delegate_tool._agent_tools = self._get_all_agent_tools()
+        delegate_tool._agent_tools: List[BaseTool] = self._get_all_agent_tools()
 
         ask_tool = AskQuestionTool(
             agents=self.agents,
             i18n=self.i18n,
             description=self.i18n.tools("ask_question").format(coworkers=coworkers),  # type: ignore
         )
-        ask_tool._agent_tools = self._get_all_agent_tools()
+        ask_tool._agent_tools: List[BaseTool] = self._get_all_agent_tools()
 
         return [delegate_tool, ask_tool]
-        
+
     def _get_all_agent_tools(self) -> list[BaseTool]:
-        """Get all tools from all agents for recursive invocation"""
-        all_tools = []
+        """
+        Get all tools from all agents for recursive invocation.
+
+        Returns:
+            list[BaseTool]: A deduplicated list of all tools from all agents.
+        """
+        seen_tools: Set[int] = set()
+        unique_tools: List[BaseTool] = []
+
         for agent in self.agents:
             if agent.tools:
-                all_tools.extend(agent.tools)
-        return all_tools
+                for tool in agent.tools:
+                    tool_id = id(tool)
+                    if tool_id not in seen_tools:
+                        seen_tools.add(tool_id)
+                        unique_tools.append(tool)
+
+        return unique_tools

src/crewai/tools/agent_tools/ask_question_tool.py

@@ -1,8 +1,9 @@
-from typing import Optional
+from typing import List, Optional
 
 from pydantic import BaseModel, Field
 
 from crewai.tools.agent_tools.base_agent_tools import BaseAgentTool
+from crewai.tools.base_tool import BaseTool
 
 
 class AskQuestionToolSchema(BaseModel):
@@ -16,6 +17,7 @@ class AskQuestionTool(BaseAgentTool):
 
     name: str = "Ask question to coworker"
     args_schema: type[BaseModel] = AskQuestionToolSchema
+    _agent_tools: Optional[List[BaseTool]] = None
 
     def _run(
         self,
@@ -25,5 +27,5 @@ class AskQuestionTool(BaseAgentTool):
         **kwargs,
     ) -> str:
         coworker = self._get_coworker(coworker, **kwargs)
-        tools = getattr(self, '_agent_tools', None) or kwargs.get('tools')
+        tools = self._get_tools(**kwargs)
         return self._execute(coworker, question, context, tools)

src/crewai/tools/agent_tools/base_agent_tools.py

@@ -18,6 +18,7 @@ class BaseAgentTool(BaseTool):
     i18n: I18N = Field(
         default_factory=I18N, description="Internationalization settings"
     )
+    MAX_RECURSION_DEPTH: int = 5
 
     def sanitize_agent_name(self, name: str) -> str:
         """
@@ -46,12 +47,22 @@ class BaseAgentTool(BaseTool):
                 coworker = coworker[1:-1].split(",")[0]
         return coworker
 
+    def _get_tools(self, **kwargs) -> Optional[List[BaseTool]]:
+        """
+        Get tools from instance or kwargs.
+
+        Returns:
+            Optional[List[BaseTool]]: The tools to use for recursive invocation.
+        """
+        return getattr(self, "_agent_tools", None) or kwargs.get("tools")
+
     def _execute(
         self,
         agent_name: Optional[str],
         task: str,
         context: Optional[str] = None,
-        tools: Optional[List[Any]] = None
+        tools: Optional[List[BaseTool]] = None,
+        recursion_depth: int = 0,
     ) -> str:
         """
         Execute delegation to an agent with case-insensitive and whitespace-tolerant matching.
@@ -61,11 +72,24 @@ class BaseAgentTool(BaseTool):
             task: The specific question or task to delegate
             context: Optional additional context for the task execution
             tools: Optional tools to pass to the delegated agent for recursive invocation
+            recursion_depth: Current recursion depth to prevent infinite loops
 
         Returns:
             str: The execution result from the delegated agent or an error message
                  if the agent cannot be found
         """
+        if tools is not None and not all(isinstance(tool, BaseTool) for tool in tools):
+            return self.i18n.errors("agent_tool_execution_error").format(
+                agent_role="unknown",
+                error="Invalid tools provided: all tools must inherit from BaseTool",
+            )
+
+        if recursion_depth >= self.MAX_RECURSION_DEPTH:
+            return self.i18n.errors("agent_tool_execution_error").format(
+                agent_role="unknown",
+                error=f"Maximum recursion depth ({self.MAX_RECURSION_DEPTH}) exceeded",
+            )
+
         try:
             if agent_name is None:
                 agent_name = ""
@@ -110,6 +134,7 @@ class BaseAgentTool(BaseTool):
 
         agent = agent[0]
         try:
+            logger.debug(f"Executing task with {len(tools) if tools else 0} tools at recursion depth {recursion_depth}")
             task_with_assigned_agent = Task(
                 description=task,
                 agent=agent,
@@ -118,7 +143,9 @@ class BaseAgentTool(BaseTool):
                 tools=tools,
             )
             logger.debug(f"Created task for agent '{self.sanitize_agent_name(agent.role)}': {task}")
-            return agent.execute_task(task_with_assigned_agent, context, tools)
+            return agent.execute_task(
+                task_with_assigned_agent, context, tools, recursion_depth + 1
+            )
         except Exception as e:
             # Handle task creation or execution errors
             return self.i18n.errors("agent_tool_execution_error").format(

src/crewai/tools/agent_tools/delegate_work_tool.py

@@ -1,8 +1,9 @@
-from typing import Optional
+from typing import List, Optional
 
 from pydantic import BaseModel, Field
 
 from crewai.tools.agent_tools.base_agent_tools import BaseAgentTool
+from crewai.tools.base_tool import BaseTool
 
 
 class DelegateWorkToolSchema(BaseModel):
@@ -18,6 +19,7 @@ class DelegateWorkTool(BaseAgentTool):
 
     name: str = "Delegate work to coworker"
     args_schema: type[BaseModel] = DelegateWorkToolSchema
+    _agent_tools: Optional[List[BaseTool]] = None
 
     def _run(
         self,
@@ -27,5 +29,5 @@ class DelegateWorkTool(BaseAgentTool):
         **kwargs,
     ) -> str:
         coworker = self._get_coworker(coworker, **kwargs)
-        tools = getattr(self, '_agent_tools', None) or kwargs.get('tools')
+        tools = self._get_tools(**kwargs)
         return self._execute(coworker, task, context, tools)