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

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

1. Введение

Протокол 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 и включите в нем все необходимые API.
Настройте рабочее пространство для вашей среды кодирования
Подготовьте переменные окружения для услуг агентов по доставке бургеров и пиццы и попробуйте локально.
Развертывание агента по доставке бургеров и пиццы в Cloud Run
Проверьте подробности о том, как был установлен сервер A2A.
Подготовьте переменные окружения для покупки консьержа и попробуйте локально
Внедрение консьержа по закупкам в Agent Engine
Подключитесь к агентскому движку через локальный интерфейс
Проверьте детали создания клиента A2A и его моделирования данных.
Проверка полезной нагрузки и взаимодействия между клиентом и сервером A2A

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

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

Мы развернем два сервиса, которые будут выступать в качестве сервера 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. Прежде чем начать

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

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

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

Настройка облачного проекта в Cloud Shell Terminal

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

После подключения к Cloud Shell вы проверяете, что вы уже прошли аутентификацию и что проекту присвоен ваш идентификатор проекта, с помощью следующей команды:

gcloud auth list

Выполните следующую команду в Cloud Shell, чтобы подтвердить, что команда gcloud знает о вашем проекте.

gcloud config list project

Если ваш проект не настроен, используйте следующую команду для его настройки:

gcloud config set project <YOUR_PROJECT_ID>

Кроме того, вы также можете увидеть идентификатор PROJECT_ID в консоли.

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

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

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

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

Operation "operations/..." finished successfully.

Альтернативой команде gcloud является поиск каждого продукта через консоль или использование этой ссылки .

Если какой-либо API отсутствует, вы всегда можете включить его в ходе реализации.

Информацию о командах и использовании gcloud см. в документации .

Перейдите в редактор Cloud Shell и настройте рабочий каталог приложения.

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

Нажмите кнопку «Открыть редактор», откроется редактор Cloud Shell, где мы можем написать наш код.
Убедитесь, что проект Cloud Code выбран в левом нижнем углу (строке состояния) редактора Cloud Shell, как показано на изображении ниже, и соответствует активному проекту Google Cloud, для которого включена оплата. При необходимости авторизуйтесь . Если вы уже выполнили предыдущую команду, кнопка может указывать непосредственно на ваш активированный проект, а не на кнопку входа.

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

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

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

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

Настройка среды

Следующий шаг — подготовка среды разработки. Ваш текущий активный терминал должен находиться в рабочем каталоге buying-concierge-a2a . В этой лабораторной работе мы будем использовать Python 3.12 и менеджер проектов UV Python для упрощения создания и управления версиями Python и виртуальными средами.

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

Теперь инициализируем виртуальную среду консьержа по закупкам с помощью uv (уже предустановленного в облачном терминале). Выполните эту команду:

uv sync --frozen

Это создаст каталог .venv и установит зависимости. Быстрый просмотр файла pyproject.toml даст вам информацию о зависимостях, показанных следующим образом.

dependencies = [
    "a2a-sdk>=0.2.16",
    "google-adk>=1.8.0",
    "gradio>=5.38.2",
]

Чтобы протестировать виртуальное окружение, создайте новый файл main.py и скопируйте следующий код.

def main():
   print("Hello from purchasing-concierge-a2a!")

if __name__ == "__main__":
   main()

Затем выполните следующую команду

uv run main.py

Вы получите вывод, как показано ниже.

Using CPython 3.12
Creating virtual environment at: .venv
Hello from purchasing-concierge-a2a!

Это показывает, что проект Python настраивается правильно.

Теперь мы можем перейти к следующему шагу — настройке и развертыванию нашего удаленного агента-продавца.

3. Развертывание удаленного агента продавца — сервера A2A в облаке

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

Развертывание удаленного агента Burger

Исходный код агента Burger находится в каталоге remote_seller_agents/burger_agent . Инициализацию агента можно проверить в скрипте 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

...

Всех файлов в каталоге 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 через браузер. Это URL для доступа к карточке развёрнутого агента сервера A2A.

В случае успешного развертывания вы увидите в своем браузере ответ, подобный показанному ниже, при доступе к https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json :

Это информация о карте агента Burger Agent, которая должна быть доступна для целей обнаружения. Мы обсудим это позже. Обратите внимание, что значение 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 уже будет правильно настроено.

Развертывание удаленного агента пиццы

Аналогично, исходный код агента доставки пиццы находится в каталоге remote_seller_agents/pizza_agent . Инициализацию агента можно проверить в скрипте agent.py . Ниже представлен фрагмент кода инициализированного агента.

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,
)

...

Как и в предыдущем шаге развертывания 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. Теперь давайте обсудим основные компоненты сервера A2A.

4. Основные компоненты сервера 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 ).

5. Развертывание Purchasing Concierge — A2A Client to Agent Engine

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

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

...

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

Сначала нам нужно создать наше промежуточное хранилище в 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 , содержимое которого показано ниже.

"""
Copyright 2025 Google LLC

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    https://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""

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

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

uv run deploy_to_agent_engine.py

После успешного развертывания появится следующий лог. Обратите внимание, что xxxx — это идентификатор вашего проекта, а 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

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

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

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://us-central1-aiplatform.googleapis.com/v1/projects/{YOUR_PROJECT_ID}/locations/us-central1/reasoningEngines/{YOUR_AGENT_ENGINE_RESOURCE_ID}:streamQuery?alt=sse -d '{
  "class_method": "stream_query",
  "input": {
    "user_id": "user_123",
    "message": "List available burger menu please",
  }
}'

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

{
  "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.

6. Основные компоненты 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 можно гибко настроить для работы с этими стандартами определения задач. Подробности этих методов можно найти в этом документе .

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

Теперь давайте рассмотрим взаимодействие нашего консьержа по закупкам с удаленным агентом с помощью веб-интерфейса.

Сначала нам нужно обновить 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

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

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. Вызов

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

9. Уборка

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

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