Agent2Agent (A2A) Protocol

Transcript

1. Mete Atamel Developer Advocate @ Google @meteatamel atamel.dev speakerdeck.com/meteatamel github.com/meteatamel/genai-beyond-basics/tree/main/samples/protocols/a2a Agent2Agent (A2A) Protocol
2. Agent Tools Model An AI agent is an application that uses models to make decisions and act independently in order to achieve a user-specified goal.
4. root_agent = Agent( model="gemini-2.5-flash", name="software_assistant", instruction=agent_instruction, tools=[get_current_date, search_tool, langchain_tool, *toolbox_tools, mcp_tools], )
5. Agents need to talk 🏨 Hotel agent ✈ Flight agent 🎒 Travel agent 🏄 Activity agent 📌 Guide agent …but how?
6. Agent2Agent (A2A) Protocol is an open protocol from Google that standardizes how agents running on diverse frameworks and platforms communicate Agent2Agent (A2A) Protocol a2a-protocol.org
8. Imagine a simple agent scenario… Client User Agent A (Client) Agent B (Remote)
9. Agent Discovery Client User Agent A (Client) Agent B (Remote) What can you do? How can I contact you?
10. Agent Card Client User Agent A (Client) Agent B (Remote) /.well-known/agent-card.json Agent Discovery Agent Card - A JSON metadata document describing an agent's identity, capabilities, endpoint, skills, and authentication requirements
12. AgentSkill skill = AgentSkill( id='get_exchange_rate', name='Currency Exchange Rates Tool', description='Gets exchange rates between currencies', tags=['currency conversion', 'currency exchange'], examples=['What is 1 GBP in USD?'], )
13. AgentCard agent_card = AgentCard( name='Currency Agent', description='Helps with exchange rates for currencies', url=f'http://{host}:{port}/', version='1.0.0', defaultInputModes=["text"], defaultOutputModes=["text"], capabilities=AgentCapabilities(streaming=True), skills=[skill] )
14. In ADK: AgentCard is automatically generated root_agent = Agent( name="currency_agent", model="gemini-2.0-flash", description="Agent to convert currencies.", instruction=instruction_prompt, tools=[convert_currency]) # Expose the agent over A2A protocol locally a2a_app = to_a2a(root_agent, host="0.0.0.0", port=int(os.getenv('PORT', '8080')))
15. Agent Interactions Client User Agent A (Client) Agent B (Remote) How can Agent A interact with Agent B?
16. Agent Response: Message or Task 1. Respond with a Stateless Message 2. Initiate a Stateful Task
17. Client User Agent A (Client) Agent B (Remote) HTTP(S) Message Role (user, agent) Parts (text, file, or JSON) Message Role (user, agent) Part (text, file, or JSON) HTTP(S) Option 1: Messages for trivial interactions Message - A single turn of communication between a client and an agent, containing content and a role ("user" or "agent")
18. AgentExecutor Bridge between Agent and A2A Client User Agent A (Client) Agent B (Remote) HTTP(S) Message Role (user, agent) Parts (text, file, or JSON) Message Role (user, agent) Part (text, file, or JSON) HTTP(S) AgentExecutor A2A Starlette Application
19. AgentExecutor class AgentExecutor(ABC): """ Implementations of this interface contain the core logic of the agent, executing tasks based on requests and publishing updates to an event queue. """ @abstractmethod async def execute(self, context: RequestContext, event_queue: EventQueue ) -> None: """Execute the agent's logic for a given request context. ) @abstractmethod async def cancel( self, context: RequestContext, event_queue: EventQueue ) -> None:
20. In ADK: AgentExecutor is already implemented root_agent = Agent( name="currency_agent", model="gemini-2.0-flash", description="Agent to convert currencies.", instruction=instruction_prompt, tools=[convert_currency]) # Expose the agent over A2A protocol locally a2a_app = to_a2a(root_agent, host="0.0.0.0", port=int(os.getenv('PORT', '8080')))
21. Option 2: Tasks for stateful interactions Task - A stateful unit of work initiated by an agent, with a unique ID and defined lifecycle Client User Agent A (Client) Message Role (user, agent) Parts (text, file, or JSON) Task ID Status HTTP(S) Agent B (Remote) ⚙ Processing…. Task Status: - Unspecified - Submitted - Working - Completed - Failed - Cancelled - Input required - Rejected - Auth Required
22. Client User Agent A (Client) Agent B (Remote) Message Role (user, agent) Parts (text, file, or JSON) JSON-RPC Task ID Status HTTP(S) Artifact Parts (text, file, or JSON) JSON-RPC Option 2: Tasks for stateful interactions Artifact - A tangible output generated by an agent during a task (for example, a document, image, or structured data) Task Status: - Unspecified - Submitted - Working - Completed - Failed - Cancelled - Input required - Rejected - Auth Required
23. async def execute(self, context: RequestContext, event_queue: EventQueue) -> None: ... query = context.get_user_input() task = context.current_task if not task: task = new_task(context.message) event_queue.enqueue_event(task) updater = TaskUpdater(event_queue, task.id, task.context_id) 23 Agent Executor Example
24. async def execute(self, context: RequestContext, event_queue: EventQueue) -> None: ... query = context.get_user_input() task = context.current_task if not task: task = new_task(context.message) event_queue.enqueue_event(task) updater = TaskUpdater(event_queue, task.id, task.context_id) 24 Agent Executor Example
25. async def execute(self, context: RequestContext, event_queue: EventQueue) -> None: ... query = context.get_user_input() task = context.current_task if not task: task = new_task(context.message) event_queue.enqueue_event(task) updater = TaskUpdater(event_queue, task.id, task.context_id) 25 Agent Executor Example
26. async def execute(self, context: RequestContext, event_queue: EventQueue) -> None: ... query = context.get_user_input() task = context.current_task if not task: task = new_task(context.message) event_queue.enqueue_event(task) updater = TaskUpdater(event_queue, task.id, task.contextId) 26 Agent Executor Example
27. async def execute(self, context: RequestContext, event_queue: EventQueue) -> None: ... async for item in self.agent.stream(query, task.contextId): is_task_complete = item['is_task_complete'] require_user_input = item['require_user_input'] if not is_task_complete and not require_user_input: updater.update_status(...) elif require_user_input: updater.update_status(..., final=True) break else: updater.add_artifact(...) updater.complete() break 27 Agent Executor Example
28. async def execute(self, context: RequestContext, event_queue: EventQueue) -> None: ... async for item in self.agent.stream(query, task.contextId): is_task_complete = item['is_task_complete'] require_user_input = item['require_user_input'] if not is_task_complete and not require_user_input: updater.update_status(...) elif require_user_input: updater.update_status(..., final=True) break else: updater.add_artifact(...) updater.complete() break 28 Agent Executor Example
29. async def execute(self, context: RequestContext, event_queue: EventQueue) -> None: ... async for item in self.agent.stream(query, task.contextId): is_task_complete = item['is_task_complete'] require_user_input = item['require_user_input'] if not is_task_complete and not require_user_input: updater.update_status(...) elif require_user_input: updater.update_status(..., final=True) break else: updater.add_artifact(...) updater.complete() break 29 Agent Executor Example
30. async def execute(self, context: RequestContext, event_queue: EventQueue) -> None: ... async for item in self.agent.stream(query, task.contextId): is_task_complete = item['is_task_complete'] require_user_input = item['require_user_input'] if not is_task_complete and not require_user_input: updater.update_status(...) elif require_user_input: updater.update_status(..., final=True) break else: updater.add_artifact(...) updater.complete() break 30 Agent Executor Example
31. async def execute(self, context: RequestContext, event_queue: EventQueue) -> None: ... async for item in self.agent.stream(query, task.contextId): is_task_complete = item['is_task_complete'] require_user_input = item['require_user_input'] if not is_task_complete and not require_user_input: updater.update_status(...) elif require_user_input: updater.update_status(..., final=True) break else: updater.add_artifact(...) updater.complete() break 31 Agent Executor Example
32. Agent Interactions 1. Request/Response (Polling) 2. Streaming with Server-Sent Events (SSE) 3. Push Notifications
33. Agent Interactions Request/Response (Polling) Client User Agent A (Client) Agent B (Remote) Message Role (user, agent) Parts (text, file, or JSON) Task ID Status HTTP(S)
34. Agent Interactions Streaming with SSE Client User Agent A (Client) Agent B (Remote) HTTPS Message Role (user, agent) Parts (text, file, or JSON) JSON-RPC Agent Card streaming: true Push updates Initial task Messages Artifacts Best suited for: - Real-time progress monitoring of long-running tasks - Receiving large results (artifacts) incrementally. - Interactive, conversational exchanges where immediate feedback or partial responses are beneficial with low latency
35. Agent Interactions Push Notifications Client User Agent A (Client) Agent B (Remote) HTTPS Message Role (user, agent) Parts (text, file, or JSON) JSON-RPC Agent Card pushNotifications : true Best suited for: - Very long-running tasks that can take minutes, hours, or days to complete. - Clients that cannot or prefer not to maintain persistent connections, such as mobile applications. - Clients only need to be notified of significant state changes rather than continuous updates.
36. # 1. Get the agent card resolver = A2ACardResolver( httpx_client=httpx_client, base_url=BASE_URL ) agent_card = await resolver.get_agent_card() A2A Client 36
37. # 2. Create an A2A client to communicate with the agent using url from the agent card client = A2AClient( httpx_client=httpx_client, agent_card=agent_card ) 37 A2A Client
38. # 3. Create a message to send to the agent send_message_payload: dict[str, Any] = { 'message': { 'role': 'user', 'parts': [ {'kind': 'text', 'text': '"Hello from the A2A client!"'} ], 'messageId': uuid4().hex, }, } request = SendMessageRequest( id=str(uuid4()), params=MessageSendParams(**send_message_payload) ) 38 A2A Client
39. # 4. Send the message using non-streaming API response = await client.send_message(request) print(response.model_dump(mode='json', exclude_none=True)) # 5. Send the message using streaming API streaming_request = SendStreamingMessageRequest( id=str(uuid4()), params=MessageSendParams(**send_message_payload) ) stream_response = client.send_message_streaming(streaming_request) async for chunk in stream_response: print(chunk.model_dump(mode='json', exclude_none=True))) 39 A2A Client
40. # Currency agent is a remote agent available over A2A currency_agent = RemoteA2aAgent( name="currency_agent", description="Agent that can convert from one currency to another.", agent_card=f"http://localhost:8080/{AGENT_CARD_WELL_KNOWN_PATH}" ) 40 In ADK, use RemoteA2aAgent
41. A2A Extensions: Empowering Custom Agent Functionality
42. A possible agentic stack Agent2Agent (A2A) protocol Open standard designed to enable communication & collaboration between AI agents. Vertex AI Agent Engine Managed platform to deploy, manage, and scale AI agents in production. Model Context Protocol Open protocol that standardizes how applications provide context to LLMs. Agent Development Kit Open-source, code-first toolkit for building, evaluating, and deploying AI agents.
43. Thank you! Mete Atamel Developer Advocate at Google @meteatamel atamel.dev speakerdeck.com/meteatamel github.com/meteatamel/genai-beyond-basics/tree/main/samples/protocols/a2a