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

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

1. 📖 Введение

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

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

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

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

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

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

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

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

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

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

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

Вы развернете следующую архитектуру сервиса

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

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

Предпосылки

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

Чему вы научитесь

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

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

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

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

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

Шаг 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`

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

Далее нам также необходимо проверить, настроена ли оболочка на правильный идентификатор ПРОЕКТА , который у вас есть. Если вы видите значение внутри ( ) перед значком $ в терминале (на снимке экрана ниже это значение «a2a-agent-engine» ), это значение показывает настроенный проект для вашего активного сеанса оболочки.

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

gcloud config set project <YOUR_PROJECT_ID>

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

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

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

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

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

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

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

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

Ваш текущий активный терминал должен находиться в рабочем каталоге buying-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 Server в Cloud Run

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

4. 🚀 Развертывание Burger Seller Agent — A2A Server

Исходный код агента 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}

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

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.

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

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

После этого нажмите «Добавить переменную» и задайте для 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 здесь будет уникальным идентификатором при развертывании службы.

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

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

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

Найдите Cloud Run в строке поиска в верхней части вашей облачной консоли.

Нажмите на ранее развернутую облачную службу Pizza-Agent.

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

Скопируйте 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. 🚀 Развертывание Purchasing Concierge — A2A Client to Agent Engine

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

Исходный код нашего консьержа-агента по закупкам находится в каталоге buying_concierge . Инициализацию агента можно проверить в скрипте buying_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_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 на терминале или нажмите кнопку веб-предварительного просмотра, чтобы открыть веб-интерфейс.

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

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

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

Мы видим, что взаимодействие с двумя разными агентами приводит к двум разным моделям поведения, и 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.

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

{
  "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, предоставляя агенту клиента контекстную информацию о возможностях и навыках удаленного агента, которые необходимо задействовать. Более подробную информацию о полях этой карточки агента можно найти в этой документации .

В нашем коде реализация карты агента реализована с использованием A2A Python SDK, проверьте фрагмент 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. 💡 [Объяснение кода] Развертывание Agent Engine

Вот фрагмент кода консьерж-агента по закупкам в файле buying_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. Затем мы можем запустить метод agent_engines.create , предоставив объект adk_app , указав требования в поле requirements , указав путь к каталогу агента в extra_packages (при необходимости вы также можете указать другие каталоги и файлы) и указав необходимые переменные окружения.

10. 💡 [Объяснение кода] Концепция и реализация A2A Client

На изображении выше показан типичный поток взаимодействий A2A:

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

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

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

Открытие карты

При запуске сервиса A2A Client типичный процесс заключается в получении информации о карте агента и её сохранении для быстрого доступа при необходимости. В этой лабораторной работе мы реализуем это в методе 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)

...

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

Инструмент «Запросить и отправить задачу»

Это подсказка и инструмент, которые мы предоставляем нашему агенту 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 на консоли, выбрать службу, которую вы только что развернули, и удалить ее.