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 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 through HTTP with out customized integration logic. A2A makes it simpler to construct scalable, interoperable multi-agent programs 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 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 are going to then create a brand new venture listing and initialize it with uv

uv init a2a-demo
cd a2a-demo

We will now create and activate a digital setting. 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 chargeable 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 end result into the occasion queue as a standardized A2A message. This setup kinds the backend logic that A2A purchasers can work together with. Take a look at 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):
        end result = await self.agent.invoke()
        await event_queue.enqueue_event(new_agent_text_message(end result))

    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 referred to as the Agent Card. Consider it because the agent’s enterprise card, containing info like its title, description, out there abilities, enter/output varieties, and model.

We additionally register the agent’s abilities, which outline the type of duties it might probably deal with. In our case, it features a talent 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 method — following the A2A protocol. Take a look at 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.varieties import AgentCapabilities, AgentCard, AgentSkill
from agent_executor import RandomNumberAgentExecutor


def most important():
    # Outline the talent metadata
    talent = 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__":
    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 incorporates 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 will likely be chargeable 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 shipped as a part of a SendMessageRequest, which wraps it with a singular request ID. As soon as the message is shipped, the agent processes it and responds with a generated random quantity, which is then printed in JSON format. Take a look at the Full Codes on GitHub

import uuid
import httpx
from a2a.shopper import A2ACardResolver, A2AClient
from a2a.varieties 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)
        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(
            position=Function.person,
            messageId=str(uuid.uuid4()),
            components=[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(most important())

Working the Agent and querying the identical

To check our A2A setup, we’ll begin by operating the agent server. That is carried out by executing the primary.py file, which initializes the agent, exposes its agent card, and begins listening for incoming requests on port 9999. Take a look at 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.

Take a look at the Full Codes on GitHub. All credit score for this analysis goes to the researchers of this venture. Additionally, be happy to observe us on Twitter and don’t overlook to affix 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 Knowledge Science, particularly Neural Networks and their utility in varied areas.