Build a random number agent in accordance with A2A: a step -by -step guide to implement the low -level executor model with Python

The agent agent protocol (A2A) is a new Google standard which allows AI agents – constantly their frame or underlying developer – to communicate and collaborate transparently. It works using standardized messages, agent cards (which describe what an agent can do) and an execution based on tasks, allowing agents to interact via HTTP without logic of personalized integration. A2A facilitates the creation of evolutionary and interoperable multi-agent systems in abstraction of the complexity of communication.

In this tutorial, we will implement a simple demonstration agent which returns a random number, helping you to understand the nucleus structure and the flow of the A2A protocol via a practical code.

Dependencies configuration

We will first configure our environment and start with the installation of the UV package manager. For Mac or Linux:

curl -LsSf https://astral.sh/uv/install.sh | sh 

For Windows (PowerShell):

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

We will then create a new project repertoire and initialize it with UV

uv init a2a-demo
cd a2a-demo

We can now create and activate a virtual environment. For Mac or Linux:

uv venv
source .venv/bin/activate

For Windows:

uv venv
.venv\Scripts\activate

We will now install the required dependencies

uv add a2a-sdk python-a2a uvicorn

Implementation of basic elements

Executor agent (agent_executor.py)

In this stage, we implement the basic logic of our agent by creating an agent executor, who is responsible for the management of incoming requests and return responses to A2A format. THE Rumanningexcutor Envelop a simple Random This generates a random number between 1 and 100. When a request arrives, the execution method calls the logic of the agent and pushes the result in the queue of events as a standardized A2A message. This configuration constitutes the backend logic with which A2A customers can interact. Discover the Complete GitHub codes

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


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

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


class RandomNumberAgentExecutor(AgentExecutor):

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

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

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

Configuration of the A2A server and the agent card (Main.Py)

In this section, we define the metadata that describe what our agent can do – this is called the Agent card. Consider it as the agent's business card, containing information such as its name, description, available skills, input / output types and version.

We also record the skills of the agent, which define the type of tasks he can manage. In our case, it includes a skill to generate a random number, labeled appropriately and with examples of prompts.

Once the metadata is ready, we configure the A2A server using A2astarlettePplication. We provide the agent card and connect it with our personalized agent logic using a Defaultrequesthandlerwhich uses the Rumanningexcutor We implemented earlier. Finally, we execute the server using Uvicorn so that the agent can start listening to A2A messages incoming on the port 9999.

This configuration allows our agent to receive standardized A2A messages, to process them and to respond in a structured manner – by following the A2A protocol. Discover the Complete GitHub codes

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


def main():
    # Define the skill metadata
    skill = AgentSkill(
        id="random_number",
        name="Random Number Generator",
        description="Generates a random number between 1 and 100",
        tags=("random", "number", "utility"),
        examples=("Give me a random number", "Roll a number", "Random"),
    )

    # Define the agent metadata
    agent_card = AgentCard(
        name="Random Number Agent",
        description="An agent that returns a random number between 1 and 100",
        url="http://localhost:9999/",
        defaultInputModes=("text"),
        defaultOutputModes=("text"),
        skills=(skill),
        version="1.0.0",
        capabilities=AgentCapabilities(),
    )

    # Configure the request handler with our custom 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.build(), host="0.0.0.0", port=9999)


if __name__ == "__main__":
    main()

Interaction with the agent using A2aclient (client.py)

Then we create the customer who will interact with our A2A agent. This customer script performs three main tasks:

• Find the agent's card: We start by resolving the agent's public metadata using A2acardresolver. This recovers the agent.json file from the .well-connu termination point, which contains essential details such as the name, description, skills and communication capacities of the agent.

• Initializing the A2A customer: Using the recovered agent card, we have configured an A2aclient, which manages the communication protocol. This customer will be responsible for sending structured messages to the agent and receiving answers.

Send a message and receive an answer: We build a message with the text “Give me a random number” using the A2A message structure (message, part, textpart). The message is sent as part of a Sendmessagerequest, which envelops it with a single request ID. Once the message is sent, the agent treats it and responds with a random number generated, which is then printed in JSON format. Discover the Complete GitHub codes

import uuid
import httpx
from a2a.client import A2ACardResolver, A2AClient
from a2a.types import (
    AgentCard,
    Message,
    MessageSendParams,
    Part,
    Role,
    SendMessageRequest,
    TextPart,
)

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


async def main() -> None:
    async with httpx.AsyncClient() as httpx_client:
        # Fetch the agent card
        resolver = A2ACardResolver(httpx_client=httpx_client, base_url=BASE_URL)
        try:
            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 successfully:")
            print(agent_card.model_dump_json(indent=2))
        except Exception as e:
            print(f"Error fetching public agent card: {e}")
            return

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

        # Build message
        message_payload = Message(
            role=Role.user,
            messageId=str(uuid.uuid4()),
            parts=(Part(root=TextPart(text="Give me a random number"))),
        )
        request = SendMessageRequest(
            id=str(uuid.uuid4()),
            params=MessageSendParams(message=message_payload),
        )

        # Send message
        print("Sending message...")
        response = await client.send_message(request)

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


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

Execute the agent and question

To test our A2A configuration, we will start by running the agent server. This is done by performing the Main.Py file, which initializes the agent, exposes his agent card and begins to listen to the incoming requests on port 9999. Consult the Complete GitHub codes

Once the agent is running, we will go to the customer script. The customer will recover the agent's metadata, send a structured request using the A2A protocol and receive an answer. In our case, the request is a simple message like “give me a random number”, and the agent will send a number between 1 and 100.

Discover the Complete GitHub codes. All the merit of this research goes to researchers in this project. Also, don't hesitate to follow us Twitter And don't forget to join our Subseubdredit 100k + ml and subscribe to Our newsletter.

I graduated in Civil Engineering (2022) by Jamia Millia Islamia, New Delhi, and I have a great interest in data science, in particular neural networks and their application in various fields.