Простой

13 мин

8.3K

У вас работает AI-агент. У соседней команды — свой, на другом фреймворке, в другом сервисе. Рано или поздно вашему агенту понадобится позвать их агента:

"Сходи найди факты, я подожду, дальше сам"

Сначала — что такое агент

ChatGPT умеет только сказать: на входе промпт, на выходе текст. Агент — следующий шаг: модель не просто отвечает, а действует. Даёшь ей:

"Посчитай выручку за квартал"

— и она сама решает, сходить в базу, дёрнуть API, запустить расчёт, и только потом ответить.

Сама LLM при этом по-прежнему ничего не делает руками — она лишь говорит

"вызовите calc_add с такими аргументами"

Важно одно. Пока агент живёт в вашем процессе, позвать его — обычный вызов функции: просто и предсказуемо. И кажется, что соседнего агента, из другого сервиса, можно подключить так же легко.

Так вот: нельзя.

Когда агентов становится двое

Приходит соседняя команда. У них свой агент — ищет факты во внешних источниках, написан на LangGraph, крутится отдельным сервисом в их кластере. Вам нужно из своего агента позвать их агента: отправить запрос, дождаться ответа, использовать результат.

Проблема в том, что агент — не ручка GET /price, которая мгновенно отдаёт число. Он ведёт себя не как "один запрос — один ответ", и контракт между сервисами обязан это учитывать:

- Ответ бывает долгим — секунды, иногда минуты: агент ходит в LLM, дёргает свои инструменты. Держите всё это в одном HTTP-запросе — и работа висит на живом соединении. Таймаут на прокси, обрыв сети — и задача, которая уже выполняется, потеряна, переподключиться к ней нечем.

- Результат хочется стримить. Одному агенту удобно вернуть всё разом, другому — отдавать по мере готовности: частичный текст, догружаемые файлы. Контракт должен уметь оба режима.

- Агент может переспросить прямо посреди работы: «уточни, за какой именно квартал». Это не ошибка, это нормальный ход событий. Голый REST такого не предполагает.

+ Сверху — авторизация, отмена задачи, повторы.

Как звать чужого агента, чтобы всё это работало? У вас три дороги.

Дорога 1: свой REST между сервисами. Договариваетесь, как слать сообщения, как возвращать статус задачи, как стримить частичный ответ, что делать с retry, какой токен слать. На второй интеграции с третьей командой этот контракт превращается в спагетти, которое уже никто не помнит целиком.

Дорога 2: OpenAPI или gRPC-контракт руками. Лучше, но агентскую специфику из списка выше он не знает. Long-running задачи, частичные артефакты, переспросить у человека — пилите заново.

Дорога 3: загнать всех в один фреймворк. Мультиагентность уже встроена в AutoGen, CrewAI, LangGraph — только у каждого по-своему. Всё это отлично работает, пока агенты живут в одном фреймворке и обычно в одном процессе.

A2A придумали ровно для случая "агент ↔ агент". Открытый стандарт, не привязанный к фреймворку, и вся специфика из списка — долгие задачи, стриминг, переспросить у человека, авторизация — в нём уже есть.
Под капотом обычный HTTP, какой именно — JSON-RPC, REST или gRPC — решает сервер.

Что такое A2A, если в двух словах

Несколько дат, чтобы прикинуть зрелость:

• Апрель 2025 — Google анонсирует протокол, на старте около 50 партнёров.

• Июнь 2025 — отдаёт его в Linux Foundation. Учредители — AWS, Cisco, Google, Microsoft, Salesforce, SAP, ServiceNow; к запуску поддержку заявили больше сотни компаний.

• Март 2026 — версия 1.0 stable, первая production-ready. К первой годовщине (апрель 2026) — больше 150 организаций.

A2A — не библиотека, а протокол. Открытый стандарт в том же смысле, что HTTP или DHCP: есть письменная спецификация, а реализаций уже несколько — Python, JS, Go.

Примеры в статье — на AG2, это Python-фреймворк для агентов. В нём есть модуль autogen.beta.a2a: он берёт готового AG2-агента и публикует его по A2A, а заодно умеет вызывать чужих A2A-агентов из вашего кода.

Одно предложение, ради которого всё затевалось: A2A задаёт единый контракт — как сервис публикует своего AI-агента и как сторонний код его вызывает, ничего не зная о внутренностях.

Четыре сущности, на которых стоит весь протокол

Выкиньте детали — останется четыре понятия. Запомните их, дальше всё вертится вокруг.

Agent Card — паспорт агента. JSON, который агент публикует о себе: имя, версия, описание, навыки (что умеет), транспорты (как обращаться), схемы авторизации (что предъявить на входе). Лежит по стандартному пути /.well-known/agent-card.json. Клиент сходил туда один раз — и знает про агента всё, чтобы начать с ним работать.

Message — реплика в разговоре. У неё role (USER или AGENT) и parts — массив частей. Часть бывает текстом, structured-JSON или файлом. То есть сообщение — это контейнер для разнородного контента, а не просто строка.

Task — долгоиграющая задача. Главное отличие от обычного REST. Шлёте агенту сообщение — он не обязан ответить сразу. Может завести задачу со своим id, выполнять её асинхронно, копить history и менять state. Клиент задачу создаёт (message/send), читает (tasks/get), отменяет (tasks/cancel) или подписывается на обновления.

TaskState — состояние задачи. Задача движется по фиксированному набору состояний: SUBMITTED → WORKING → COMPLETED (либо FAILED / CANCELED). Плюс особые состояния: INPUT_REQUIRED (агент посреди работы просит уточнение у человека), AUTH_REQUIRED, REJECTED. В a2a-sdk они живут как proto-enum (TASK_STATE_WORKING и т.д.), но по смыслу это те же статусы.

Как эта четвёрка работает в типичном обмене:

┌─ Client ─┐    GET /.well-known/agent-card.json     ┌─ Server ─┐
│          │ ─────────────────────────────────────►  │          │
│          │ ◄────── AgentCard (JSON) ────────────   │          │
│          │                                         │          │
│          │    POST / { method: "message/send" }    │          │
│          │ ─────────────────────────────────────►  │ ┌──────┐ │
│          │                                         │ │ Task │ │
│          │ ◄──── Task { state: WORKING } ───────   │ │ store│ │
│          │                                         │ └──────┘ │
│          │  SSE: status_update + artifacts (...)   │          │
│          │ ◄═══════════════════════════════════    │          │
│          │                                         │          │
│          │ ◄──── Task { state: COMPLETED } ─────   │          │
└──────────┘                                         └──────────┘

Слева клиент, справа сервер. Сначала discovery — забрали карточку. Потом отправили сообщение, получили задачу в WORKING. Дальше сервер стримит обновления, пока задача не дойдёт до терминального состояния. Всё остальное в протоколе — вариации поверх этой картинки.

Жизненный цикл задачи

Состояния — не просто лейблы, между ними есть разрешённые переходы:

                       │ message/send
                       ▼
                ┌─────────────┐    не подходит
                │  SUBMITTED  │ ──────────────► REJECTED  (terminal)
                └──────┬──────┘
                       │
                       ▼
                ┌─────────────┐
            ┌─► │   WORKING   │
            │   └──┬───┬───┬──┘
            │      │   │   │
            │      │   │   ├───► COMPLETED (terminal)
            │      │   │   ├───► FAILED    (terminal)
            │      │   │   └───► CANCELED  (terminal)
            │      │   │
            │      │   ▼
            │   ┌──────────────┐   
            └───│INPUT_REQUIRED│ ◄─── человек дал ввод
                │/AUTH_REQUIRED│
                └──────────────┘

Ради двух вещей эта схема состояний и нужна.

Первое — INPUT_REQUIRED. Задача в середине выполнения упирается в нехватку данных: агенту нужно уточнение, которое есть только у человека. В голом REST это пришлось бы городить руками — вернуть особый код, распарсить на клиенте, переспросить, дослать. Здесь это часть контракта: агент переводит задачу в INPUT_REQUIRED, клиент видит статус и понимает, что от него ждут ввод.

Второе — терминальные состояния отделены от рабочих. COMPLETED, FAILED, CANCELED, REJECTED означают "всё, дальше ничего не будет". По ним клиент понимает, когда закрывать SSE или прекращать polling. Без явной границы пришлось бы гадать.

Discovery: как клиент находит агента

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

GET https://agent.example.com/.well-known/agent-card.json

Клиент забирает Agent Card, смотрит supported_interfaces, выбирает транспорт — и шлёт запросы туда. В AG2 этот шаг спрятан за A2AConfig. Вот как выглядит вызов удалённого агента целиком:

import asyncio

from autogen.beta import Agent
from autogen.beta.a2a import A2AConfig


async def main() -> None:
    remote = Agent(
        "remote",
        config=A2AConfig(card_url="http://127.0.0.1:8000", prefer="jsonrpc"),
    )
    reply = await remote.ask("Add 17 and 25 with calc_add. Just the number.")
    print(reply.response.content)


if __name__ == "__main__":
    asyncio.run(main())

Ни слова про HTTP, JSON-RPC или задачи — просто remote.ask(...). А под капотом случилось всё, что мы разбирали: A2AConfig сходил на card_url + /.well-known/agent-card.json, скачал карточку, по prefer="jsonrpc" выбрал JSON-RPC-endpoint, отправил message/send, дождался COMPLETED и вернул текст. Для вызывающего удалённый агент неотличим от локального.

Серверная сторона: опубликовать агента

Симметричная половина — поднять своего агента так, чтобы его звали по A2A:

import asyncio

import uvicorn

from autogen.beta import Agent
from autogen.beta.a2a import A2AServer, build_card
from autogen.beta.config import AnthropicConfig
from autogen.beta.tools import tool


@tool(description="Add two integers.")
async def calc_add(a: int, b: int) -> str:
    return str(a + b)


agent = Agent(
    name="claude",
    config=AnthropicConfig(model="claude-sonnet-4-6"),
    tools=[calc_add],
)


async def main() -> None:
    server = A2AServer(agent)
    card = build_card(agent)
    asgi = server.build_jsonrpc(url="http://127.0.0.1:8000", card=card)

    await uvicorn.Server(uvicorn.Config(asgi, host="127.0.0.1", port=8000)).serve()


if __name__ == "__main__":
    asyncio.run(main())

Содержательного кода тут три строки. A2AServer(agent) оборачивает обычного агента; build_card(agent) собирает Agent Card — имя, навыки, транспорты подтягиваются автоматически из самого агента и его инструментов; build_jsonrpc(...) отдаёт готовое ASGI-приложение под uvicorn.

Транспорты: JSON-RPC, REST, gRPC

A2A не привязан к одному способу передачи. Спека описывает три транспорта, выбор за вами:

- JSON-RPC 2.0 — дефолт. Компактно, методы вроде message/send, tasks/get, tasks/cancel, tasks/pushNotificationConfig/*.

- HTTP+JSON (REST) — обычные ручки, если команде так привычнее.

- gRPC — строгая типизация, бинарный формат, классический межсервисный RPC.

Фишка в том, что один сервер поднимает все три сразу, а клиент сам выбирает, по какому говорить. В AG2 это выглядит так:

server = A2AServer(agent)

card = build_card(
    agent,
    url="http://127.0.0.1:8000",
    rest_url="http://127.0.0.1:8001",
    grpc_url="127.0.0.1:50051",
    transports=("jsonrpc", "rest", "grpc"),
)

jsonrpc_app = server.build_jsonrpc(url="http://127.0.0.1:8000", card=card)
rest_app    = server.build_rest(url="http://127.0.0.1:8001", card=card)
grpc_server = server.build_grpc(
    bind="127.0.0.1:50051", 
    grpc_url="127.0.0.1:50051", 
    card=card
)

Один и тот же агент обслуживает все три транспорта — снаружи это три двери в один сервис. В Agent Card появляются три записи в supported_interfaces, каждая со своим URL и транспортом. Клиент читает карточку и идёт в ту дверь, что ему удобнее.

Из чего состоит сообщение: Parts

Самое важное в Message — это parts: одно сообщение несёт не только текст, но и структурированные данные, и файлы. Каждое сообщение — это массив частей, и часть бывает трёх видов:

- TextPart — обычный текст.

- DataPart — структурированный JSON: заполненная форма, результат в машинном формате.

- FilePart — файл, как байты или как URI.

Мультимодальность зашита с первого дня: одним сообщением летят и текст вопроса, и приложенная картинка, и JSON-контекст. Получатель разбирает parts по типам.

Стриминг через SSE

Агент думает минутами — ходит в LLM, дёргает инструменты, копит результат. Заставлять клиента в это время пуллить tasks/get по таймеру — неудобно и расточительно. Поэтому в A2A есть стриминг через Server-Sent Events: клиент держит одно соединение и получает события по мере того, как они случаются на сервере.

Событий четыре вида — это четыре варианта payload в потоке:

• Task — полный снимок задачи целиком.

• Message — новое сообщение от агента.

• TaskStatusUpdateEvent — смена состояния, например WORKING → COMPLETED.

• TaskArtifactUpdateEvent — кусок результата: текст по чанкам или файл.

Этой четвёрки хватает на весь стриминг. По TaskStatusUpdateEvent клиент видит, как задача движется по состояниям; по TaskArtifactUpdateEvent — собирает результат из чанков; Task и Message дают полные снимки, когда нужен не инкремент, а вся картина. Технически это обычный SSE поверх того же HTTP-эндпоинта — отдельного протокола поверх протокола заводить не пришлось.

Push-уведомления: когда соединение держать нельзя

У SSE есть цена: пока задача считается, клиент обязан держать открытый HTTP-стрим. Для задачи на пару минут — нормально. Дальше начинаются сценарии, где это ломается. Клиент сам короткоживущий — serverless-функция или CI-джоба, которая отработала и завершилась, а не висит полчаса в ожидании. Или клиентов тысячи, и тысячи постоянно открытых соединений превращаются в отдельную инфраструктурную головную боль.

Push переворачивает схему. При SSE соединение держит клиент и тянет из него поток. При push соединение инициирует сервер — сам, когда есть что сообщить. Клиент один раз регистрирует webhook (куда стучаться и как себя аутентифицировать), после чего может отключаться и заниматься своими делами. Меняется состояние задачи — сервер делает HTTP POST на этот URL. Постоянный канал с агентом больше не нужен, достаточно своего поднятого эндпоинта.

Регистрация конфига:

from autogen.beta.a2a.push import (
    A2APushAuthentication,
    A2APushConfig,
    create_push_notification_config,
)

push = A2APushConfig(
    url="https://hooks.example.com/a2a",
    token="webhook-token",
    authentication=A2APushAuthentication(scheme="bearer", credentials="abc..."),
)

created = await create_push_notification_config(config, task_id, push)

Передаём URL вебхука, токен для сверки и схему авторизации, которой сервер подпишет свои POST-запросы — чтобы вы убедились, что стучится именно он, а не кто попало. Дальше живёте без открытого соединения: сервер достучится сам, когда задача сдвинется. Кроме регистрации, протокол умеет и обратные операции над этими конфигами — получить, перечислить, удалить.

Что ещё умеет протокол

Базовый цикл разобрали. Дальше — четыре возможности, которые на практике всплывают чаще всего. По каждой сначала что говорит протокол, потом как это выглядит в коде на AG2

Human-in-the-loop

На уровне протокола это одно состояние — INPUT_REQUIRED (в proto TASK_STATE_INPUT_REQUIRED). Агент, которому посреди работы не хватает данных, переводит Task в него и кладёт в статус сообщение с тем, что хочет узнать. Дальше важная деталь контракта: чтобы продолжить, клиент шлёт новое сообщение с тем же task_id (поле есть прямо в Message). Оно не заводит новую задачу, а прицепляется к старой, и та возвращается в WORKING с того же места. task_id тут — нитка, которая связывает доуточнение с исходной задачей.

То есть «переспросить человека» в A2A — штатный переход по состояниям, а не отдельный механизм. В AG2 клиентская сторона сводится к одному хуку, который фреймворк сам дёрнет на INPUT_REQUIRED:

async def hitl_hook() -> str:
    # в примере — консоль; в проде здесь был бы запрос в UI или мессенджер
    return await asyncio.to_thread(input, "server asks input> ")

remote = Agent(
    "remote", 
    config=A2AConfig(
        card_url="...", 
        input_required_timeout=30.0
    ), 
    hitl_hook=hitl_hook
)

Multi-tenant

Здесь интересно, что мультиарендность зашита прямо в контракт. В спеке у каждого интерфейса в Agent Card есть поле tenant, и у каждого запроса (SendMessageRequest, GetTaskRequest, …) — тоже; в REST-биндинге это видно даже в пути /{tenant}/message:send. Сервер выставляет разные интерфейсы под разных арендаторов, а каждый запрос явно несёт, от чьего имени идёт. Изоляция получается на уровне протокола: тенанты — это разные адреса и разные записи в карточке, перепутать их нельзя.

В AG2 обе стороны короткие:

# сервер: привязываем транспорты к тенантам
card = build_card(
    agent, 
    url="...", 
    transports=("jsonrpc", "grpc"),
    tenants={"jsonrpc": "tenant-A", "grpc": "tenant-B"}
)

# клиент: по умолчанию ходит как tenant-A, 
# но тенанта можно переопределить на один вызов
remote = Agent(
    "remote", 
    config=A2AConfig(
        card_url="...", 
        tenant="tenant-A"
    )
)
await remote.ask("ping", variables={"a2a:tenant": "tenant-Z"})

Security

Авторизацию A2A описывает декларативно и ровно в духе OpenAPI. В Agent Card два поля: security_schemes — словарь именованных схем (API Key, HTTP Bearer, OAuth2, OpenID Connect, mutual TLS), и security_requirements — список требований. Комбинируются они так: список требований — это ИЛИ (достаточно выполнить любое), а схемы внутри одного требования — это И (нужны все сразу). Клиент читает карточку и заранее знает, что предъявлять, без проб и ошибок на живых запросах.

В AG2 это собирается хелперами, где один require(...) — одна ИЛИ-альтернатива:

# bearer, api, oauth — заранее объявленные схемы
card = build_card(
    agent, 
    url="...", 
    security=[
        require(bearer),                              # либо просто Bearer
        require(oauth.with_scopes("read", "write")),  # либо OAuth с нужными scope
        require(bearer, api),                         # либо Bearer И API-ключ вместе
])

Список читается буквально: «пускаем, если есть Bearer, либо OAuth с нужными scope, либо Bearer вместе с API-ключом».

Агент как инструмент

А вот это уже не отдельная фича, а следствие симметрии протокола. Раз любой агент одновременно и сервер, и потенциальный клиент, ничто не мешает агенту посреди своей задачи самому сходить к другому агенту по A2A. Для вызывающей модели чужой агент выглядит обычным инструментом, а под капотом идёт полный A2A-обмен: discovery, message/send, ожидание COMPLETED. Так из независимых агентов собираются конвейеры, и каждый узел остаётся отдельным сервисом на своём стеке.

В AG2 удалённый агент оборачивается в инструмент одним .as_tool():

# researcher живёт в другом сервисе — для нас это просто card_url
researcher = Agent(
    "researcher", 
    config=A2AConfig(card_url="http://research.internal:8000")
)

writer = Agent(
    "writer",
    config=AnthropicConfig(model="claude-sonnet-4-6"),
    tools=[
        researcher.as_tool(
            description="Delegate research questions to the remote researcher."),
    ],
)

A2A или MCP — в чём разница

MCP (Model Context Protocol) — про инструменты для одной LLM. «Дай этой модели доступ к моей файловой системе / БД / API». MCP-сервер — набор tools, resources и prompts, которые подключаются к модели, чтобы расширить её возможности.

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

И они не конфликтуют. Нормальная картина: ваш агент ходит в несколько MCP-серверов за инструментами и одновременно общается с другими агентами по A2A. Разные уровни, разные задачи.

Когда A2A нужен, а когда нет

Граница проходит не там, где «один фреймворк против многих». Она проходит там, где кончается общий процесс. Вся внутрифреймворковая мультиагентность — общий стейт в LangGraph, групповой чат в AutoGen, делегирование в CrewAI — держится на общей памяти: агенты видят друг друга, потому что живут в одном процессе и дёргают друг друга обычными вызовами. Как только агент уезжает в отдельный процесс, этот механизм обрывается: у соседнего процесса нет доступа к вашим объектам в памяти, соединять их придётся по проводу. А раз нужен провод — нужен контракт: формат сообщений, статусы задач, стриминг, авторизация. Свой контракт — это возврат к «Дороге 1» со всеми её граблями; готовый — это и есть A2A.

Отсюда практический критерий.

Нужен, когда: агенты — отдельные сервисы; стеки или фреймворки разные; есть long-running задачи; нужен human-in-the-loop; нужна общая стандартизация для команды или внешних потребителей. И чем больше сервисов и команд, тем сильнее A2A окупается: каждый новый агент подключается по той же карточке, а не очередной самописной интеграцией «все со всеми».

Не нужен, когда: все агенты живут в одном процессе и одном фреймворке; «агент» по сути просто функция. Тогда хватает внутреннего механизма фреймворка или обычного вызова, а A2A — лишний слой.

Что в итоге

A2A сводится к одной договорённости: как агенты находят друг друга (Agent Card), как обмениваются репликами (Message с Parts), как ведут задачи (Task + TaskState) и как стримят результат (SSE и push). Всё остальное — авторизация, мульти-транспорт, multi-tenant, композиция агентов — надстройки поверх этих четырёх сущностей.

Никакой магии тут нет: A2A просто один раз и аккуратно записал то, что мы и так пишем руками каждый раз, когда двум сервисам нужно поговорить. Лучший способ это проверить — поднять у себя. Возьмите server_basic.py и client_basic.py: около 40 строк Python на клиент и сервер вместе, и весь цикл "discovery → message/send → Task → COMPLETED" проходит у вас на глазах.

Ссылки:

• Спека: a2a-protocol.org

• Примеры на Python: github.com/ag2ai/ag2/tree/main/examples/a2a

• Реализация в AG2: github.com/ag2ai/ag2/tree/main/autogen/beta/a2a