Constructing an A2A-Compliant Random Quantity Agent: A Step-by-Step Information to Implementing the Low-Degree Executor Sample with Python

The Agent-to-Agent (A2A) protocol is a brand new commonplace by Google that allows AI brokers—no matter their underlying framework or developer—to speak and collaborate seamlessly. It really works by utilizing standardized messages, agent playing cards (which describe what an agent can do), and task-based execution, permitting brokers to work together through HTTP with out customized integration logic. A2A makes it simpler to construct scalable, interoperable multi-agent techniques 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 movement of the A2A protocol via hands-on code.

Establishing the dependencies

We are going to first arrange our surroundings and begin with putting in the uv package deal 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 are going to 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 surroundings. For Mac or Linux:

uv venv
supply .venv/bin/activate

For Home windows:

uv venv
.venvScriptsactivate

We are going to 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 technique calls the agent’s logic and pushes the consequence 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):
        consequence = await self.agent.invoke()
        await event_queue.enqueue_event(new_agent_text_message(consequence))

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

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

On this part, we outline the metadata that describes what our agent can do — that is known 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 may 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 DefaultRequestHandlerwhich makes use of the RandomNumberAgentExecutor we applied earlier. Lastly, we run the server utilizing uvicorn so the agent can begin listening for incoming A2A messages on port 9999.

This setup allows our agent to obtain standardized A2A messages, course of them, and reply in a structured means — 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 most important():
    # Outline the ability metadata
    ability = AgentSkill(
        id="random_number",
        title="Random Quantity Generator",
        description="Generates a random quantity between 1 and 100",
        tags=("random", "quantity", "utility"),
        examples=("Give me a random quantity", "Roll a quantity", "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=("textual content"),
        defaultOutputModes=("textual content"),
        abilities=(ability),
        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__":
    most important()

Interacting with the Agent Utilizing A2AClient (shopper.py)

Subsequent, we create the shopper that can work together with our A2A agent. This shopper script performs three most important 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 comprises important particulars just like the agent’s title, description, abilities, and communication capabilities.

• Initialize the A2A Consumer: Utilizing the fetched AgentCard, we arrange an A2AClient, which handles the communication protocol. This shopper can 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 most important() -> None:
    async with httpx.AsyncClient() as httpx_client:
        # Fetch the agent card
        resolver = A2ACardResolver(httpx_client=httpx_client, base_url=BASE_URL)
        attempt:
            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(
            position=Function.consumer,
            messageId=str(uuid.uuid4()),
            elements=(Half(root=TextPart(textual content="Give me a random quantity"))),
        )
        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(most important())

Operating the Agent and querying the identical

To check our A2A setup, we’ll begin by working the agent server. That is achieved by executing the primary.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 working, 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 an easy 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 happy to comply with us on Twitter and don’t overlook to affix our 100k+ ML SubReddit and Subscribe to our E-newsletter.

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 utility in varied areas.