Эта страница переведена с помощью Cloud Translation API.

Начало работы с протоколом Agent2Agent (A2A): взаимодействие консьержа по закупкам и удаленного агента-продавца в Cloud Run и Agent Engine

1. 📖 Введение

Протокол Agent2Agent (A2A) предназначен для стандартизации связи между агентами ИИ, особенно для тех, которые развернуты во внешних системах. Ранее подобные протоколы были разработаны для инструментов, называемых протоколом контекста модели (MCP) , который является новым стандартом для связи LLM с данными и ресурсами. A2A пытается дополнить MCP, поскольку A2A фокусируется на другой проблеме: MCP направлен на снижение сложности связи агентов с инструментами и данными, а A2A — на то, как обеспечить взаимодействие агентов в их естественных модальностях. Он позволяет агентам общаться как агенты (или как пользователи), а не как инструменты; например, обеспечивает двустороннюю связь при заказе чего-либо.

A2A позиционируется как дополнение к MCP; в официальной документации рекомендуется использовать MCP для инструментов и A2A для агентов, представленных AgentCard (мы обсудим это позже). Затем фреймворки могут использовать A2A для связи со своим пользователем, удаленными агентами и другими агентами.

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

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

Клиент A2A сначала выполнит обнаружение всех доступных карт агентов сервера A2A и использует полученную информацию для установления соединения.
При необходимости A2A-клиент отправляет сообщение на A2A-сервер, который оценивает это как задачу, подлежащую выполнению. Если на A2A-клиенте настроен URL-адрес получателя push-уведомлений, и он поддерживается A2A-сервером, сервер также сможет публиковать состояние выполнения задачи на принимающую конечную точку на клиенте.
После завершения задачи A2A-сервер отправит ответный артефакт A2A-клиенту.

В ходе выполнения практического задания вы будете использовать следующий пошаговый подход:

Подготовка проекта Google Cloud
Настройка рабочего каталога для среды программирования
Разверните агент Burger в Cloud Run.
Разверните агент для запуска пиццы в Cloud Run.
Внедрить систему управления закупками в Agent Engine.
Взаимодействуйте с консультантом по закупкам через локальный интерфейс.

Обзор архитектуры

Вам необходимо развернуть следующую архитектуру сервисов.

Вы развернете 2 сервиса, которые будут выступать в качестве A2A-серверов: агент Burger (на базе фреймворка CrewAI) и агент Pizza (на базе фреймворка Langgraph). Пользователь будет напрямую взаимодействовать только с консьержем отдела закупок, который будет работать с использованием фреймворка Agent Development Kit (ADK), выступающего в качестве A2A-клиента.

Каждый из этих агентов будет иметь свою собственную среду и систему развертывания.

Предварительные требования

Уверенно работаю с Python.
Понимание базовой архитектуры полного стека с использованием HTTP-сервисов.

Что вы узнаете

Основная структура A2A-сервера
Основная структура клиента A2A
Развертывание службы агента в Cloud Run
Развертывание службы агента в Agent Engine
Как клиент A2A подключается к серверу A2A
Структура запроса и ответа при непотоковом соединении

Что вам понадобится

Веб-браузер Chrome
Аккаунт Gmail
Облачный проект с включенным платежным аккаунтом.

Этот практический урок, разработанный для разработчиков всех уровней (включая начинающих), использует Python в качестве примера приложения. Однако знание Python не требуется для понимания представленных концепций.

2. 🚀 Подготовка к настройке процесса разработки в мастерской.

ВАЖНЫЕ ЗАМЕЧАНИЯ:

Для выполнения этого руководства необходим проект Google Cloud с активным платёжным аккаунтом . Если у вас его ещё нет, выполните следующие шаги, чтобы получить пробный платёжный аккаунт , который можно использовать для этого руководства:

Открыть окно в режиме инкогнито
Если вы присоединяетесь к мастер-классу, URL-адрес портала активации обычно предоставляется во время занятия; в противном случае вы можете перейти на портал активации этой учетной записи для оплаты пробной версии.
Войдите в систему, используя свой личный аккаунт Gmail (Google Workspace / другие адреса электронной почты не подойдут!).
Следуйте пошаговым инструкциям по активации , чтобы успешно создать новый проект Google Cloud, связанный с ранее активированной пробной учетной записью, использованной в этом руководстве.

Шаг 1: Выберите активный проект в облачной консоли.

В консоли Google Cloud на странице выбора проекта выберите или создайте проект Google Cloud (см. раздел в левом верхнем углу консоли).

Нажмите на него, и вы увидите список всех ваших проектов, как в этом примере.

Значение, обозначенное красной рамкой, — это идентификатор проекта , и это значение будет использоваться на протяжении всего урока.

Убедитесь, что для вашего облачного проекта включена оплата. Для этого нажмите на значок меню (гамбургер) ☰ в верхнем левом углу панели, где отображается меню навигации, и найдите пункт «Оплата».

Если вы видите , что привязана пробная учетная запись Google Cloud Platform , ваш проект готов к использованию в этом руководстве. В противном случае вернитесь к началу руководства и активируйте учетную запись.

Шаг 2: Ознакомьтесь с Cloud Shell.

В большинстве руководств вы будете использовать Cloud Shell . Нажмите «Активировать Cloud Shell» в верхней части консоли Google Cloud. Если система запросит авторизацию, нажмите «Авторизовать».

После подключения к Cloud Shell нам потребуется проверить, авторизована ли оболочка (или терминал) уже с использованием нашей учетной записи.

gcloud auth list

Если вы видите в своей личной почте Gmail результат, как в приведенном ниже примере, значит, все в порядке.

Credentialed Accounts

ACTIVE: *
ACCOUNT: alvinprayuda@gmail.com

To set the active account, run:
    $ gcloud config set account `ACCOUNT`

Если это не поможет, попробуйте обновить страницу в браузере и обязательно нажмите кнопку « Авторизовать» , когда появится соответствующий запрос (возможно, авторизация прервалась из-за проблем с подключением).

Далее нам также необходимо проверить, настроена ли оболочка уже для правильного идентификатора проекта (PROJECT ID ). Если вы видите значение внутри скобок ( ) перед значком $ в терминале (на скриншоте ниже значение равно "a2a-agent-engine" ), это значение показывает настроенный проект для вашей активной сессии оболочки.

Если отображаемое значение уже верное , вы можете пропустить следующую команду . Однако, если оно неверно или отсутствует, выполните следующую команду.

gcloud config set project <YOUR_PROJECT_ID>

Затем клонируйте рабочую директорию шаблона для этого практического занятия из Github и выполните следующую команду. Она создаст рабочую директорию в каталоге purchasing-concierge-a2a.

git clone https://github.com/alphinside/purchasing-concierge-intro-a2a-codelab-starter.git purchasing-concierge-a2a

Шаг 3: Ознакомьтесь с редактором Cloud Shell и настройте рабочий каталог приложения.

Теперь мы можем настроить наш редактор кода для выполнения некоторых действий по программированию. Для этого мы будем использовать редактор Cloud Shell.

Нажмите кнопку «Открыть редактор» , это откроет редактор Cloud Shell.

После этого перейдите в верхнюю часть редактора Cloud Shell и нажмите «Файл» -> «Открыть папку», найдите каталог с вашим именем пользователя и найдите каталог purchasing-concierge-a2a, затем нажмите кнопку «ОК». Это сделает выбранный каталог основным рабочим каталогом. В этом примере имя пользователя — alvinprayuda , поэтому путь к каталогу показан ниже.

Теперь ваш редактор Cloud Shell должен выглядеть так.

Теперь откройте терминал для редактора. Это можно сделать, щелкнув «Терминал» -> «Новый терминал» в строке меню, или используя Ctrl + Shift + C — это откроет окно терминала в нижней части браузера.

Ваш текущий активный терминал должен находиться в рабочей директории purchasing-concierge-a2a . В этом практическом занятии мы будем использовать Python 3.12 и менеджер проектов Python uv для упрощения создания и управления версиями Python и виртуальными средами. Этот пакет uv уже предустановлен в Cloud Shell.

Выполните эту команду, чтобы установить необходимые зависимости в виртуальное окружение в каталоге .venv.

uv sync --frozen

Проверьте файл pyproject.toml , чтобы увидеть объявленные зависимости для этого руководства: a2a-sdk, google-adk, and gradio .

Теперь нам нужно будет включить необходимые API с помощью команды, показанной ниже. Это может занять некоторое время.

gcloud services enable aiplatform.googleapis.com \
                       run.googleapis.com \
                       cloudbuild.googleapis.com \
                       cloudresourcemanager.googleapis.com

После успешного выполнения команды вы должны увидеть сообщение, похожее на показанное ниже:

Operation "operations/..." finished successfully.

3. 🚀 Развертывание агентов удаленной продажи A2A-сервера в облаке

На этом этапе мы развернем двух удаленных агентов-продавцов, отмеченных красной рамкой. Агент, отвечающий за бургеры, будет работать на основе фреймворка CrewAI, а агент, отвечающий за пиццу, — на основе фреймворка Langgraph.

4. 🚀 Развертывание агента продавца бургеров - A2A-сервера

Исходный код агента Burger находится в каталоге remote_seller_agents/burger_agent .

Все файлы, находящиеся в каталоге remote_seller_agents/burger_agent, уже достаточны для развертывания нашего агента в Cloud Run, чтобы он был доступен как сервис. Выполните следующую команду для его развертывания.

gcloud run deploy burger-agent \
    --source remote_seller_agents/burger_agent \
    --port=8080 \
    --allow-unauthenticated \
    --min 1 \
    --region us-central1 \
    --update-env-vars GOOGLE_CLOUD_LOCATION=us-central1 \
    --update-env-vars GOOGLE_CLOUD_PROJECT={your-project-id}

Если появится запрос о создании репозитория контейнеров для развертывания из исходного кода, ответьте «Да». После успешного развертывания отобразится примерно такой лог.

Service [burger-agent] revision [burger-agent-xxxxx-xxx] has been deployed and is serving 100 percent of traffic.
Service URL: https://burger-agent-xxxxxxxxx.us-central1.run.app

Часть xxxx здесь будет служить уникальным идентификатором при развертывании сервиса.

Откройте новую вкладку браузера и перейдите по адресу https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json , указанному в файле, содержащем адрес развернутых служб Burger Agent. Это URL-адрес для доступа к карточке агента развернутого сервера A2A.

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

Это информация с карты агента сети ресторанов быстрого питания, которая должна быть доступна в целях раскрытия информации.

Обратите внимание, что значение url по-прежнему установлено на http://0.0.0.0:8080/ . Это значение url должно содержать основную информацию для A2A-клиента, необходимую для отправки сообщений извне, и оно настроено неправильно.

Нам необходимо обновить это значение, указав URL-адрес нашего сервиса Burger Agent, добавив дополнительную переменную среды HOST_OVERRIDE .

Обновление значения URL-адреса агента Burger в карточке агента через переменную среды.

Чтобы добавить HOST_OVERRIDE в службу Burger Agent, выполните следующие действия.

Введите в строку поиска в верхней части консоли облачных сервисов «Найдите Cloud Run».

Нажмите на ранее развернутую службу burger-agent cloud run.

Скопируйте URL-адрес сервиса Burger, затем нажмите «Редактировать» и разверните новую версию.

Затем нажмите на раздел «Переменные и секреты».

После этого нажмите «Добавить переменную» и установите значение HOST_OVERRIDE равным URL-адресу службы (тому, который имеет шаблон https://burger-agent-xxxxxxxxx.us-central1.run.app ).

Наконец, нажмите кнопку развертывания , чтобы повторно развернуть вашу службу.

При повторном доступе к карточке агента burger-agent в браузере по адресу https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json значение url будет уже правильно настроено.

5. 🚀 Развертывание агента продавца пиццы - A2A-сервера

Аналогично, исходный код приложения для продажи пиццы находится в каталоге remote_seller_agents/pizza_agent .

Аналогично предыдущему шагу развертывания burger-agent, всех файлов, находящихся в каталоге remote_seller_agents/pizza_agent, уже достаточно для развертывания нашего агента в Cloud Run, чтобы он был доступен как сервис. Выполните следующую команду для его развертывания.

gcloud run deploy pizza-agent \
    --source remote_seller_agents/pizza_agent \
    --port=8080 \
    --allow-unauthenticated \
    --min 1 \
    --region us-central1 \
    --update-env-vars GOOGLE_CLOUD_LOCATION=us-central1 \
    --update-env-vars GOOGLE_CLOUD_PROJECT={your-project-id}

После успешного развертывания в журнале отобразится примерно следующее.

Service [pizza-agent] revision [pizza-agent-xxxxx-xxx] has been deployed and is serving 100 percent of traffic.
Service URL: https://pizza-agent-xxxxxxxxx.us-central1.run.app

Часть xxxx здесь будет служить уникальным идентификатором при развертывании сервиса.

Аналогичная ситуация и с агентом Burger: при попытке перейти по адресу https://pizza-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json развернутых сервисов Pizza Agent через браузер для доступа к карточке агента A2A-сервера, значение url Pizza Agent на его карточке агента еще не настроено должным образом. Также необходимо добавить HOST_OVERRIDE в его переменную окружения.

Обновление значения URL-адреса агента Pizza Agent на карточке агента через переменную среды.

Чтобы добавить HOST_OVERRIDE в службу агента Pizza, выполните следующие действия.

Введите в строку поиска в верхней части консоли облачных сервисов «Найдите Cloud Run».

Нажмите на ранее развернутую службу pizza-agent cloud run.

Нажмите «Редактировать» и разверните новую версию.

Скопируйте URL-адрес службы доставки пиццы, затем перейдите в раздел «Переменные и секреты».

После этого нажмите «Добавить переменную» и установите значение HOST_OVERRIDE равным URL-адресу службы (тому, который имеет шаблон https://pizza-agent-xxxxxxxxx.us-central1.run.app ).

Наконец, нажмите кнопку развертывания , чтобы повторно развернуть вашу службу.

Теперь, когда вы снова откроете карточку агента pizza-agent в браузере по адресу https://pizza-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json , значение url будет уже правильно настроено.

На данный момент мы уже успешно развернули сервисы для приготовления бургеров и пиццы на платформе Cloud Run.

6. 🚀 Внедрение системы «Консьерж по закупкам» — механизм взаимодействия клиента и агента (A2A Client to Agent Engine).

На этом этапе мы задействуем агента по сопровождению покупателей. Именно с этим агентом мы будем взаимодействовать.

Исходный код нашего агента по сопровождению закупок находится в каталоге purchasing_concierge . Инициализацию агента можно посмотреть в скрипте purchasing_concierge/purchasing_agent.py .

Для развертывания выполните следующие шаги:

Для начала нам нужно создать временное хранилище в Cloud Storage.

gcloud storage buckets create gs://purchasing-concierge-{your-project-id} --location=us-central1

Теперь нам нужно сначала подготовить переменную .env. Давайте скопируем файл .env.example в файл .env.

cp .env.example .env

Теперь откройте файл .env , и вы увидите следующее содержимое.

GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT={your-project-id}
GOOGLE_CLOUD_LOCATION=us-central1
STAGING_BUCKET=gs://purchasing-concierge-{your-project-id}
PIZZA_SELLER_AGENT_URL={your-pizza-agent-url}
BURGER_SELLER_AGENT_URL={your-burger-agent-url}
AGENT_ENGINE_RESOURCE_NAME={your-agent-engine-resource-name}

Этот агент будет взаимодействовать с агентами, отвечающими за бургеры и пиццу, поэтому нам необходимо предоставить соответствующие учетные данные для обоих. Нам потребуется обновить поля PIZZA_SELLER_AGENT_URL и BURGER_SELLER_AGENT_URL , указав URL-адрес Cloud Run из предыдущих шагов.

Если вы забыли об этом, зайдите в консоль Cloud Run. Введите «Cloud Run» в строку поиска в верхней части консоли и щелкните правой кнопкой мыши по значку Cloud Run, чтобы открыть его в новой вкладке.

Вы должны увидеть наши ранее развернутые сервисы удаленных агентов продавцов, как показано ниже.

Чтобы увидеть общедоступные URL-адреса этих сервисов, щелкните по одному из них, и вы будете перенаправлены на страницу с подробной информацией о сервисе. URL-адрес можно увидеть в верхней части страницы, справа от информации о регионе.

Итоговая переменная окружения должна выглядеть примерно так.

GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT={your-project-id}
GOOGLE_CLOUD_LOCATION=us-central1
STAGING_BUCKET=gs://purchasing-concierge-{your-project-id}
PIZZA_SELLER_AGENT_URL=https://pizza-agent-xxxxx.us-central1.run.app
BURGER_SELLER_AGENT_URL=https://burger-agent-xxxxx.us-central1.run.app
AGENT_ENGINE_RESOURCE_NAME={your-agent-engine-resource-name}

Теперь мы готовы развернуть нашего агента по сопровождению покупок. Мы развернем его в системе агентов, а код развертывания находится в скрипте deploy_to_agent_engine.py .

Мы можем развернуть его, запустив скрипт:

uv run deploy_to_agent_engine.py

После успешного развертывания в журнале отобразится примерно следующее. В нем будет указано имя ресурса Agent Engine как "projects/xxxx/locations/us-central1/reasoningEngines/yyyy".

AgentEngine created. Resource name: projects/xxxx/locations/us-central1/reasoningEngines/yyyy
To use this AgentEngine in another session:
agent_engine = vertexai.agent_engines.get('projects/xxxx/locations/us-central1/reasoningEngines/yyyy)
Deployed remote app resource: projects/xxxx/locations/us-central1/reasoningEngines/xxxx

А когда мы проверим это на панели управления Agent Engine (найдем «Agent Engine» в строке поиска), там отобразится наше предыдущее развертывание.

Вы также можете проверить, отображается ли там имя ресурса Agent Engine. Затем мы можем использовать это имя ресурса для его тестирования.

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

GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT={your-project-id}
GOOGLE_CLOUD_LOCATION=us-central1
STAGING_BUCKET=gs://purchasing-concierge-{your-project-id}
PIZZA_SELLER_AGENT_URL=https://pizza-agent-xxxxx.us-central1.run.app
BURGER_SELLER_AGENT_URL=https://burger-agent-xxxxx.us-central1.run.app
AGENT_ENGINE_RESOURCE_NAME=projects/xxxx/locations/us-central1/reasoningEngines/yyyy

Тестирование развернутого агента на Agent Engine

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

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

bash test_agent_engine.sh

Вы можете ознакомиться со скриптом и увидеть, что мы пытаемся задать агенту вопрос : «Пожалуйста, перечислите доступные блюда из меню бургеров».

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

{
  "content": {
    "parts": [
      {
        "text": "Here is our burger menu:\n- Classic Cheeseburger: IDR 85K\n- Double Cheeseburger: IDR 110K\n- Spicy Chicken Burger: IDR 80K\n- Spicy Cajun Burger: IDR 85K"
      }
    ],
    "role": "model"
  },
  "usage_metadata": {
    "candidates_token_count": 51,
    "candidates_tokens_details": [
      {
        "modality": "TEXT",
        "token_count": 51
      }
    ],
    "prompt_token_count": 907,
    "prompt_tokens_details": [
      {
        "modality": "TEXT",
        "token_count": 907
      }
    ],
    "total_token_count": 958,
    "traffic_type": "ON_DEMAND"
  },
  "invocation_id": "e-14679918-af68-45f1-b942-cf014368a733",
  "author": "purchasing_agent",
  "actions": {
    "state_delta": {},
    "artifact_delta": {},
    "requested_auth_configs": {}
  },
  "id": "dbe7fc43-b82a-4f3e-82aa-dd97afa8f15b",
  "timestamp": 1754287348.941454
}

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

7. 🚀 Интеграционное тестирование и проверка полезной нагрузки

Теперь давайте изучим работу нашего консьержа по закупкам с помощью удаленного агента через веб-интерфейс. Выполните следующую команду для развертывания приложения Gradio. Для запуска этого приложения необходимо предварительно правильно заполнить файл .env

uv run purchasing_concierge_ui.py

В случае успеха будет отображен следующий результат.

* Running on local URL:  http://0.0.0.0:8080
* To create a public link, set `share=True` in `launch()`.

Затем, удерживая клавишу Ctrl, щелкните по URL-адресу http://0.0.0.0:8080 в терминале или нажмите кнопку предварительного просмотра веб-интерфейса, чтобы открыть веб-интерфейс.

Попробуйте провести примерно такой разговор:

Покажите мне меню бургеров и пиццы
Я хочу заказать одну пиццу с курицей барбекю и один острый бургер по-каджунски.

И продолжайте разговор, пока не завершите заказ. Проанализируйте, как проходит взаимодействие, и каковы вызов и ответ инструмента. На следующем изображении представлен пример результата взаимодействия.

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

Теперь мы рассмотрели основные концепции A2A и посмотрим, как она реализована в виде клиент-серверной архитектуры.

8. 💡 [Пояснение к коду] Концепция и реализация A2A-сервера

Инициализацию удаленного агента продавца можно посмотреть в скрипте remote_seller_agents/*/agent.py . Вот фрагмент кода агентов продавца.

Агент по бургерам

from crewai import Agent, Crew, LLM, Task, Process
from crewai.tools import tool

...

       model = LLM(
            model="vertex_ai/gemini-2.5-flash-lite",  # Use base model name without provider prefix
        )
        burger_agent = Agent(
            role="Burger Seller Agent",
            goal=(
                "Help user to understand what is available on burger menu and price also handle order creation."
            ),
            backstory=("You are an expert and helpful burger seller agent."),
            verbose=False,
            allow_delegation=False,
            tools=[create_burger_order],
            llm=model,
        )

        agent_task = Task(
            description=self.TaskInstruction,
            agent=burger_agent,
            expected_output="Response to the user in friendly and helpful manner",
        )

        crew = Crew(
            tasks=[agent_task],
            agents=[burger_agent],
            verbose=False,
            process=Process.sequential,
        )

        inputs = {"user_prompt": query, "session_id": sessionId}
        response = crew.kickoff(inputs)
        return response

...

Пицца-агент

from langchain_google_vertexai import ChatVertexAI
from langgraph.prebuilt import create_react_agent

...

self.model = ChatVertexAI(
    model="gemini-2.5-flash-lite",
    location=os.getenv("GOOGLE_CLOUD_LOCATION"),
    project=os.getenv("GOOGLE_CLOUD_PROJECT"),
)
self.tools = [create_pizza_order]
self.graph = create_react_agent(
    self.model,
    tools=self.tools,
    checkpointer=memory,
    prompt=self.SYSTEM_INSTRUCTION,
)

...

Как видите, эти два агента созданы с использованием совершенно разных фреймворков (CrewAI и Langgraph) по сравнению с клиентским агентом (ADK). В случае с A2A это не проблема, нам не нужно, чтобы они обменивались своим внутренним кодом для взаимодействия друг с другом, не имеет значения, какие фреймворки используются, какой язык программирования применяется или где они развернуты.

Основные компоненты A2A-сервера

Теперь давайте обсудим основную концепцию и компоненты A2A-сервера.

Карта агента

Каждый сервер A2A должен иметь карточку агента, доступную по адресу /.well-known/agent.json . Это необходимо для поддержки этапа обнаружения на клиенте A2A, который должен предоставлять полную информацию и контекст о том, как получить доступ к агенту и узнать обо всех его возможностях. Это чем-то похоже на хорошо документированную документацию API с использованием Swagger или Postman.

Это содержимое нашей развернутой карты агента Burger Agent.

{
  "capabilities": {
    "streaming": true
  },
  "defaultInputModes": [
    "text",
    "text/plain"
  ],
  "defaultOutputModes": [
    "text",
    "text/plain"
  ],
  "description": "Helps with creating burger orders",
  "name": "burger_seller_agent",
  "protocolVersion": "0.2.6",
  "skills": [
    {
      "description": "Helps with creating burger orders",
      "examples": [
        "I want to order 2 classic cheeseburgers"
      ],
      "id": "create_burger_order",
      "name": "Burger Order Creation Tool",
      "tags": [
        "burger order creation"
      ]
    }
  ],
  "url": "https://burger-agent-109790610330.us-central1.run.app",
  "version": "1.0.0"
}

Эти карточки агентов содержат информацию о многих важных компонентах, таких как навыки агента, возможности потоковой передачи, поддерживаемые режимы, версия протокола и другие параметры.

Вся эта информация может быть использована для разработки надлежащего механизма связи, чтобы клиент A2A мог корректно взаимодействовать. Поддерживаемые режимы и механизмы аутентификации гарантируют правильное установление связи, а информация skills агента может быть встроена в системное приглашение клиента A2A, чтобы предоставить агенту клиента контекст о возможностях и навыках удаленного агента, которые необходимо вызвать. Более подробные поля для этой карточки агента можно найти в этой документации .

В нашем коде реализация агентской карты осуществляется с использованием Python SDK A2A. Пример реализации можно посмотреть в приведенном ниже фрагменте кода remote_seller_agents/burger_agent/main.py.

...

        capabilities = AgentCapabilities(streaming=True)
        skill = AgentSkill(
            id="create_burger_order",
            name="Burger Order Creation Tool",
            description="Helps with creating burger orders",
            tags=["burger order creation"],
            examples=["I want to order 2 classic cheeseburgers"],
        )
        agent_host_url = (
            os.getenv("HOST_OVERRIDE")
            if os.getenv("HOST_OVERRIDE")
            else f"http://{host}:{port}/"
        )
        agent_card = AgentCard(
            name="burger_seller_agent",
            description="Helps with creating burger orders",
            url=agent_host_url,
            version="1.0.0",
            defaultInputModes=BurgerSellerAgent.SUPPORTED_CONTENT_TYPES,
            defaultOutputModes=BurgerSellerAgent.SUPPORTED_CONTENT_TYPES,
            capabilities=capabilities,
            skills=[skill],
        )

...

Там мы видим несколько полей, например:

AgentCapabilities : Объявление дополнительных необязательных функций, поддерживаемых службой агента, таких как возможность потоковой передачи и/или поддержка push-уведомлений.
AgentSkill : Инструменты или функции, поддерживаемые агентом.
Input/OutputModes : поддерживаемые типы ввода/вывода.
Url : Адрес для связи с агентом

В этой конфигурации мы обеспечиваем динамическое создание URL-адреса хоста агента, что упрощает переключение между локальным тестированием и развертыванием в облаке, поэтому на предыдущем шаге нам необходимо добавить переменную HOST_OVERRIDE .

Очередь задач и исполнитель агентов

Сервер A2A может обрабатывать запросы от разных агентов или пользователей и идеально изолировать каждую задачу. Для лучшего понимания контекста этих процессов вы можете посмотреть изображение ниже.

Таким образом, каждый A2A-сервер должен уметь отслеживать входящие задачи и хранить соответствующую информацию о них. A2A SDK предоставляет модули для решения этой задачи на A2A-сервере. Во-первых, мы можем создать логику обработки входящих запросов. Наследуя абстрактный класс AgentExecutor, мы можем контролировать выполнение и отмену задач. Пример реализации можно посмотреть в модуле remote_seller_agents/burger_agent/agent_executor.py (аналогичный путь для случая с продавцом пиццы).

...

class BurgerSellerAgentExecutor(AgentExecutor):
    """Burger Seller AgentExecutor."""

    def __init__(self):
        self.agent = BurgerSellerAgent()

    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        query = context.get_user_input()
        try:
            result = self.agent.invoke(query, context.context_id)
            print(f"Final Result ===> {result}")

            parts = [Part(root=TextPart(text=str(result)))]
            await event_queue.enqueue_event(
                completed_task(
                    context.task_id,
                    context.context_id,
                    [new_artifact(parts, f"burger_{context.task_id}")],
                    [context.message],
                )
            )
        except Exception as e:
            print("Error invoking agent: %s", e)
            raise ServerError(error=ValueError(f"Error invoking agent: {e}")) from e

    async def cancel(
        self, request: RequestContext, event_queue: EventQueue
    ) -> Task | None:
        raise ServerError(error=UnsupportedOperationError())

...

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

После создания исполнителя мы можем напрямую использовать встроенные функции DefaultRequestHandler, InMemoryTaskStore и A2AStarletteApplication для запуска HTTP-сервера. Эту реализацию можно изучить в remote_seller_agents/burger_agent/__main__.py

...

        request_handler = DefaultRequestHandler(
            agent_executor=BurgerSellerAgentExecutor(),
            task_store=InMemoryTaskStore(),
        )
        server = A2AStarletteApplication(
            agent_card=agent_card, http_handler=request_handler
        )

        uvicorn.run(server.build(), host=host, port=port)

...

Этот модуль предоставит вам реализацию маршрута /.well-known/agent.json для доступа к карточке агента, а также конечную точку POST для поддержки протокола A2A.

Краткое содержание

Вкратце, на данный момент наш развернутый A2A-сервер использует Python SDK и поддерживает две следующие функции:

Публикация карточки агента по маршруту /.well-known/agent.json
Обработка JSON-RPC запросов с использованием очереди задач в оперативной памяти.

Точку входа для запуска этих функций можно посмотреть в скрипте __main__.py (в remote_seller_agents/burger_agent или remote_seller_agents/pizza_agent ).

9. 💡 [Пояснение к коду] Развертывание агента-движка

Вот фрагмент кода агента по сопровождению покупок из файла purchasing_concierge/purchasing_agent.py:

from google.adk import Agent

...

def create_agent(self) -> Agent:
        return Agent(
            model="gemini-2.5-flash-lite",
            name="purchasing_agent",
            instruction=self.root_instruction,
            before_model_callback=self.before_model_callback,
            before_agent_callback=self.before_agent_callback,
            description=(
                "This purchasing agent orchestrates the decomposition of the user purchase request into"
                " tasks that can be performed by the seller agents."
            ),
            tools=[
                self.send_task,
            ],
        )

...

Этот агент создан с использованием ADK и развернут на Agent Engine.

Vertex AI Agent Engine — это набор сервисов, позволяющий разработчикам развертывать, управлять и масштабировать ИИ-агентов в производственной среде. Он обеспечивает инфраструктуру для масштабирования агентов в производственной среде, позволяя нам сосредоточиться на создании приложений. Подробнее об этом можно прочитать в этом документе . Если ранее нам требовалось подготовить файлы, необходимые для развертывания нашего агентского сервиса (например, основной скрипт сервера и Dockerfile), то в этом случае мы можем развернуть нашего агента непосредственно из скрипта Python без необходимости разработки собственного бэкэнд-сервиса, используя комбинацию ADK и Agent Engine.

В этом руководстве мы выполняем развертывание с помощью скрипта deploy_to_agent_engine.py , содержимое которого показано ниже.

import vertexai
from vertexai.preview import reasoning_engines
from vertexai import agent_engines
from dotenv import load_dotenv
import os
from purchasing_concierge.agent import root_agent

load_dotenv()

PROJECT_ID = os.getenv("GOOGLE_CLOUD_PROJECT")
LOCATION = os.getenv("GOOGLE_CLOUD_LOCATION")
STAGING_BUCKET = os.getenv("STAGING_BUCKET")

vertexai.init(
    project=PROJECT_ID,
    location=LOCATION,
    staging_bucket=STAGING_BUCKET,
)

adk_app = reasoning_engines.AdkApp(
    agent=root_agent,
)

remote_app = agent_engines.create(
    agent_engine=adk_app,
    display_name="purchasing-concierge",
    requirements=[
        "google-cloud-aiplatform[adk,agent_engines]",
        "a2a-sdk==0.2.16",
    ],
    extra_packages=[
        "./purchasing_concierge",
    ],
    env_vars={
        "GOOGLE_GENAI_USE_VERTEXAI": os.environ["GOOGLE_GENAI_USE_VERTEXAI"],
        "PIZZA_SELLER_AGENT_URL": os.environ["PIZZA_SELLER_AGENT_URL"],
        "BURGER_SELLER_AGENT_URL": os.environ["BURGER_SELLER_AGENT_URL"],
    },
)

print(f"Deployed remote app resource: {remote_app.resource_name}")

Вот шаги, необходимые для развертывания нашего агента ADK в движке агентов. Во-первых, нам нужно создать объект AdkApp из нашего root_agent ADK (root_agent). Затем мы можем запустить метод agent_engines.create , указав объект adk_app , требования в поле requirements , путь к каталогу агента в extra_packages (при необходимости можно указать и другие каталоги и файлы) и необходимые переменные окружения.

10. 💡 [Пояснение к коду] Концепция и реализация клиента A2A

Изображение выше демонстрирует типичную последовательность взаимодействий типа «от человека к человеку»:

Клиент попытается найти любую опубликованную карточку агента в указанном URL-адресе удаленного агента по адресу /.well-known/agent.json
Затем, при необходимости, сервер отправит этому агенту сообщение, содержащее текст сообщения и необходимые параметры метаданных (например, идентификатор сессии, исторический контекст и т. д.). Сервер воспримет это сообщение как задачу, которую необходимо выполнить.
Сервер A2A обрабатывает запрос, и если он поддерживает push-уведомления, он также сможет публиковать некоторые уведомления на протяжении всей обработки задачи (эта функциональность выходит за рамки данного практического занятия).
После завершения обработки A2A-сервер отправит ответ обратно клиенту.

К основным объектам для описанных выше взаимодействий относятся следующие элементы (более подробную информацию можно прочитать здесь ):

Сообщение: диалог между клиентом и удалённым агентом.
Задача : Основная единица работы, управляемая A2A, идентифицируемая уникальным идентификатором.
Артефакт: Результат (например, документ, изображение, структурированные данные), сгенерированный агентом в результате выполнения задачи, состоящий из частей.
Часть: наименьшая единица содержимого в сообщении или артефакте. Частью может быть текст, изображение, видео, файл и т. д.

Обнаружение карт

При запуске клиентского сервиса A2A обычно происходит попытка получить информацию о карточке агента и сохранить её для быстрого доступа при необходимости. В этом практическом задании мы реализуем это в функции before_agent_callback , вы можете увидеть реализацию в purchasing_concierge/purchasing_agent.py (см. фрагмент кода ниже).

...

async def before_agent_callback(self, callback_context: CallbackContext):
        if not self.a2a_client_init_status:
            httpx_client = httpx.AsyncClient(timeout=httpx.Timeout(timeout=30))
            for address in self.remote_agent_addresses:
                card_resolver = A2ACardResolver(
                    base_url=address, httpx_client=httpx_client
                )
                try:
                    card = await card_resolver.get_agent_card()
                    remote_connection = RemoteAgentConnections(
                        agent_card=card, agent_url=card.url
                    )
                    self.remote_agent_connections[card.name] = remote_connection
                    self.cards[card.name] = card
                except httpx.ConnectError:
                    print(f"ERROR: Failed to get agent card from : {address}")
            agent_info = []
            for ra in self.list_remote_agents():
                agent_info.append(json.dumps(ra))
            self.agents = "\n".join(agent_info)

...

Здесь мы пытаемся получить доступ ко всем доступным карточкам агентов, используя встроенный модуль A2ACardResolver клиента A2A, затем получаем соединение, необходимое для отправки сообщения агенту, после чего нам также необходимо вывести список всех доступных агентов и их характеристики в командную строку, чтобы наш агент знал, что он может взаимодействовать с этими агентами.

Инструмент для отправки и запроса задач

Это подсказка и инструмент, которые мы предоставляем нашему агенту ADK.

...

def root_instruction(self, context: ReadonlyContext) -> str:
    current_agent = self.check_active_agent(context)
    return f"""You are an expert purchasing delegator that can delegate the user product inquiry and purchase request to the
appropriate seller remote agents.

Execution:
- For actionable tasks, you can use `send_task` to assign tasks to remote agents to perform.
- When the remote agent is repeatedly asking for user confirmation, assume that the remote agent doesn't have access to user's conversation context. 
So improve the task description to include all the necessary information related to that agent
- Never ask user permission when you want to connect with remote agents. If you need to make connection with multiple remote agents, directly
connect with them without asking user permission or asking user preference
- Always show the detailed response information from the seller agent and propagate it properly to the user. 
- If the remote seller is asking for confirmation, rely the confirmation question to the user if the user haven't do so. 
- If the user already confirmed the related order in the past conversation history, you can confirm on behalf of the user
- Do not give irrelevant context to remote seller agent. For example, ordered pizza item is not relevant for the burger seller agent
- Never ask order confirmation to the remote seller agent 

Please rely on tools to address the request, and don't make up the response. If you are not sure, please ask the user for more details.
Focus on the most recent parts of the conversation primarily.

If there is an active agent, send the request to that agent with the update task tool.

Agents:
{self.agents}

Current active seller agent: {current_agent["active_agent"]}
"""

...

async def send_task(self, agent_name: str, task: str, tool_context: ToolContext):
        """Sends a task to remote seller agent

        This will send a message to the remote agent named agent_name.

        Args:
            agent_name: The name of the agent to send the task to.
            task: The comprehensive conversation context summary
                and goal to be achieved regarding user inquiry and purchase request.
            tool_context: The tool context this method runs in.

        Yields:
            A dictionary of JSON data.
        """
        if agent_name not in self.remote_agent_connections:
            raise ValueError(f"Agent {agent_name} not found")
        state = tool_context.state
        state["active_agent"] = agent_name
        client = self.remote_agent_connections[agent_name]
        if not client:
            raise ValueError(f"Client not available for {agent_name}")
        session_id = state["session_id"]
        task: Task
        message_id = ""
        metadata = {}
        if "input_message_metadata" in state:
            metadata.update(**state["input_message_metadata"])
            if "message_id" in state["input_message_metadata"]:
                message_id = state["input_message_metadata"]["message_id"]
        if not message_id:
            message_id = str(uuid.uuid4())

        payload = {
            "message": {
                "role": "user",
                "parts": [
                    {"type": "text", "text": task}
                ],  # Use the 'task' argument here
                "messageId": message_id,
                "contextId": session_id,
            },
        }

        message_request = SendMessageRequest(
            id=message_id, params=MessageSendParams.model_validate(payload)
        )
        send_response: SendMessageResponse = await client.send_message(
            message_request=message_request
        )
        print(
            "send_response",
            send_response.model_dump_json(exclude_none=True, indent=2),
        )

        if not isinstance(send_response.root, SendMessageSuccessResponse):
            print("received non-success response. Aborting get task ")
            return None

        if not isinstance(send_response.root.result, Task):
            print("received non-task response. Aborting get task ")
            return None

        return send_response.root.result

...

В запросе мы указываем нашему агенту по сопровождению закупок имена и описания всех доступных удаленных агентов, а в функции self.send_task предоставляем механизм для получения соответствующего клиента для подключения к агенту и отправки необходимых метаданных с помощью объекта SendMessageRequest .

Протоколы связи

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

Обмен сообщениями и задачами реализован с использованием формата полезной нагрузки поверх стандарта JSON-RPC , как показано в приведенном ниже примере протокола message/send :

{
  # identifier for this request
  "id": "abc123",
  # version of JSON-RPC protocol
  "jsonrpc": "2.0",
  # method name
  "method": "message/send",
  # parameters/arguments of the method
  "params": {
    "message": "hi, what can you help me with?"
  }  
}

Доступны различные методы, например, для поддержки разных типов связи (например, синхронная, потоковая, асинхронная) или для настройки уведомлений о состоянии задачи. Сервер A2A может быть гибко настроен для обработки этих стандартов определения задач. Подробности об этих методах можно прочитать в этом документе .

11. 🎯 Вызов

Итак, сможете ли вы подготовить необходимые файлы и самостоятельно развернуть приложение Gradio в облаке? Пришло время принять вызов!

12. 🧹 Уборка

Чтобы избежать списания средств с вашего аккаунта Google Cloud за ресурсы, использованные в этом практическом задании, выполните следующие действия:

В консоли Google Cloud перейдите на страницу «Управление ресурсами» .
В списке проектов выберите проект, который хотите удалить, и нажмите кнопку «Удалить» .
В диалоговом окне введите идентификатор проекта, а затем нажмите «Завершить» , чтобы удалить проект.
В качестве альтернативы вы можете перейти в Cloud Run и Agent Engine в консоли, выбрать только что развернутую службу и удалить ее.