Протокол Agent-to-Agent (A2A) — это новый стандарт от Google, который позволяет агентам искусственного интеллекта (ИИ), независимо от их базовой платформы или разработчика, беспрепятственно взаимодействовать и сотрудничать. Он работает с использованием стандартизированных сообщений, карточек агентов (которые описывают возможности агента) и выполнения задач на основе задач, позволяя агентам взаимодействовать через HTTP без необходимости специальной логики интеграции. A2A упрощает создание масштабируемых, совместимых мультиагентных систем, абстрагируя сложности коммуникации.
В этом руководстве мы реализуем простого демонстрационного агента, который возвращает случайное число, помогая вам понять основную структуру и порядок работы протокола A2A с помощью практического кода.
Настройка зависимостей
Сначала мы настроим нашу среду и начнём с установки менеджера пакетов `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`:
“`
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. Когда поступает запрос, метод `execute` вызывает логику агента и помещает результат в очередь событий в виде стандартизированного сообщения A2A. Эта настройка формирует внутреннюю логику, с которой могут взаимодействовать клиенты A2A.
“`python
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 newagenttext_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 eventqueue.enqueueevent(newagenttext_message(result))
async def cancel(self, context: RequestContext, event_queue: EventQueue):
raise Exception(“Cancel not supported”)
“`
Настройка сервера A2A и карточки агента (main.py)
В этом разделе мы определяем метаданные, которые описывают возможности нашего агента — это называется карточкой агента. Мы также регистрируем навыки агента, которые определяют тип задач, которые он может обрабатывать. В нашем случае это включает навык генерации случайного числа, помеченный соответствующим образом и с примерами подсказок.
“`python
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(
httphandler=requesthandler,
agentcard=agentcard,
)
# Run the server
uvicorn.run(server.build(), host=”0.0.0.0″, port=9999)
if name == “main“:
main()
“`
Взаимодействие с агентом с помощью A2AClient (client.py)
Далее мы создаём клиента, который будет взаимодействовать с нашим агентом A2A. Этот клиентский скрипт выполняет три основные задачи:
1. Извлечение карточки агента: мы начинаем с разрешения общедоступного метаданных агента с помощью `A2ACardResolver`. Это извлекает файл `agent.json` из конечной точки `.well-known`, который содержит важные сведения, такие как имя агента, описание, навыки и возможности связи.
2. Инициализация клиента A2A: используя извлечённую `AgentCard`, мы настраиваем `A2AClient`, который обрабатывает протокол связи. Этот клиент будет отвечать за отправку структурированных сообщений агенту и получение ответов.
3. Отправка сообщения и получение ответа: мы конструируем сообщение с текстом «Give me a random number» с помощью структуры сообщений A2A (`Message`, `Part`, `TextPart`). Сообщение отправляется как часть `SendMessageRequest`, которое оборачивает его уникальным идентификатором запроса. После отправки сообщения агент обрабатывает его и отвечает сгенерированным случайным числом, которое затем выводится в формате JSON.
“`python
import uuid
import httpx
from a2a.client import A2ACardResolver, A2AClient
from a2a.types import (
AgentCard,
Message,
MessageSendParams,
Part,
Role,
SendMessageRequest,
TextPart,
)
PUBLICAGENTCARD_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(httpxclient=httpxclient, baseurl=BASEURL)
try:
print(f”Fetching public agent card from: {BASEURL}{PUBLICAGENTCARDPATH}”)
agentcard: AgentCard = await resolver.getagent_card()
print(“Agent card fetched successfully:”)
print(agentcard.modeldump_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(httpxclient=httpxclient, agentcard=agentcard)
# 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.modeldumpjson(indent=2))
if name == “main“:
import asyncio
asyncio.run(main())
“`
Запуск агента и запрос
Чтобы протестировать нашу настройку A2A, мы сначала запустим сервер агента. Это делается путём выполнения файла `main.py`, который инициализирует агента, выставляет его карточку агента и начинает прослушивание входящих запросов на порту 9999.
“`
uv run main.py
“`
После запуска агента мы перейдём к клиентскому скрипту. Клиент извлечёт метаданные агента, отправит структурированный запрос с помощью протокола A2A и получит ответ. В нашем случае запрос представляет собой простое сообщение типа «Give me a random number», и агент вернёт число от 1 до 100.
“`
uv run client.py
“`