Anleitung: Schritt-für-Schritt-Erstellung von zwei Python-Agenten mit Googles A2A-Protokoll

Name: Elwynn AI

Updated on 10.4.2025

Erlernen Sie, wie man zwei einfache Python-Agenten mit Googles A2A-Protokoll erstellt. Diese Schritt-für-Schritt-Anleitung deckt die Einrichtung, Code-Beispiele und die Kommunikation zwischen den Agenten ab. A2A ist ein offener Standard für die Kommunikation zwischen Agenten und ermöglicht die Interoperabilität zwischen verschiedenen Frameworks und Anbietern.

Das Agent-to-Agent (A2A) Protokoll von Google ist ein offener Standard, der es ermöglicht, dass unterschiedliche KI-Agenten miteinander kommunizieren und zusammenarbeiten, unabhängig davon, aus welchem Framework oder von welchem Anbieter sie stammen (GitHub - google/A2A (opens in a new tab)). Einfacher ausgedrückt gibt A2A Ihren Agenten eine gemeinsame Sprache für die Interaktion. Diese Anleitung führt Sie durch die Erstellung von zwei einfachen Python-Agenten – einer agiert als A2A Server und einer als A2A Client –, die mithilfe des A2A-Protokolls miteinander kommunizieren. Wir werden die Einrichtung abdecken, erklären, wie A2A auf grundlegender Ebene funktioniert, und kommentierte Code-Beispiele für beide Agenten bereitstellen. Vorkenntnisse mit A2A sind nicht erforderlich, und wir werden fortgeschrittene Konzepte wie komplexe Toolchains oder Multi-Agenten-Orchestrierung meiden, um die Dinge anfängerfreundlich zu halten.

Was ist das A2A-Protokoll?

Im Kern ist A2A ein Protokoll (auf HTTP aufgebaut), das standardisiert, wie Agenten Nachrichten, Anfragen und Daten austauschen. Hier sind einige Schlüsselkonzepte in A2A:

Agent Card: Eine öffentliche Metadaten-Datei (normalerweise unter /.well-known/agent.json gehostet), die die Fähigkeiten, Kompetenzen, die URL des Endpunkts und die Authentifizierungsanforderungen eines Agenten beschreibt. Andere Agenten oder Clients holen sich diese Karte, um zu entdecken, was der Agent kann (GitHub - google/A2A (opens in a new tab)). Sie ist im Wesentlichen die „Visitenkarte“ des Agenten im A2A-Netzwerk.
A2A Server: Ein Agent, der einen HTTP-API-Endpunkt bereitstellt, der A2A-Methoden implementiert (zum Beispiel Endpunkte zum Senden von Aufgaben). Er empfängt Anfragen und führt Aufgaben im Namen anderer Agenten aus (GitHub - google/A2A (opens in a new tab)).
A2A Client: Eine Anwendung oder ein Agent, der Anfragen an die URL eines A2A-Servers sendet, um Aufgaben oder Gespräche zu initiieren (GitHub - google/A2A (opens in a new tab)). In einem Szenario von Agent-zu-Agent wird eine Client-Komponente eines Agenten die Server-Komponente eines anderen Agenten über A2A aufrufen.
Task: Die grundlegende Arbeitseinheit in A2A. Ein Client startet eine Aufgabe, indem er eine Nachricht (mithilfe eines tasks/send Anrufs) an einen Agenten sendet. Jede Aufgabe hat eine eindeutige ID und durchläuft einen Lebenszyklus von Zuständen (z. B. submitted, working, input-required, completed, etc.) (GitHub - google/A2A (opens in a new tab)).
Message: Ein einzelner Schritt in der Kommunikation zwischen dem Client (Rolle "user") und dem Agenten (Rolle "agent"). Die Anfrage des Clients ist eine Nachricht (vom „user“), und die Antwort des Agenten ist eine Nachricht (vom „agent“) (GitHub - google/A2A (opens in a new tab)).
Part: Ein Inhaltsteil innerhalb einer Nachricht. Eine Nachricht besteht aus einem oder mehreren Teilen. Teile können Text, Dateien oder strukturierte Daten sein. Beispielsweise trägt ein Textteil (TextPart) einfachen Text, während ein Dateiteil ein Bild oder andere Binärdaten transportieren kann (GitHub - google/A2A (opens in a new tab)). In dieser Anleitung bleiben wir der Klarheit halber bei einfachen Textteilen.

Wie A2A-Kommunikation funktioniert (Grundlegender Ablauf): Wenn ein Agent über A2A mit einem anderen kommunizieren möchte, folgt die Interaktion typischerweise einem standardisierten Ablauf:

Entdeckung: Der Client-Agent entdeckt zunächst den anderen Agenten, indem er seine Agent Card abruft (normalerweise von der /.well-known/agent.json URL des Servers) (GitHub - google/A2A (opens in a new tab)). Diese Karte sagt dem Client, wie der Agent heißt, was er kann und wohin Anfragen gesendet werden sollen.
Initiation: Der Client sendet dann eine Anfrage an den A2A-Endpunkt des Server-Agenten, um eine Aufgabe zu starten – üblicherweise über ein POST an den tasks/send Endpunkt. Diese erste Anfrage enthält eine eindeutige Aufgaben-ID und die erste Benutzernachricht (z. B. eine Frage oder ein Befehl) (GitHub - google/A2A (opens in a new tab)).
Verarbeitung: Der Server-Agent empfängt die Aufgabenanfrage und verarbeitet sie. Wenn der Agent Streaming-Antworten unterstützt, kann er Zwischenergebnisse zurückstreamen (mithilfe von Server-Sent Events), während er arbeitet, obwohl wir in unserem einfachen Beispiel kein Streaming verwenden werden. Andernfalls verarbeitet der Agent die Aufgabe und sendet eine endgültige Antwort zurück, wenn er fertig ist (GitHub - google/A2A (opens in a new tab)). Die Antwort enthält das Ergebnis der Aufgabe – typischerweise eine oder mehrere Nachrichten vom Agenten (und optional Artefakte wie Dateien, wenn das Teil der Aufgabe ist). Für einfache Textinteraktionen wird die endgültige Antwort nur die Antwortnachricht des Agenten enthalten.

Zusammengefasst definiert A2A einen klaren Anforderungs-Antwort-Zyklus für Agenten: ein Client findet die Fähigkeiten eines Agenten (Agent Card), sendet ihm eine Aufgabe (mit einer Anfangsnachricht) und erhält die Antwort des Agenten (als Nachrichten und Artefakte), alles in einem konsistenten JSON-Format. Jetzt richten wir unsere Umgebung ein und erstellen ein minimales Beispiel, um dies in Aktion zu sehen.

Installation und Einrichtung

Bevor Sie Code schreiben, stellen Sie sicher, dass Sie die erforderlichen Werkzeuge installiert haben:

Python 3.12 oder höher – die A2A-Beispiele empfehlen Python 3.12+ (sogar Python 3.13) für Kompatibilität (A2A/samples/python/hosts/cli at main · google/A2A · GitHub (opens in a new tab)).
UV (Python-Paketmanager) – dies ist ein optionales, aber empfohlenes Werkzeug vom A2A-Projekt zur Verwaltung von Abhängigkeiten und zum Ausführen der Beispiele (A2A/samples/python/hosts/cli at main · google/A2A · GitHub (opens in a new tab)). UV ist ein moderner Paketmanager; wenn Sie jedoch bevorzugen, können Sie auch pip/venv verwenden, um die Abhängigkeiten manuell zu installieren. In dieser Anleitung werden wir die Dinge einfach halten und Standard-Python-Tools verwenden.
Google A2A Repository – erhalten Sie den offiziellen A2A-Code, indem Sie das GitHub-Repository klonen:
```
git clone https://github.com/google/A2A.git
```
Dies lädt das Projekt auf Ihren Rechner herunter. Unser Tutorial-Code wird Teile dieses Repositories referenzieren (insbesondere die Beispiel-A2A-Client-Dienstprogramme).
Abhängigkeiten – wenn Sie UV verwenden, können Sie es die Abhängigkeiten automatisch verwalten lassen. Beispielsweise können Sie im Verzeichnis samples/python des Repositorys uv sync ausführen (um die Abhängigkeiten zu installieren) oder Befehle wie uv run ... laut der offiziellen README ausführen. Wenn Sie es bevorzugen, pip zu verwenden, stellen Sie sicher, dass Sie alle Bibliotheken installieren, die für unseren Beispielcode benötigt werden:
```
pip install flask requests
```
Die Flask-Bibliothek wird verwendet, um einen einfachen HTTP-Server für unseren Agenten zu erstellen, und requests wird verwendet, damit der Client den Agenten aufrufen kann. (Der offizielle A2A-Beispielcode verwendet FastAPI/UVICorn und eine interne Client-Bibliothek, aber für einen anfängerfreundlichen Ansatz verwenden wir Flask und requests der Klarheit halber.)

Erstellen eines einfachen A2A-Serveragenten

Zuerst erstellen wir die Serverseite: einen Agenten, der auf eingehende Aufgaben wartet und antwortet. Unser Agent wird extrem einfach sein – er wird ein Textprompt akzeptieren und nur mit einer freundlichen Nachricht zurückgeben. Trotz seiner Einfachheit wird er die A2A-Protokollstruktur befolgen, sodass jeder A2A-Client mit ihm kommunizieren kann.

Agentenfähigkeiten und Agent Card: Damit unser Agent entdeckt werden kann, muss er eine Agent Card bereitstellen. In einer echten Bereitstellung würden wir eine JSON-Datei unter /.well-known/agent.json auf dem Server bereitstellen. Für unsere Flask-Anwendung können wir diese Daten über einen Endpunkt bereitstellen. Die Agent Card enthält typischerweise Felder wie den Namen des Agenten, die Beschreibung, die URL seines A2A-Endpunkts, unterstützte Funktionen (Fähigkeiten) usw. (GitHub - google/A2A (opens in a new tab)). Wir erstellen eine minimale Agent Card mit nur wenigen Schlüsselwerten.

tasks/send Endpoint: A2A erwartet, dass der Agent bestimmte API-Endpunkte implementiert. Der wichtigste ist tasks/send, den der Client aufrufen wird (HTTP POST), um eine neue Aufgabe (mit einer Anfangsnachricht) an den Agenten zu senden (GitHub - google/A2A (opens in a new tab)). Unsere Flask-App wird eine Route für /tasks/send definieren, um diese Anfragen zu bearbeiten. Sie wird das eingehende JSON (das die Aufgaben-ID und die Benutzeransage enthält) analysieren, verarbeiten (in unserem Fall nur eine Echoantwort generieren) und eine JSON-Antwort gemäß dem A2A-Format zurückgeben.

Lassen Sie uns den Code für unseren einfachen A2A-Serveragenten mithilfe von Flask schreiben:

from flask import Flask, request, jsonify
 
app = Flask(__name__)
 
# Definiere die Agent Card Daten (Metadaten zu diesem Agenten)
AGENT_CARD = {
    "name": "EchoAgent",
    "description": "Ein einfacher Agent, der Benutzernachrichten zurückgibt.",
    "url": "http://localhost:5000",  # Die Basis-URL, wo dieser Agent gehostet wird
    "version": "1.0",
    "capabilities": {
        "streaming": False,           # Dieser Agent unterstützt keine Streaming-Antworten
        "pushNotifications": False    # Keine Push-Benachrichtigungen in diesem einfachen Beispiel
    }
    # (In einer vollständigen Agent Card könnten mehr Felder wie Authentifizierungsinformationen enthalten sein.)
}
 
# Serve die Agent Card an der bekannten URL.
@app.get("/.well-known/agent.json")
def get_agent_card():
    """Endpunkt, um die Metadaten dieses Agenten (Agent Card) bereitzustellen."""
    return jsonify(AGENT_CARD)
 
# Behandeln eingehende Aufgabenanfragen am A2A-Endpunkt.
@app.post("/tasks/send")
def handle_task():
    """Endpunkt für A2A-Clients, um eine neue Aufgabe (mit einer anfänglichen Benutzernachricht) zu senden."""
    task_request = request.get_json()  # parse die eingehende JSON-Anfrage
    # Extrahiere die Aufgaben-ID und den Nachrichtentext des Benutzers aus der Anfrage.
    task_id = task_request.get("id")
    user_message = ""
    try:
        # Gemäß A2A-Spezifikation befindet sich die Benutzernachricht in task_request["message"]["parts"][0]["text"]
        user_message = task_request["message"]["parts"][0]["text"]
    except Exception as e:
        return jsonify({"error": "Ungültiges Anfrageformat"}), 400
 
    # Für diesen einfachen Agenten ist die "Verarbeitung" nur das Zurückgeben der Nachricht.
    agent_reply_text = f"Hallo! Sie haben gesagt: '{user_message}'"
 
    # Formuliere die Antwort im A2A Task Format.
    # Wir geben ein Task-Objekt mit dem finalen Status = 'completed' und der Nachricht des Agenten zurück.
    response_task = {
        "id": task_id,
        "status": {"state": "completed"},
        "messages": [
            task_request.get("message", {}),             # beinhalte die ursprüngliche Benutzernachricht in der Historie
            {
                "role": "agent",                        # die Antwort des Agenten
                "parts": [{"text": agent_reply_text}]   # Agentennachrichteninhalt als Textteil
            }
        ]
        # Wir könnten auch ein "artifacts" Feld einfügen, wenn der Agent Dateien oder andere Daten zurückgab.
    }
    return jsonify(response_task)
 
# Führe die Flask-App (A2A-Server) aus, wenn dieses Skript direkt ausgeführt wird.
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

Verständnis des Servercodes: Lassen Sie uns aufschlüsseln, was im obigen Code passiert:

Wir haben eine Flask-App erstellt und ein globales AGENT_CARD-Dictionary definiert. Dieses enthält grundlegende Informationen über unseren Agenten. Besonders erwähnenswert sind der Name, die Beschreibung, eine URL (die auf sich selbst auf localhost und Port 5000 zeigt) und ein Fähigkeitenbereich. Wir haben streaming: False gesetzt, da unser Agent nur einfache synchrone Antworten zurückgeben wird (wenn ein Agent Streaming unterstützte, würde er tasks/sendSubscribe behandeln und Zwischenergebnisse streamen, was wir hier nicht abdecken).
Wir haben eine Route /.well-known/agent.json eingerichtet, die die Agent Card JSON zurückgibt. Dies spiegelt den standardmäßigen Ort wider, an dem eine Agentenkarte für die Erkennung erwartet wird (GitHub - google/A2A (opens in a new tab)).
Wir definieren eine POST-Route für /tasks/send. Wenn ein A2A-Client eine Aufgabe sendet, wird Flask handle_task() aufrufen. Innerhalb dieser Funktion:
- Wir parsieren den JSON-Body der Anfrage in task_request. Gemäß dem A2A-Schema enthält dies eine "id" (Aufgaben-ID) und ein "message"-Objekt, das die anfängliche Benutzernachricht darstellt.
- Wir extrahieren den Text der Benutzernachricht. In A2A kann eine Nachricht mehrere Teile haben (z. B. Text, Dateien). Hier gehen wir davon aus, dass der erste Teil Text ist und greifen auf task_request["message"]["parts"][0]["text"] zu. In einer robusteren Implementierung würden wir den Typ des Teils überprüfen und Fehler behandeln (unser Code führt einen einfachen try/except aus und gibt einen 400-Fehler zurück, wenn das Format nicht wie erwartet ist).
- Dann generieren wir die Antwort des Agenten. Unsere Logik ist trivial: Wir nehmen den Benutzertext und fügen "Hallo! Sie haben gesagt: ..." hinzu. In einem echten Agenten würde hier die Kern-KI-Logik stattfinden (ein LLM aufrufen, Berechnungen durchführen usw.).
- Als nächstes erstellen wir das Antwort-Task-Objekt. Wir müssen die id zurückgeben (damit der Client weiß, um welche Aufgabe es sich handelt), den status.state auf "completed" setzen (da wir die Aufgabe abgeschlossen haben) und eine Liste von messages bereitstellen. Wir beinhalten die ursprüngliche Benutzernachricht (zur Kontext/Historie) und fügen dann die Antwort unseres Agenten als eine weitere Nachricht hinzu. Jede Nachricht hat eine role ("user" oder "agent") und eine Liste von parts. Wir verwenden einen einzelnen TextPart, der unseren Antworttext für die Nachricht des Agenten enthält.
- Schließlich geben wir dies als JSON zurück. Flask serialisiert das Python-Dict für uns in JSON.

Mit diesem laufenden Server ist unser Agent effektiv „live“ auf http://localhost:5000. Er kann seine Agent Card bereitstellen und Aufgabenanfragen bearbeiten.

Erstellen eines einfachen A2A-Clientagenten

Da wir nun einen Agentenserver haben, benötigen wir einen Client, der mit ihm kommunizieren kann. Dieser Client könnte ein anderer Agent oder einfach eine benutzerseitige Anwendung sein. Im A2A-Kontext ist er ein A2A-Client, weil er die A2A-API unseres Server-Agenten nutzt (GitHub - google/A2A (opens in a new tab)). Für unser Tutorial schreiben wir ein kleines Python-Skript, das die Rolle eines zweiten Agenten (oder Benutzers) übernimmt, der eine Frage an unseren EchoAgent sendet und die Antwort erhält.

Schritte für den Client:

Agenten erkennen: Der Client sollte die Agent Card des Serveragenten abrufen, um über dessen Endpunkt und Fähigkeiten zu erfahren. Wir wissen, dass unsere Agentenkarte unter http://localhost:5000/.well-known/agent.json ist. Wir können die Python-Bibliothek requests verwenden, um diese URL zu GETten.
Eine Aufgabe senden: Als nächstes wird der Client eine POST-Anfrage an den tasks/send Endpunkt des Agenten mit einem JSON-Payload senden, das die Aufgabe darstellt. Dieses Payload sollte enthalten:
- eine eindeutige "id" für die Aufgabe (jede eindeutige Zeichenfolge oder Zahl; z.B. "task1"),
- ein "message"-Objekt mit role: "user" und einer parts-Liste, die die Benutzerabfrage als TextPart enthält.
Antwort empfangen: Der Agent wird mit einem JSON antworten, das die Task darstellt (einschließlich der Antwortnachricht des Agenten). Der Client muss dies lesen und die Antwort des Agenten extrahieren.

Lassen Sie uns den Client-Code schreiben, um diese Schritte auszuführen:

import requests
import uuid
 
# 1. Agenten durch Abrufen seiner Agent Card entdecken
AGENT_BASE_URL = "http://localhost:5000"
agent_card_url = f"{AGENT_BASE_URL}/.well-known/agent.json"
response = requests.get(agent_card_url)
if response.status_code != 200:
    raise RuntimeError(f"Agent Card konnte nicht abgerufen werden: {response.status_code}")
agent_card = response.json()
print("Entdeckter Agent:", agent_card["name"], "-", agent_card.get("description", ""))
 
# 2. Eine Aufgabenanfrage für den Agenten vorbereiten
task_id = str(uuid.uuid4())  # generiere eine zufällige eindeutige Aufgaben-ID
user_text = "Was ist der Sinn des Lebens?"
task_payload = {
    "id": task_id,
    "message": {
        "role": "user",
        "parts": [
            {"text": user_text}
        ]
    }
}
print(f"Sende Aufgabe {task_id} an Agenten mit Nachricht: '{user_text}'")
 
# 3. Die Aufgabe an den tasks/send Endpunkt des Agenten senden
tasks_send_url = f"{AGENT_BASE_URL}/tasks/send"
result = requests.post(tasks_send_url, json=task_payload)
if result.status_code != 200:
    raise RuntimeError(f"Anfrage fehlgeschlagen: {result.status_code}, {result.text}")
task_response = result.json()
 
# 4. Die Antwort des Agenten verarbeiten
# Die Antwort sollte die Aufgaben-ID, den Status und die Nachrichten (einschließlich der Antwort des Agenten) enthalten.
if task_response.get("status", {}).get("state") == "completed":
    # Die letzte Nachricht in der Liste sollte die Antwort des Agenten sein (da unser Agent Historie in Nachrichten beinhaltete).
    messages = task_response.get("messages", [])
    if messages:
        agent_message = messages[-1]  # letzte Nachricht (vom Agenten)
        agent_reply_text = ""
        # Text aus den Nachrichtenteilen des Agenten extrahieren
        for part in agent_message.get("parts", []):
            if "text" in part:
                agent_reply_text += part["text"]
        print("Antwort des Agenten:", agent_reply_text)
    else:
        print("Keine Nachrichten in der Antwort!")
else:
    print("Aufgabe wurde nicht abgeschlossen. Status:", task_response.get("status"))

Verständnis des Clientcodes: Hier ist, was der Client tut:

Er verwendet requests.get, um die Agent Card von http://localhost:5000/.well-known/agent.json abzurufen. Bei Erfolg analysieren wir das JSON und drucken den Namen und die Beschreibung des Agenten aus, um zu überprüfen, ob wir den richtigen Agenten gefunden haben.
Wir generieren eine eindeutige Aufgaben-ID mit uuid.uuid4(). Dies dient nur zur Sicherstellung, dass wir keine ID erneut verwenden (in realen Szenarien muss der Client sicherstellen, dass jede gesendete Aufgabe eine eindeutige ID hat).
Wir erstellen das Aufgaben-Payload als ein Python-Dictionary. Dieses Payload folgt dem A2A-Schema: es hat eine "id" (unsere generierte ID) und eine "message" für die Benutzerabfrage. Die Nachricht hat role: "user" und einen Teil, der den Text "Was ist der Sinn des Lebens?" enthält (dieser Text kann geändert werden – der EchoAgent wird ihn nur zurückgeben).
Wir senden diesen Payload an den tasks/send-Endpunkt des Agenten mithilfe von requests.post. Beachten Sie, dass wir den json=... Parameter verwenden, der die requests-Bibliothek anweist, das Dictionary als JSON zu senden.
Wir überprüfen auf einen 200 OK-Status. Sollte der Status nicht 200 sein, werfen wir einen Fehler. (In einem nachsichtigeren Skript könnten Sie nicht-200-Antworten freundlicher handhaben.)
Wenn der Aufruf erfolgreich war, analysieren wir das JSON der Antwort in task_response. Dies müsste das Task-Objekt sein, das unser Agent zurückgegeben hat. Wir untersuchen es dann:
- Wir überprüfen, ob status.state "completed" ist – in unserem Design markiert der Agent es nach einer Antwort als abgeschlossen. Wenn der Agent beispielsweise mehr Eingaben benötigte ("input-required") oder fehlschlug, würden wir diese Fälle behandeln. Zum Zwecke der Einfachheit gehen wir von einer Vollendung aus.
- Wir holen dann die Liste der messages aus der Antwort. Wir erwarten, dass die letzte Nachricht in dieser Liste die Antwort des Agenten ist (da unser Servercode die Nachricht des Agenten nach der Benutzeranfrage hinzugefügt hat). Wir extrahieren den Text aus den Nachrichtenteilen des Agenten. In diesem Fall sollte es die Echoantwort sein, die wir konstruiert haben.
- Schließlich drucken wir die Antwort des Agenten aus.

Zu diesem Zeitpunkt wird unser Client die Antwort des Agenten ausgedruckt haben. Im nächsten Abschnitt werden wir sowohl den Server als auch den Client ausführen, um den gesamten Austausch zu sehen.

Ausführen und Testen der Agenten

Um unsere beiden Agenten zu testen, gehen Sie wie folgt vor:

Starte den Server-Agenten: Führen Sie das Flask-Server-Skript aus (wir nennen es echo_agent.py). Beispielsweise in Ihrem Terminal:
```
python echo_agent.py
```
Dies sollte den Flask-Entwicklungsserver auf http://localhost:5000 starten. Sie sollten eine Nachricht wie „Running on http://0.0.0.0:5000/“ (opens in a new tab) sehen. Der Agent hört jetzt auf Anfragen. (Stellen Sie sicher, dass nichts anderes bereits Port 5000 verwendet, oder ändern Sie den Port im Code, falls erforderlich.)
Führen Sie das Client-Skript aus: Öffnen Sie ein weiteres Terminal und führen Sie das Client-Skript aus (z. B. client_agent.py):
```
python client_agent.py
```
Der Client sollte etwas Folgendes ausdrucken:
```
Entdeckter Agent: EchoAgent – Ein einfacher Agent, der Benutzernachrichten zurückgibt.
Sende Aufgabe 3f8ae8ac-... an Agenten mit Nachricht: 'Was ist der Sinn des Lebens?'
Antwort des Agenten: Hallo! Sie haben gesagt: 'Was ist der Sinn des Lebens?'
```
Dies zeigt an, dass der Client erfolgreich den Agenten entdeckt, eine Frage gesendet und die Antwort erhalten hat. Die Antwort des Agenten ist die erwartete Echo-Nachricht.
Interaktion überprüfen: Sie können auch die Konsolenausgabe des Servers überprüfen. Flask protokolliert die Anfragen, die es behandelt hat. Sie sollten eine GET-Anfrage für /.well-known/agent.json und ein POST an /tasks/send sehen, jeweils mit einem 200-Status. Dies bestätigt den Protokollablauf:
- Entdeckung (Client GET der Agent Card) → Initiation (Client POST der Aufgabe) → Verarbeitung/Vollendung (Server beantwortet mit Antwort).

Herzlichen Glückwunsch! Sie haben gerade eine grundlegende Agent-zu-Agent-Kommunikation unter Verwendung des A2A-Protokolls implementiert. Ein Agent fungierte als Dienst (EchoAgent) und der andere als Client, der ihn anfragt. Während dieses Beispiel nur ein einfaches Echo darstellt (der „KI“ gibt die Eingabe nur zurück), demonstriert es die notwendige Architektur für jede A2A-Interaktion.

Verwendung der A2A-Client-Bibliothek (Optional)

In unserem Client-Code haben wir manuell HTTP-Anfragen mithilfe von requests erstellt. Das offizielle A2A-GitHub-Repository bietet Hilfsklassen, um diesen Prozess zu vereinfachen. Zum Beispiel gibt es eine A2AClient-Klasse, die das Abrufen der Agent Card und das Senden von Aufgaben für Sie übernehmen kann, sowie einen A2ACardResolver, um die Agentenkarte zu erkennen. Hier ist, wie wir diese verwenden könnten (für die Interessierten an einem idiomatischeren Ansatz):

import asyncio
from samples.python.common.client import A2AClient, A2ACardResolver
from samples.python.common.types import TaskSendParams, Message, TextPart
 
async def query_agent(agent_url, user_text):
    # Agentenkarten automatisch abrufen
    card_resolver = A2ACardResolver(agent_url)
    agent_card = card_resolver.get_agent_card()
    print("Entdeckter Agent:", agent_card.name)
    # Erstellen Sie einen A2A-Client mit der Agent Card
    client = A2AClient(agent_card=agent_card)
    # Task-Parameter vorbereiten (mit A2A-Typklassen)
    payload = TaskSendParams(
        id=str(uuid.uuid4()),
        message=Message(role="user", parts=[TextPart(text=user_text)])
    )
    # Die Aufgabe senden und auf Vollendung warten
    result_task = await client.send_task(payload)  # send_task ist eine asynchrone Methode
    # Die Nachricht des Agenten aus dem result_task extrahieren
    if result_task.status.state.value == "completed":
        # Das A2A Task-Objekt kann auf Nachrichten und Artefakte geprüft werden
        for msg in result_task.messages:
            if msg.role == "agent":
                # Textteile aus der Nachricht des Agenten ausdrucken
                print("Antwort des Agenten:", " ".join(part.text for part in msg.parts if hasattr(part, "text")))

In dem obigen Snippet stammen A2ACardResolver und A2AClient aus der A2A-Beispielbibliothek (im Verzeichnis samples/python/common des Repos zu finden). Die TaskSendParams, Message und TextPart sind Datenklassen (wahrscheinlich basierend auf Pydantic-Modellen), die Teilen des A2A JSON-Schemas entsprechen. Mit diesen müssen wir keine Dictionnaires manuell erstellen; wir erstellen Python-Objekte und die Bibliothek übernimmt die JSON-Serialisierung und die HTTP-Abfragen. Die send_task Methode des Clients ist asynchron (daher verwenden wir await), und sie gibt ein Task-Objekt zurück. Unser Beispiel zeigt, wie Sie die Antwort des Agenten aus diesem Objekt abrufen könnten.

Hinweis: Der obige Code ist zur Veranschaulichung und erfordert, dass der A2A-Repository-Code in Ihrem Python-Pfad zugänglich ist. Wenn Sie das Repo geklont und dessen Anforderungen installiert haben (über UV oder pip), könnten Sie dies in Ihren Client integrieren. Das offizielle CLI-Tool im A2A-Repo imitiert im Wesentlichen diese Schritte – es liest die Agent Card und tritt in eine Schleife ein, um Aufgaben zu senden und Antworten auszudrucken (A2A/samples/python/hosts/cli at main · google/A2A · GitHub (opens in a new tab)).

📚

Fazit

In diesem Tutorial haben wir die Grundlagen des Google A2A-Protokolls behandelt und eine einfache Zwei-Agenten-Kommunikation gezeigt. Wir haben einen minimalen A2A-Server (EchoAgent) und einen Client erstellt, der mit ihm interagiert. Auf dem Weg haben wir etwas über Agent Cards (zur Erkennung) gelernt und wie Aufgaben und Nachrichten in A2A strukturiert sind.

Obwohl unser Beispiel nur ein trivialer Echo-Agent war, haben Sie nun die Grundlage, um komplexere Agenten zu entwickeln. Beispielsweise könnten Sie die Echo-Logik durch einen Aufruf eines Sprachmodells oder einer API ersetzen, um den Agenten tatsächlich ein Problem lösen oder Fragen beantworten zu lassen. Das Schöne an A2A ist, dass, solange Ihr Agent das Protokoll einhält (eine Agent Card bereitstellt und die Aufgabenendpunkte implementiert), jeder andere A2A-kompatible Agent oder jedes A2A-kompatible Werkzeug sich mit ihm integrieren kann.

Nächste Schritte: Sie könnten die offiziellen A2A-Repository-Beispiele für fortgeschrittenere Muster erkunden. Beispielsweise enthalten Googles Beispiele einen Agenten, der Bilder aus Text generiert und sie als Bildartefakte zurückgibt (A2A/samples/python/agents/crewai/README.md at main · google/A2A · GitHub (opens in a new tab)), und einen anderen, der mehrteilige Formulare ausfüllen kann. Sie können auch das bereitgestellte Befehlszeilen-Interface (CLI) oder die Web-Demo ausprobieren, um mit Ihrem Agenten zu chatten. Um das CLI mit unserem EchoAgent zu verwenden, könnten Sie uv run hosts/cli --agent http://localhost:5000 (wenn Sie UV und das Repo-Setup haben) ausführen, um ein REPL-ähnliches Gespräch mit dem Agenten zu beginnen (A2A/samples/python/hosts/cli at main · google/A2A · GitHub (opens in a new tab)).

Durch die Standardisierung der Agentenkommunikation eröffnet A2A die Tür für ein reichhaltiges Ökosystem interoperabler KI-Agenten. Wir haben hier nur an der Oberfläche mit einem einfachen Beispiel gekratzt. Viel Spaß beim Experimentieren mit A2A – und mögen Ihre Agenten effektiv zusammenarbeiten!

Quellen:

Google A2A GitHub – Agent2Agent Protocol README (Konzeptioneller Überblick) (GitHub - google/A2A (opens in a new tab)) (GitHub - google/A2A (opens in a new tab)) (GitHub - google/A2A (opens in a new tab)) (GitHub - google/A2A (opens in a new tab))
Google A2A GitHub – Beispiele und CLI-Nutzung (A2A/samples/python/hosts/cli at main · google/A2A · GitHub (opens in a new tab)) (A2A/samples/python/hosts/cli at main · google/A2A · GitHub (opens in a new tab))
Google A2A GitHub – CrewAI Beispiel README (Agentenfähigkeiten via A2A) (A2A/samples/python/agents/crewai/README.md at main · google/A2A · GitHub (opens in a new tab)) (für weiterführende Erkundungen)

📚

Umfragehinweis: Erkundung von OpenAIs GPT-4.5-Veröffentlichung