Source URL: https://cloud.google.com/blog/products/ai-machine-learning/unlock-ai-agent-collaboration-convert-adk-agents-for-a2a/
Source: Cloud Blog
Title: A guide to converting ADK agents with MCP to the A2A framework
Feedly Summary: The evolution of AI agents has led to powerful, specialized models capable of complex tasks. The Google Agent Development Kit (ADK) – a toolkit designed to simplify the construction and management of language model-based applications – makes it easy for developers to build agents, usually equipped with tools via the Model Context Protocol (MCP) for tasks like web scraping. However, to unlock their full potential, these agents must be able to collaborate. The Agent-to-Agent (A2A) framework – a standardized communication protocol that allows disparate agents to discover each other, understand their capabilities, and interact securely – provides the standard for this interoperability.
This guide provides a step-by-step process for converting a standalone ADK agent that uses an MCP tool into a fully A2A-compatible component, ready to participate in a larger, multi-agent ecosystem. We will use a MultiURLBrowser agent, designed to scrape web content, as a practical example
Step 1: Define the core agent and its MCP tool (agent.py)
The foundation of your agent remains its core logic. The key is to properly initialize the ADK LlmAgent and configure its MCPToolset to connect with its external tool.
In agent.py, the _build_agent method is where you specify the LLM and its tools. The MCPToolset is configured to launch the firecrawl-mcp tool, passing the required API key through its environment variables
code_block
Step 2: Establish a public identity (__main__.py)
For other agents to discover and understand your agent, it needs a public identity. This is achieved through the AgentSkill and AgentCard in the __main__.py file, which also serves as the entry point for the A2A server.
1. Define AgentSkill: This object acts as a declaration of the agent’s capabilities. It includes a unique ID, a human-readable name, a description, and examples
code_block
<ListValue: [StructValue([(‘code’, ‘# agents/search_agent/__main__.py\r\nfrom a2a.skills.skill_declarations import AgentSkill\r\n\r\nskill = AgentSkill(\r\n id="MultiURLBrowser",\r\n name="MultiURLBrowser_Agent",\r\n description="Agent to scrape content from the URLs specified by the user.",\r\n tags=["multi-url", "browser", "scraper", "web"],\r\n examples=[\r\n "Scrape the URL: https://example.com/page1",\r\n "Extract data from: https://example.com/page1 and https://example.com/page2"\r\n ]\r\n)’), (‘language’, ‘lang-py’), (‘caption’, <wagtail.rich_text.RichText object at 0x3ebb4af37c10>)])]>
2. Define AgentCard: This is the agent’s primary metadata for discovery. It includes the agent’s name, URL, version, and, crucially, the list of skills it possesses.
code_block
<ListValue: [StructValue([(‘code’, ‘# agents/search_agent/__main__.py\r\nfrom a2a.cards.agent_card import AgentCard, AgentCapabilities\r\n\r\nagent_card = AgentCard(\r\n name="MultiURLBrowser",\r\n description="Agent designed to efficiently scrape content from URLs.",\r\n url=f"http://{host}:{port}/",\r\n version="1.0.0",\r\n defaultInputModes=[\’text\’],\r\n defaultOutputModes=[\’text\’],\r\n capabilities=AgentCapabilities(streaming=True),\r\n skills=[skill],\r\n supportsAuthenticatedExtendedCard=True,\r\n)’), (‘language’, ‘lang-py’), (‘caption’, <wagtail.rich_text.RichText object at 0x3ebb4af37700>)])]>
Step 3: Implement the A2A task manager (task_manager.py)
The AgentTaskManager is the bridge between the A2A framework and your agent’s logic. It implements the AgentExecutor interface, which requires execute and cancel methods.
The execute method is triggered by the A2A server upon receiving a request. It manages the task’s lifecycle, invokes the agent, and streams status updates and results back to the server via an EventQueue and TaskUpdater.
code_block
<ListValue: [StructValue([(‘code’, ‘# agents/search_agent/task_manager.py\r\nfrom a2a.server.task_manager import AgentExecutor, RequestContext, EventQueue, TaskUpdater\r\nfrom a2a.server.task_protocols import TaskState, new_task, new_agent_text_message\r\nfrom .agent import MultiURLBrowser\r\n\r\nclass AgentTaskManager(AgentExecutor):\r\n def __init__(self):\r\n self.agent = MultiURLBrowser()\r\n\r\n async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:\r\n query = context.get_user_input()\r\n task = context.current_task or new_task(context.message)\r\n await event_queue.enqueue_event(task)\r\n updater = TaskUpdater(event_queue, task.id, task.contextId)\r\n\r\n try:\r\n async for item in self.agent.invoke(query, task.contextId):\r\n if not item.get(\’is_task_complete\’, False):\r\n await updater.update_status(\r\n TaskState.working,\r\n new_agent_text_message(item.get(\’updates\’), task.contextId, task.id)\r\n )\r\n else:\r\n message = new_agent_text_message(item.get(\’content\’), task.contextId, task.id)\r\n await updater.update_status(TaskState.completed, message)\r\n break\r\n except Exception as e:\r\n error_message = f"An error occurred: {str(e)}"\r\n await updater.update_status(\r\n TaskState.failed,\r\n new_agent_text_message(error_message, task.contextId, task.id)\r\n )\r\n raise’), (‘language’, ‘lang-py’), (‘caption’, <wagtail.rich_text.RichText object at 0x3ebb4af37fd0>)])]>
Step 4: Create the agent’s invoke method (agent.py)
The invoke method is the entry point into the agent’s core ADK logic. It is called by the AgentTaskManager and is responsible for running the ADK Runner. As the runner processes the query, this asynchronous generator yields events, allowing for streaming of progress updates and the final response.
code_block
<ListValue: [StructValue([(‘code’, ‘# agents/search_agent/agent.py\r\n# …\r\nfrom adk.runner import Runner\r\n# …\r\n\r\nclass MultiURLBrowser:\r\n # … __init__ and _build_agent …\r\n\r\n async def invoke(self, query: str, session_id: str) -> AsyncIterable[dict]:\r\n # … session management …\r\n\r\n user_content = types.Content(role="user", parts=[types.Part.from_text(text=query)])\r\n\r\n async for event in self._runner.run_async(\r\n user_id=self._user_id,\r\n session_id=session.id,\r\n new_message=user_content\r\n ):\r\n if event.is_final_response():\r\n response_text = event.content.parts[-1].text if event.content and event.content.parts else ""\r\n yield {\’is_task_complete\’: True, \’content\’: response_text}\r\n else:\r\n yield {\’is_task_complete\’: False, \’updates\’: "Processing request…"}’), (‘language’, ‘lang-py’), (‘caption’, <wagtail.rich_text.RichText object at 0x3ebb4af37c70>)])]>
With all components correctly configured, the MultiURLBrowser agent is now a fully operational A2A agent. When a client sends it a request to scrape content, it processes the task and returns the final result. The terminal output below shows a successful interaction, where the agent has received a mission and provided the extracted information as its final response.
Once you have A2A-compatible agents, you can create an "Orchestrator Agent" that delegates sub-tasks to them. This allows for the completion of complex, multi-step workflows.
Step 1: Discover available agents
An orchestrator must first know what other agents are available. This can be achieved by querying a known registry endpoint that lists the AgentCard for all registered agents.
code_block
<ListValue: [StructValue([(‘code’, ‘# Scrap_Translate/agent.py\r\nimport httpx\r\n\r\nAGENT_REGISTRY_BASE_URL = "http://localhost:10000"\r\n\r\nasync with httpx.AsyncClient() as httpx_client:\r\n base_url = AGENT_REGISTRY_BASE_URL.rstrip("/")\r\n resolver = A2ACardResolver(\r\n httpx_client=httpx_client,\r\n base_url=base_url,\r\n # agent_card_path and extended_agent_card_path use defaults if not specified\r\n )\r\n final_agent_card_to_use: AgentCard | None = None\r\n\r\n try:\r\n # Fetches the AgentCard from the standard public path.\r\n public_card = await resolver.get_agent_card()\r\n final_agent_card_to_use = public_card\r\n except Exception as e:\r\n # Handle exceptions as needed for your specific use case.\r\n # For a blog post, you might simplify or omit detailed error handling\r\n # if the focus is purely on the successful path.\r\n print(f"An error occurred: {e}")’), (‘language’, ‘lang-py’), (‘caption’, <wagtail.rich_text.RichText object at 0x3ebb4af37f70>)])]>
Step 2: Call other agents as tools
The orchestrator interacts with other agents using the a2a.client. The call_agent function demonstrates how to construct a SendMessageRequest and dispatch it to a target agent.
code_block
<ListValue: [StructValue([(‘code’, "# Scrap_Translate/agent.py\r\nfrom a2a.client import A2AClient\r\nfrom a2a.client.protocols import SendMessageRequest, MessageSendParams\r\nfrom uuid import uuid4\r\n\r\nasync def call_agent(agent_name: str, message: str) -> str:\r\n # In a real implementation, you would resolve the agent’s URL first\r\n # using its card from list_agents().\r\n client = A2AClient(httpx_client=httpx.AsyncClient(timeout=300), agent_card=cards)\r\n\r\n payload = {\r\n ‘message’: {\r\n ‘role’: ‘user’,\r\n ‘parts’: [{‘kind’: ‘text’, ‘text’: message}],\r\n ‘messageId’: uuid4().hex,\r\n },\r\n }\r\n request = SendMessageRequest(id=str(uuid4()), params=MessageSendParams(**payload))\r\n\r\n response_record = await client.send_message(request)\r\n # Extract the text content from the response record\r\n response_model = response_record.model_dump(mode=’json’, exclude_none=True)\r\n return response_model[‘result’][‘status’][‘message’][‘parts’][0][‘text’]"), (‘language’, ‘lang-py’), (‘caption’, <wagtail.rich_text.RichText object at 0x3ebb556f7e20>)])]>
Step 3: Configure the orchestrator’s LLM
Finally, configure the orchestrator’s LlmAgent to use the discovery and delegation functions as tools. Provide a system instruction that guides the LLM on how to use these tools to break down user requests and coordinate with other agents
code_block
<ListValue: [StructValue([(‘code’, ‘# Scrap_Translate/agent.py\r\nfrom adk.agent import LlmAgent\r\nfrom adk.tools import FunctionTool\r\n\r\nsystem_instr = (\r\n "You are a root orchestrator agent. You have two tools:\\n"\r\n "1) list_agents() → Use this tool to see available agents.\\n"\r\n "2) call_agent(agent_name: str, message: str) → Use this tool to send a task to another agent.\\n"\r\n "Fulfill user requests by discovering and interacting with other agents."\r\n)\r\n\r\nroot_agent = LlmAgent(\r\n model="gemini-1.5-pro-preview-0514",\r\n name="root_orchestrator",\r\n instruction=system_instr,\r\n tools=[\r\n FunctionTool(list_agents),\r\n FunctionTool(call_agent),\r\n ],\r\n)’), (‘language’, ‘lang-py’), (‘caption’, <wagtail.rich_text.RichText object at 0x3ebb556f7460>)])]>
By following these steps, you can create both specialized, A2A-compatible agents and powerful orchestrators that leverage them, forming a robust and collaborative multi-agent system.
The true power of this architecture becomes visible when the orchestrator agent is run. Guided by its instructions, the LLM correctly interprets a user’s complex request and uses its specialized tools to coordinate with other agents. The screenshot below from a debugging UI shows the orchestrator in action: it first calls list_agents to discover available capabilities and then proceeds to call_agent to delegate the web-scraping task, perfectly illustrating the multi-agent workflow we set out to build.
Get started
This guide details the conversion of a standalone ADK/MCP agent into an A2A-compatible component and demonstrates how to build an orchestrator to manage such agents. The complete source code for all examples, along with official documentation, is available at the links below.
Project source code
Official A2A Python SDK
Official A2A sample projects
AI Summary and Description: Yes
Summary: The provided text serves as a comprehensive guide for developing A2A-compatible agents using the Google Agent Development Kit (ADK). It introduces a framework for facilitating communication and collaboration among AI agents and emphasizes the significance of interoperability in performing complex tasks through orchestrator agents. This is relevant for AI security professionals and developers focusing on secure and efficient AI-based applications.
Detailed Description: This text outlines the process of transitioning a basic language model (LM)-based agent into an Agent-to-Agent (A2A)-compatible component, enabling enhanced functionality and collaboration amongst multiple agents. The key components of the guide include:
– **Agent Development Kit (ADK)**: A toolkit that simplifies the creation and management of language model-based applications.
– **Model Context Protocol (MCP)**: A framework that allows agents to utilize external tools efficiently.
### Major Points:
– **A2A Framework**:
– Standardized communication protocol for agent interoperability.
– Enables agents to discover each other, understand capabilities, and interact securely.
– **Steps for Development**:
1. **Define the Core Agent**:
– Initialization of the ADK LlmAgent and setup of MCP toolset.
2. **Establish a Public Identity**:
– The **AgentSkill** and **AgentCard** are used for agent capabilities declaration and public metadata, respectively.
3. **Implement the A2A Task Manager**:
– The **AgentTaskManager** acts as the interface between the A2A framework and agent logic.
4. **Create the Agent’s Invoke Method**:
– This method facilitates interaction with the agent’s core ADK functions.
– **Multi-Agent Ecosystem**: The agent can be further enhanced to act as an orchestrator, managing additional agents and executing multi-step workflows.
### Practical Implications:
– **Security Considerations**:
– As agents interact with one another, maintaining secure communication through the A2A framework is crucial to prevent unauthorized access and data breaches.
– **Enhanced Collaboration**:
– The ability to build orchestrators can enable complex multi-agent systems, potentially increasing system efficiency and functionality.
– **Source Code and Documentation**:
– The guide refers to official source code and documentation, providing practical resources for developers looking to implement the A2A framework.
This guide is valuable for professionals in AI security as it not only enhances the practical deployment of AI agents but also addresses security aspects involved in their operation and collaboration.