Sample Page Title

The Agent-to-Agent (A2A) protocol is a brand new customary by Google that allows AI brokers—no matter their underlying framework or developer—to speak and collaborate seamlessly. It really works through the use of standardized messages, agent playing cards (which describe what an agent can do), and task-based execution, permitting brokers to work together by way of HTTP with out customized integration logic. A2A makes it simpler to construct scalable, interoperable multi-agent methods by abstracting away the complexities of communication. 

On this tutorial, we’ll implement a easy demo agent that returns a random quantity, serving to you perceive the core construction and circulate of the A2A protocol by hands-on code.

Organising the dependencies

We’ll first arrange the environment and begin with putting in the uv bundle supervisor. For Mac or Linux:

curl -LsSf https://astral.sh/uv/set up.sh | sh 

For Home windows (PowerShell):

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/set up.ps1 | iex"

We’ll then create a brand new undertaking listing and initialize it with uv

uv init a2a-demo
cd a2a-demo

We will now create and activate a digital atmosphere. For Mac or Linux:

uv venv
supply .venv/bin/activate

For Home windows:

uv venv
.venvScriptsactivate

We’ll now set up the required dependencies

uv add a2a-sdk python-a2a uvicorn

Implementing the Core Constructing Blocks

Agent Executor (agent_executor.py)

On this step, we implement the core logic of our agent by creating an Agent Executor, which is answerable for dealing with incoming requests and returning responses within the A2A format. The RandomNumberAgentExecutor wraps a easy RandomNumberAgent that generates a random quantity between 1 and 100. When a request is available in, the execute methodology calls the agent’s logic and pushes the outcome into the occasion queue as a standardized A2A message. This setup kinds the backend logic that A2A purchasers can work together with. Try the Full Codes on GitHub

import random
from a2a.server.agent_execution import AgentExecutor
from a2a.server.agent_execution.context import RequestContext
from a2a.server.occasions.event_queue import EventQueue
from a2a.utils import new_agent_text_message
from pydantic import BaseModel


class RandomNumberAgent(BaseModel):
    """Generates a random quantity between 1 and 100"""

    async def invoke(self) -> str:
        quantity = random.randint(1, 100)
        return f"Random quantity generated: {quantity}"


class RandomNumberAgentExecutor(AgentExecutor):

    def __init__(self):
        self.agent = RandomNumberAgent()

    async def execute(self, context: RequestContext, event_queue: EventQueue):
        outcome = await self.agent.invoke()
        await event_queue.enqueue_event(new_agent_text_message(outcome))

    async def cancel(self, context: RequestContext, event_queue: EventQueue):
        increase Exception("Cancel not supported")

Setting Up the A2A Server and Agent Card (predominant.py)

On this part, we outline the metadata that describes what our agent can do — that is referred to as the Agent Card. Consider it because the agent’s enterprise card, containing info like its title, description, obtainable abilities, enter/output sorts, and model.

We additionally register the agent’s abilities, which outline the form of duties it might probably deal with. In our case, it features a ability to generate a random quantity, tagged appropriately and with instance prompts.

As soon as the metadata is prepared, we configure the A2A server utilizing A2AStarletteApplication. We offer the agent card and join it with our customized agent logic utilizing a DefaultRequestHandler, which makes use of the RandomNumberAgentExecutor we carried out earlier. Lastly, we run the server utilizing uvicorn so the agent can begin listening for incoming A2A messages on port 9999.

This setup permits our agent to obtain standardized A2A messages, course of them, and reply in a structured approach — following the A2A protocol. Try the Full Codes on GitHub

import uvicorn
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.duties import InMemoryTaskStore
from a2a.sorts import AgentCapabilities, AgentCard, AgentSkill
from agent_executor import RandomNumberAgentExecutor


def predominant():
    # Outline the ability metadata
    ability = AgentSkill(
        id="random_number",
        title="Random Quantity Generator",
        description="Generates a random quantity between 1 and 100",
        tags=["random", "number", "utility"],
        examples=["Give me a random number", "Roll a number", "Random"],
    )

    # Outline the agent metadata
    agent_card = AgentCard(
        title="Random Quantity Agent",
        description="An agent that returns a random quantity between 1 and 100",
        url="http://localhost:9999/",
        defaultInputModes=["text"],
        defaultOutputModes=["text"],
        abilities=[skill],
        model="1.0.0",
        capabilities=AgentCapabilities(),
    )

    # Configure the request handler with our customized agent executor
    request_handler = DefaultRequestHandler(
        agent_executor=RandomNumberAgentExecutor(),
        task_store=InMemoryTaskStore(),
    )

    # Create the A2A app server
    server = A2AStarletteApplication(
        http_handler=request_handler,
        agent_card=agent_card,
    )

    # Run the server
    uvicorn.run(server.construct(), host="0.0.0.0", port=9999)


if __name__ == "__main__":
    predominant()

Interacting with the Agent Utilizing A2AClient (shopper.py)

Subsequent, we create the shopper that may work together with our A2A agent. This shopper script performs three predominant duties:

• Fetch the Agent Card: We begin by resolving the agent’s public metadata utilizing A2ACardResolver. This fetches the agent.json file from the .well-known endpoint, which accommodates important particulars just like the agent’s title, description, abilities, and communication capabilities.

• Initialize the A2A Shopper: Utilizing the fetched AgentCard, we arrange an A2AClient, which handles the communication protocol. This shopper will probably be answerable for sending structured messages to the agent and receiving responses.

Ship a Message and Obtain a Response: We assemble a message with the textual content “Give me a random quantity” utilizing A2A’s message construction (Message, Half, TextPart). The message is distributed as a part of a SendMessageRequest, which wraps it with a novel request ID. As soon as the message is distributed, the agent processes it and responds with a generated random quantity, which is then printed in JSON format. Try the Full Codes on GitHub

import uuid
import httpx
from a2a.shopper import A2ACardResolver, A2AClient
from a2a.sorts import (
    AgentCard,
    Message,
    MessageSendParams,
    Half,
    Function,
    SendMessageRequest,
    TextPart,
)

PUBLIC_AGENT_CARD_PATH = "/.well-known/agent.json"
BASE_URL = "http://localhost:9999"


async def predominant() -> None:
    async with httpx.AsyncClient() as httpx_client:
        # Fetch the agent card
        resolver = A2ACardResolver(httpx_client=httpx_client, base_url=BASE_URL)
        strive:
            print(f"Fetching public agent card from: {BASE_URL}{PUBLIC_AGENT_CARD_PATH}")
            agent_card: AgentCard = await resolver.get_agent_card()
            print("Agent card fetched efficiently:")
            print(agent_card.model_dump_json(indent=2))
        besides Exception as e:
            print(f"Error fetching public agent card: {e}")
            return

        # Initialize A2A shopper with the agent card
        shopper = A2AClient(httpx_client=httpx_client, agent_card=agent_card)

        # Construct message
        message_payload = Message(
            function=Function.consumer,
            messageId=str(uuid.uuid4()),
            elements=[Part(root=TextPart(text="Give me a random number"))],
        )
        request = SendMessageRequest(
            id=str(uuid.uuid4()),
            params=MessageSendParams(message=message_payload),
        )

        # Ship message
        print("Sending message...")
        response = await shopper.send_message(request)

        # Print response
        print("Response:")
        print(response.model_dump_json(indent=2))


if __name__ == "__main__":
    import asyncio
    asyncio.run(predominant())

Operating the Agent and querying the identical

To check our A2A setup, we’ll begin by operating the agent server. That is executed by executing the principle.py file, which initializes the agent, exposes its agent card, and begins listening for incoming requests on port 9999. Try the Full Codes on GitHub

As soon as the agent is up and operating, we’ll transfer to the shopper script. The shopper will fetch the agent’s metadata, ship a structured question utilizing the A2A protocol, and obtain a response. In our case, the question is a straightforward message like “Give me a random quantity”, and the agent will return a quantity between 1 and 100.

Try the Full Codes on GitHub. All credit score for this analysis goes to the researchers of this undertaking. Additionally, be at liberty to comply with us on Twitter and don’t neglect to hitch our 100k+ ML SubReddit and Subscribe to our Publication.

I’m a Civil Engineering Graduate (2022) from Jamia Millia Islamia, New Delhi, and I’ve a eager curiosity in Information Science, particularly Neural Networks and their software in numerous areas.