Construyendo un agente de números aleatorios que cumplen con A2A: una guía paso a paso para implementar el patrón de ejecutor de bajo nivel con Python

El protocolo de agente a agente (A2A) es un nuevo estándar de Google que permite a los agentes de IA, independientemente de su marco subyacente o desarrollador, comunicarse y colaborar sin problemas. Funciona utilizando mensajes estandarizados, tarjetas de agente (que describen lo que puede hacer un agente) y la ejecución basada en tareas, lo que permite a los agentes interactuar a través de HTTP sin una lógica de integración personalizada. A2A facilita la construcción de sistemas de agentes múltiples escalables e interoperables abstraiendo las complejidades de la comunicación.

En este tutorial, implementaremos un agente de demostración simple que devuelva un número aleatorio, ayudándole a comprender la estructura y el flujo básicos del protocolo A2A a través del código práctico.

Configuración de las dependencias

Primero configuraremos nuestro entorno y comenzaremos con la instalación del Administrador de paquetes UV. Para Mac o Linux:

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

Para Windows (PowerShell):

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

Luego crearemos un nuevo directorio de proyectos e inicializaremos con UV

uv init a2a-demo
cd a2a-demo

Ahora podemos crear y activar un entorno virtual. Para Mac o Linux:

uv venv
source .venv/bin/activate

Para Windows:

uv venv
.venv\Scripts\activate

Ahora instalaremos las dependencias requeridas

uv add a2a-sdk python-a2a uvicorn

Implementación de los bloques de construcción principales

Ejecutor del agente (agente_executor.py)

En este paso, implementamos la lógica central de nuestro agente creando un ejecutor de agentes, que es responsable de manejar solicitudes entrantes y devolver las respuestas en el formato A2A. El RandomNumberAgentExecutor envuelve un simple Randomnumberagent Eso genera un número aleatorio entre 1 y 100. Cuando entra una solicitud, el método de ejecución llama a la lógica del agente y empuja el resultado a la cola de eventos como un mensaje A2A estandarizado. Esta configuración forma la lógica de backend con la que los clientes de A2A pueden interactuar. Mira el Lleno Códigos en 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")

Configuración del servidor A2A y la tarjeta de agente (main.py)

En esta sección, definimos los metadatos que describen lo que nuestro agente puede hacer, esto se llama el Tarjeta de agente. Piense en ello como la tarjeta de presentación del agente, que contiene información como su nombre, descripción, habilidades disponibles, tipos de entrada/salida y versión.

También registramos las habilidades del agente, que definen el tipo de tareas que puede manejar. En nuestro caso, incluye una habilidad para generar un número aleatorio, etiquetado adecuadamente y con indicaciones de ejemplo.

Una vez que los metadatos están listos, configuramos el servidor A2A usando A2ASTARLETTEAPLAPLACIÓN. Proporcionamos la tarjeta de agente y la conectamos con nuestra lógica de agente personalizado utilizando un Defaultequesthandlerque usa el RandomNumberAgentExecutor Implementamos anteriormente. Finalmente, ejecutamos el servidor usando Uvicorn para que el agente pueda comenzar a escuchar mensajes A2A entrantes en el puerto 9999.

Esta configuración permite a nuestro agente recibir mensajes A2A estandarizados, procesarlos y responder de manera estructurada, siguiendo el protocolo A2A. Mira el Lleno Códigos en 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="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()

Interactuar con el agente usando A2ACLient (Client.py)

A continuación, creamos el cliente que interactuará con nuestro agente A2A. Este script de cliente realiza tres tareas principales:

• Buscar la tarjeta de agente: Comenzamos resolviendo los metadatos públicos del agente usando A2ACardResolver. Esto obtiene el archivo agente.json desde el punto final. Bien conocido, que contiene detalles esenciales como el nombre, la descripción, las habilidades y las capacidades de comunicación del agente.

• Inicializar el cliente A2A: Utilizando el agente obtenido, configuramos un A2AClient, que maneja el protocolo de comunicación. Este cliente será responsable de enviar mensajes estructurados al agente y recibir respuestas.

Enviar un mensaje y recibir una respuesta: Construimos un mensaje con el texto “Dame un número aleatorio” usando la estructura de mensajes de A2A (mensaje, parte, TextPart). El mensaje se envía como parte de un SendMessagequest, que lo envuelve con una ID de solicitud única. Una vez que se envía el mensaje, el agente lo procesa y responde con un número aleatorio generado, que luego se imprime en formato JSON. Mira el Lleno Códigos en 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 = "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())

Ejecutar el agente y consultar lo mismo

Para probar nuestra configuración de A2A, comenzaremos ejecutando el servidor del agente. Esto se hace ejecutando el archivo main.py, que inicializa al agente, expone su tarjeta de agente y comienza a escuchar las solicitudes entrantes en el puerto 9999. Consulte el Lleno Códigos en Github

Una vez que el agente esté en funcionamiento, nos pasaremos al script del cliente. El cliente obtendrá los metadatos del agente, enviará una consulta estructurada utilizando el protocolo A2A y recibirá una respuesta. En nuestro caso, la consulta es un mensaje simple como “Dame un número aleatorio”, y el agente devolverá un número entre 1 y 100.

Mira el Lleno Códigos en Github. Todo el crédito por esta investigación va a los investigadores de este proyecto. Además, siéntete libre de seguirnos Gorjeo Y no olvides unirte a nuestro Subreddit de 100k+ ml y suscribirse a Nuestro boletín.

Soy un graduado de ingeniería civil (2022) de Jamia Millia Islamia, Nueva Delhi, y tengo un gran interés en la ciencia de datos, especialmente las redes neuronales y su aplicación en varias áreas.