If under the rock and work with AI, you may have heard about agent2Agent (A2A) protocol, “an open standard designed to enable communication between Agents. It's a good tounge, but it's already getting a lot of buzz. As it plays well with MCP (looks like improving the industry), A2A builds This page Go-to Standard For Multi-Alent Connectivity in the industry.
When Google starts winning protocol clarity, my first reaction was: “All right, cool … But what should I do with this?” Fortunately, this week they issued the official Python Sdk of Protocol, so it finally speaks the language I understand.
In this article we will actually move on how to make link between agents and customers. Spoiler: Everything is on the way to work. Making things wrong, let's build a small pattern of toy together.
Communication between an event-to-party machine and A2A client
In our plans we have Detector AI event agent (responsible for receiving events) and Alert a agent (responsible for awareness of the user of events). As I focus on A2A protocol here, both agents are ridiculed as simple Python ropes. But in real life you can create your agents for any framework you prefer (LangGraph, Google Adk, crewai, etc.).
We have three letters in our system, the userthe Event agent once Alert. All contacts using Messages. A Message represents one chance of communication in A2A protocol. We are wrapped up the agents A2A servers. Servers express the fate of HTTP EXTPONT to use the law. Each A2A server has Event queues That works as a buffer between the killing of agencchronous and the management of the server response.
This page A2A client begins to communicate, and if two agents need to contact someone A2A server and can play the role of A2A client. The diagram below shows that the client and the server interacts between the Protocol.
This page EventQueue house Messages, Tasks, TaskStatusUpdateEvent, TaskArtifactUpdateEvent, A2AError including JSONRPCError things. This page Task It may be the most important thing to understand how to create various agents for A2A. According to A2A documents:
- When the client sends a message to an agent, the agent may determine that fulfilling the request requires that a powerful function is completed (eg.
- Each employee has a unique ID described by an agent and developed a description of defined (eg.
submitted,working,input-required,completed,failed). - Functions flourish and may include multiple exchange (messages) between client and server.
Think about Task As something in your Multi-Any Agent Schedule fine including unique purpose. We have two tasks in our program:
- See the event
- Login
Each agent does something of them (work). Let's build an A2A server for an event so things are more glorious.
To create an Event A2A server
First: Agent card. The agent card by JSON is a document used to know other agents available. :
- Identity and the identity
- Skills
- Skills
- ENDPOINT service
- Sweat
- Customers should ensure and participate with an agent
Let us first explain the agent's agent card for an agent agent (explain the skills based on this example from Google):
agent_card = AgentCard(
name='Event Detection Agent',
description='Detects relevant events and alerts the user',
url='
version='1.0.0',
defaultInputModes=['text'],
defaultOutputModes=['text'],
capabilities=AgentCapabilities(streaming=False),
authentication={ "schemes": ["basic"] },
skills=[
AgentSkill(
id='detect_events',
name='Detect Events',
description='Detects events and alert the user',
tags=['event'],
),
],
)You can learn more about the Agent Card Card Building here:
The agent itself will be an Uvicorn server, so let's build main() how to take and work. The entire request will be managed by DefaultRequestHandler of A2A-Python SDK. The manager needs a TaskStore to maintain jobs with AgentExecutor The implementation of an agent's concept (we will build EventAgentExecutor per minute).
The last component of main() The way is the A2AStarletteApplicationStarlette application uses A2A Protocol servers. We need to give Agent Card once DefaultRequestHandler to start it. Now the last step is to use the app using Uvicorn. Here is the full code of main() Method:
import click
import uvicorn
from a2a.types import (
AgentCard, AgentCapabilities, AgentSkill
)
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.server.apps import A2AStarletteApplication
@click.command()
@click.option('--host', default='localhost')
@click.option('--port', default=10008)
def main(host: str, port: int):
agent_executor = EventAgentExecutor()
agent_card = AgentCard(
name='Event Detection Agent',
description='Detects relevant events and alerts the user',
url='
version='1.0.0',
defaultInputModes=['text'],
defaultOutputModes=['text'],
capabilities=AgentCapabilities(streaming=False),
authentication={ "schemes": ["basic"] },
skills=[ AgentSkill( id='detect_events', name='Detect Events', description='Detects events and alert the user', tags=['event'],
),
],
)
request_handler = DefaultRequestHandler(
agent_executor=agent_executor,
task_store=InMemoryTaskStore()
)
a2a_app = A2AStarletteApplication(
agent_card=agent_card,
http_handler=request_handler
)
uvicorn.run(a2a_app.build(), host=host, port=port)Creating an Event event
Now it's time to build our agent's root and finally see how to use jobs to make the agents associated. This page EventAgentExecutor Class Fatest from AgentExecutor Interface and as a result we need to use execute() once cancel() Means. Both take a RequestContext with EventQueue thing as parameters. This page RequestContext holds information about current request is processed by server and EventQueue Actions as a buffer between the execution of the agencchronous and the management of the server response.
Our agent will only consider whether the string “event“Kiss ✨ messages. If”event“Then we should call a notice agent. We will do that by sending a Message to the other anarchy. This is a direct configuration strategy, which means we will prepare an agent for the URL of downloading the Agency Agent's Agency. To make our event provider you will serve as a2A client.
Let's build a male base of steps. First of all let's develop a primary job (work of finding events). We need to include a TaskUpdater Item (Section of Configuration Services for updates on the Service event line), and send the work and announces we apply with start_work() Method:
from a2a.server.agent_execution import AgentExecutor
class EventAgentExecutor(AgentExecutor):
async def execute(self, context: RequestContext, event_queue: EventQueue):
task_updater = TaskUpdater(event_queue, context.task_id, context.context_id)
task_updater.submit()
task_updater.start_work()The user's message will send to the agent you will look like:
send_message_payload = {
'message': {
'role': 'user',
'parts': [{'type': 'text', 'text': f'it has an event!'}],
'messageId': uuid4().hex,
}
}A Part represents a different piece of content within a Messagerepresenting the content of the post without TextPart, FilePartor DataPart. We will use a TextPart So we need to get you out of Excer Maceb:
from a2a.server.agent_execution import AgentExecutor
class EventAgentExecutor(AgentExecutor):
async def execute(self, context: RequestContext, event_queue: EventQueue):
task_updater = TaskUpdater(event_queue, context.task_id, context.context_id)
task_updater.submit()
task_updater.start_work()
await asyncio.sleep(1) #let's pretend we're actually doing something
user_message = context.message.parts[0].root.text # unwraping the TextPartTime to make the advanced view of our agent. If the message does not have a string “event“We do not have to call awareness agent and work done:
from a2a.server.agent_execution import AgentExecutor
class EventAgentExecutor(AgentExecutor):
async def execute(self, context: RequestContext, event_queue: EventQueue):
task_updater = TaskUpdater(event_queue, context.task_id, context.context_id)
task_updater.submit()
task_updater.start_work()
await asyncio.sleep(1) #let's pretend we're actually doing something
user_message = context.message.parts[0].root.text # unwraping the TextPart
if "event" not in user_message:
task_updater.update_status(
TaskState.completed,
message=task_updater.new_agent_message(parts=[TextPart(text=f"No event detected")]),
)Creating a User A2A client
Let's build a2A client to check the agent as it is. The client uses the get_client_from_agent_card_url() path from A2AClient Class to (guess what) get an agent card. Then we wrap the message to SendMessageRequest item and send it to agent using send_message() the client method. Here is the perfect code:
import httpx
import asyncio
from a2a.client import A2AClient
from a2a.types import SendMessageRequest, MessageSendParams
from uuid import uuid4
from pprint import pprint
async def main():
send_message_payload = {
'message': {
'role': 'user',
'parts': [{'type': 'text', 'text': f'nothing happening here'}],
'messageId': uuid4().hex,
}
}
async with httpx.AsyncClient() as httpx_client:
client = await A2AClient.get_client_from_agent_card_url(
httpx_client, '
)
request = SendMessageRequest(
params=MessageSendParams(**send_message_payload)
)
response = await client.send_message(request)
pprint(response.model_dump(mode='json', exclude_none=True))
if __name__ == "__main__":
asyncio.run(main())This happens in terminal using an event server:
And this is the Message You see:
The event of the event was created and no event received, beautiful! But the whole point of A2A to make agents in contact with each other, so let's make an event agent talk to an awareness agent.
Making an event provider to talk to an awareness agent
To make an event agent talk to an agency awareness agent will serve as a customer:
from a2a.server.agent_execution import AgentExecutor
ALERT_AGENT_URL = "
class EventAgentExecutor(AgentExecutor):
async def execute(self, context: RequestContext, event_queue: EventQueue):
task_updater = TaskUpdater(event_queue, context.task_id, context.context_id)
task_updater.submit()
task_updater.start_work()
await asyncio.sleep(1) #let's pretend we're actually doing something
user_message = context.message.parts[0].root.text # unwraping the TextPart
if "event" not in user_message:
task_updater.update_status(
TaskState.completed,
message=task_updater.new_agent_message(parts=[TextPart(text=f"No event detected")]),
)
else:
alert_message = task_updater.new_agent_message(parts=[TextPart(text="Event detected!")])
send_alert_payload = SendMessageRequest(
params=MessageSendParams(
message=alert_message
)
)
async with httpx.AsyncClient() as client:
alert_agent = A2AClient(httpx_client=client, url=ALERT_AGENT_URL)
response = await alert_agent.send_message(send_alert_payload)
if hasattr(response.root, "result"):
alert_task = response.root.result
# Polling until the task is done
while alert_task.status.state not in (
TaskState.completed, TaskState.failed, TaskState.canceled, TaskState.rejected
):
await asyncio.sleep(0.5)
get_resp = await alert_agent.get_task(
GetTaskRequest(params=TaskQueryParams(id=alert_task.id))
)
if isinstance(get_resp.root, GetTaskSuccessResponse):
alert_task = get_resp.root.result
else:
break
# Complete the original task
if alert_task.status.state == TaskState.completed:
task_updater.update_status(
TaskState.completed,
message=task_updater.new_agent_message(parts=[TextPart(text="Event detected and alert sent!")]),
)
else:
task_updater.update_status(
TaskState.failed,
message=task_updater.new_agent_message(parts=[TextPart(text=f"Failed to send alert: {alert_task.status.state}")]),
)
else:
task_updater.update_status(
TaskState.failed,
message=task_updater.new_agent_message(parts=[TextPart(text=f"Failed to create alert task")]),
)We call a warning agent just as we call an event provider as a user, and in the process of awareness agent, we complete the original performance of the original event. Let's call an event agent and but this time for the event:
Beauty here simply calls notification agent and we do not need to know anything about how to bind the user. We just send a message to it and wait for us to finish.
The notice agent is very similar to the event agent. You can check the entire code here:
The last thoughts
Understanding how much agents Systems are A2A may be bad at first, but at your end Send messages to allow agents to do something. All you need to do is combine your agents with A2A to build a classic class of the age agent AgentExecutor and run a agent as a server.
I hope this article helped you on your A2A journey, thank you for reading!
Progress
[1] Padham, Lin, and Michael Wikoff. Developing intelligent Agent Apps: An effective guide. John Wiley & Sons, 2005.
[2]
[3] / Tree / Main / Examples / Google_adik
[4]
