Пошаговое руководство по созданию агента генерации случайных чисел с использованием протокола A2A на Python

Что такое протокол Agent-to-Agent (A2A)

Протокол Agent-to-Agent (A2A) — это новый стандарт от Google, который позволяет ИИ-агентам независимо от их платформы или разработчика общаться и сотрудничать без проблем. Он использует стандартизированные сообщения, карточки агентов, описывающие их возможности, и выполнение задач, обеспечивая взаимодействие через HTTP без необходимости в кастомной логике интеграции. Такой подход упрощает создание масштабируемых и совместимых мультиагентных систем, скрывая сложности коммуникации.

Настройка окружения

Сначала установите менеджер пакетов uv. Для Mac или Linux выполните:

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

Для Windows (PowerShell):

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

Создайте каталог проекта и инициализируйте его:

uv init a2a-demo
cd a2a-demo

Создайте и активируйте виртуальное окружение:

Mac/Linux:

uv venv
source .venv/bin/activate

Windows:

uv venv
.venv\Scripts\activate

Установите необходимые зависимости:

uv add a2a-sdk python-a2a uvicorn

Реализация Agent Executor

Основная логика агента реализована в Agent Executor, который обрабатывает входящие запросы и возвращает ответы в формате A2A. Класс RandomNumberAgentExecutor оборачивает агента, генерирующего случайное число от 1 до 100. При вызове метода execute агент генерирует число и отправляет стандартизированное сообщение A2A.

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):
    """Генерирует случайное число от 1 до 100"""
 
    async def invoke(self) -> str:
        number = random.randint(1, 100)
        return f"Сгенерировано случайное число: {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("Отмена не поддерживается")

Настройка сервера A2A и карточки агента

Карточка агента описывает метаданные: имя, описание, навыки, режимы ввода/вывода и версию. Навык генерации случайного числа регистрируется с примерами запросов.

Сервер создаётся с помощью A2AStarletteApplication, который принимает карточку агента и соединяет её с кастомным executor через DefaultRequestHandler. Сервер запускается на порту 9999 с помощью uvicorn.

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="Генератор случайных чисел",
        description="Генерирует случайное число от 1 до 100",
        tags=["random", "number", "utility"],
        examples=["Дай случайное число", "Брось число", "Случайное"],
    )
 
    agent_card = AgentCard(
        name="Агент генерации случайных чисел",
        description="Агент, возвращающий случайное число от 1 до 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()

Взаимодействие с агентом через A2AClient

Клиент получает публичную карточку агента с помощью A2ACardResolver, создаёт A2AClient и отправляет структурированное сообщение с просьбой сгенерировать число. Агент обрабатывает запрос и возвращает ответ.

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"Получение публичной карточки агента: {BASE_URL}{PUBLIC_AGENT_CARD_PATH}")
            agent_card: AgentCard = await resolver.get_agent_card()
            print("Карточка агента получена:")
            print(agent_card.model_dump_json(indent=2))
        except Exception as e:
            print(f"Ошибка получения карточки агента: {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="Дай случайное число"))],
        )
        request = SendMessageRequest(
            id=str(uuid.uuid4()),
            params=MessageSendParams(message=message_payload),
        )
 
        print("Отправка сообщения...")
        response = await client.send_message(request)
 
        print("Ответ:")
        print(response.model_dump_json(indent=2))
 
if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

Запуск и тестирование агента

Запустите сервер агента командой:

uv run main.py

После запуска выполните клиентский скрипт:

uv run client.py

Таким образом демонстрируется создание и взаимодействие с агентом, соответствующим протоколу A2A на Python.

Для полного кода и подробностей смотрите репозиторий на GitHub, ссылка в оригинальной статье.