Erste Schritte mit dem Agent-zu-Agent-Protokoll (A2A): Interaktionen zwischen Kundenservicemitarbeitern und Remote-Verkäufern mit Gemini in Cloud Run

Informationen zu diesem Codelab

0 Minuten

Zuletzt aktualisiert: 8. Mai 2025

Verfasst von Alvin Prayuda Juniarta Dwiyantoro

1. Einführung

Das Agent-zu-Agent-Protokoll (A2A) wurde entwickelt, um die Kommunikation zwischen KI-Agenten zu standardisieren, insbesondere für solche, die in externen Systemen bereitgestellt werden. Bisher wurden solche Protokolle für Tools namens Model Context Protocol (MCP) eingeführt. Dies ist ein neuer Standard, um LLMs mit Daten und Ressourcen zu verbinden. A2A soll MCP ergänzen, wobei A2A sich auf ein anderes Problem konzentriert. Während MCP darauf abzielt, die Komplexität zu verringern, um Kundenservicemitarbeiter mit Tools und Daten zu verbinden, liegt der Schwerpunkt von A2A darauf, Kundenservicemitarbeitern die Zusammenarbeit in ihren natürlichen Modalitäten zu ermöglichen. So können Kundenservicemitarbeiter als Kundenservicemitarbeiter (oder als Nutzer) und nicht als Tools kommunizieren. So ist beispielsweise eine Unterhaltung möglich, wenn Sie etwas bestellen möchten.

A2A soll MCP ergänzen. In der offiziellen Dokumentation wird empfohlen, dass Anwendungen A2A-Kundenservicemitarbeiter als MCP-Ressourcen modellieren, die durch die AgentCard dargestellt werden. (Wir werden später darauf zurückkommen.) Die Frameworks können dann A2A verwenden, um mit ihren Nutzern, den Remote-Kundenservicemitarbeitern und anderen Kundenservicemitarbeitern zu kommunizieren.

In dieser Demo beginnen wir mit der Implementierung von A2A von Grund auf. Anhand dieser Beispiel-Repositories sehen wir uns einen Anwendungsfall an, bei dem wir einen persönlichen Einkaufsassistenten haben, der uns bei der Kommunikation mit den Mitarbeitern von Burger- und Pizzabuden helfen kann, um unsere Bestellung zu bearbeiten.

A2A nutzt das Client-Server-Prinzip. Hier ist der typische A2A-Ablauf, den wir in dieser Demo erwarten.

Der A2A-Client führt zuerst eine Suche auf allen zugänglichen A2A-Server-Agentkarten durch und verwendet die Informationen, um einen Verbindungsclient zu erstellen.
Bei Bedarf sendet der A2A-Client Aufgaben an den A2A-Server. Wenn die Empfänger-URL der Push-Benachrichtigung auf dem A2A-Client konfiguriert ist, kann der A2A-Server auch den Status des Aufgabenfortschritts an den Empfängerendpunkt veröffentlichen.
Nach Abschluss der Aufgabe sendet der A2A-Server das Antwort-Artefakt an den A2A-Client.

Im Codelab gehen Sie so vor:

Google Cloud-Projekt vorbereiten und alle erforderlichen APIs aktivieren
Arbeitsbereich für Ihre Programmierumgebung einrichten
Env-Variablen für Burger- und Pizza-Kundenservice vorbereiten
Burger- und Pizza-Agent in Cloud Run bereitstellen
Details zur Einrichtung des A2A-Servers prüfen
Umgebungsvariablen für den Shopping-Concierge vorbereiten
Shopping Concierge in Cloud Run bereitstellen
Details zur Einrichtung des A2A-Clients und zur Datenmodellierung prüfen
Payload 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-Agenten ( gestützt auf das CrewAI-Agent-Framework) und den Pizza-Agenten ( gestützt auf das Langgraph-Agent-Framework). Der Nutzer interagiert direkt nur mit dem Kundenservicemitarbeiter für den Kauf, der mit dem Agent Development Kit (ADK)-Framework ausgeführt wird, das als A2A-Client fungiert.

Jeder dieser Kundenservicemitarbeiter hat seine eigene Umgebung und Bereitstellung.

Voraussetzungen

Gute Python-Kenntnisse
Grundlegende Kenntnisse der Full-Stack-Architektur mit HTTP-Dienst

Lerninhalte

Kernstruktur des A2A-Servers
Kernstruktur des A2A-Clients
Dienst in Cloud Run bereitstellen
So stellt der A2A-Client eine Verbindung zum A2A-Server her
Anfrage- und Antwortstruktur bei einer nicht streamenden Verbindung

Voraussetzungen

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

Dieses Codelab richtet sich an Entwickler aller Stufen (einschließlich Anfänger) und verwendet Python in der Beispielanwendung. 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 keine haben, können Sie der Anleitung unten folgen, um loszulegen.

Wählen Sie in der Google Cloud Console auf der Seite der 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 bq bereits vorinstalliert hat. Klicken Sie oben in der Google Cloud Console auf „Cloud Shell aktivieren“. Wenn Sie zur Autorisierung aufgefordert werden, klicken Sie auf Autorisieren.

Nachdem Sie eine Verbindung zu Cloud Shell hergestellt haben, prüfen Sie mit dem folgenden Befehl, ob Sie bereits authentifiziert sind und das Projekt auf Ihre Projekt-ID festgelegt ist:

gcloud auth list

Führen Sie in Cloud Shell den folgenden Befehl aus, um zu prüfen, ob 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 Ihr gesamtes Projekt und die Projekt-ID.

Aktivieren Sie die erforderlichen APIs mit dem folgenden Befehl. Das kann einige Minuten dauern.

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

Wenn der Befehl erfolgreich ausgeführt wurde, sollte eine Meldung ähnlich der folgenden angezeigt werden:

Operation "operations/..." finished successfully.

Alternativ können Sie in der Console nach den einzelnen Produkten suchen oder diesen Link verwenden.

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

Weitere 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 die Programmierung einrichten. Dazu verwenden wir den Cloud Shell-Editor.

Klicken Sie auf die Schaltfläche „Editor öffnen“. Dadurch wird der Cloud Shell-Editor geöffnet, in dem Sie Ihren Code schreiben können.
Achten Sie darauf, dass das Cloud Code-Projekt links unten (Statusleiste) im Cloud Shell-Editor festgelegt ist, wie im Bild unten hervorgehoben. Es muss sich dabei um das aktive Google Cloud-Projekt handeln, in dem Sie die Abrechnung aktiviert haben. Klicken Sie auf Autorisieren, wenn Sie dazu aufgefordert werden. Wenn Sie den vorherigen Befehl bereits ausgeführt haben, verweist die Schaltfläche möglicherweise direkt auf Ihr aktiviertes Projekt und nicht auf die Anmeldeschaltfläche.

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,gehen Sie zu purchasing-concierge-a2a und klicken Sie auf die Schaltfläche „Ok“. Dadurch wird das ausgewählte Verzeichnis zum Haupt-Arbeitsverzeichnis. In diesem Beispiel lautet der Nutzername alvinprayuda. Der Verzeichnispfad wird unten angezeigt.

Der Cloud Shell-Editor sollte jetzt so aussehen:

Umgebung einrichten

Im nächsten Schritt wird die Entwicklungsumgebung vorbereitet. Ihr aktuelles aktives Terminal sollte sich im Arbeitsverzeichnis purchasing-concierge-a2a befinden. In diesem Codelab verwenden wir Python 3.12 und den uv Python Project Manager, um das Erstellen und Verwalten der Python-Version und der virtuellen Umgebung zu vereinfachen.

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

Laden Sie uv herunter und installieren Sie Python 3.12 mit dem folgenden Befehl:

curl -LsSf https://astral.sh/uv/0.7.2/install.sh | sh && \
source $HOME/.local/bin/env && \
uv python install 3.12

Jetzt initialisieren wir die virtuelle Umgebung des Kundenservicemitarbeiters mit uv. 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 pyproject.toml liefert Informationen zu den Abhängigkeiten, die so dargestellt werden:

dependencies = [
    "google-adk>=0.3.0",
    "gradio>=5.28.0",
    "httpx>=0.28.1",
    "jwcrypto>=1.5.6",
    "pydantic>=2.10.6",
    "pyjwt>=2.10.1",
    "sse-starlette>=2.3.3",
    "starlette>=0.46.2",
    "typing-extensions>=4.13.2",
    "uvicorn>=0.34.0",
]

Erstellen Sie zum Testen der virtuellen Umgebung 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 Python-Projekt wird also richtig eingerichtet.

Jetzt können wir mit dem nächsten Schritt fortfahren: dem Konfigurieren und Bereitstellen des Remote-Kundenservicemitarbeiters.

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

In diesem Schritt stellen wir diese beiden Remote-Kundenservicemitarbeiter bereit, die durch das rote Feld gekennzeichnet sind. Der Burger-Agent wird vom CrewAI-Agent-Framework und der Pizza-Agent vom Langgraph-Agenten unterstützt. Beide basieren 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 Script 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.0-flash",  # 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,
    output_pydantic=ResponseFormat,
    agent=burger_agent,
    expected_output=(
        "A JSON object with 'status' and 'message' fields."
        "Set response status to input_required if asking for user order confirmation."
        "Set response status to error if there is an error while processing the request."
        "Set response status to completed if the request is complete."
    ),
)

crew = Crew(
    tasks=[agent_task],
    agents=[burger_agent],
    verbose=False,
    process=Process.sequential,
)

inputs = {"user_prompt": query, "session_id": sessionId}
response = crew.kickoff(inputs)

...

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

cp remote_seller_agents/burger_agent/.env.example remote_seller_agents/burger_agent/.env

Öffnen Sie jetzt die Datei remote_seller_agents/burger_agent/.env. Sie sehen dann den folgenden Inhalt:

AUTH_USERNAME=burgeruser123
AUTH_PASSWORD=burgerpass123
GCLOUD_LOCATION=us-central1
GCLOUD_PROJECT_ID={your-project-id}

Der A2A-Server des Burger-Agenten wird mit der Basis-HTTP-Authentifizierung ( mit Base64-codiertem Nutzernamen und Passwort) ausgeführt. Für diese Demo konfigurieren wir den zulässigen Nutzernamen und das zulässige Passwort hier in der Datei .env. Aktualisieren Sie nun die Variable GCLOUD_PROJECT_ID auf Ihre aktuelle aktive Projekt-ID.

Denken Sie daran, Ihre Änderungen zu speichern. Als Nächstes können wir die Dienste direkt bereitstellen. Den Codeinhalt prüfen wir später. Führen Sie den folgenden Befehl aus, um die Bereitstellung vorzunehmen:

gcloud run deploy burger-agent \
           --source remote_seller_agents/burger_agent \
           --port=8080 \
           --allow-unauthenticated \
           --min 1 \
           --region us-central1

Wenn Sie aufgefordert werden, ein Container-Repository für die Bereitstellung aus der Quelle zu erstellen, antworten Sie mit Ja. Nach erfolgreicher Bereitstellung wird ein Protokoll 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 den Pfad /.well-known/agent.json zu diesen bereitgestellten Burger-Kundenservicediensten aufzurufen. Die Ausgabe sollte in etwa so aussehen:

Dies sind die Informationen auf der Infokarte des Kundenservicemitarbeiters, die für Discovery-Zwecke zugänglich sein sollten. Darauf gehen wir später noch genauer ein. Merken Sie sich fürs Erste nur die URL unseres Burger-Agent-Dienstes, die wir später verwenden werden.

Remote Pizza Agent bereitstellen

Der Quellcode für Pizzalieferanten befindet sich ebenfalls im Verzeichnis remote_seller_agents/pizza_agent. Die Agent-Initialisierung kann im Script 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.0-flash",
    location=os.getenv("GCLOUD_LOCATION"),
    project=os.getenv("GCLOUD_PROJECT_ID"),
)
self.tools = [create_pizza_order]
self.graph = create_react_agent(
    self.model,
    tools=self.tools,
    checkpointer=memory,
    prompt=self.SYSTEM_INSTRUCTION,
    response_format=ResponseFormat,
)

...

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

cp remote_seller_agents/pizza_agent/.env.example remote_seller_agents/pizza_agent/.env

Öffnen Sie nun die Datei remote_seller_agents/pizza_agent/.env. Sie sehen dann den folgenden Inhalt:

API_KEY=pizza123
GCLOUD_LOCATION=us-central1
GCLOUD_PROJECT_ID={your-project-id}

Der A2A-Server des Pizza-Agenten wird mit der Bearer-HTTP-Authentifizierung (mit API-Schlüssel) ausgeführt. Für diese Demo konfigurieren wir den zulässigen API-Schlüssel hier in der Datei .env. Aktualisieren Sie nun die Variable GCLOUD_PROJECT_ID auf Ihre aktuelle aktive Projekt-ID.

gcloud run deploy pizza-agent \
           --source remote_seller_agents/pizza_agent \
           --port=8080 \
           --allow-unauthenticated \
           --min 1 \
           --region us-central1

Nach erfolgreicher Bereitstellung wird ein Protokoll 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.

Versuchen wir nun, den /.well-known/agent.json-Pfad dieser bereitgestellten Pizza-Kundenservicedienste über den Browser aufzurufen. Die Ausgabe sollte in etwa so aussehen:

Dies sind die Informationen auf der Pizzalieferantenkarte, die für Suchzwecke zugänglich sein sollten. Darauf gehen wir später noch genauer ein. Notieren Sie sich fürs Erste nur die URL unseres Pizza-Agenten-Dienstes.

Wir haben bereits sowohl Burger- als auch Pizzadienste in Cloud Run bereitgestellt. Sehen wir uns nun die Hauptkomponenten des A2A-Servers an.

4. Hauptkomponenten des A2A-Servers

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

Kundenservicemitarbeiterkarte

Jeder A2A-Server muss eine Kundenservicemitarbeiterkarte haben, auf die über die /.well-known/agent.json-Ressource zugegriffen werden kann. Dies soll die Discovery-Phase auf dem A2A-Client unterstützen, die vollständige Informationen und Kontexte zum Zugriff auf den Bot und alle seine Funktionen enthalten sollte. Ähnlich verhält es sich mit einer gut dokumentierten API-Dokumentation mit Swagger oder Postman.

Das ist der Inhalt der Karte des eingesetzten Burger-Kundenservicemitarbeiters.

{
  "name": "burger_seller_agent",
  "description": "Helps with creating burger orders",
  "url": "http://0.0.0.0:8080/",
  "version": "1.0.0",
  "capabilities": {
    "streaming": false,
    "pushNotifications": true,
    "stateTransitionHistory": false
  },
  "authentication": {
    "schemes": [
      "Basic"
    ]
  },
  "defaultInputModes": [
    "text",
    "text/plain"
  ],
  "defaultOutputModes": [
    "text",
    "text/plain"
  ],
  "skills": [
    {
      "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"
      ]
    }
  ]
}

Auf diesen Kundenservicemitarbeiterkarten werden viele wichtige Komponenten hervorgehoben, z. B. die Fähigkeiten von Kundenservicemitarbeitern, Streamingfunktionen, unterstützte Modalitäten und Authentifizierung.

Anhand dieser Informationen kann ein geeigneter Kommunikationsmechanismus entwickelt werden, damit der A2A-Client ordnungsgemäß kommunizieren kann. Die unterstützte Modalität und der Authentifizierungsmechanismus sorgen dafür, dass die Kommunikation richtig hergestellt werden kann. Die Informationen zum Kundenservicemitarbeiter skills können in den A2A-Clientsystemprompt eingebettet werden, um dem Kundenservicemitarbeiter des Kunden Kontext zu den Fähigkeiten und Kompetenzen des Remote-Kundenservicemitarbeiters zu geben, die aufgerufen werden sollen. Ausführlichere Felder für diese Kundenservicemitarbeiterkarte findest du in dieser Dokumentation.

In unserem Code wird die Implementierung der Kundenservicemitarbeiterkarte mit Pydantic in a2a_types.py ( auf remote_seller_agents/burger_agent oder remote_seller_agents/pizza_agent) festgelegt.

...

class AgentProvider(BaseModel):
    organization: str
    url: str | None = None


class AgentCapabilities(BaseModel):
    streaming: bool = False
    pushNotifications: bool = False
    stateTransitionHistory: bool = False


class AgentAuthentication(BaseModel):
    schemes: List[str]
    credentials: str | None = None


class AgentSkill(BaseModel):
    id: str
    name: str
    description: str | None = None
    tags: List[str] | None = None
    examples: List[str] | None = None
    inputModes: List[str] | None = None
    outputModes: List[str] | None = None


class AgentCard(BaseModel):
    name: str
    description: str | None = None
    url: str
    provider: AgentProvider | None = None
    version: str
    documentationUrl: str | None = None
    capabilities: AgentCapabilities
    authentication: AgentAuthentication | None = None
    defaultInputModes: List[str] = ["text"]
    defaultOutputModes: List[str] = ["text"]
    skills: List[AgentSkill]

...

Die Objektkonstruktion ist im unten dargestellten remote_seller_agents/burger_agent/__main__.py zu sehen.

...

def main(host, port):
    """Starts the Burger Seller Agent server."""
    try:
        capabilities = AgentCapabilities(pushNotifications=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_card = AgentCard(
            name="burger_seller_agent",
            description="Helps with creating burger orders",
            # The URL provided here is for the sake of demo,
            # in production you should use a proper domain name
            url=f"http://{host}:{port}/",
            version="1.0.0",
            authentication=AgentAuthentication(schemes=["Basic"]),
            defaultInputModes=BurgerSellerAgent.SUPPORTED_CONTENT_TYPES,
            defaultOutputModes=BurgerSellerAgent.SUPPORTED_CONTENT_TYPES,
            capabilities=capabilities,
            skills=[skill],
        )

        notification_sender_auth = PushNotificationSenderAuth()
        notification_sender_auth.generate_jwk()
        server = A2AServer(
            agent_card=agent_card,
            task_manager=AgentTaskManager(
                agent=BurgerSellerAgent(),
                notification_sender_auth=notification_sender_auth,
            ),
            host=host,
            port=port,
            auth_username=os.environ.get("AUTH_USERNAME"),
            auth_password=os.environ.get("AUTH_PASSWORD"),
        )

...

Aufgabendefinition und Task-Manager

Eine der Hauptkomponenten in A2A ist die Definition von Aufgaben. Das Nutzlastformat wird an den JSON-RPC-Standard angepasst. In dieser Demo wird dies in a2a_types.py ( auch in remote_seller_agents/burger_agent oder remote_seller_agents/pizza_agent) mit Pydantic implementiert.

...

## RPC Messages


class JSONRPCMessage(BaseModel):
    jsonrpc: Literal["2.0"] = "2.0"
    id: int | str | None = Field(default_factory=lambda: uuid4().hex)


class JSONRPCRequest(JSONRPCMessage):
    method: str
    params: dict[str, Any] | None = None

...

class SendTaskRequest(JSONRPCRequest):
    method: Literal["tasks/send"] = "tasks/send"
    params: TaskSendParams

class SendTaskStreamingRequest(JSONRPCRequest):
    method: Literal["tasks/sendSubscribe"] = "tasks/sendSubscribe"
    params: TaskSendParams

class GetTaskRequest(JSONRPCRequest):
    method: Literal["tasks/get"] = "tasks/get"
    params: TaskQueryParams

class CancelTaskRequest(JSONRPCRequest):
    method: Literal["tasks/cancel",] = "tasks/cancel"
    params: TaskIdParams

class SetTaskPushNotificationRequest(JSONRPCRequest):
    method: Literal["tasks/pushNotification/set",] = "tasks/pushNotification/set"
    params: TaskPushNotificationConfig

class GetTaskPushNotificationRequest(JSONRPCRequest):
    method: Literal["tasks/pushNotification/get",] = "tasks/pushNotification/get"
    params: TaskIdParams

class TaskResubscriptionRequest(JSONRPCRequest):
    method: Literal["tasks/resubscribe",] = "tasks/resubscribe"
    params: TaskIdParams

...

Es gibt verschiedene Aufgabenmethoden, die verschiedene Arten der Kommunikation unterstützen (z.B. Synchronisierung, Streaming, asynchron) und Benachrichtigungen für den Aufgabenstatus konfigurieren. A2A-Server können flexibel für die Verarbeitung dieser Standards für Aufgabendefinitionen konfiguriert werden.

Ein A2A-Server kann Anfragen von verschiedenen Kundenservicemitarbeitern oder Nutzern verarbeiten und jede Aufgabe perfekt isolieren. Das Bild unten veranschaulicht die Kontexte dieser Begriffe.

Daher sollte jeder A2A-Server eingehende Aufgaben verfolgen und die entsprechenden Informationen dazu speichern können. Normalerweise haben eingehende Anfragen eine Task-ID und eine Sitzungs-ID. In unserem Code befindet sich die Implementierung dieses Task-Managers auf der remote_seller_agents/burger_agent/task_manager.py (der Pizza-Kundenservicemitarbeiter verwendet ebenfalls einen ähnlichen Task-Manager).

...

class AgentTaskManager(InMemoryTaskManager):
    def __init__(
        self,
        agent: BurgerSellerAgent,
        notification_sender_auth: PushNotificationSenderAuth,
    ):
        super().__init__()
        self.agent = agent
        self.notification_sender_auth = notification_sender_auth

    ...

    async def on_send_task(self, request: SendTaskRequest) -> SendTaskResponse:
        """Handles the 'send task' request."""
        validation_error = self._validate_request(request)
        if validation_error:
            return SendTaskResponse(id=request.id, error=validation_error.error)

        await self.upsert_task(request.params)

        if request.params.pushNotification:
            if not await self.set_push_notification_info(
                request.params.id, request.params.pushNotification
            ):
                return SendTaskResponse(
                    id=request.id,
                    error=InvalidParamsError(
                        message="Push notification URL is invalid"
                    ),
                )

        task = await self.update_store(
            request.params.id, TaskStatus(state=TaskState.WORKING), None
        )
        await self.send_task_notification(task)

        task_send_params: TaskSendParams = request.params
        query = self._get_user_query(task_send_params)
        try:
            agent_response = self.agent.invoke(query, task_send_params.sessionId)
        except Exception as e:
            logger.error(f"Error invoking agent: {e}")
            raise ValueError(f"Error invoking agent: {e}")
        return await self._process_agent_response(request, agent_response)

    ...

    async def _process_agent_response(
        self, request: SendTaskRequest, agent_response: dict
    ) -> SendTaskResponse:
        """Processes the agent's response and updates the task store."""
        task_send_params: TaskSendParams = request.params
        task_id = task_send_params.id
        history_length = task_send_params.historyLength
        task_status = None

        parts = [{"type": "text", "text": agent_response["content"]}]
        artifact = None
        if agent_response["require_user_input"]:
            task_status = TaskStatus(
                state=TaskState.INPUT_REQUIRED,
                message=Message(role="agent", parts=parts),
            )
        else:
            task_status = TaskStatus(state=TaskState.COMPLETED)
            artifact = Artifact(parts=parts)
        task = await self.update_store(
            task_id, task_status, None if artifact is None else [artifact]
        )
        task_result = self.append_task_history(task, history_length)
        await self.send_task_notification(task)
        return SendTaskResponse(id=request.id, result=task_result)

   ...

Aus dem obigen Code geht hervor, dass bei der Verarbeitung einer eingehenden Aufgabe ( die Methode on_send_task wird ausgeführt, wenn die Methode der eingehenden Anfrage tasks/send ist) mehrere Vorgänge zum Aktualisieren des Aufgabenspeichers ( self.update_store-Methodenaufruf) und zum Senden einer Benachrichtigung ( self.send_task_notification-Methodenaufruf) ausgeführt werden. Dies ist eines der Beispiele dafür, wie der A2A-Server die Aktualisierung des Aufgabenstatus und die Benachrichtigung während der synchronen Sendeaufgabenanfrage verwaltet.

Zusammenfassung

Kurz gesagt: Unser bereitgestellter A2A-Server unterstützt derzeit die folgenden zwei Funktionen:

Die Kundenservicemitarbeiterkarte auf der Route /.well-known/agent.json veröffentlichen
JSON-RPC-Anfrage mit der Methode tasks/send verarbeiten

Der Einstiegspunkt zum Starten dieser Funktionen kann im Script main.py ( auf dem remote_seller_agents/burger_agent oder remote_seller_agents/pizza_agent) geprüft werden. Wir sehen, dass wir zuerst die Agent Card konfigurieren und dann den Server im folgenden Code-Snippet starten müssen.

...

capabilities = AgentCapabilities(pushNotifications=True)
skill = AgentSkill(
    id="create_pizza_order",
    name="Pizza Order Creation Tool",
    description="Helps with creating pizza orders",
    tags=["pizza order creation"],
    examples=["I want to order 2 pepperoni pizzas"],
)
agent_card = AgentCard(
    name="pizza_seller_agent",
    description="Helps with creating pizza orders",
    # The URL provided here is for the sake of demo,
    # in production you should use a proper domain name
    url=f"http://{host}:{port}/",
    version="1.0.0",
    authentication=AgentAuthentication(schemes=["Bearer"]),
    defaultInputModes=PizzaSellerAgent.SUPPORTED_CONTENT_TYPES,
    defaultOutputModes=PizzaSellerAgent.SUPPORTED_CONTENT_TYPES,
    capabilities=capabilities,
    skills=[skill],
)

...

server = A2AServer(
    agent_card=agent_card,
    task_manager=AgentTaskManager(
        agent=PizzaSellerAgent(),
        notification_sender_auth=notification_sender_auth,
    ),
    host=host,
    port=port,
    api_key=os.environ.get("API_KEY"),
)

...

logger.info(f"Starting server on {host}:{port}")
server.start()

...

5. Purchasing Concierge – A2A-Client in Cloud Run bereitstellen

In diesem Schritt stellen wir den Concierge-Kundenservicemitarbeiter für den Kauf bereit. Mit diesem Kundenservicemitarbeiter werden wir interagieren.

Der Quellcode unseres Shopping-Concierge-Agenten befindet sich im Verzeichnis purchasing_concierge. Die Agent-Initialisierung kann im Script 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.0-flash-001",
        name="purchasing_agent",
        instruction=self.root_instruction,
        before_model_callback=self.before_model_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.list_remote_agents,
            self.send_task,
        ],
    )

...

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

cp purchasing_concierge/.env.example purchasing_concierge/.env

Öffnen Sie jetzt die Datei purchasing_concierge/.env. Sie sehen dann den folgenden Inhalt:

PIZZA_SELLER_AGENT_AUTH=pizza123
PIZZA_SELLER_AGENT_URL=http://localhost:10000
BURGER_SELLER_AGENT_AUTH=burgeruser123:burgerpass123
BURGER_SELLER_AGENT_URL=http://localhost:10001
GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT={your-project-id}
GOOGLE_CLOUD_LOCATION=us-central1

Dieser Kundenservicemitarbeiter kommuniziert mit dem Kundenservicemitarbeiter für Burger und Pizza. Daher müssen wir die richtigen Anmeldedaten für beide angeben. Aktualisieren Sie nun die Variable GCLOUD_PROJECT_ID auf Ihre derzeit aktive Projekt-ID.

Jetzt müssen wir auch die PIZZA_SELLER_AGENT_URL und BURGER_SELLER_AGENT_URL mit der Cloud Run-URL aus den vorherigen Schritten aktualisieren. Falls Sie das vergessen, rufen Sie einfach die Cloud Run-Konsole auf. Geben Sie „Cloud Run“ in die Suchleiste oben in der Konsole ein und klicken Sie mit der rechten Maustaste auf das Cloud Run-Symbol, um es in einem neuen Tab zu öffnen.

Sie sollten unsere zuvor bereitgestellten Remote-Kundenservicemitarbeiter-Dienste wie unten dargestellt sehen.

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

Kopieren Sie den Wert dieser URL und fügen Sie ihn in die Felder PIZZA_SELLER_AGENT_URL und BURGER_SELLER_AGENT_URL ein.

Die endgültige Umgebungsvariable sollte in etwa so aussehen:

PIZZA_SELLER_AGENT_AUTH=pizza123
PIZZA_SELLER_AGENT_URL=https://pizza-agent-xxxxx.us-central1.run.app
BURGER_SELLER_AGENT_AUTH=burgeruser123:burgerpass123
BURGER_SELLER_AGENT_URL=https://burger-agent-xxxxx.us-central1.run.app
GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT={your-project-id}
GOOGLE_CLOUD_LOCATION=us-central1

Jetzt können wir unseren Concierge-Kundenservice für Käufe bereitstellen. Führen Sie den folgenden Befehl aus, um diesen Agenten bereitzustellen:

gcloud run deploy purchasing-concierge \
           --source . \
           --port=8080 \
           --allow-unauthenticated \
           --min 1 \
           --region us-central1 \
           --memory 1024Mi

Nach erfolgreicher Bereitstellung wird ein Protokoll wie dieses angezeigt.

Service [purchasing-concierge] revision [purchasing-concierge-xxxxx-xxx] has been deployed and is serving 100 percent of traffic.
Service URL: https://purchasing-concierge-xxxxxx.us-central1.run.app

Der Teil xxxx ist eine eindeutige Kennung, wenn wir den Dienst bereitstellen.

Jetzt können wir versuchen, über die Benutzeroberfläche mit dem Concierge-Kundenservicemitarbeiter zu interagieren. Wenn Sie auf die Dienst-URL zugreifen, sollte die Weboberfläche von Gradio wie unten dargestellt angezeigt werden.

Sehen wir uns nun die Hauptkomponenten und den typischen Ablauf von A2A-Clients an.

6. Hauptkomponenten des A2A-Clients

Die Abbildung oben zeigt den typischen Ablauf von A2A-Interaktionen:

Der Client versucht, eine veröffentlichte Kundenservicemitarbeiterkarte in der angegebenen Remote-Kundenservicemitarbeiter-URL in der Route zu finden. /.well-known/agent.json
Falls erforderlich, wird an diesen Kundenservicemitarbeiter eine Aufgabe mit der Nachricht und den erforderlichen Metadatenparametern ( z. B. Sitzungs-ID, bisheriger Kontext) gesendet.
Der A2A-Server authentifiziert und verarbeitet die Anfrage. Wenn der Server Push-Benachrichtigungen unterstützt, versucht er auch, während der Aufgabenverarbeitung einige Benachrichtigungen zu veröffentlichen.
Danach sendet der A2A-Server das Antwort-Artefakt an den Client zurück.

Die Hauptobjekte für die oben genannten Interaktionen sind die folgenden Elemente (Details finden Sie hier) :

Aufgabe: zustandsabhängige Entität, mit der Kunden und Remote-Kundenservicemitarbeiter ein bestimmtes Ergebnis erzielen und Ergebnisse erzielen können
Artefakt: Endergebnis einer Aufgabe
Nachricht:Alle Inhalte, die kein Artefakt sind. Zum Beispiel: Gedanken des Kundenservicemitarbeiters, Nutzerkontext, Anweisungen, Fehler, Status oder Metadaten
Teil: Ein vollständiger Inhalt, der zwischen einem Client und einem Remote-Agenten als Teil einer Nachricht oder eines Artefakts ausgetauscht wird. Ein Teil kann Text, Bild, Video oder Datei sein.
Push-Benachrichtigungen (optional): Ein sicherer Benachrichtigungsmechanismus, mit dem Kundenservicemitarbeiter einen Kunden außerhalb einer verbundenen Sitzung über ein Update informieren können.

Kartenerkennung

Wenn der A2A-Clientdienst gestartet wird, wird in der Regel versucht, die Informationen zur Kundenkarte abzurufen und zu speichern, damit bei Bedarf problemlos darauf zugegriffen werden kann. Wir können das hier im Script purchasing_concierge/purchasing_agent.py prüfen.

...

class PurchasingAgent:
    """The purchasing agent.

    This is the agent responsible for choosing which remote seller agents to send
    tasks to and coordinate their work.
    """

    def __init__(
        self,
        remote_agent_addresses: List[str],
        task_callback: TaskUpdateCallback | None = None,
    ):
        self.task_callback = task_callback
        self.remote_agent_connections: dict[str, RemoteAgentConnections] = {}
        self.cards: dict[str, AgentCard] = {}
        for address in remote_agent_addresses:
            card_resolver = A2ACardResolver(address)
            try:
                card = card_resolver.get_agent_card()
                # The URL accessed here should be the same as the one provided in the agent card
                # However, in this demo we are using the URL provided in the key arguments
                remote_connection = RemoteAgentConnections(
                    agent_card=card, agent_url=address
                )
                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)

...

Tool zum Senden von Aufgaben

Wir stellen den Remote-Kundenservicemitarbeitern dann den Kontext in unserem Prompt-System für den Kundenservice zur Verfügung und stellen auch Tools bereit, mit denen die Aufgabe an den Kundenservicemitarbeiter gesendet werden kann. Das ist die Aufforderung und das Tool, das wir unseren ADK-Kunden 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}")
    if "task_id" in state:
        taskId = state["task_id"]
    else:
        taskId = str(uuid.uuid4())
    sessionId = state["session_id"]
    task: Task
    messageId = ""
    metadata = {}
    if "input_message_metadata" in state:
        metadata.update(**state["input_message_metadata"])
        if "message_id" in state["input_message_metadata"]:
            messageId = state["input_message_metadata"]["message_id"]
    if not messageId:
        messageId = str(uuid.uuid4())
    metadata.update(**{"conversation_id": sessionId, "message_id": messageId})
    request: TaskSendParams = TaskSendParams(
        id=taskId,
        sessionId=sessionId,
        message=Message(
            role="user",
            parts=[TextPart(text=task)],
            metadata=metadata,
        ),
        acceptedOutputModes=["text", "text/plain"],
        # pushNotification=None,
        metadata={"conversation_id": sessionId},
    )
    task = await client.send_task(request, self.task_callback)
    # Assume completion unless a state returns that isn't complete
    state["session_active"] = task.status.state not in [
        TaskState.COMPLETED,
        TaskState.CANCELED,
        TaskState.FAILED,
        TaskState.UNKNOWN,
    ]
    if task.status.state == TaskState.INPUT_REQUIRED:
        # Force user input back
        tool_context.actions.escalate = True
    elif task.status.state == TaskState.COMPLETED:
        # Reset active agent is task is completed
        state["active_agent"] = "None"

    response = []
    if task.status.message:
        # Assume the information is in the task message.
        response.extend(convert_parts(task.status.message.parts, tool_context))
    if task.artifacts:
        for artifact in task.artifacts:
            response.extend(convert_parts(artifact.parts, tool_context))
    return response

...

Im Prompt geben wir unserem Kundenservicemitarbeiter den Namen und die Beschreibung aller verfügbaren Remote-Kundenservicemitarbeiter an. Im Tool self.send_task stellen wir einen Mechanismus bereit, mit dem der richtige Kunde abgerufen wird, um eine Verbindung zum Kundenservicemitarbeiter herzustellen und die erforderlichen Metadaten über das TaskSendParams-Objekt zu senden.

Im Tool können wir auch angeben, wie sich der Kundenservicemitarbeiter verhalten soll, wenn eine Aufgabe nicht abgeschlossen werden kann. Und schließlich müssen wir das Antwortartefakt verarbeiten, das zurückgegeben wird, wenn die Aufgabe abgeschlossen ist.

7. Integrationstests und Nutzlastprüfung

Sehen wir uns nun die folgende Unterhaltung an und prüfen die Benutzeroberfläche und Dienstprotokolle des Kundenservicemitarbeiters für Käufe. Versuchen Sie, eine Unterhaltung wie diese zu führen :

Zeig mir die Burger- und Pizzakarte
Ich möchte eine BBQ-Hähnchenpizza und einen scharfen Cajun-Burger bestellen.

Setzen Sie das Gespräch fort, bis Sie die Bestellung abgeschlossen haben. Sehen Sie sich an, wie die Interaktion abläuft und was der Toolaufruf und die Antwort sind. Das folgende Bild ist ein Beispiel für das Interaktionsergebnis.

Wir sehen, dass die Kommunikation mit zwei verschiedenen Kundenservicemitarbeitern zu zwei unterschiedlichen Verhaltensweisen führt. A2A kann damit gut umgehen. Der Agent des Burgerverkäufers akzeptiert die Anfrage des Einkaufsagenten direkt, während der Pizzaagent unsere Bestätigung benötigt, bevor er mit der Anfrage fortfährt. Nachdem wir die Bestätigung gegeben haben, kann der Agent sie an den Pizzaagenten weiterleiten.

Sehen wir uns nun die ausgetauschten Daten im Dienstprotokoll purchasing-agent an. Rufen Sie zuerst die Cloud Run Console auf. Geben Sie dazu „Cloud Run“ in die Suchleiste oben in der Console ein und klicken Sie mit der rechten Maustaste auf das Cloud Run-Symbol, um es in einem neuen Browsertab zu öffnen.

Jetzt sollten Sie die zuvor bereitgestellten Dienste wie unten dargestellt sehen. Klicken Sie auf purchasing-concierge.

Sie befinden sich jetzt auf der Seite „Dienstdetails“. Klicken Sie auf den Tab Protokolle.

Jetzt sehen wir uns die Protokolle unseres bereitgestellten Dienstes purchasing-concierge an. Scrollen Sie nach unten und suchen Sie nach dem aktuellen Protokoll zu unseren Interaktionen.

Sie sehen, dass die Anfrage und Antwort zwischen A2A-Client und -Server im JSON-RPC-Format vorliegen und dem A2A-Standard entsprechen.

Wir haben die grundlegenden Konzepte von A2A kennengelernt und sehen uns jetzt an, wie sie als Client- und Serverarchitektur implementiert wird.

8. Bereinigen

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

Rufen Sie in der Google Cloud Console die Seite Ressourcen verwalten auf.
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 Cloud Run aufrufen, den gerade bereitgestellten Dienst auswählen und löschen.