Constructing A2A-compliant Random Quantity Brokers: A Step-by-Step Information for Implementing Low-Stage Executor Patterns Utilizing Python

The Agent-to-Agent (A2A) protocol is a brand new normal for Google, permitting AI brokers from the underlying framework and builders to seamlessly talk and collaborate. It really works with standardized messages, agent playing cards (describe what brokers can do), and task-based execution, permitting brokers to work together over HTTP with out customized integration logic. A2A makes it straightforward to construct scalable, interoperable multi-agent programs by abstracting communication complexity.

This tutorial will implement a easy demo agent that returns random numbers and assist you perceive the core construction and circulate of the A2A protocol through sensible code.

Set dependencies

First, arrange your surroundings and begin by putting in the UV bundle supervisor. On Mac or Linux:

curl -LsSf https://astral.sh/uv/set up.sh | sh 

Home windows (PowerShell):

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

Subsequent, create a brand new venture listing and initialize it with UV

uv init a2a-demo
cd a2a-demo

Now you can create and activate a digital surroundings. On Mac or Linux:

uv venv
supply .venv/bin/activate

On Home windows:

uv venv
.venvScriptsactivate

Set up the required dependencies

uv add a2a-sdk python-a2a uvicorn

Implementing core constructing blocks

Agent executor (agent_executor.py)

On this step, you create an agent executor to implement the agent’s core logic. That is answerable for processing incoming requests and returning A2A format responses. randomnumberagentexecutor Easy wrap Random quantity fence This can generate a random quantity between 1 and 100. When a request is in, the execution methodology invokes the agent’s logic and pushes the end result into the occasion queue as a standardized A2A message. This setup kinds backend logic that permits A2A shoppers to work together with. Please test full Github code

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):
        elevate Exception("Cancel not supported")

A2A Server and Agent Card Setup (Principal.py)

This part defines metadata that explains what an agent can do. that is Agent Card. Consider it as an agent’s enterprise card that comprises data similar to its identify, description, accessible expertise, enter/output sort, model, and extra.

Additionally, you will register your agent expertise. This defines the kinds of duties that may be processed. In our case, it contains expertise to generate correctly tagged random numbers, and there are examples of prompts.

As soon as the metadata is prepared, configure it utilizing an A2A server a2astarlette software. Present agent playing cards and join utilizing customized agent logic defaultrequesthandleruse it randomnumberagentexecutor It was applied beforehand. Lastly, you may run the server utilizing uvicorn to start out listening to incoming A2A messages on the port 9999.

This setup permits brokers to obtain standardized A2A messages, course of them, and reply in a structured manner in line with the A2A protocol. Please test full Github code

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 talent metadata
    talent = AgentSkill(
        id="random_number",
        identify="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(
        identify="Random Quantity Agent",
        description="An agent that returns a random quantity between 1 and 100",
        url="http://localhost:9999/",
        defaultInputModes=["text"],
        defaultOutputModes=["text"],
        expertise=[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()

Work together with an agent utilizing A2Aclient (shopper.py)

Subsequent, create a shopper that interacts with the A2A agent. This shopper script performs three most important duties:

• Get an agent card: Begin by resolving the general public metadata for the agent utilizing A2AcardResolver. This can retrieve the Agent.json file from the .well-known Endpoint, which comprises obligatory particulars such because the agent’s identify, description, expertise, and communication capabilities.

• Initialize the A2A shopper: Use the Fetched AgentCard to configure the A2Aclient to deal with communication protocols. This shopper is answerable for sending structured messages to the agent and receiving responses.

Ship a message and obtain a response: Create a message with the textual content “Give Random Quantity” utilizing the A2A message construction (message, half, TextPart). The message is shipped as a part of a sendmessagerequest and wrapped with a singular request ID. When a message is shipped, the agent processes it, responds with the generated random quantity, and is printed in JSON format. Please test full Github code

import uuid
import httpx
from a2a.shopper import A2ACardResolver, A2AClient
from a2a.sorts import (
    AgentCard,
    Message,
    MessageSendParams,
    Half,
    Position,
    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=Position.consumer,
            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())

Run the agent and question the identical

To check your A2A setup, begin by working the agent server. That is accomplished by initializing the agent, exposing the agent card, and working the Principal.py file that begins listening to incoming requests on port 9999. full Github code

As soon as the agent is up and working, go to the shopper script. The shopper retrieves the agent’s metadata, sends a structured question utilizing the A2A protocol, and receives a response. On this case, the question is an easy message, similar to “give random numbers”, and the agent returns a quantity between 1 and 100.

Please test full Github code. All credit for this examine will likely be directed to researchers on this venture. Additionally, please be at liberty to comply with us Twitter And remember to affix us 100k+ ml subreddit And subscribe Our Newsletter.

I’m a civil engineering graduate (2022) from Jamia Milia Islamia, New Delhi, and have a powerful curiosity in knowledge science, significantly neural networks and purposes in quite a lot of fields.