AI

Building A2A Compliant Random Number Agent: A Step-by-Step Guide to Implementing Low-Level Executor Patterns with Python

The proxy-to-proxy (A2A) protocol is a new standard for Google that enables AI proxy (without the need to communicate and collaborate with its underlying framework or developers. It works by using standardized messages, proxy cards (describing what a proxy can do), and task-based execution, allowing proxy interaction over HTTP without custom integration logic. It makes it easier for A2A to build a scalable, interoperable multi-proxy system through the complexity of abstract communication.

In this tutorial, we will implement a simple demo agent that returns a random number that helps you understand the core structure and process of the A2A protocol through hands-on code.

Set up dependencies

We will first set up the environment and then start by installing the UV package manager. For Mac or Linux:

curl -LsSf  | sh 

For Windows (PowerShell):

powershell -ExecutionPolicy ByPass -c "irm  | iex"

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

uv init a2a-demo
cd a2a-demo

Now we can create and activate a virtual environment. For Mac or Linux:

uv venv
source .venv/bin/activate

For Windows:

uv venv
.venvScriptsactivate

Now we will install the required dependencies

uv add a2a-sdk python-a2a uvicorn

Implement core building blocks

Agent executor (agent_executor.py)

In this step, we implement the core logic of the agent by creating a proxy executor, which is responsible for handling incoming requests and returning the response in A2A format. this RandomNumberagentExecutor Pack a simple Randomnumberagent This generates a random number between 1 and 100. When a request comes in, the logic of the method call agent is executed and the result is pushed into the event queue as a standardized A2A message. This setting constitutes the backend logic that the A2A client can interact with. Check Full code on GitHub

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")

Set up A2A server and proxy card (main.py)

In this section, we define metadata that describes what a proxy can do – this is called Agent Card. Think of it as an agent’s business card containing information such as its name, description, available skills, input/output types and versions.

We also registered an agent’s skill that defines the tasks it can handle. In our case, it includes the skill to generate a random number, tagged appropriately and tagged under the example prompt.

Once the metadata is ready, we use the A2A server configuration a2astarletteapplication. We provide proxy cards and connect them using our custom proxy logic DefaultrequestHandler,use RandomNumberagentExecutor We implemented it before. Finally, we run the server using Uvicorn so that the proxy can start listening for A2A messages on the port 9999.

This setup allows our agent to receive standardized A2A messages, process them, and respond in a structured manner – in accordance with the A2A protocol. Check Full code on GitHub

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="
        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()

Interact with the proxy using A2ACLIENT (client.py)

Next, we create a client that will interact with our A2A proxy. This client script performs three main tasks:

  • Get a proxy card: We first use A2AcardResolver to resolve the agent’s public metadata. This will get the Agent.json file from the endpoint of .Well, which contains basic details such as the name, description, skills, and communication capabilities of the agent.
  • Initialize the A2A client: Using the acquired proxy card, we set up an A2ACLIENT that handles the communication protocol. The client will be responsible for sending structured messages to the agent and receiving responses.

Send a message and receive a reply: We use A2A’s message structure (message, part, textpart) to construct a message with the text “Give me a random number”. The message is sent as part of the SendMessagereQuest, which wraps it in a unique request ID. After sending the message, the agent will process and respond with the generated random number and then print it in JSON format. Check Full code on GitHub

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 = "


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())

Run the agent and query the same query

To test our A2A setup, we will start by running Agent Server. This is done by executing the Main.py file, which initializes the proxy, reveals its proxy card and starts listening to incoming requests from port 9999. Check Full code on GitHub

Once the agent is up and running, we will go to the client script. The client will get the metadata of the proxy, send a structured query using the A2A protocol, and then receive a response. In our case, the query is a simple message like “Give me a random number” and the proxy will return a number between 1 and 100.


Check Full code on GitHub. All credits for this study are to the researchers on the project. Also, please stay tuned for us twitter And don’t forget to join us 100K+ ml reddit And subscribe Our newsletter.


I am a civil engineering graduate in Islamic Islam in Jamia Milia New Delhi (2022) and I am very interested in data science, especially neural networks and their applications in various fields.

Related Articles

Leave a Reply

Your email address will not be published. Required fields are marked *

Back to top button