Implement A2A protocol support for recursive agent invocation

- Modified BaseAgentTool._execute to accept and pass tools parameter
- Updated DelegateWorkTool and AskQuestionTool to pass agent tools
- Modified AgentTools to inject tools into delegation tools
- Added test cases to verify recursive agent invocation works

Fixes #2900

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

src/crewai/tools/agent_tools/agent_tools.py

@@ -22,11 +22,21 @@ 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()
 
         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()
 
         return [delegate_tool, ask_tool]
+        
+    def _get_all_agent_tools(self) -> list[BaseTool]:
+        """Get all tools from all agents for recursive invocation"""
+        all_tools = []
+        for agent in self.agents:
+            if agent.tools:
+                all_tools.extend(agent.tools)
+        return all_tools

src/crewai/tools/agent_tools/ask_question_tool.py

@@ -25,4 +25,5 @@ class AskQuestionTool(BaseAgentTool):
         **kwargs,
     ) -> str:
         coworker = self._get_coworker(coworker, **kwargs)
-        return self._execute(coworker, question, context)
+        tools = getattr(self, '_agent_tools', None) or kwargs.get('tools')
+        return self._execute(coworker, question, context, tools)

src/crewai/tools/agent_tools/base_agent_tools.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Optional
+from typing import Any, List, Optional
 
 from pydantic import Field
 
@@ -50,7 +50,8 @@ class BaseAgentTool(BaseTool):
         self,
         agent_name: Optional[str],
         task: str,
-        context: Optional[str] = None
+        context: Optional[str] = None,
+        tools: Optional[List[Any]] = None
     ) -> str:
         """
         Execute delegation to an agent with case-insensitive and whitespace-tolerant matching.
@@ -59,6 +60,7 @@ class BaseAgentTool(BaseTool):
             agent_name: Name/role of the agent to delegate to (case-insensitive)
             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
 
         Returns:
             str: The execution result from the delegated agent or an error message
@@ -113,9 +115,10 @@ class BaseAgentTool(BaseTool):
                 agent=agent,
                 expected_output=agent.i18n.slice("manager_request"),
                 i18n=agent.i18n,
+                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)
+            return agent.execute_task(task_with_assigned_agent, context, tools)
         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

@@ -27,4 +27,5 @@ class DelegateWorkTool(BaseAgentTool):
         **kwargs,
     ) -> str:
         coworker = self._get_coworker(coworker, **kwargs)
-        return self._execute(coworker, task, context)
+        tools = getattr(self, '_agent_tools', None) or kwargs.get('tools')
+        return self._execute(coworker, task, context, tools)

tests/tools/agent_tools/a2a_protocol_test.py

@@ -0,0 +1,137 @@
+"""Test for A2A protocol support in CrewAI."""
+
+import pytest
+from unittest.mock import patch, MagicMock
+
+from crewai.agent import Agent
+from crewai.crew import Crew
+from crewai.task import Task
+from crewai.tools.agent_tools.agent_tools import AgentTools
+from crewai.tools.agent_tools.delegate_work_tool import DelegateWorkTool
+from crewai.tools.agent_tools.ask_question_tool import AskQuestionTool
+from crewai.tools.base_tool import BaseTool
+
+
+def test_tools_passed_to_execute():
+    """Test that tools are properly passed to the _execute method."""
+    original_execute = DelegateWorkTool._execute
+    
+    tools_passed = {"value": False}
+    
+    def mock_execute(self, agent_name, task, context=None, tools=None):
+        assert tools is not None, "Tools should not be None"
+        assert len(tools) > 0, "Tools should not be empty"
+        assert any(isinstance(tool, DelegateWorkTool) for tool in tools), "DelegateWorkTool should be in tools"
+        assert any(isinstance(tool, AskQuestionTool) for tool in tools), "AskQuestionTool should be in tools"
+        tools_passed["value"] = True
+        return "Task executed successfully"
+    
+    researcher = Agent(
+        role="researcher",
+        goal="research and analyze content",
+        backstory="You're an expert researcher",
+        allow_delegation=True,
+    )
+    
+    writer = Agent(
+        role="writer",
+        goal="write content based on research",
+        backstory="You're an expert writer",
+        allow_delegation=True,
+    )
+    
+    agent_tools = AgentTools(agents=[researcher, writer])
+    delegation_tools = agent_tools.tools()
+    
+    with patch.object(DelegateWorkTool, '_execute', mock_execute):
+        delegate_tool = delegation_tools[0]  # DelegateWorkTool is the first tool
+        assert isinstance(delegate_tool, DelegateWorkTool), "First tool should be DelegateWorkTool"
+        
+        delegate_tool._run(
+            task="Test task",
+            context="Test context",
+            coworker="writer",
+            tools=delegation_tools
+        )
+        
+        assert tools_passed["value"], "Tools should be passed to _execute method"
+
+
+def test_tools_passed_from_ask_question_tool():
+    """Test that tools are properly passed from AskQuestionTool to _execute."""
+    original_execute = AskQuestionTool._execute
+    
+    tools_passed = {"value": False}
+    
+    def mock_execute(self, agent_name, question, context=None, tools=None):
+        assert tools is not None, "Tools should not be None"
+        assert len(tools) > 0, "Tools should not be empty"
+        assert any(isinstance(tool, DelegateWorkTool) for tool in tools), "DelegateWorkTool should be in tools"
+        assert any(isinstance(tool, AskQuestionTool) for tool in tools), "AskQuestionTool should be in tools"
+        tools_passed["value"] = True
+        return "Question answered successfully"
+    
+    researcher = Agent(
+        role="researcher",
+        goal="research and analyze content",
+        backstory="You're an expert researcher",
+        allow_delegation=True,
+    )
+    
+    writer = Agent(
+        role="writer",
+        goal="write content based on research",
+        backstory="You're an expert writer",
+        allow_delegation=True,
+    )
+    
+    agent_tools = AgentTools(agents=[researcher, writer])
+    delegation_tools = agent_tools.tools()
+    
+    with patch.object(AskQuestionTool, '_execute', mock_execute):
+        ask_tool = delegation_tools[1]  # AskQuestionTool is the second tool
+        assert isinstance(ask_tool, AskQuestionTool), "Second tool should be AskQuestionTool"
+        
+        ask_tool._run(
+            question="Test question",
+            context="Test context",
+            coworker="writer",
+            tools=delegation_tools
+        )
+        
+        assert tools_passed["value"], "Tools should be passed to _execute method"
+
+
+def test_agent_tools_injects_tools():
+    """Test that AgentTools injects tools into delegation tools."""
+    researcher = Agent(
+        role="researcher",
+        goal="research and analyze content",
+        backstory="You're an expert researcher",
+        allow_delegation=True,
+    )
+    
+    writer = Agent(
+        role="writer",
+        goal="write content based on research",
+        backstory="You're an expert writer",
+        allow_delegation=True,
+    )
+    
+    class CustomTool(BaseTool):
+        name: str = "Custom Tool"
+        description: str = "A custom tool for testing"
+        
+        def _run(self, *args, **kwargs):
+            return "Custom tool executed"
+    
+    custom_tool = CustomTool()
+    researcher.tools = [custom_tool]
+    
+    agent_tools = AgentTools(agents=[researcher, writer])
+    delegation_tools = agent_tools.tools()
+    
+    for tool in delegation_tools:
+        assert hasattr(tool, '_agent_tools'), "Tool should have _agent_tools attribute"
+        assert len(tool._agent_tools) > 0, "Tool should have agent tools injected"
+        assert any(isinstance(t, CustomTool) for t in tool._agent_tools), "Custom tool should be injected"