Создание A2A-совместимого агента случайных чисел: пошаговое руководство по реализации низкоуровневого шаблона исполнителя с Python

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

В этом уроке мы реализуем простой демонстрационный агент, который возвращает случайное число, помогая вам понять основную структуру и поток протокола A2A через практический код.

Настройка зависимости

Сначала мы настроим нашу среду и начнем с установки менеджера УФ -пакета. Для 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.py)

На этом этапе мы реализуем основную логику нашего агента, создав исполнителя агента, который отвечает за обработку входящих запросов и возврат ответов в формате A2A. А Randomnumberagentexecutor Окутает простой Randomnumberagent Это генерирует случайное число от 1 до 100. Когда входит запрос, метод выполнения вызывает логику агента и вкладывает результат в очередь событий в качестве стандартизированного сообщения A2A. Эта настройка формирует логику бэкэнд, с которой могут взаимодействовать клиенты A2A. Проверьте Полные коды на 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")

Настройка сервера A2A Server и Agent Card (main.py)

В этом разделе мы определяем метаданные, которые описывают, что может сделать наш агент – это называется Агент картаПолем Думайте об этом как о визитной карточке агента, содержащей информацию, такую ​​как ее имя, описание, доступные навыки, типы ввода/вывода и версию.

Мы также регистрируем навыки агента, которые определяют, какие задачи, с которыми он может выполнить. В нашем случае он включает навык для генерации случайного числа, помеченного соответствующим образом и с примерами подсказок.

Как только метаданные будут готовы, мы настроим сервер A2A, используя A2astarletteapplicationПолем Мы предоставляем карту агента и подключаем ее с нашей пользовательской логикой агента, используя Defaultrequesthandlerкоторый использует Randomnumberagentexecutor Мы реализовали ранее. Наконец, мы запускаем сервер, используя Uvicorn, чтобы агент мог начать прослушивание входящих сообщений A2A в порту 9999Полем

Эта настройка позволяет нашему агенту получать стандартизированные сообщения A2A, обрабатывать их и реагировать структурированным образом – после протокола A2A. Проверьте Полные коды на 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()

Взаимодействие с агентом с использованием a2aclient (client.py)

Затем мы создаем клиента, который будет взаимодействовать с нашим агентом A2A. Этот сценарий клиента выполняет три основных задачи:

• Принесите карту агента: Мы начинаем с разрешения публичных метаданных агента с помощью A2acardresolver. Это получает файл Agent.json из известной конечной точки.

• Инициализировать клиент A2A: Используя полученную AgentCard, мы установили A2Client, который обрабатывает протокол связи. Этот клиент будет нести ответственность за отправку структурированных сообщений агенту и получение ответов.

Отправить сообщение и получить ответ: Мы строим сообщение с текстом «Дайте мне случайное число», используя структуру сообщения A2A (сообщение, часть, текстовая часть). Сообщение отправляется как часть Sendmessagerequest, который завершает его уникальным идентификатором запроса. После того, как сообщение отправлено, агент обрабатывает его и отвечает сгенерированным случайным числом, которое затем печатается в формате JSON. Проверьте Полные коды на 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())

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

Чтобы проверить нашу настройку A2A, мы начнем с запуска сервера агента. Это делается путем выполнения файла main.py, который инициализирует агент, раскрывает свою карту агента и начинает прислушиваться к входящим запросам в порту 9999. Проверьте Полные коды на GitHub

Как только агент запущен и запускается, мы перейдем к сценарию клиента. Клиент принесет метаданные агента, отправляет структурированный запрос с помощью протокола A2A и получит ответ. В нашем случае запрос – это простое сообщение, например, «Дайте мне случайное число», и агент вернет число от 1 до 100.

Проверьте Полные коды на GitHubПолем Весь кредит на это исследование направлено на исследователей этого проекта. Кроме того, не стесняйтесь следить за нами Twitter И не забудьте присоединиться к нашему 100K+ ML Subreddit и подписаться на Наша информационный бюллетеньПолем

Я выпускник гражданского строительства (2022) из ​​Jamia Millia Islamia, Нью -Дели, и у меня интерес к науке о данных, особенно в нейронных сетях и их применении в различных областях.