Протокол 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
«`