Практическое руководство по протоколу Agent-to-Agent (A2A) от Google: архитектура, настройка агентов, обмен задачами, карточки агентов и потоковая передача.
Google A2A (Agent-to-Agent) — это открытый протокол, разработанный Google для стандартизации взаимодействия между AI-агентами. Представьте себе мир, где ваш персональный AI-ассистент может делегировать задачу специализированному агенту по анализу данных, который, в свою очередь, обращается к агенту визуализации — и всё это через единый, хорошо документированный протокол.
В основе A2A лежит простая идея: каждый агент публикует свою «карточку» (Agent Card) — JSON-документ, описывающий его возможности, навыки и точку входа. Другие агенты могут обнаружить эту карточку, понять, что умеет агент, и отправить ему задачу. Протокол использует знакомые веб-технологии: HTTP, JSON-RPC, REST и Server-Sent Events (SSE) для потоковой передачи.
Ключевое преимущество A2A — децентрализация. Вам не нужен центральный оркестратор. Агенты могут общаться напрямую, формируя динамические «созвездия» (constellations) специализированных агентов, каждый из которых решает свою часть общей задачи. Протокол уже поддерживается в экосистеме Google — включая интеграцию с Gemini, Vertex AI Agent Builder и Agent Development Kit (ADK).
В этом руководстве мы пройдём полный цикл: от установки SDK до создания полноценного созвездия из трёх агентов, обменивающихся задачами в реальном времени. Весь код рабочий — вы можете скопировать его и запустить на своей машине через 20 минут.
Google предоставляет официальный Python SDK для работы с A2A — пакет a2a-sdk. Он включает серверную и клиентскую части, утилиты для создания Agent Card, типы для задач и артефактов, а также встроенную поддержку потоковой передачи через Server-Sent Events. Установка тривиальна — один pip, и вы готовы к работе.
Для полноценной работы рекомендуется Python 3.10 или выше. Если вы планируете использовать A2A с Gemini (как LLM-бэкенд для агентов), дополнительно потребуется google-genai. Однако сам протокол A2A работает с любым LLM — OpenAI, Anthropic, локальные модели через Ollama.
# Создаём виртуальное окружение python3.11 -m venv a2a-env source a2a-env/bin/activate # Устанавливаем A2A SDK pip install a2a-sdk # Для интеграции с Gemini (опционально) pip install google-genai google-adk # Проверяем установку python -c "import a2a; print(a2a.__version__)" 0.2.0
SDK устанавливает все необходимые зависимости: pydantic для валидации моделей данных, httpx для асинхронных HTTP-запросов, sse-starlette для Server-Sent Events. После установки вы получаете полный инструментарий для создания как клиентских, так и серверных агентов.
Agent Card — краеугольный камень протокола A2A. Это JSON-документ, который описывает вашего агента: его имя, описание, список навыков (skills), поддерживаемые способы взаимодействия (текст, файлы, потоковая передача), а также URL-эндпоинты для отправки задач. По соглашению Agent Card размещается по пути /.well-known/agent.json на домене вашего агента.
Другие агенты «обнаруживают» (discover) вашего агента, запрашивая этот файл. На основе Agent Card они понимают, какие задачи ваш агент способен решать, в каком формате принимать входные данные и какие артефакты ожидать на выходе. Это похоже на OpenAPI-спецификацию, но адаптированную специально для AI-агентов.
from a2a.types import AgentCard, AgentSkill, AgentCapabilities from a2a.utils import AgentCardBuilder # Способ 1: ручное создание AgentCard (полный контроль) # Все поля валидируются через Pydantic — опечатки не пройдут agent_card = AgentCard( name="analyst-agent", description="Агент-аналитик: анализ данных, SQL-запросы, отчёты, визуализация. Специализируется на Pandas и Matplotlib.", url="https://agents.example.com/analyst", provider=dict( organization="QantCore AI", url="https://qantcore.space", ), version="1.0.0", capabilities=AgentCapabilities( streaming=True, # Поддержка SSE push_notifications=False, # Webhooks пока нет state_transition_history=True, # История смены состояний ), skills=[ AgentSkill( id="data_analysis", name="Анализ данных", description="Выполняет анализ CSV/JSON данных, строит сводные таблицы и графики", tags=["data", "pandas", "analytics", "sql"], examples=["Проанализируй sales.csv и найди топ-5 продуктов"], input_modes=["text", "file"], output_modes=["text", "file", "chart"], ), AgentSkill( id="sql_queries", name="SQL-запросы", description="Генерирует и выполняет SQL-запросы к базам данных", tags=["sql", "database", "query"], examples=["Напиши запрос для подсчёта MAU за последний квартал"], input_modes=["text"], output_modes=["text", "file"], ), ], default_input_modes=["text"], default_output_modes=["text"], supports_authenticated_extended_card=False, ) # Способ 2: AgentCardBuilder — удобный fluent-интерфейс card = ( AgentCardBuilder() .set_name("analyst-agent") .set_description("Аналитический агент") .set_url("https://agents.example.com/analyst") .add_skill("sql", lambda s: s .set_name("SQL Expert") .add_tag("database") ) .enable_streaming() .build() ) # Экспорт в JSON (готово для .well-known/agent.json) import json print(json.dumps(agent_card.model_dump(), indent=2, ensure_ascii=False))
Agent Card может быть статическим JSON-файлом или генерироваться динамически — например, если навыки агента зависят от загруженных плагинов или доступных инструментов. В production-сценариях карточку обычно отдают через веб-сервер (FastAPI, Flask) на эндпоинте GET /.well-known/agent.json. Поле skills — критически важное: именно по тегам и описаниям навыков другие агенты решают, кому делегировать задачу.
Сервер A2A — это HTTP-приложение, которое принимает задачи от клиентских агентов, обрабатывает их (самостоятельно или делегируя LLM) и возвращает результаты. SDK предоставляет готовый A2AServer, который легко интегрируется с FastAPI или Starlette.
Ключевые эндпоинты, которые автоматически создаются сервером: POST /tasks/send (создание задачи), GET /tasks/{task_id}/status (проверка статуса), а также SSE-поток для получения обновлений в реальном времени. Сервер управляет жизненным циклом задачи: от состояния pending до completed или failed.
import asyncio import uvicorn from a2a.server import A2AServer from a2a.server.tasks import InMemoryTaskStore from a2a.types import ( AgentCard, AgentCapabilities, AgentSkill, Task, TaskState, Message, Part, Artifact, ) from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware # --- Agent Card --- agent_card = AgentCard( name="weather-helper", description="Агент погоды: предоставляет прогноз для любого города мира через Open-Meteo API. Бесплатно, без ключей.", url="http://localhost:8000", version="1.0.0", capabilities=AgentCapabilities( streaming=True, push_notifications=False, state_transition_history=True, ), skills=[ AgentSkill( id="weather", name="Прогноз погоды", description="Получает текущую погоду и прогноз для указанного города", tags=["weather", "forecast"], examples=["Какая погода в Токио?", "Прогноз на неделю для Берлина"], input_modes=["text"], output_modes=["text"], ), ], default_input_modes=["text"], default_output_modes=["text"], ) # --- Обработчик задач (ядро логики агента) --- async def handle_task(task: Task) -> Task: """Выполняет задачу и возвращает результат. Вызывается сервером автоматически.""" import httpx, json # Извлекаем текст запроса из сообщения user_message = task.messages[-1].content.parts[0].text city = user_message.strip() # Запрашиваем Open-Meteo API (геокодинг города) async with httpx.AsyncClient() as client: geo_resp = await client.get( "https://geocoding-api.open-meteo.com/v1/search", params={"name": city, "count": 1, "language": "ru"} ) geo_data = geo_resp.json() if not geo_data.get("results"): task.state = TaskState.FAILED task.artifacts = [Artifact( parts=[Part(text=f"Город '{city}' не найден. Уточните название.")] )] return task loc = geo_data["results"][0] lat, lon = loc["latitude"], loc["longitude"] # Запрашиваем прогноз погоды weather_resp = await client.get( "https://api.open-meteo.com/v1/forecast", params={ "latitude": lat, "longitude": lon, "current": "temperature_2m,wind_speed_10m,weather_code", "timezone": "auto", } ) weather = weather_resp.json()["current"] result = ( f"🌍 Погода в {loc['name']}, {loc.get('country', '')}:\n" f"🌡 Температура: {weather['temperature_2m']}°C\n" f"💨 Ветер: {weather['wind_speed_10m']} км/ч\n" f"🔮 Код погоды: {weather['weather_code']}" ) task.state = TaskState.COMPLETED task.artifacts = [Artifact(parts=[Part(text=result)])] return task # --- Собираем приложение --- app = FastAPI(title="Weather A2A Agent") app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) task_store = InMemoryTaskStore() # Для продакшена — RedisTaskStore server = A2AServer( agent_card=agent_card, task_store=task_store, task_handler=handle_task, host="0.0.0.0", port=8000, ) # Монтируем A2A-роуты на FastAPI server.configure_app(app, base_path="") if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000, log_level="info")
Запустите сервер командой python server.py и откройте http://localhost:8000/.well-known/agent.json — вы увидите Agent Card вашего агента. Сервер готов принимать задачи через POST /tasks/send. Для production замените InMemoryTaskStore на персистентное хранилище задач.
Клиент A2A отвечает за обнаружение удалённых агентов (discovery), формирование задач и отправку их на сервер. SDK предоставляет класс A2AClient с методами для синхронного и асинхронного взаимодействия. Клиент умеет читать Agent Card удалённого агента, валидировать совместимость навыков и отслеживать статус задачи вплоть до её завершения.
Типичный сценарий: клиентский агент получает запрос от пользователя, анализирует его, понимает, что нужен специализированный навык (например, «погода»), находит подходящего агента через его Agent Card и отправляет ему подзадачу. Результат возвращается пользователю — прозрачно, как будто всё сделал один агент.
import asyncio from a2a.client import A2AClient from a2a.types import ( Message, Part, SendTaskRequest, SendTaskResponse, TaskState, TaskStatus, ) async def ask_weather_agent(city: str, server_url: str = "http://localhost:8000"): """Отправляет задачу агенту погоды и ждёт результат.""" async with A2AClient(server_url) as client: # Шаг 1: Discover — читаем Agent Card удалённого агента agent_card = await client.get_agent_card() print(f"🔍 Обнаружен агент: {agent_card.name} v{agent_card.version}") # Проверяем, что агент умеет работать с погодой skill_ids = [s.id for s in agent_card.skills] if "weather" not in skill_ids: raise ValueError("Агент не поддерживает прогноз погоды!") print(f"✅ Навыки агента: {skill_ids}") # Шаг 2: Формируем задачу task_request = SendTaskRequest( message=Message( role="user", content=Part(text=city), ), # Опционально: указать конкретный навык skill_id="weather", ) # Шаг 3: Отправляем задачу (синхронный режим — ждём ответа) response: SendTaskResponse = await client.send_task(task_request) task = response.task print(f"📤 Задача отправлена: {task.id}") print(f"📊 Статус: {task.state}") # Шаг 4: Если задача ещё выполняется — ждём (polling) while task.state in (TaskState.PENDING, TaskState.WORKING): await asyncio.sleep(0.5) task_status: TaskStatus = await client.get_task_status(task.id) task = task_status.task print(f"⏳ Состояние: {task.state}") # Шаг 5: Извлекаем результат if task.state == TaskState.COMPLETED and task.artifacts: result_text = task.artifacts[0].parts[0].text print(f"\n📋 Результат:\n{result_text}") return result_text elif task.state == TaskState.FAILED: error = task.artifacts[0].parts[0].text if task.artifacts else "Неизвестная ошибка" print(f"❌ Ошибка: {error}") return None if __name__ == "__main__": asyncio.run(ask_weather_agent("Токио"))
Клиент поддерживает несколько стратегий ожидания: синхронный polling (как в примере выше), колбэки и потоковую передачу через SSE (рассмотрим в следующей секции). Для продакшена рекомендуется использовать SSE — это снижает нагрузку на сеть и даёт мгновенную реакцию на изменение статуса задачи.
Server-Sent Events (SSE) — это механизм, при котором сервер отправляет клиенту поток событий через одно долгоживущее HTTP-соединение. В контексте A2A это идеальный способ получать обновления о ходе выполнения задачи: переходы между состояниями, частичные результаты (логи, промежуточные артефакты), прогресс-бары и финальный ответ.
Когда вы отправляете задачу с флагом streaming=True, сервер возвращает SSE-поток. Клиент может читать события по мере их поступления и реагировать мгновенно — показывать промежуточные результаты пользователю, обновлять UI, логировать прогресс. Это особенно важно для длительных задач: анализа больших данных, генерации отчётов, многошаговых цепочек рассуждений.
import asyncio, json from a2a.client import A2AClient from a2a.types import Message, Part, SendTaskRequest, TaskState async def stream_task_to_agent(server_url: str, prompt: str): """Отправляет задачу и читает поток SSE-событий от агента.""" async with A2AClient(server_url) as client: # Отправляем задачу с явным запросом streaming request = SendTaskRequest( message=Message(role="user", content=Part(text=prompt)), streaming=True, # 🔑 Ключевой флаг для SSE ) # Открываем SSE-поток print("📡 Открыт SSE-поток...") async for event in client.send_task_streaming(request): event_type = event.get("event", "unknown") data = json.loads(event.get("data", "{}")) if event_type == "task_state_update": state = data.get("state") task_id = data.get("task_id") print(f"🔄 [{task_id[:8]}] Состояние → {state}") if state == TaskState.COMPLETED: print("✅ Задача завершена!") if "artifacts" in data: for artifact in data["artifacts"]: for part in artifact.get("parts", []): if "text" in part: print(f"📋 Результат:\n{part['text']}") elif state == TaskState.FAILED: print("❌ Задача провалилась") elif event_type == "artifact_update": # Частичный артефакт — промежуточный результат artifact = data.get("artifact", {}) for part in artifact.get("parts", []): if "text" in part: print(f"📝 Частичный результат: {part['text'][:100]}...") elif event_type == "log": # Логи выполнения (полезно для отладки) print(f"📜 Лог: {data.get('message', '')}") elif event_type == "progress": pct = data.get("percentage", 0) print(f"⏳ Прогресс: {pct}%") print("🔌 SSE-поток закрыт") if __name__ == "__main__": asyncio.run(stream_task_to_agent( "http://localhost:8000", "Проанализируй данные о продажах за последний квартал" ))
Обратите внимание на типы событий в SSE-потоке: task_state_update (смена состояния — pending → working → completed), artifact_update (промежуточные результаты), log (отладочные сообщения) и progress (процент выполнения). Эти события позволяют строить насыщенные интерактивные интерфейсы, где пользователь видит живой прогресс работы агента.
Самое мощное применение A2A — «созвездия» (constellations): несколько агентов с разными навыками, работающих вместе над сложной задачей. В этом примере мы создадим трёх агентов и одного оркестратора, который распределяет подзадачи между ними.
Сценарий: пользователь спрашивает «Какой город лучше для отпуска в июле — Барселона или Стамбул?». Оркестратор разбивает запрос на две подзадачи (погода в Барселоне, погода в Стамбуле), отправляет их параллельно двум погодным агентам, затем передаёт результаты агенту-советнику, который сравнивает и выдаёт рекомендацию. Всё это — через A2A-протокол, с реальными HTTP-запросами между контейнерами.
import asyncio from a2a.client import A2AClient from a2a.types import Message, Part, SendTaskRequest, SendTaskResponse # ============================================================ # КОНФИГУРАЦИЯ АГЕНТОВ В СОЗВЕЗДИИ # ============================================================ AGENTS = { "weather_eu": { "url": "http://localhost:8001", "description": "Погода в Европе (Open-Meteo, с геокодингом)", }, "weather_asia": { "url": "http://localhost:8002", "description": "Погода в Азии (Open-Meteo, метрическая система)", }, "travel_advisor": { "url": "http://localhost:8003", "description": "Советник по путешествиям — сравнивает города и даёт рекомендации", }, } async def discover_agents(): """Обнаруживает всех агентов в созвездии и проверяет их навыки.""" discovered = {} for name, cfg in AGENTS.items(): async with A2AClient(cfg["url"]) as client: try: card = await client.get_agent_card() discovered[name] = {"client": client, "card": card, "url": cfg["url"]} print(f"✅ [{name}] {card.name} — навыки: {[s.id for s in card.skills]}") except Exception as e: print(f"❌ [{name}] Недоступен: {e}") return discovered async def orchestrate(user_query: str): """Оркестратор: разбирает запрос, делегирует агентам, собирает ответ.""" # Шаг 1: Обнаружение агентов agents = await discover_agents() if len(agents) < 3: return "❌ Не все агенты доступны — проверьте серверы" # Шаг 2: Распознаём города из запроса (упрощённо) cities = [] for city_candidate in ["Барселона", "Стамбул", "Париж", "Токио", "Лондон"]: if city_candidate.lower() in user_query.lower(): cities.append(city_candidate) print(f"🏙 Найденные города: {cities}") # Шаг 3: Параллельный запрос погоды для каждого города async def fetch_weather(city: str): """Выбираем подходящего погодного агента и запрашиваем.""" # Простая эвристика: Европа → weather_eu, остальное → weather_asia eu_cities = {"барселона", "париж", "лондон", "берлин", "рим"} agent_key = "weather_eu" if city.lower() in eu_cities else "weather_asia" agent = agents[agent_key] async with A2AClient(agent["url"]) as client: request = SendTaskRequest( message=Message(role="user", content=Part(text=city)), skill_id="weather", ) response: SendTaskResponse = await client.send_task(request) task = response.task # Ждём завершения (в реальном коде — SSE) while task.state.value in ("pending", "working"): await asyncio.sleep(0.3) status = await client.get_task_status(task.id) task = status.task if task.state.value == "completed" and task.artifacts: return {"city": city, "weather": task.artifacts[0].parts[0].text} return {"city": city, "weather": "❌ Ошибка получения погоды"} # Параллельный запуск — все города одновременно! weather_results = await asyncio.gather(*[fetch_weather(c) for c in cities]) print("\n🌤 Результаты погоды:") for wr in weather_results: print(f" {wr['city']}: {wr['weather'][:80]}...") # Шаг 4: Отправляем результаты советнику advisor_prompt = ( "Сравни следующие города для отпуска в июле и выбери лучший. " "Учитывай температуру, ветер и общие погодные условия.\n\n" ) for wr in weather_results: advisor_prompt += f"=== {wr['city']} ===\n{wr['weather']}\n\n" async with A2AClient(agents["travel_advisor"]["url"]) as client: request = SendTaskRequest( message=Message(role="user", content=Part(text=advisor_prompt)), skill_id="travel_advice", ) response = await client.send_task(request) task = response.task while task.state.value in ("pending", "working"): await asyncio.sleep(0.5) status = await client.get_task_status(task.id) task = status.task if task.state.value == "completed" and task.artifacts: recommendation = task.artifacts[0].parts[0].text print(f"\n✈️ ИТОГОВАЯ РЕКОМЕНДАЦИЯ:\n{recommendation}") return recommendation return "❌ Не удалось получить рекомендацию" if __name__ == "__main__": result = asyncio.run(orchestrate( "Какой город лучше для отпуска в июле — Барселона или Стамбул?" )) print("\n" + "=" * 50) print(result)
Этот пример демонстрирует ключевые паттерны A2A-оркестрации: discovery (обнаружение агентов через Agent Card), routing (маршрутизация подзадач к нужному агенту по навыкам), parallelization (параллельное выполнение независимых подзадач через asyncio.gather), и aggregation (сбор результатов и передача следующему агенту в цепочке). В реальном продакшене вы бы заменили хардкод-эвристики на LLM-роутер, а polling — на SSE-потоки.
Когда вы переходите от прототипа к production-развёртыванию A2A-агентов, на первый план выходят вопросы безопасности, надёжности и наблюдаемости. Протокол A2A спроектирован с учётом enterprise-требований, но правильная настройка — ваша ответственность.
A2A поддерживает несколько уровней безопасности. На транспортном уровне — стандартный HTTPS с mutual TLS (mTLS) для аутентификации обеих сторон. На уровне протокола — JWT-токены в заголовке Authorization: Bearer, которые передаются с каждым запросом к задачам. Для конфиденциальных Agent Card (с внутренними эндпоинтами) предусмотрен механизм supports_authenticated_extended_card=True, при котором полная карточка отдаётся только аутентифицированным клиентам.
# Production-конфигурация A2A-сервера с безопасностью from a2a.server import A2AServer from a2a.server.tasks import RedisTaskStore from a2a.server.auth import JWTAuthProvider server = A2AServer( agent_card=agent_card, task_store=RedisTaskStore( redis_url="redis://redis-cluster:6379/0", task_ttl=3600, # Храним задачи 1 час max_retries=3, # Retry при сбоях Redis ), task_handler=handle_task, auth_provider=JWTAuthProvider( jwks_url="https://auth.qantcore.space/.well-known/jwks.json", audience="a2a-agents", issuer="https://auth.qantcore.space", ), rate_limiter="token_bucket", # Ограничение: 100 задач/мин max_concurrent_tasks=10, # Не более 10 одновременных задач host="0.0.0.0", port=443, ssl_certfile="/etc/ssl/certs/agent.pem", ssl_keyfile="/etc/ssl/private/agent.key", )
Каждый A2A-сервер автоматически экспортирует метрики в формате OpenTelemetry. Вы можете подключить Prometheus для сбора метрик и Grafana для визуализации. Ключевые метрики: a2a_tasks_total (общее количество задач), a2a_task_duration_seconds (длительность выполнения), a2a_tasks_failed_total (ошибки), a2a_active_connections (активные SSE-соединения).
Задачи в A2A идемпотентны по дизайну: каждая задача имеет уникальный UUID, и повторная отправка той же задачи не создаёт дубликат. Используйте RedisTaskStore с персистентностью для сохранения задач между перезагрузками сервера. Настройте retry-логику на клиенте с экспоненциальной задержкой (exponential backoff). Если сервер временно недоступен, клиент A2A автоматически повторяет запрос.
Самая сильная сторона A2A — нативная интеграция с экосистемой Google AI. Gemini можно использовать как «мозг» вашего агента: он принимает решения о маршрутизации задач, генерирует ответы и даже dynamically создаёт новые A2A-задачи для других агентов. Vertex AI Agent Builder позволяет собирать A2A-агентов визуально, без написания кода.
Google Agent Development Kit (ADK) идёт ещё дальше: вы описываете агента в YAML-конфигурации, указываете его Agent Card, навыки и инструменты — и ADK автоматически генерирует полноценный A2A-совместимый сервер с поддержкой streaming, безопасности и мониторинга.
from google.adk import Agent from google.adk.tools import FunctionTool from google.genai import Client as GenAIClient # Создаём агента с Gemini-бэкендом и A2A-совместимостью agent = Agent( name="gemini-analyst", model="gemini-2.5-flash", # Быстрая и мощная модель description="Аналитический агент на Gemini с A2A-интерфейсом", instruction="""Ты — агент-аналитик с доступом к инструментам. При получении сложного запроса разбей его на подзадачи. Если подзадача требует специализированного навыка (погода, SQL, перевод), делегируй её другому агенту через A2A-протокол.""", tools=[ FunctionTool(my_pandas_analysis_tool), FunctionTool(my_sql_query_tool), ], # A2A-конфигурация — Agent Card создаётся автоматически! a2a_config={ "enabled": True, "skills": ["data_analysis", "sql", "reporting"], "streaming": True, "require_auth": True, "known_peers": [ # Заранее известные агенты {"url": "https://weather.qantcore.space", "skills": ["weather"]}, {"url": "https://sql.qantcore.space", "skills": ["sql_queries"]}, ], }, ) # Деплой одной командой — ADK поднимает A2A-сервер автоматически # agent.deploy(target="vertex-ai", region="us-central1")
За 20 минут мы прошли полный путь A2A-разработчика: установили SDK, создали Agent Card, подняли сервер с реальной бизнес-логикой, написали клиент с обнаружением агентов, освоили потоковую передачу через SSE и собрали созвездие из трёх агентов с оркестрацией. Теперь вы можете:
/.well-known/agent.jsonПротокол A2A — это не просто спецификация, а фундамент для новой экосистемы взаимодействующих AI-агентов. Google активно развивает его, и мы увидим всё больше A2A-совместимых агентов — от погодных ассистентов до сложных исследовательских систем. Лучшее время начать — сейчас.