Ta strona została przetłumaczona przez Cloud Translation API.

Wprowadzenie do protokołu Agent2Agent (A2A): interakcje agenta ds. zakupów i zdalnego sprzedawcy w Cloud Run i Agent Engine

1. 📖 Wprowadzenie

Protokół Agent2Agent (A2A) został zaprojektowany w celu standaryzacji komunikacji między agentami AI, zwłaszcza tymi, które są wdrażane w systemach zewnętrznych. Wcześniej takie protokoły były ustanawiane dla narzędzi o nazwie Model Context Protocol (MCP), który jest nowym standardem łączenia dużych modeli językowych z danymi i zasobami. A2A ma uzupełniać MCP. A2A koncentruje się na innym problemie niż MCP. MCP skupia się na zmniejszeniu złożoności połączenia agentów z narzędziami i danymi, a A2A – na umożliwieniu agentom współpracy w ich naturalnych trybach. Umożliwia to agentom komunikowanie się jako agenci (lub użytkownicy), a nie jako narzędzia. Na przykład umożliwia to dwukierunkową komunikację, gdy chcesz coś zamówić.

A2A ma uzupełniać MCP. W oficjalnej dokumentacji zaleca się, aby aplikacje używały MCP do obsługi narzędzi, a A2A do obsługi agentów reprezentowanych przez AgentCard ( omówimy to później). Frameworki mogą następnie używać A2A do komunikowania się z użytkownikiem, zdalnymi agentami i innymi agentami.

W tej wersji demonstracyjnej zaczniemy od wdrożenia A2A przy użyciu pakietu SDK w języku Python. Rozważymy przypadek użycia, w którym mamy osobistego asystenta zakupów, który może pomóc nam w komunikacji z agentami sprzedającymi burgery i pizzę w celu obsługi naszego zamówienia.

A2A wykorzystuje zasadę klient-serwer. Oto typowy przepływ A2A, którego możesz się spodziewać w tym samouczku.

Klient A2A najpierw wykrywa wszystkie dostępne karty agenta serwera A2A i wykorzystuje informacje z nich do utworzenia klienta połączenia.
W razie potrzeby klient A2A wyśle wiadomość do serwera A2A, który potraktuje ją jako zadanie do wykonania. Jeśli adres URL odbiorcy powiadomień push jest skonfigurowany na kliencie A2A i obsługiwany przez serwer A2A, serwer będzie też mógł publikować stan postępu zadania w punkcie końcowym odbiorcy na kliencie.
Po zakończeniu zadania serwer A2A wyśle artefakt odpowiedzi do klienta A2A.

W ramach ćwiczeń z programowania będziesz wykonywać kolejne czynności:

Przygotowywanie projektu Google Cloud
Konfigurowanie katalogu roboczego dla środowiska kodowania
Wdrażanie agenta do obsługi zamówień na burgery w Cloud Run
Wdrażanie agenta do zamawiania pizzy w Cloud Run
Wdrażanie usługi zakupów Concierge w Agent Engine
Interakcja z asystentem zakupów za pomocą lokalnego interfejsu

Omówienie architektury

Wdrożysz tę architekturę usługi

Wdrożysz 2 usługi, które będą pełnić funkcję serwera A2A: agenta Burger ( opartego na platformie agentów CrewAI) i agenta Pizza ( opartego na platformie agentów Langgraph). Użytkownik będzie wchodzić w interakcję bezpośrednio tylko z usługą Purchasing Concierge, która będzie działać na platformie Agent Development Kit (ADK) i pełnić funkcję klienta A2A.

Każdy z tych agentów będzie miał własne środowisko i wdrożenie.

Wymagania wstępne

Komfortowa praca z Pythonem
Znajomość podstawowej architektury pełnego stosu z użyciem usługi HTTP

Czego się nauczysz

Podstawowa struktura serwera A2A
Podstawowa struktura klienta A2A
Wdrażanie usługi agenta w Cloud Run
Wdrażanie usługi agenta w Agent Engine
Sposób łączenia się klienta A2A z serwerem A2A
Struktura żądania i odpowiedzi w przypadku połączenia niestrumieniowego

Czego potrzebujesz

przeglądarki Chrome,
konto Gmail,
Projekt w chmurze z włączonym kontem rozliczeniowym

Ten przewodnik, przeznaczony dla deweloperów na wszystkich poziomach zaawansowania (w tym dla początkujących), wykorzystuje w przykładowej aplikacji język Python. Znajomość Pythona nie jest jednak wymagana do zrozumienia przedstawionych koncepcji.

2. 🚀 Przygotowywanie konfiguracji programistycznej warsztatów

Krok 1. Wybierz aktywny projekt w Cloud Console

W konsoli Google Cloud na stronie selektora projektów wybierz lub utwórz projekt Google Cloud (patrz lewa górna sekcja konsoli).

Kliknij go, a zobaczysz listę wszystkich projektów, jak w tym przykładzie:

Wartość wskazana przez czerwone pole to IDENTYFIKATOR PROJEKTU. Będzie ona używana w całym samouczku.

Sprawdź, czy w projekcie Cloud włączone są płatności. Aby to sprawdzić, kliknij ikonę menu ☰ na pasku w lewym górnym rogu, aby wyświetlić menu nawigacyjne, a następnie znajdź menu Płatności.

Jeśli widzisz komunikat „Z tym projektem połączone jest konto rozliczeniowe Google Cloud Platform w wersji próbnej”, możesz użyć projektu w tym samouczku. Jeśli nie, wróć na początek tego samouczka i zrealizuj konto rozliczeniowe.

Krok 2. Zapoznaj się z Cloud Shell

W większości samouczków będziesz używać Cloud Shell. Kliknij Aktywuj Cloud Shell u góry konsoli Google Cloud. Jeśli pojawi się prośba o autoryzację, kliknij Autoryzuj.

Po połączeniu z Cloud Shell musimy sprawdzić, czy powłoka ( lub terminal) jest już uwierzytelniona na naszym koncie.

gcloud auth list

Jeśli widzisz swój osobisty adres Gmail, jak w przykładzie poniżej, wszystko jest w porządku.

Credentialed Accounts

ACTIVE: *
ACCOUNT: alvinprayuda@gmail.com

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

Jeśli nie, spróbuj odświeżyć przeglądarkę i kliknąć Autoryzuj, gdy pojawi się odpowiedni komunikat ( może to zostać przerwane z powodu problemu z połączeniem).

Następnie musisz sprawdzić, czy powłoka jest już skonfigurowana z prawidłowym IDENTYFIKATOREM PROJEKTU. Jeśli w terminalu przed ikoną $ widzisz wartość w nawiasach ( ) ( na zrzucie ekranu poniżej wartość to „a2a-agent-engine”), oznacza to, że w aktywnej sesji powłoki skonfigurowany jest projekt.

Jeśli wyświetlana wartość jest już prawidłowa, możesz pominąć następne polecenie. Jeśli jednak jest nieprawidłowy lub go brakuje, uruchom to polecenie:

gcloud config set project <YOUR_PROJECT_ID>

Następnie skopiuj z GitHuba katalog roboczy szablonu dla tego laboratorium, wykonując to polecenie: Utworzy katalog roboczy w katalogu purchasing-concierge-a2a.

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

Krok 3. Zapoznaj się z edytorem Cloud Shell i skonfiguruj katalog roboczy aplikacji

Teraz możemy skonfigurować edytor kodu, aby wykonywać różne czynności związane z kodowaniem. W tym celu użyjemy edytora Cloud Shell.

Kliknij przycisk Otwórz edytor. Spowoduje to otwarcie edytora Cloud Shell .

Następnie przejdź do górnej sekcji edytora Cloud Shell i kliknij File (Plik) –> Open Folder (Otwórz folder). Znajdź katalog username i katalog purchasing-concierge-a2a, a potem kliknij OK. Spowoduje to ustawienie wybranego katalogu jako głównego katalogu roboczego. W tym przykładzie nazwa użytkownika to alvinprayuda, więc ścieżka do katalogu jest widoczna poniżej.

Edytor Cloud Shell powinien teraz wyglądać tak:

Teraz otwórz terminal edytora. Możesz to zrobić, klikając na pasku menu Terminal -> Nowy terminal lub używając skrótu Ctrl + Shift + C. W dolnej części przeglądarki otworzy się okno terminala.

Aktywny terminal powinien znajdować się w katalogu roboczym purchasing-concierge-a2a. W tym samouczku użyjemy Pythona 3.12 i menedżera projektów Pythona uv, aby uprościć tworzenie wersji Pythona i środowiska wirtualnego oraz zarządzanie nimi. Pakiet uv jest już wstępnie zainstalowany w Cloud Shell.

Uruchom to polecenie, aby zainstalować wymagane zależności w środowisku wirtualnym w katalogu .venv.

uv sync --frozen

Sprawdź plik pyproject.toml, aby zobaczyć zadeklarowane zależności na potrzeby tego samouczka, czyli a2a-sdk, google-adk, and gradio.

Teraz musimy włączyć wymagane interfejsy API za pomocą polecenia pokazanego poniżej. To może chwilę potrwać.

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

Po pomyślnym wykonaniu polecenia powinien wyświetlić się komunikat podobny do tego poniżej:

Operation "operations/..." finished successfully.

3. 🚀 Wdrażanie agentów zdalnych sprzedawców A2A na Cloud Run

W tym kroku wdrożymy 2 agenty sprzedawcy zdalnego oznaczone czerwonym polem. Agent do burgerów będzie korzystać z platformy agentów CrewAI, a agent do pizzy – z platformy Langgraph.

4. 🚀 Wdrażanie agenta sprzedającego burgery – serwer A2A

Kod źródłowy agenta burgerów znajduje się w katalogu remote_seller_agents/burger_agent.

Wszystkie pliki znajdujące się w katalogu remote_seller_agents/burger_agent wystarczą do wdrożenia agenta w Cloud Run, aby był dostępny jako usługa. Aby go wdrożyć, uruchom to polecenie:

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}

Jeśli pojawi się pytanie, czy utworzyć repozytorium kontenera na potrzeby wdrażania z kodu źródłowego, odpowiedz Y. Po udanym wdrożeniu pojawi się log podobny do tego.

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

Część xxxx będzie unikalnym identyfikatorem po wdrożeniu usługi.

Otwórz nową kartę przeglądarki i przejdź do trasy https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json wdrożonych usług agenta burgera w przeglądarce. Jest to adres URL, pod którym można uzyskać dostęp do wdrożonej karty agenta serwera A2A.

Jeśli wdrożenie się powiedzie, po otwarciu karty agenta w przeglądarce zobaczysz odpowiedź podobną do tej poniżej.

Są to informacje o karcie agenta burgera, które powinny być dostępne do celów wykrywania.

Zwróć uwagę, że wartość url jest nadal ustawiona na http://0.0.0.0:8080/. Ta wartość url powinna być główną informacją dla klienta A2A, która umożliwia wysyłanie wiadomości do świata zewnętrznego. Nie jest ona jednak prawidłowo skonfigurowana.

Musimy zaktualizować tę wartość, aby wskazywała adres URL naszej usługi agenta ds. burgerów, dodając dodatkową zmienną środowiskową HOST_OVERRIDE.

Aktualizowanie wartości adresu URL agenta Burger na karcie agenta za pomocą zmiennej środowiskowej

Aby dodać HOST_OVERRIDE do usługi agenta burgerów, wykonaj te czynności:

Wyszukaj Cloud Run na pasku wyszukiwania u góry konsoli chmury.

Kliknij wdrożoną wcześniej usługę Cloud Run burger-agent.

Skopiuj adres URL usługi burger-service, a następnie kliknij Edytuj i wdroż nową wersję.

Następnie kliknij sekcję Zmienne i obiekty tajne.

Następnie kliknij Dodaj zmienną i ustaw wartość HOST_OVERRIDE na adres URL usługi ( ten ze wzorcem https://burger-agent-xxxxxxxxx.us-central1.run.app).

Na koniec kliknij przycisk Wdróż, aby ponownie wdrożyć usługę.

Gdy ponownie otworzysz kartę agenta burger-agent w przeglądarce https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json , wartość url będzie już prawidłowo skonfigurowana.

5. 🚀 Wdrażanie agenta sprzedawcy pizzy – serwer A2A

Podobnie kod źródłowy agenta do pizzy znajduje się w katalogu remote_seller_agents/pizza_agent.

Podobnie jak w przypadku poprzedniego kroku wdrażania agenta do zamawiania burgerów, wszystkie pliki znajdujące się w katalogu remote_seller_agents/pizza_agent wystarczą do wdrożenia agenta w Cloud Run, aby był dostępny jako usługa. Aby go wdrożyć, uruchom to polecenie:

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}

Po udanym wdrożeniu pojawi się log podobny do tego.

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

Część xxxx będzie unikalnym identyfikatorem po wdrożeniu usługi.

Podobnie jest w przypadku agenta burgerów. Jeśli spróbujesz przejść do trasy https://pizza-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json wdrożonych usług agenta pizzy za pomocą przeglądarki, aby uzyskać dostęp do karty agenta serwera A2A, wartość url agenta pizzy na jego karcie agenta nie jest jeszcze prawidłowo skonfigurowana. Musimy też dodać HOST_OVERRIDE do zmiennej środowiskowej

Aktualizowanie wartości adresu URL agenta Pizza na karcie agenta za pomocą zmiennej środowiskowej

Aby dodać HOST_OVERRIDE do usługi agenta ds. pizzy, wykonaj te czynności:

Wyszukaj Cloud Run na pasku wyszukiwania u góry konsoli chmury.

Kliknij wdrożoną wcześniej usługę Cloud Run pizza-agent.

Kliknij Edytuj i wdrażaj nową wersję.

Skopiuj adres URL usługi pizza, a następnie kliknij sekcję Zmienne i obiekty tajne.

Następnie kliknij Dodaj zmienną i ustaw wartość HOST_OVERRIDE na adres URL usługi ( ten ze wzorcem https://pizza-agent-xxxxxxxxx.us-central1.run.app).

Na koniec kliknij przycisk Wdróż, aby ponownie wdrożyć usługę.

Teraz, gdy ponownie otworzysz kartę agenta pizza-agent w przeglądarce https://pizza-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json, wartość url będzie już prawidłowo skonfigurowana.

Na tym etapie udało nam się już wdrożyć w Cloud Run usługi burgerów i pizzy.

6. 🚀 Wdrażanie usługi Purchasing Concierge – klienta A2A w Agent Engine

W tym kroku wdrożymy agenta ds. zakupów. To z tym agentem będziemy wchodzić w interakcję.

Kod źródłowy naszego agenta zakupowego znajduje się w katalogu purchasing_concierge. Inicjowanie agenta można sprawdzić w skrypcie purchasing_concierge/purchasing_agent.py.

Aby go wdrożyć, wykonaj te czynności :

Najpierw musimy utworzyć w Cloud Storage miejsce na dane tymczasowe.

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

Teraz musimy najpierw przygotować zmienną .env. Skopiujmy plik .env.example do pliku .env.

cp .env.example .env

Teraz otwórz plik .env. Zobaczysz w nim te treści:

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}

Ten agent będzie komunikować się z agentem od burgerów i pizzy, dlatego musimy podać odpowiednie dane logowania dla obu agentów. Musimy zaktualizować zmienne PIZZA_SELLER_AGENT_URL i BURGER_SELLER_AGENT_URL, podając adres URL Cloud Run z poprzednich kroków.

Jeśli o tym zapomnisz, otwórz konsolę Cloud Run. Wpisz „Cloud Run” na pasku wyszukiwania u góry konsoli i kliknij prawym przyciskiem myszy ikonę Cloud Run, aby otworzyć ją w nowej karcie.

Powinny pojawić się wdrożone wcześniej usługi agenta sprzedawcy zdalnego, jak pokazano poniżej.

Aby wyświetlić publiczny adres URL tych usług, kliknij jedną z nich. Przekierujemy Cię na stronę Szczegóły usługi. Adres URL znajdziesz w górnej części ekranu, obok informacji o regionie.

Ostateczna zmienna środowiskowa powinna wyglądać podobnie do tej:

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}

Teraz możemy wdrożyć naszego agenta ds. zakupów. Wdrożymy go w silniku agenta, a kod wdrożenia znajduje się w skrypcie deploy_to_agent_engine.py.

Możemy go wdrożyć, uruchamiając skrypt:

uv run deploy_to_agent_engine.py

Po udanym wdrożeniu pojawi się log podobny do tego. Wyświetli nazwę zasobu silnika agenta w formacie "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

Gdy sprawdzimy go w panelu silnika agenta (wyszukaj „silnik agenta” na pasku wyszukiwania), zobaczymy poprzednie wdrożenie.

Możesz też sprawdzić, czy wyświetla się tam nazwa zasobu silnika agenta. Następnie możesz użyć tej nazwy zasobu do przetestowania.

Następnie zaktualizuj wartość AGENT_ENGINE_RESOURCE_NAME w pliku .env. Sprawdź, czy podajesz prawidłową nazwę zasobu silnika agenta. Plik .env powinien wyglądać tak:

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

Testowanie wdrożonego agenta w silniku agenta

Interakcja z silnikiem agenta może odbywać się za pomocą curlpolecenia i pakietu SDK. Aby na przykład spróbować interakcji z wdrożonym agentem, uruchom to polecenie.

Możesz wysłać to zapytanie, aby sprawdzić, czy agent został wdrożony. Uruchom ten skrypt test_agent_engine.sh:

bash test_agent_engine.sh

Możesz sprawdzić skrypt i zobaczyć, że próbujemy zadać agentowi pytanie „List available burger menu please” (Poproś o listę dostępnych zestawów z burgerem).

Jeśli operacja się uda, w konsoli pojawi się kilka zdarzeń odpowiedzi, np. tak:

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

W następnym kroku spróbujemy użyć interfejsu, ale najpierw omówmy główne komponenty i typowe działanie klientów A2A.

7. 🚀 Testowanie integracji i sprawdzanie ładunku

Przyjrzyjmy się teraz naszemu asystentowi zakupów z interakcją zdalnego agenta za pomocą interfejsu internetowego. Aby wdrożyć aplikację Gradio, uruchom to polecenie. Aby uruchomić tę aplikację, musisz wcześniej prawidłowo wypełnić plik .env.

uv run purchasing_concierge_ui.py

Jeśli operacja się powiedzie, zobaczysz te dane wyjściowe:

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

Następnie kliknij z naciśniętym klawiszem Ctrl adres http://0.0.0.0:8080 w terminalu lub kliknij przycisk podglądu w przeglądarce, aby otworzyć interfejs internetowy.

Spróbuj przeprowadzić rozmowę w ten sposób :

Pokaż menu burgerów i pizzy
Chcę zamówić 1 pizzę z kurczakiem w sosie BBQ i 1 burgera z kurczakiem w sosie cajun.

Kontynuuj rozmowę, aż zamówienie zostanie zrealizowane. Sprawdź, jak przebiega interakcja, oraz jakie jest wywołanie narzędzia i odpowiedź. Obraz poniżej przedstawia przykład wyniku interakcji.

Widzimy, że komunikacja z 2 różnymi agentami wywołuje 2 różne zachowania, a A2A dobrze sobie z tym radzi. Agent sprzedawcy pizzy bezpośrednio akceptuje naszą prośbę o zakup, natomiast agent sprzedawcy burgerów potrzebuje naszego potwierdzenia, zanim przystąpi do realizacji naszej prośby. Po potwierdzeniu agent może przekazać je agentowi sprzedawcy burgerów.

Omówiliśmy już podstawowe pojęcia związane z A2A. Zobaczmy teraz, jak jest ona wdrażana w architekturze klient-serwer.

8. 💡 [Wyjaśnienie kodu] Koncepcja i implementacja serwera A2A

Inicjowanie agenta sprzedawcy zdalnego można sprawdzić w skrypcie remote_seller_agents/*/agent.py. Oto fragment kodu agentów sprzedawcy.

Burger Agent

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

...

Agent do pizzy

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

...

Jak widać, te 2 agenty zostały utworzone przy użyciu zupełnie innych platform ( CrewAI i Langgraph) niż agent klienta ( ADK). W przypadku A2A nie stanowi to problemu. Aby się ze sobą komunikować, nie muszą udostępniać sobie kodu wewnętrznego. Nie ma znaczenia, jakich platform używają, w jakim języku działają ani gdzie są wdrożone.

Główne komponenty serwera A2A

Omówmy teraz podstawowe pojęcie i komponenty serwera A2A.

Karta agenta

Każdy serwer A2A musi mieć kartę agenta dostępną w zasobie /.well-known/agent.json. Ma to na celu wsparcie etapu odkrywania w przypadku klienta A2A, który powinien zawierać pełne informacje i konteksty dotyczące dostępu do agenta oraz jego możliwości. Jest to podobne do dobrze udokumentowanej dokumentacji interfejsu API z użyciem Swaggera lub Postmana.

To jest zawartość karty agenta burgera.

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

Karty agentów zawierają wiele ważnych komponentów, takich jak umiejętności agenta, możliwości przesyłania strumieniowego, obsługiwane tryby, wersja protokołu i inne informacje.

Wszystkie te informacje można wykorzystać do opracowania odpowiedniego mechanizmu komunikacji, aby klient A2A mógł się prawidłowo komunikować. Obsługiwany tryb i mechanizm uwierzytelniania zapewniają prawidłowe nawiązanie komunikacji, a informacje o agencie skills można umieścić w prompcie systemowym klienta A2A, aby przekazać agentowi klienta kontekst dotyczący możliwości i umiejętności zdalnego agenta, które mają zostać wywołane. Bardziej szczegółowe pola tej karty agenta znajdziesz w tej dokumentacji.

W naszym kodzie implementacja karty agenta jest realizowana za pomocą pakietu A2A Python SDK. Implementację znajdziesz we fragmencie kodu remote_seller_agents/burger_agent/main.py poniżej.

...

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

...

Widzimy tam kilka pól, takich jak:

AgentCapabilities : deklaracja dodatkowych funkcji opcjonalnych obsługiwanych przez usługę agenta,takich jak możliwość przesyłania strumieniowego lub obsługa powiadomień push.
AgentSkill : narzędzia lub funkcje obsługiwane przez agenta.
Input/OutputModes : obsługiwany typ danych wejściowych/wyjściowych
Url : adres do komunikacji z agentem;

W tej konfiguracji zapewniamy dynamiczne tworzenie adresu URL hosta agenta, co ułatwia przełączanie się między testowaniem lokalnym a wdrażaniem w chmurze. Dlatego w poprzednim kroku musieliśmy dodać zmienną HOST_OVERRIDE.

Kolejka zadań i wykonawca agenta

Serwer A2A może obsługiwać żądania od różnych agentów lub użytkowników i doskonale izolować poszczególne zadania. Aby lepiej zrozumieć kontekst tych przykładów, możesz przyjrzeć się poniższemu obrazowi.

Dlatego każdy serwer A2A powinien śledzić przychodzące zadania i przechowywać odpowiednie informacje na ich temat. Pakiet SDK A2A zawiera moduły, które pomagają rozwiązać ten problem na serwerze A2A. Najpierw możemy utworzyć instancję logiki, która określa, jak chcemy obsługiwać żądanie przychodzące. Dziedzicząc klasę abstrakcyjną AgentExecutor, możemy kontrolować sposób zarządzania wykonywaniem i anulowaniem zadań. Przykładową implementację można sprawdzić w remote_seller_agents/burger_agent/agent_executor.py module ( podobna ścieżka w przypadku sprzedawcy pizzy ).

...

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

...

W powyższym kodzie implementujemy podstawowy schemat przetwarzania, w którym agent jest wywoływany bezpośrednio po nadejściu żądania i wysyła zdarzenia ukończonego zadania po zakończeniu wywołania. Nie wdrożyliśmy tutaj metody anulowania, ponieważ uznaliśmy, że jest to operacja krótkotrwała.

Po utworzeniu wykonawcy możemy bezpośrednio użyć wbudowanych klas DefaultRequestHandler, InMemoryTaskStore i A2AStarletteApplication, aby uruchomić serwer HTTP. Tę implementację możesz sprawdzić w 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)

...

W tym module znajdziesz implementację /.well-known/agent.json, która umożliwia dostęp do karty agenta, a także punkt końcowy POST obsługujący protokół A2A.

Podsumowanie

Krótko mówiąc, wdrożony przez nas serwer A2A korzystający z pakietu SDK Pythona obsługuje 2 funkcje:

Publikowanie karty agenta na trasie /.well-known/agent.json
Obsługa żądania JSON-RPC z kolejkowaniem zadań w pamięci

Punkt wejścia do tych funkcji można sprawdzić w skrypcie __main__.py ( na remote_seller_agents/burger_agent lub remote_seller_agents/pizza_agent) .

9. 💡 [Wyjaśnienie kodu] Wdrażanie silnika agenta

Oto fragment kodu agenta ds. zakupów w pliku 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,
            ],
        )

...

Ten agent został utworzony za pomocą platformy ADK i wdrożony w Agent Engine.

Vertex AI Agent Engine to zestaw usług, które umożliwiają programistom wdrażanie agentów AI w środowisku produkcyjnym, zarządzanie nimi i ich skalowanie. Obsługuje infrastrukturę do skalowania agentów w środowisku produkcyjnym, dzięki czemu możemy skupić się na tworzeniu aplikacji. Więcej informacji na ten temat znajdziesz w tym dokumencie . Wcześniej musieliśmy przygotować pliki potrzebne do wdrożenia usługi agenta (np. skrypt serwera main i plik Dockerfile). W tym przypadku możemy wdrożyć agenta bezpośrednio ze skryptu w języku Python bez konieczności tworzenia własnej usługi backendu, używając kombinacji ADK i Agent Engine.

W tym samouczku wdrożymy skrypt deploy_to_agent_engine.py, którego treść jest widoczna poniżej.

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}")

Oto czynności potrzebne do wdrożenia agenta pakietu ADK w silniku agenta. Najpierw musimy utworzyć obiekt AdkApp z naszego ADK root_agent. Następnie możemy uruchomić metodę agent_engines.create, podając obiekt adk_app, określając wymagania w polu requirements, określając ścieżkę do katalogu agenta w polu extra_packages ( w razie potrzeby możesz tu też podać inne katalogi i pliki) oraz podając niezbędne zmienne środowiskowe.

10. 💡 [Wyjaśnienie kodu] Koncepcja i implementacja klienta A2A

Obraz powyżej przedstawia typowy przebieg interakcji A2A:

Klient spróbuje znaleźć dowolną opublikowaną kartę agenta pod podanym adresem URL agenta zdalnego na trasie /.well-known/agent.json
W razie potrzeby wyśle do tego agenta wiadomość z tekstem i niezbędnymi parametrami metadanych ( np. identyfikatorem sesji, kontekstem historycznym itp.). Serwer potraktuje tę wiadomość jako zadanie do wykonania.
Serwer A2A przetwarza żądanie. Jeśli obsługuje powiadomienia push, może też publikować niektóre powiadomienia podczas przetwarzania zadania ( ta funkcja wykracza poza zakres tego laboratorium).
Po zakończeniu serwer A2A wyśle artefakt odpowiedzi z powrotem do klienta.

Niektóre z głównych obiektów w przypadku powyższych interakcji to te elementy (więcej informacji znajdziesz tutaj) :

Wiadomość: wymiana informacji między klientem a pracownikiem obsługi klienta.
Zadanie: podstawowa jednostka pracy zarządzana przez A2A, identyfikowana za pomocą unikalnego identyfikatora.
Artefakt: dane wyjściowe (np.dokument, obraz, dane strukturalne) wygenerowane przez agenta w wyniku wykonania zadania, składające się z części.
Część: najmniejsza jednostka treści w wiadomości lub artefakcie. Może to być tekst, obraz, film, plik itp.

Odkrywanie kart

Podczas uruchamiania usługi A2A Client typowy proces polega na próbie uzyskania informacji o karcie agenta i zapisaniu ich, aby w razie potrzeby mieć do nich łatwy dostęp. W tym samouczku implementujemy go na before_agent_callback. Możesz zobaczyć implementację w purchasing_concierge/purchasing_agent.py. Fragment kodu znajdziesz poniżej.

...

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)

...

W tym miejscu próbujemy uzyskać dostęp do wszystkich dostępnych kart agentów za pomocą wbudowanego modułu klienta A2AA2ACardResolver. Następnie zbieramy połączenie potrzebne do wysłania wiadomości do agenta. Musimy też wymienić w prompcie wszystkich dostępnych agentów i ich specyfikacje, aby nasz agent wiedział, że może się z nimi komunikować.

Narzędzie Prompt and Send Task

To prompt i narzędzie, które udostępniamy naszemu agentowi 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

...

W prompcie podajemy naszemu asystentowi ds. zakupów imiona i nazwiska oraz opisy wszystkich dostępnych zdalnych agentów, a w narzędziu self.send_task udostępniamy mechanizm pobierania odpowiedniego klienta do połączenia z agentem i wysyłania wymaganych metadanych za pomocą obiektu SendMessageRequest.

Protokoły komunikacyjne

Definicja zadania to domena należąca do serwera A2A. Z perspektywy klienta A2A jest to jednak wiadomość wysyłana na serwer. To serwer decyduje, jak zdefiniować przychodzące wiadomości od klienta jako zadanie i czy do jego wykonania potrzebna jest interakcja ze strony klienta. Więcej informacji o cyklu życia zadania znajdziesz w tej dokumentacji. Koncepcję wyższego poziomu można przedstawić w ten sposób:

Wymiana komunikatów na zadania jest realizowana za pomocą formatu ładunku na podstawie standardu JSON-RPC, jak pokazano w poniższym przykładzie protokołu 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?"
  }  
}

Dostępne są różne metody, np.do obsługi różnych typów komunikacji (np. synchronicznej, strumieniowej, asynchronicznej) lub do konfigurowania powiadomień o stanie zadania. Serwer A2A można elastycznie skonfigurować tak, aby obsługiwał te standardy definicji zadań. Szczegółowe informacje o tych metodach znajdziesz w tym dokumencie.

11. 🎯 Wyzwanie

Czy możesz teraz przygotować niezbędny plik i samodzielnie wdrożyć aplikację Gradio w Cloud Run? Czas podjąć wyzwanie!

12. 🧹 Czyszczenie

Aby uniknąć obciążenia konta Google Cloud opłatami za zasoby użyte w tym laboratorium, wykonaj te czynności:

W konsoli Google Cloud otwórz stronę Zarządzanie zasobami.
Z listy projektów wybierz projekt do usunięcia, a potem kliknij Usuń.
W oknie wpisz identyfikator projektu i kliknij Wyłącz, aby usunąć projekt.
Możesz też otworzyć Cloud Run i Agent Engine w konsoli, wybrać wdrożoną usługę i ją usunąć.