<RETURN_TO_BASE

Step-by-Step Guide to Building a Random Number Agent Using the A2A Protocol with Python

This tutorial guides you through building a random number agent following Google's A2A protocol with Python, covering setup, implementation, and client interaction.

Understanding the Agent-to-Agent (A2A) Protocol

The Agent-to-Agent (A2A) protocol is an innovative standard developed by Google that allows AI agents to communicate and collaborate seamlessly, regardless of their underlying frameworks or developers. It achieves this by using standardized messages, agent cards that describe agent capabilities, and task-based execution, enabling agents to interact over HTTP without the need for custom integration logic. This protocol simplifies the development of scalable and interoperable multi-agent systems by abstracting complex communication details.

Setting Up the Environment

To begin, install the uv package manager. On Mac or Linux, run:

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

On Windows PowerShell:

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

Create a new project directory and initialize it:

uv init a2a-demo
cd a2a-demo

Create and activate a virtual environment:

  • Mac/Linux:
uv venv
source .venv/bin/activate
  • Windows:
uv venv
.venv\Scripts\activate

Install required dependencies:

uv add a2a-sdk python-a2a uvicorn

Implementing the Agent Executor

The core logic resides in an Agent Executor, which handles incoming requests and responds following the A2A protocol. The RandomNumberAgentExecutor wraps a simple agent generating a random number between 1 and 100. When executed, it generates the number and sends a standardized A2A message back.

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

Configuring the A2A Server and Agent Card

The Agent Card defines metadata describing the agent's capabilities, such as its name, description, skills, input/output modes, and version. We register a skill for generating a random number with example prompts.

The server uses A2AStarletteApplication, which receives the agent card and connects it with the custom executor through a DefaultRequestHandler. Uvicorn runs the server on port 9999, enabling the agent to listen for A2A messages.

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():
    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"],
    )
 
    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(),
    )
 
    request_handler = DefaultRequestHandler(
        agent_executor=RandomNumberAgentExecutor(),
        task_store=InMemoryTaskStore(),
    )
 
    server = A2AStarletteApplication(
        http_handler=request_handler,
        agent_card=agent_card,
    )
 
    uvicorn.run(server.build(), host="0.0.0.0", port=9999)
 
if __name__ == "__main__":
    main()

Interacting with the Agent via A2AClient

The client script fetches the agent's public metadata using A2ACardResolver, initializes an A2AClient with the agent card, and sends a structured message requesting a random number. The agent processes the request and responds accordingly.

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:
        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
 
        client = A2AClient(httpx_client=httpx_client, agent_card=agent_card)
 
        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),
        )
 
        print("Sending message...")
        response = await client.send_message(request)
 
        print("Response:")
        print(response.model_dump_json(indent=2))
 
if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

Running and Testing the Agent

Start the agent server by running:

uv run main.py

Once the server is running, execute the client to send the request and receive the random number:

uv run client.py

This completes the demonstration of building and interacting with an A2A-compliant random number agent using Python.

For complete code and further details, check the GitHub repository linked in the original article.

🇷🇺

Сменить язык

Читать эту статью на русском

Переключить на Русский