Diese Seite wurde von der Cloud Translation API übersetzt.

Erste Schritte mit dem Agent-to-Agent-Protokoll (A2A): Interaktionen zwischen einem Einkaufs-Concierge und einem Remote-Verkäufer-Agenten mit Gemini in Cloud Run und Agent Engine

1. Einführung

Das Agent-to-Agent-Protokoll (A2A) wurde entwickelt, um die Kommunikation zwischen KI-Agents zu standardisieren, insbesondere für Agents, die in externen Systemen bereitgestellt werden. Bisher wurden solche Protokolle für Tools eingerichtet, die als Model Context Protocol (MCP) bezeichnet werden. Das MCP ist ein aufstrebender Standard, um LLMs mit Daten und Ressourcen zu verbinden. A2A versucht, MCP zu ergänzen. Während sich MCP darauf konzentriert, die Komplexität zu verringern, um Kundenservicemitarbeiter mit Tools und Daten zu verbinden, konzentriert sich A2A darauf, wie Kundenservicemitarbeiter in ihren natürlichen Modalitäten zusammenarbeiten können. So können Agents als Agents (oder als Nutzer) und nicht als Tools kommunizieren. Das ist z. B. nützlich, wenn Sie etwas bestellen möchten.

A2A soll MCP ergänzen. In der offiziellen Dokumentation wird empfohlen, dass Anwendungen MCP für Tools und A2A für Agents verwenden, die durch AgentCard dargestellt werden ( darauf gehen wir später ein). Die Frameworks können dann A2A verwenden, um mit ihren Nutzern, den Remote-Agents und anderen Agents zu kommunizieren.

In dieser Demo beginnen wir mit der Implementierung von A2A mithilfe des Python SDK. Wir werden einen Anwendungsfall untersuchen, in dem wir einen persönlichen Einkaufsassistenten haben, der uns helfen kann, mit den Verkäufern von Burgern und Pizzen zu kommunizieren, um unsere Bestellung zu bearbeiten.

A2A basiert auf dem Client-Server-Prinzip. Hier sehen Sie den typischen A2A-Ablauf, den wir in dieser Demo erwarten.

Der A2A-Client führt zuerst die Suche auf allen zugänglichen A2A-Server-Agent-Karten durch und verwendet die Informationen, um einen Verbindungsclient zu erstellen.
Bei Bedarf sendet der A2A-Client eine Nachricht an den A2A-Server. Der Server wertet diese als zu erledigende Aufgabe aus. Wenn die URL des Push-Benachrichtigungsempfängers auf dem A2A-Client konfiguriert ist und vom A2A-Server unterstützt wird, kann der Server auch den Status des Aufgabenfortschritts am Empfangsendpunkt auf dem Client veröffentlichen.
Nach Abschluss der Aufgabe sendet der A2A-Server das Antwortartefakt an den A2A-Client.

In diesem Codelab gehen Sie schrittweise so vor:

Google Cloud-Projekt vorbereiten und alle erforderlichen APIs aktivieren
Arbeitsbereich für Ihre Programmierumgebung einrichten
Umgebungsvariablen für Burger- und Pizza-Agent-Dienste vorbereiten und lokal testen
Burger- und Pizza-Agent in Cloud Run bereitstellen
Details zur Einrichtung des A2A-Servers ansehen
Umgebungsvariablen für den Kauf-Concierge vorbereiten und lokal testen
Einkaufs-Concierge in Agent Engine bereitstellen
Über die lokale Schnittstelle mit der Agent-Engine verbinden
Details zur Einrichtung des A2A-Clients und zur Datenmodellierung ansehen
Nutzlast und Interaktion zwischen A2A-Client und ‑Server prüfen

Architekturübersicht

Wir stellen die folgende Dienstarchitektur bereit:

Wir stellen zwei Dienste bereit, die als A2A-Server fungieren: den Burger-Agent ( basierend auf dem CrewAI-Agent-Framework) und den Pizza-Agent ( basierend auf dem Langgraph-Agent-Framework). Der Nutzer interagiert nur direkt mit dem Einkaufs-Concierge, der mit dem Agent Development Kit (ADK)-Framework ausgeführt wird und als A2A-Client fungiert.

Jeder dieser Agents hat eine eigene Umgebung und wird separat bereitgestellt.

Voraussetzungen

Vertrautheit mit Python
Grundkenntnisse der Full-Stack-Architektur mit HTTP-Dienst

Lerninhalte

Kernstruktur des A2A-Servers
Kernstruktur des A2A-Clients
Agent-Dienst in Cloud Run bereitstellen
Agent-Dienst in Agent Engine bereitstellen
So stellt der A2A-Client eine Verbindung zum A2A-Server her
Struktur von Anfragen und Antworten bei Nicht-Streaming-Verbindungen

Voraussetzungen

Chrome-Webbrowser
Ein Gmail-Konto
Ein Cloud-Projekt mit aktivierter Abrechnung

In diesem Codelab, das sich an Entwickler aller Erfahrungsstufen (auch Anfänger) richtet, wird Python in der Beispielanwendung verwendet. Python-Kenntnisse sind jedoch nicht erforderlich, um die vorgestellten Konzepte zu verstehen.

2. Hinweis

Aktives Projekt in der Cloud Console auswählen

In diesem Codelab wird davon ausgegangen, dass Sie bereits ein Google Cloud-Projekt mit aktivierter Abrechnung haben. Wenn Sie noch kein Konto haben, können Sie der Anleitung unten folgen, um eines zu erstellen.

Wählen Sie in der Google Cloud Console auf der Seite zur Projektauswahl ein Google Cloud-Projekt aus oder erstellen Sie eines.
Die Abrechnung für das Cloud-Projekt muss aktiviert sein. So prüfen Sie, ob die Abrechnung für ein Projekt aktiviert ist.

Cloud-Projekt im Cloud Shell-Terminal einrichten

Sie verwenden Cloud Shell, eine Befehlszeilenumgebung, die in Google Cloud ausgeführt wird und in der bq vorinstalliert ist. Klicken Sie oben in der Google Cloud Console auf „Cloud Shell aktivieren“. Wenn Sie zur Autorisierung aufgefordert werden, klicken Sie auf Autorisieren.

Wenn Sie mit Cloud Shell verbunden sind, können Sie mit dem folgenden Befehl prüfen, ob Sie bereits authentifiziert sind und das Projekt auf Ihre Projekt-ID festgelegt ist:

gcloud auth list

Führen Sie den folgenden Befehl in Cloud Shell aus, um zu bestätigen, dass der gcloud-Befehl Ihr Projekt kennt.

gcloud config list project

Wenn Ihr Projekt nicht festgelegt ist, verwenden Sie den folgenden Befehl, um es festzulegen:

gcloud config set project <YOUR_PROJECT_ID>

Alternativ können Sie die PROJECT_ID-ID auch in der Console aufrufen.

Klicken Sie darauf. Rechts sehen Sie dann alle Ihre Projekte und die Projekt-ID.

Aktivieren Sie die erforderlichen APIs mit dem unten gezeigten Befehl. Dies kann einige Minuten dauern.

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

Bei erfolgreicher Ausführung des Befehls sollte eine Meldung wie die unten gezeigte angezeigt werden:

Operation "operations/..." finished successfully.

Alternativ zum gcloud-Befehl können Sie in der Konsole nach den einzelnen Produkten suchen oder diesen Link verwenden.

Wenn eine API fehlt, können Sie sie jederzeit während der Implementierung aktivieren.

Informationen zu gcloud-Befehlen und deren Verwendung finden Sie in der Dokumentation.

Cloud Shell-Editor aufrufen und Arbeitsverzeichnis der Anwendung einrichten

Jetzt können wir unseren Code-Editor für einige Programmieraufgaben einrichten. Dazu verwenden wir den Cloud Shell-Editor.

Klicken Sie auf die Schaltfläche „Editor öffnen“, um den Cloud Shell-Editor zu öffnen. Hier können Sie Ihren Code schreiben.
Achten Sie darauf, dass das Cloud Code-Projekt in der unteren linken Ecke (Statusleiste) des Cloud Shell-Editors festgelegt ist, wie im Bild unten dargestellt. Es muss auf das aktive Google Cloud-Projekt festgelegt sein, für das die Abrechnung aktiviert ist. Autorisieren, wenn Sie dazu aufgefordert werden. Wenn Sie den vorherigen Befehl bereits ausgeführt haben, wird die Schaltfläche möglicherweise direkt zu Ihrem aktivierten Projekt weitergeleitet und nicht zur Schaltfläche „Anmelden“.

Als Nächstes klonen wir das Arbeitsverzeichnis der Vorlage für dieses Codelab von GitHub. Führen Sie dazu den folgenden Befehl aus. Das Arbeitsverzeichnis wird im Verzeichnis purchasing-concierge-a2a erstellt.

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

Klicken Sie dann oben im Cloud Shell-Editor auf Datei > Ordner öffnen, suchen Sie nach dem Verzeichnis username und dann nach dem Verzeichnis purchasing-concierge-a2a. Klicken Sie auf die Schaltfläche „OK“. Dadurch wird das ausgewählte Verzeichnis zum Hauptarbeitsverzeichnis. In diesem Beispiel ist der Nutzername alvinprayuda. Der Verzeichnispfad wird unten angezeigt.

Ihr Cloud Shell Editor sollte jetzt so aussehen:

Umgebung einrichten

Im nächsten Schritt bereiten Sie die Entwicklungsumgebung vor. Ihr aktuelles aktives Terminal sollte sich im Arbeitsverzeichnis purchasing-concierge-a2a befinden. In diesem Codelab verwenden wir Python 3.12 und den uv-Python-Projektmanager, um das Erstellen und Verwalten von Python-Versionen und virtuellen Umgebungen zu vereinfachen.

Wenn Sie das Terminal noch nicht geöffnet haben, klicken Sie auf Terminal > Neues Terminal oder verwenden Sie Strg + Umschalt + C. Dadurch wird ein Terminalfenster im unteren Bereich des Browsers geöffnet.

Initialisieren Sie nun die virtuelle Umgebung des Einkaufs-Concierge mit uv (bereits im Cloud-Terminal vorinstalliert). Führen Sie dazu diesen Befehl aus:

uv sync --frozen

Dadurch wird das Verzeichnis .venv erstellt und die Abhängigkeiten werden installiert. Ein kurzer Blick in die Datei pyproject.toml gibt Ihnen Informationen zu den Abhängigkeiten, die so angezeigt werden:

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

Um die virtuelle Umgebung zu testen, erstellen Sie die neue Datei main.py und kopieren Sie den folgenden Code.

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

if __name__ == "__main__":
   main()

Führen Sie dann den folgenden Befehl aus:

uv run main.py

Die Ausgabe sollte in etwa so aussehen:

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

Das zeigt, dass das Python-Projekt richtig eingerichtet wird.

Jetzt können wir mit dem nächsten Schritt fortfahren: Konfigurieren und Bereitstellen unseres Agenten für Remote-Verkäufer.

3. Remote Seller Agent – A2A-Server in Cloud Run bereitstellen

In diesem Schritt stellen wir die beiden Agenten für Remote-Verkäufer bereit, die durch das rote Kästchen gekennzeichnet sind. Der Burger-Agent basiert auf dem CrewAI-Agent-Framework und der Pizza-Agent auf dem Langgraph-Agent, beide auf dem Gemini Flash 2.0-Modell.

Remote Burger-Agent bereitstellen

Der Quellcode des Burger-Agents befindet sich im Verzeichnis remote_seller_agents/burger_agent. Die Agent-Initialisierung kann im Skript agent.py geprüft werden. Hier ist das Code-Snippet des initialisierten Agents:

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

...

Alle Dateien im Verzeichnis remote_seller_agents/burger_agent reichen bereits aus, um unseren Agent in Cloud Run bereitzustellen, sodass er als Dienst zugänglich ist. Darauf gehen wir später noch ein. Führen Sie den folgenden Befehl aus, um sie bereitzustellen:

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}

Wenn Sie aufgefordert werden, ein Container-Repository für die Bereitstellung aus der Quelle zu erstellen, antworten Sie mit Y. Dies ist nur der Fall, wenn Sie noch nie Quellcode in Cloud Run bereitgestellt haben. Nach erfolgreicher Bereitstellung wird ein Log wie dieses angezeigt.

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

Der Teil xxxx ist eine eindeutige Kennung, wenn wir den Dienst bereitstellen. Versuchen wir nun, über den Browser auf die bereitgestellten Burger-Agent-Dienste zuzugreifen, die über https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json bereitgestellt wurden. Dies ist die URL für den Zugriff auf die bereitgestellte A2A-Server-Agent-Karte.

Wenn die Bereitstellung erfolgreich war, wird beim Zugriff auf https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json in Ihrem Browser die folgende Antwort angezeigt :

Dies sind die Informationen auf der Burger-Agent-Karte, die für die Suche zugänglich sein sollten. Darauf gehen wir später noch ein. Beachten Sie, dass der Wert url hier weiterhin auf http://0.0.0.0:8080/ festgelegt ist. Dieser url-Wert sollte die Hauptinformation für den A2A-Client sein, um Nachrichten von außerhalb zu senden. Er ist jedoch nicht richtig konfiguriert. Für diese Demo müssen wir diesen Wert auf die URL unseres Burger-Agent-Dienstes aktualisieren, indem wir eine zusätzliche Umgebungsvariable HOST_OVERRIDE hinzufügen.

Aktualisieren des Burger Agent-URL-Werts auf der Agent-Karte über die Umgebungsvariable

So fügen Sie HOST_OVERRIDE dem Burger-Agent-Dienst hinzu:

Suchen Sie in der Suchleiste oben in der Cloud Console nach Cloud Run.

Klicken Sie auf den zuvor bereitgestellten Cloud Run-Dienst burger-agent.

Kopieren Sie die URL des Burger-Dienstes und klicken Sie dann auf Neue Überarbeitung bearbeiten und bereitstellen.

Klicken Sie dann auf den Bereich Variablen und Secrets.

Klicken Sie dann auf Variable hinzufügen und legen Sie für HOST_OVERRIDE den Wert auf die Dienst-URL fest ( die mit dem https://burger-agent-xxxxxxxxx.us-central1.run.app-Muster).

Klicken Sie abschließend auf die Schaltfläche Bereitstellen, um Ihren Dienst noch einmal bereitzustellen.

Wenn Sie jetzt im Browser https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json wieder auf die Agent-Karte für den Burger-Agent zugreifen, ist der Wert url bereits richtig konfiguriert.

Remote Pizza Agent bereitstellen

Der Quellcode des Pizza-Agents befindet sich im Verzeichnis remote_seller_agents/pizza_agent. Die Agent-Initialisierung kann im Skript agent.py geprüft werden. Hier ist das Code-Snippet des initialisierten Agents:

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

...

Ähnlich wie beim vorherigen Schritt zur Bereitstellung des Burger-Agents reichen alle Dateien im Verzeichnis remote_seller_agents/pizza_agent bereits aus, um unseren Agent in Cloud Run bereitzustellen, sodass er als Dienst zugänglich ist. Führen Sie den folgenden Befehl aus, um sie bereitzustellen:

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}

Nach erfolgreicher Bereitstellung wird ein Log wie dieses angezeigt.

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

Der Teil xxxx ist eine eindeutige Kennung, wenn wir den Dienst bereitstellen. Das ist auch beim Burger-Agenten der Fall. Wenn Sie versuchen, über den Browser auf die https://pizza-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json-Route der bereitgestellten Pizza-Agent-Dienste zuzugreifen, um die A2A-Server-Agent-Karte aufzurufen, ist der url-Wert des Pizza-Agenten auf der Agent-Karte noch nicht richtig konfiguriert. Außerdem müssen wir HOST_OVERRIDE in die Umgebungsvariable einfügen.

Aktualisieren des URL-Werts des Pizza-Agents auf der Agent-Karte über die Umgebungsvariable

So fügen Sie HOST_OVERRIDE dem Pizza-Agent-Dienst hinzu:

Suchen Sie in der Suchleiste oben in der Cloud Console nach Cloud Run.

Klicken Sie auf den zuvor bereitgestellten Cloud Run-Dienst pizza-agent.

Klicken Sie auf Neue Überarbeitung bearbeiten und bereitstellen.

Kopieren Sie die URL des Pizzadienstes und klicken Sie dann auf den Bereich Variablen und Secrets.

Klicken Sie dann auf Variable hinzufügen und legen Sie für HOST_OVERRIDE den Wert auf die Dienst-URL fest ( die mit dem https://pizza-agent-xxxxxxxxx.us-central1.run.app-Muster).

Klicken Sie abschließend auf die Schaltfläche Bereitstellen, um Ihren Dienst noch einmal bereitzustellen.

Wenn Sie jetzt im Browser https://pizza-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json wieder auf die Agent-Karte „pizza-agent“ zugreifen, ist der Wert url bereits richtig konfiguriert.

An diesem Punkt haben wir bereits sowohl den Burger- als auch den Pizza-Dienst erfolgreich in Cloud Run bereitgestellt. Sehen wir uns nun die Kernkomponenten des A2A-Servers an.

4. Kernkomponenten des A2A-Servers

Sehen wir uns nun das Kernkonzept und die Komponenten des A2A-Servers an.

Agent-Karte

Jeder A2A-Server muss eine Agent-Karte haben, auf die über die /.well-known/agent.json-Ressource zugegriffen werden kann. Dies soll die Ermittlungsphase im A2A-Client unterstützen, die vollständige Informationen und Kontexte zum Zugriff auf den Agenten und zu allen seinen Funktionen liefern soll. Das ist ähnlich wie bei gut dokumentierter API-Dokumentation mit Swagger oder Postman.

Das ist der Inhalt der Agent-Karte unseres bereitgestellten Burger-Agents.

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

Auf diesen Agentenkarten werden viele wichtige Komponenten wie Agentenfähigkeiten, Streamingfunktionen, unterstützte Modalitäten, Protokollversion und andere Dinge hervorgehoben.

Alle diese Informationen können verwendet werden, um einen geeigneten Kommunikationsmechanismus zu entwickeln, damit der A2A-Client ordnungsgemäß kommunizieren kann. Die unterstützte Modalität und der Authentifizierungsmechanismus sorgen dafür, dass die Kommunikation ordnungsgemäß hergestellt werden kann. Die skills-Informationen des Agents können in den Systemprompt des A2A-Clients eingebettet werden, um dem Agent des Clients Kontext zu den Funktionen und Fähigkeiten des Remote-Agents zu geben, die aufgerufen werden sollen. Weitere Informationen zu den Feldern dieser Agent-Karte

In unserem Code wird die Implementierung der Agent-Karte mit dem A2A Python SDK eingerichtet. Die Implementierung finden Sie im folgenden Snippet 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],
        )

...

Dort sind mehrere Felder zu sehen, z. B.:

AgentCapabilities :Deklaration zusätzlicher optionaler Funktionen, die vom Agent-Dienst unterstützt werden, z. B. Streaming- und/oder Push-Benachrichtigungsunterstützung
AgentSkill :Tools oder Funktionen, die vom Agent unterstützt werden
Input/OutputModes :Unterstützte Modalität des Ein-/Ausgabetyps
Url :Adresse für die Kommunikation mit dem Agent

In dieser Konfiguration wird eine dynamische Agent-Host-URL erstellt, sodass das Wechseln zwischen lokalen Tests und der Cloud-Bereitstellung einfacher ist. Aus diesem Grund müssen wir im vorherigen Schritt die Variable HOST_OVERRIDE hinzufügen.

Aufgabenwarteschlange und Agent Executor

Ein A2A-Server kann Anfragen von verschiedenen Agents oder Nutzern verarbeiten und jede Aufgabe perfekt isolieren. Die Kontexte sind auf dem Bild unten dargestellt.

Daher sollte jeder A2A-Server eingehende Aufgaben verfolgen und entsprechende Informationen dazu speichern können. Das A2A SDK bietet Module, um diese Herausforderung auf dem A2A-Server zu bewältigen. Zuerst können wir die Logik dafür instanziieren, wie wir die eingehende Anfrage verarbeiten möchten. Durch die Übernahme der abstrakten Klasse „AgentExecutor“ können wir steuern, wie wir die Ausführung und das Abbrechen von Aufgaben verwalten möchten. Diese Beispielimplementierung kann im remote_seller_agents/burger_agent/agent_executor.py-Modul eingesehen werden ( ähnlicher Pfad für den Fall des Pizzaverkäufers).

...

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

...

Im obigen Code implementieren wir ein einfaches Verarbeitungsschema, bei dem der Agent direkt aufgerufen wird, wenn eine Anfrage eingeht, und Ereignisse für abgeschlossene Aufgaben gesendet werden, nachdem der Aufruf beendet wurde. Die Methode zum Abbrechen wurde hier jedoch nicht implementiert, da es sich um einen kurz laufenden Vorgang handelt.

Nachdem wir den Executor erstellt haben, können wir direkt die integrierten DefaultRequestHandler, InMemoryTaskStore und A2AStarletteApplication verwenden,um den HTTP-Server zu starten. Diese Implementierung kann in remote_seller_agents/burger_agent/__main__.py eingesehen werden.

...

        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)

...

In diesem Modul wird die Implementierung der /.well-known/agent.json-Route für den Zugriff auf die Agent-Karte sowie des POST-Endpunkts zur Unterstützung des A2A-Protokolls beschrieben.

Zusammenfassung

Kurz gesagt: Bisher haben wir einen A2A-Server mit dem Python SDK bereitgestellt, der die beiden folgenden Funktionen unterstützt:

Agent-Karte auf der Route /.well-known/agent.json veröffentlichen
JSON-RPC-Anfrage mit In-Memory-Aufgabenwarteschlange verarbeiten

Der Einstiegspunkt beim Start dieser Funktionen kann im __main__.py-Skript ( auf der remote_seller_agents/burger_agent oder remote_seller_agents/pizza_agent) eingesehen werden.

5. Bereitstellung des Purchasing Concierge – A2A-Client für Agent Engine

In diesem Schritt stellen wir den Einkaufs-Concierge-Agent bereit. Mit diesem Agenten interagieren wir.

Der Quellcode unseres Purchasing Concierge-Agents befindet sich im Verzeichnis purchasing_concierge. Die Agent-Initialisierung kann im Skript purchasing_agent.py geprüft werden. Hier ist das Code-Snippet des initialisierten Agents.

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

...

Wir stellen diesen Agent in der Agent Engine bereit. Vertex AI Agent Engine ist eine Reihe von Diensten, mit denen Entwickler KI-Agents in der Produktion bereitstellen, verwalten und skalieren können. Die Infrastruktur zum Skalieren von KI-Agenten in der Produktion wird verwaltet, sodass wir uns auf die Entwicklung von Anwendungen konzentrieren können. Weitere Informationen Bisher mussten wir Dateien vorbereiten, die für die Bereitstellung unseres Agent-Dienstes erforderlich sind, z. B. das main-Serverskript und die Dockerfile. In diesem Fall können wir unseren Agent direkt über das Python-Skript bereitstellen, ohne einen eigenen Back-End-Dienst entwickeln zu müssen. Dazu verwenden wir eine Kombination aus ADK und Agent Engine. So stellen Sie die Funktion bereit :

Zuerst müssen wir den Staging-Speicher in Cloud Storage erstellen.

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

Jetzt müssen wir zuerst die Variable .env vorbereiten. Kopieren wir die Datei .env.example in die Datei .env.

cp .env.example .env

Öffnen Sie nun die Datei .env. Sie sehen den folgenden Inhalt:

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}

Dieser Agent kommuniziert mit dem Burger- und Pizza-Agenten. Daher müssen wir die entsprechenden Anmeldedaten für beide angeben. Wir müssen die PIZZA_SELLER_AGENT_URL und BURGER_SELLER_AGENT_URL mit der Cloud Run-URL aus den vorherigen Schritten aktualisieren.

Wenn Sie das vergessen haben, rufen Sie die Cloud Run-Konsole auf. Geben Sie in der Suchleiste oben in der Konsole „Cloud Run“ ein und klicken Sie mit der rechten Maustaste auf das Cloud Run-Symbol, um es auf einem neuen Tab zu öffnen.

Sie sollten die zuvor bereitgestellten Dienste für den Agenten des Remote-Verkäufers wie unten dargestellt sehen.

Wenn Sie die öffentliche URL dieser Dienste aufrufen möchten, klicken Sie auf einen der Dienste. Sie werden dann zur Seite „Dienstdetails“ weitergeleitet. Die URL wird oben direkt neben den Regionsinformationen angezeigt.

Die endgültige Umgebungsvariable sollte so aussehen:

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}

Jetzt können wir unseren Einkaufs-Concierge-Agent bereitstellen. In dieser Demo stellen wir die Inhalte mit dem Skript deploy_to_agent_engine.py bereit, das unten zu sehen ist.

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

So stellen Sie Ihren ADK-KI-Agenten in der Agent Engine bereit. Zuerst müssen wir ein AdkApp-Objekt aus unserem ADK root_agent erstellen. Anschließend können wir die Methode agent_engines.create ausführen, indem wir das adk_app-Objekt, die Anforderungen im Feld requirements und den Agent-Verzeichnispfad in extra_packages angeben und die erforderlichen Umgebungsvariablen bereitstellen.

Wir können es bereitstellen, indem wir das Skript ausführen:

uv run deploy_to_agent_engine.py

Nach erfolgreicher Bereitstellung wird ein Log wie dieses angezeigt. xxxx ist Ihre Projekt-ID und yyyy die Ressourcen-ID der Agent Engine.

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

Wenn wir es im Dashboard der Agent-Engine prüfen (suchen Sie in der Suchleiste nach „Agent-Engine“), wird unsere vorherige Bereitstellung angezeigt.

Bereitgestellten Agent in Agent Engine testen

Die Interaktion mit der Agent-Engine kann über den curl-Befehl und das SDK erfolgen. Führen Sie beispielsweise den folgenden Befehl aus, um mit dem bereitgestellten Agenten zu interagieren.

Sie können diese Anfrage senden, um zu prüfen, ob der Agent erfolgreich bereitgestellt wurde.

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

Bei Erfolg werden in der Konsole mehrere Antwort-Ereignisse gestreamt, wie hier zu sehen:

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

Im nächsten Schritt werden wir die Benutzeroberfläche verwenden. Zuerst möchten wir jedoch die Kernkomponenten und den typischen Ablauf von A2A-Clients erläutern.

6. Kernkomponenten des A2A-Clients

Das Bild oben zeigt den typischen Ablauf von A2A-Interaktionen:

Der Client versucht, in der bereitgestellten Remote-Agent-URL im Pfad /.well-known/agent.json eine veröffentlichte Agent-Karte zu finden.
Bei Bedarf wird dann eine Nachricht mit der Nachricht und den erforderlichen Metadatenparametern ( z. B. Sitzungs-ID, historischer Kontext usw.) an diesen Agenten gesendet. Der Server betrachtet diese Nachricht als zu erledigende Aufgabe.
Der A2A-Server verarbeitet die Anfrage. Wenn der Server Push-Benachrichtigungen unterstützt, kann er auch während der Aufgabenverarbeitung Benachrichtigungen veröffentlichen. Diese Funktion ist jedoch nicht Teil dieses Codelabs.
Nach Abschluss sendet der A2A-Server das Antwortartefakt zurück an den Client.

Einige der wichtigsten Objekte für die oben genannten Interaktionen sind die folgenden (weitere Informationen) :

Nachricht:Eine Kommunikationsrunde zwischen einem Client und einem Remote-Agent
Aufgabe: Die von A2A verwaltete Grundeinheit der Arbeit, die durch eine eindeutige ID identifiziert wird.
Artefakt:Eine Ausgabe (z. B. ein Dokument, ein Bild oder strukturierte Daten), die vom Agent als Ergebnis einer Aufgabe generiert wird und aus Teilen besteht.
Teil:Die kleinste Inhaltseinheit in einer Nachricht oder einem Artefakt. Ein Teil kann Text, ein Bild, ein Video, eine Datei usw. sein.

Karten-Erkennung

Wenn der A2A-Clientdienst gestartet wird, wird in der Regel versucht, die Informationen der Agent-Karte abzurufen und zu speichern, damit bei Bedarf leicht darauf zugegriffen werden kann. In diesem Codelab implementieren wir sie auf before_agent_callback. Die Implementierung finden Sie in purchasing_concierge/purchasing_agent.py (siehe Code-Snippet unten).

...

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)

...

Hier versuchen wir, über das integrierte A2A-Clientmodul A2ACardResolver auf alle verfügbaren Agentenkarten zuzugreifen. Dann erfassen wir die Verbindung, die zum Senden von Nachrichten an den Agenten erforderlich ist. Außerdem müssen wir alle verfügbaren Agenten und ihre Spezifikationen in den Prompt aufnehmen, damit unser Agent weiß, dass er mit diesen Agenten kommunizieren kann.

Tool „Prompt and Send Task“

Das ist der Prompt und das Tool, das wir unserem ADK-Agenten hier zur Verfügung stellen.

...

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

...

Im Prompt geben wir unserem Purchasing Concierge-Agenten alle verfügbaren Namen und Beschreibungen der Remote-Agents. Im Tool self.send_task stellen wir einen Mechanismus zum Abrufen des entsprechenden Clients bereit, um eine Verbindung zum Agenten herzustellen und die erforderlichen Metadaten über das SendMessageRequest-Objekt zu senden.

Die Kommunikationsprotokolle

Die Task-Definition ist eine Domain, die dem A2A-Server gehört. Aus Sicht des A2A-Clients wird die Nachricht jedoch als Message betrachtet, die an den Server gesendet wird. Es liegt am Server, eingehende Nachrichten vom Client als welche Aufgabe zu definieren und ob für die Ausführung der Aufgabe eine Interaktion des Clients erforderlich ist. Weitere Informationen zum Aufgabenlebenszyklus finden Sie in dieser Dokumentation. Das übergeordnete Konzept lässt sich so veranschaulichen:

Dieser Austausch von Nachricht -> Aufgabe wird mithilfe des Nutzlastformats auf Grundlage des JSON-RPC-Standards implementiert, wie im folgenden Beispiel des message/send-Protokolls gezeigt :

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

Es gibt verschiedene Methoden, z. B. zur Unterstützung verschiedener Arten der Kommunikation (z. B. synchron, Streaming, asynchron) oder zum Konfigurieren von Benachrichtigungen für den Aufgabenstatus. Ein A2A-Server kann flexibel konfiguriert werden, um diese Standards für die Aufgabendefinition zu verarbeiten. Details zu diesen Methoden finden Sie in diesem Dokument.

7. Integrationstests und Nutzlastprüfung

Sehen wir uns nun unseren Kauf-Concierge mit der Remote-Agent-Interaktion über eine Weboberfläche an.

Zuerst müssen wir die AGENT_ENGINE_RESOURCE_NAME in der Datei aktualisieren .env-Datei. Achten Sie darauf, dass Sie den richtigen Ressourcennamen für die Agent-Engine angeben. Ihre .env-Datei sollte so aussehen:

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

Führen Sie danach den folgenden Befehl aus, um eine Gradio-App bereitzustellen.

uv run purchasing_concierge_ui.py

Bei Erfolg wird die folgende Ausgabe angezeigt:

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

Strg+Klick auf die URL „http://0.0.0.0:8080“ im Terminal oder auf die Schaltfläche „Webvorschau“, um die Weboberfläche zu öffnen

So könnte eine Unterhaltung aussehen :

Zeig mir die Speisekarte für Burger und Pizza
Ich möchte eine BBQ-Chicken-Pizza und einen Spicy Cajun Burger bestellen.

Setzen Sie die Unterhaltung fort, bis Sie die Bestellung abgeschlossen haben. Sehen Sie sich an, wie die Interaktion verläuft und was der Tool-Aufruf und die Antwort sind. Das folgende Bild zeigt ein Beispiel für das Interaktionsergebnis.

Die Kommunikation mit zwei verschiedenen Agenten führt zu zwei unterschiedlichen Verhaltensweisen. A2A kann dies gut verarbeiten. Der Pizzaverkäufer-Agent akzeptiert unsere Anfrage direkt, während der Burger-Agent unsere Bestätigung benötigt, bevor er mit unserer Anfrage fortfährt. Nachdem wir die Bestätigung gesendet haben, kann er sich darauf verlassen.

Nachdem wir nun die grundlegenden Konzepte von A2A kennengelernt haben, sehen wir uns an, wie sie als Client- und Serverarchitektur implementiert werden.

8. Herausforderung

Können Sie jetzt die erforderliche Datei vorbereiten und die Gradio-App selbst in Cloud Run bereitstellen? Zeit, die Herausforderung anzunehmen!

9. Bereinigen

So vermeiden Sie, dass Ihrem Google Cloud-Konto die in diesem Codelab verwendeten Ressourcen in Rechnung gestellt werden:

Wechseln Sie in der Google Cloud Console zur Seite Ressourcen verwalten.
Wählen Sie in der Projektliste das Projekt aus, das Sie löschen möchten, und klicken Sie auf Löschen.
Geben Sie im Dialogfeld die Projekt-ID ein und klicken Sie auf Beenden, um das Projekt zu löschen.
Alternativ können Sie in der Console zu Cloud Run wechseln, den gerade bereitgestellten Dienst auswählen und löschen.