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

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

1. Wprowadzenie

Protokół agent-agent (A2A) ma na celu ujednolicenie komunikacji między agentami AI, zwłaszcza tymi, którzy są wdrażani w systemach zewnętrznych. Wcześniej takie protokoły były ustanawiane dla narzędzi o nazwie protokół kontekstu modelu (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. Możesz na przykład prowadzić dwustronną 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 oczekujemy w tym demo

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 i włączanie w nim wszystkich wymaganych interfejsów API
Konfigurowanie obszaru roboczego dla środowiska programistycznego
Przygotowywanie zmiennych środowiskowych dla usług agentów burgerów i pizzy oraz testowanie ich lokalnie
Wdrażanie agenta do obsługi burgerów i pizzy w Cloud Run
Sprawdź szczegóły dotyczące sposobu ustanowienia serwera A2A
Przygotowywanie zmiennych środowiskowych dla usługi concierge zakupów i testowanie ich lokalnie
Wdrażanie usługi zakupów Concierge w Agent Engine
Łączenie z platformą agenta za pomocą interfejsu lokalnego
Sprawdź szczegóły dotyczące sposobu nawiązania połączenia przez klienta A2A i modelowania danych.
Sprawdzanie ładunku i interakcji między klientem A2A a serwerem

Omówienie architektury

Wdrożymy tę architekturę usługi

Wdrożymy 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 bezpośrednią interakcję tylko z konsjerżem ds. zakupów, który będzie działać na platformie Agent Development Kit (ADK) pełniącej 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łączonymi płatnościami

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. Zanim zaczniesz

Wybieranie aktywnego projektu w Cloud Console

W tym samouczku zakładamy, że masz już projekt Google Cloud z włączonymi płatnościami. Jeśli jeszcze go nie masz, możesz zacząć, wykonując te czynności.

W konsoli Google Cloud na stronie selektora projektów wybierz lub utwórz projekt Google Cloud.
Sprawdź, czy w projekcie Cloud włączone są płatności. Dowiedz się, jak sprawdzić, czy w projekcie włączone są płatności.

Konfigurowanie projektu w Cloud Shell

Będziesz używać Cloud Shell, czyli środowiska wiersza poleceń działającego w Google Cloud, które jest wstępnie załadowane narzędziem bq. U góry konsoli Google Cloud kliknij Aktywuj Cloud Shell. Jeśli pojawi się prośba o autoryzację, kliknij Autoryzuj.

Po połączeniu z Cloud Shell sprawdź, czy jesteś już uwierzytelniony i czy projekt jest ustawiony na Twój identyfikator projektu, używając tego polecenia:

gcloud auth list

Aby potwierdzić, że polecenie gcloud zna Twój projekt, uruchom w Cloud Shell to polecenie:

gcloud config list project

Jeśli projekt nie jest ustawiony, użyj tego polecenia, aby go ustawić:

gcloud config set project <YOUR_PROJECT_ID>

Możesz też zobaczyć identyfikator PROJECT_ID w konsoli.

Kliknij go, a po prawej stronie zobaczysz wszystkie swoje projekty i identyfikator projektu.

Włącz wymagane interfejsy API za pomocą polecenia pokazanego poniżej. Może to potrwać kilka minut, więc zachowaj cierpliwość.

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.

Alternatywą dla polecenia gcloud jest wyszukanie poszczególnych usług w konsoli lub skorzystanie z tego linku.

Jeśli pominiesz jakiś interfejs API, zawsze możesz go włączyć w trakcie wdrażania.

Informacje o poleceniach gcloud i ich użyciu znajdziesz w dokumentacji.

Otwórz edytor 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, w którym możesz pisać kod.
Sprawdź, czy projekt Cloud Code jest ustawiony w lewym dolnym rogu (na pasku stanu) edytora Cloud Shell, jak pokazano na ilustracji poniżej, i czy jest ustawiony jako aktywny projekt Google Cloud, w którym masz włączone płatności. W razie potrzeby kliknij Zezwól. Jeśli wykonasz poprzednie polecenie, przycisk może też prowadzić bezpośrednio do aktywowanego projektu zamiast do przycisku logowania.

Następnie sklonujmy z GitHuba katalog roboczy szablonu na potrzeby 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

Następnie przejdź do górnej sekcji edytora Cloud Shell i kliknij File->Open Folder (Plik –> Otwórz folder). Znajdź katalog username (nazwa użytkownika) 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:

Konfiguracja środowiska

Następnym krokiem jest przygotowanie środowiska programistycznego. Aktywny terminal powinien znajdować się w katalogu roboczym purchasing-concierge-a2a. W tym przewodniku wykorzystamy Pythona 3.12 i menedżera projektów Pythona uv, aby uprościć tworzenie i zarządzanie wersją Pythona oraz środowiskiem wirtualnym.

Jeśli terminal nie jest jeszcze otwarty, otwórz go, klikając Terminal –> Nowy terminal lub użyj skrótu Ctrl + Shift + C. W dolnej części przeglądarki otworzy się okno terminala.

Teraz zainicjuj środowisko wirtualne usługi Purchasing Concierge za pomocą polecenia uv (jest już wstępnie zainstalowane w terminalu w chmurze). Uruchom to polecenie:

uv sync --frozen

Spowoduje to utworzenie katalogu .venv i zainstalowanie zależności. Szybki podgląd pliku pyproject.toml pozwoli Ci uzyskać informacje o zależnościach, które będą wyświetlane w ten sposób:

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

Aby przetestować środowisko wirtualne, utwórz nowy plik main.py i skopiuj ten kod:

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

if __name__ == "__main__":
   main()

Następnie uruchom to polecenie:

uv run main.py

Otrzymasz dane wyjściowe podobne do tych poniżej.

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

Oznacza to, że projekt w Pythonie jest prawidłowo konfigurowany.

Teraz możemy przejść do następnego kroku, czyli skonfigurowania i wdrożenia agenta sprzedawcy zdalnego.

3. Wdrażanie agenta sprzedawcy zdalnego – serwer A2A w Cloud Run

W tym kroku wdrożymy 2 agenty sprzedawcy zdalnego oznaczone czerwonym polem. Agent do obsługi zamówień burgerów będzie korzystać z platformy CrewAI, a agent do obsługi zamówień pizzy – z platformy Langgraph. Oba będą oparte na modelu Gemini Flash 2.0.

Wdrażanie zdalnego agenta Burgera

Kod źródłowy agenta Burger znajduje się w katalogu remote_seller_agents/burger_agent. Inicjalizację agenta można sprawdzić w skrypcie agent.py. Oto fragment kodu zainicjowanego agenta

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

...

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. Omówimy to w dalszej części prezentacji. 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ę komunikat z prośbą o potwierdzenie utworzenia repozytorium kontenera na potrzeby wdrażania z kodu źródłowego, odpowiedz Y. Dzieje się tak tylko wtedy, gdy nigdy wcześniej nie wdrażano w Cloud Run z kodu źródłowego. 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. Teraz spróbujmy przejść https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json do wdrożonych usług agenta do burgerów za pomocą przeglądarki. 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 adresu https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json 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. Omówimy to w dalszej części prezentacji. 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. W tym przykładzie musimy zaktualizować tę wartość do adresu URL usługi agenta 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 wdróż 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ę.

Teraz, 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.

Wdrażanie zdalnego agenta Pizza

Podobnie kod źródłowy agenta do pizzy znajduje się w katalogu remote_seller_agents/pizza_agent. Inicjalizację agenta można sprawdzić w skrypcie agent.py. Oto fragment kodu zainicjowanego agenta

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

...

Podobnie jak w przypadku poprzedniego kroku wdrażania agenta do obsługi 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. Omówmy teraz podstawowe komponenty serwera A2A.

4. 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 informacji, takich jak umiejętności agenta, możliwości przesyłania strumieniowego, obsługiwane tryby, wersja protokołu i inne.

Wszystkie te informacje mogą być wykorzystywane 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 udostępnia 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 jednak 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:

Opublikuj kartę 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) .

5. Wdrażanie klienta Purchasing Concierge 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_agent.py. Oto fragment kodu zainicjowanego agenta.

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

...

Wdrożymy tego agenta w silniku agenta. 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. 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, przejdź do konsoli 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 asystenta ds. zakupów. W tej wersji demonstracyjnej wdrożymy skrypt deploy_to_agent_engine.py, którego treść jest widoczna poniżej.

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

Oto czynności, które należy wykonać, aby wdrożyć 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, podając ścieżkę katalogu agenta w polu extra_packages i podając niezbędne zmienne środowiskowe.

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. Zwróć uwagę, że xxxx to identyfikator projektu, a yyyy to identyfikator zasobu silnika agenta.

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.

Testowanie wdrożonego agenta w Agent Engine

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 spróbować wysłać to zapytanie, aby sprawdzić, czy agent został wdrożony.

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",
  }
}'

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.

6. Główne komponenty 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, a 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 Tool

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.

7. Testowanie integracji i sprawdzanie ładunku

Przyjrzyjmy się teraz naszemu asystentowi zakupów z interakcją zdalnego agenta za pomocą interfejsu internetowego.

Najpierw musimy zaktualizować AGENT_ENGINE_RESOURCE_NAME w .env pliku. 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

Następnie uruchom to polecenie, aby wdrożyć aplikację Gradio:

uv run purchasing_concierge_ui.py

Jeśli się uda, 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ż dokończysz składanie zamówienia. 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. Teraz zobaczmy, jak jest ona wdrażana w architekturze klient-serwer.

8. Wyzwanie

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

9. Czyszczenie danych

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.
Na liście projektów wybierz projekt, który chcesz usunąć, a potem kliknij Usuń.
W oknie wpisz identyfikator projektu i kliknij Wyłącz, aby usunąć projekt.
Możesz też otworzyć Cloud Run w konsoli, wybrać wdrożoną usługę i ją usunąć.