Agent-to-Agent — A2A auf Protokoll-Ebene.
Du baust zwei AI-Agents, die nie den Code des jeweils anderen gesehen haben, plus einen Orchestrator, der sie nacheinander beauftragt. Am Ende verstehst du A2A auf Protokoll-Ebene — nicht nur die SDK-Oberfläche, sondern das, was tatsächlich über das Netzwerk geht.
"Ich baue Systeme, in denen Agents sich gegenseitig beauftragen — über Team-, Framework- und Maschinen-Grenzen hinweg. A2A ist dafür der Standard. Dieser Guide ist der Pfad, den ich jedem neuen Engineer im Team gebe: erst die fünf Begriffe, dann zwei Agents, dann der Draht selbst."
Verifiziert gegen a2a-sdk 1.1.0 · Python 3.13 · Groq 1.4.0 ·
Protokoll-Fakten aus der A2A v1.0 Spec, Mai 2026.
Drei Etappen — vom Begriff bis zum Draht
Erst das mentale Modell, dann der lauffähige Code, dann das, was wirklich über das Netzwerk geht. In dieser Reihenfolge wird A2A nachvollziehbar.
1 — Die Grundbegriffe
Fünf Begriffe tragen das ganze Projekt: A2A, Agent Card, JSON-RPC, Protobuf und der Task-Lifecycle. Einmal gelesen — und der Code hört auf, Magie zu sein.
Zu den Begriffen2 — Zwei Agents bauen
Ein Research Agent und ein Writer Agent als A2A-Server, plus ein Orchestrator als Client, der beide über ihre Cards entdeckt und nacheinander beauftragt.
Agents bauen3 — Der Draht
SDK weg, und du siehst die rohe JSON-RPC-Envelope, die Task-Antwort und eine saubere Failure-Trace. Plus der v1.0-Pfad in die Produktion.
Den Draht lesenDie Grundbegriffe
Fünf Begriffe tragen das ganze Projekt. Lies sie einmal, und der Code hört auf, Magie zu sein.
A2A — Agent2Agent
Ein offenes Protokoll (Linux Foundation, beigetragen von Google), mit dem ein Agent einen anderen über HTTP aufruft — über Teams, Frameworks und Maschinen hinweg. REST für Agents: beide Seiten einigen sich auf ein Discovery-Dokument, ein Message-Format und einen Transport.
A2A vs. MCP — sideways vs. down
MCP verbindet einen Agent nach unten mit Tools und Daten — die Gegenseite ist passiv. A2A verbindet einen Agent seitwärts mit anderen Agents — die Gegenseite ist autonom und kann entscheiden, ablehnen oder rückfragen.
Agent Card
Ein kleines JSON-Dokument, das jeder Agent unter dem festen Pfad
/.well-known/agent-card.json veröffentlicht. Es beantwortet
drei Fragen: wer bist du, was kannst du (skills), wie erreiche ich dich
(URL + Protokoll). Discovery heißt: keine hartkodierten Endpoints in der
Aufruf-Logik.
JSON-RPC 2.0
Der jahrzehntealte "ruf eine Funktion auf einem Remote-Server per JSON auf"-Standard.
Ein Request nennt eine method, übergibt params, trägt eine
id; die Response spiegelt die id und liefert ein
result oder einen error. A2A nutzt ihn als einen
Transport; die Methode ist message/send.
Protobuf
Das Schema, nicht zwingend die Wire-Bytes. Jeder Typ — AgentCard,
Message, Task, Artifact — ist ein Protobuf-Objekt. Über JSON-RPC serialisiert
es zu JSON, aber mit Protobuf-Fingerprints: Felder in camelCase
(mediaType), Enums als Großbuchstaben-Strings (ROLE_USER).
Life of a Task
Stateful work durchläuft einen Lifecycle:
submitted → working → completed (oder input-required,
failed, canceled, rejected). Ergebnisse
hängen als Artifacts an. Die Reihenfolge, in der du diese Events
emittierst, ist nicht optional — mehr dazu beim Research Agent.
Architektur-Überblick
Der Orchestrator treibt alles an. Er liest die Card jedes Agents, schickt eine Message und liest das Artifact, das zurückkommt. Die Agents rufen je Groq auf, um ihre eigentliche Denkarbeit zu erledigen.
1 — Research Card holen
Der Orchestrator holt /.well-known/agent-card.json vom Research
Agent, um Adresse und Skills zu lernen. Kein hartkodierter Endpoint.
2 — Topic senden
Er sendet das Thema als Message und erhält eine Zusammenfassung als Task-Artifact zurück.
3 — Writer Card holen
Derselbe Discovery-Schritt, diesmal für den Writer Agent auf einem anderen Port.
4 — Summary senden, Rewrite empfangen
Der Output des ersten Agents wird zum Input des zweiten. Genau dieses Chaining ist die Orchestrierung.
orchestrator → research card → research agent → summary → writer card → writer agent → rewrite. Jeder Agent ruft im Hintergrund das Groq LLM auf; der Orchestrator sieht davon nichts außer dem fertigen Artifact.
Voraussetzungen & Setup
Du brauchst Python 3.13 und uv (nie pip oder
venv direkt), plus einen kostenlosen Groq-API-Key von
console.groq.com (keine Kreditkarte).
uv init a2a-agents --python 3.13
cd a2a-agentsDas wichtigste Detail: die A2A-Server-Routes brauchen Starlette,
das nur als optional extra mitkommt. Installiere über das
[http-server]-Extra, sonst importiert der Server nicht.
uv add "a2a-sdk[http-server]==1.1.0" "groq==1.4.0" uvicorn httpx python-dotenvLege eine .env mit deinem Key an, dazu ein committetes
.env.example-Template. Stelle sicher, dass .gitignore
.env, .venv/ und __pycache__/ ausschließt.
# .env
GROQ_API_KEY=gsk_your_actual_key_hereStep 1 — Verify Your Groq Key
Bevor du Agents baust, bestätige mit einem winzigen async-Call,
dass der Key funktioniert. Das lehrt zugleich die wichtigste Windows-Regel: rufe
load_dotenv() vor der Konstruktion des Groq-Clients auf —
der Client liest GROQ_API_KEY aus der Umgebung in dem Moment, in
dem er erzeugt wird, und uv run lädt .env auf Windows
nicht automatisch.
"""Throwaway check that GROQ_API_KEY works with the async Groq client."""
import asyncio
from dotenv import load_dotenv
from groq import AsyncGroq
load_dotenv() # must run before AsyncGroq()
async def main() -> None:
client = AsyncGroq()
response = await client.chat.completions.create(
model="llama-3.3-70b-versatile",
messages=[{"role": "user", "content": "Reply with exactly: Groq is working"}],
max_tokens=20,
)
print(response.choices[0].message.content)
if __name__ == "__main__":
asyncio.run(main())Jeder Run-Befehl in diesem Projekt ist mit PYTHONPATH=.
präfixiert, damit Package-Imports korrekt auflösen.
PYTHONPATH=. uv run python test_groq.pyEin Authentication-Error heißt, der Key ist falsch.
GROQ_API_KEY not found heißt, deine .env fehlt oder
load_dotenv() wurde übersprungen.
Der Research Agent
Ein A2A-Agent-Server sind drei Konzepte, miteinander verdrahtet: die
AgentCard (die Visitenkarte), der AgentExecutor
(deine Logik — zwei async-Methoden, execute und cancel)
und die Route-Factories, die aus Card + Handler die GET-Discovery-Route
und die POST-Work-Route machen.
Imports, Prompt, Card
Der String "JSONRPC" im Interface ist exakt — er muss
dem Transport-Namen des SDK entsprechen, sonst findet der Client den Endpoint nicht.
"""Research Agent — an A2A server that summarizes a topic with Groq."""
from dotenv import load_dotenv
from groq import AsyncGroq
from starlette.applications import Starlette
from a2a.helpers import new_task_from_user_message, new_text_message, new_text_part
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore, TaskUpdater
from a2a.types import AgentCapabilities, AgentCard, AgentInterface, AgentSkill
load_dotenv()
SYSTEM_PROMPT = (
"You are a research assistant. Given a topic, write a tight, factual "
"summary in 3-4 sentences. No preamble, no bullet points."
)
def build_agent_card() -> AgentCard:
"""Build this agent's public Agent Card (its discovery document)."""
skill = AgentSkill(
id="summarize_topic",
name="Summarize Topic",
description="Generates a concise 3-4 sentence research summary of any topic.",
tags=["research", "summary"],
examples=["Summarize transformers in AI"],
)
return AgentCard(
name="Research Agent",
description="Summarizes any topic into a short research brief using Groq.",
version="1.0.0",
default_input_modes=["text/plain"],
default_output_modes=["text/plain"],
capabilities=AgentCapabilities(streaming=False),
supported_interfaces=[
AgentInterface(url="http://localhost:8001", protocol_binding="JSONRPC"),
],
skills=[skill],
)Der Executor — und die Reihenfolge, die nicht optional ist
Das Task-Objekt muss die Queue vor jedem Status-Event erreichen,
sonst wirft das SDK InvalidAgentResponseError. Und das Artifact muss
vor complete() hinzugefügt werden, weil complete()
einen terminalen Zustand markiert, der weitere Updates sperrt.
1 — Enqueue Task (FIRST)
Hol oder erzeuge den Task und enqueue das Task-Objekt zuerst.
2 — start_work()
TaskUpdater umhüllt die Queue, damit Standard-Lifecycle-Events emittiert werden.
3 — _summarize() → Groq
Die eigentliche Arbeit: Topic rein, Summary raus.
4 — add_artifact(), dann complete()
Erst das Artifact anhängen, dann den terminalen Zustand setzen.
class ResearchAgentExecutor(AgentExecutor):
"""Core logic: read the topic, call Groq, publish a summary artifact."""
def __init__(self) -> None:
self.groq = AsyncGroq() # reads GROQ_API_KEY from the environment
async def _summarize(self, topic: str) -> str:
"""Call Groq and return the summary text."""
response = await self.groq.chat.completions.create(
model="llama-3.3-70b-versatile",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": topic},
],
max_tokens=512,
)
return response.choices[0].message.content
async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
"""Handle one A2A request: topic in, summary artifact out."""
# 1. Get or create the task, and enqueue the Task object FIRST.
task = context.current_task or new_task_from_user_message(context.message)
if not context.current_task:
await event_queue.enqueue_event(task)
# 2. TaskUpdater wraps the queue so we emit standard lifecycle events.
updater = TaskUpdater(event_queue=event_queue, task_id=task.id, context_id=task.context_id)
await updater.start_work(message=new_text_message("Researching the topic..."))
# 3. Do the actual work.
summary = await self._summarize(context.get_user_input())
# 4. Attach the artifact, THEN mark complete.
await updater.add_artifact(parts=[new_text_part(summary, media_type="text/plain")])
await updater.complete(message=new_text_message("Research complete."))
async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
"""No long-running work to interrupt in this project."""
return NoneIn eine App verdrahten
DefaultRequestHandler hält den Executor, einen
In-Memory-Task-Store und die Card. Die zwei Route-Factories erzeugen die GET-
(Discovery) und POST- (Work) Routes, und app ist das Objekt, das
uvicorn importiert.
def build_app() -> Starlette:
"""Wire the executor and card into a Starlette app with A2A routes."""
card = build_agent_card()
handler = DefaultRequestHandler(
agent_executor=ResearchAgentExecutor(),
task_store=InMemoryTaskStore(),
agent_card=card,
)
routes = [
*create_agent_card_routes(card),
*create_jsonrpc_routes(handler, "/"),
]
return Starlette(routes=routes)
app = build_app()Du brauchst zusätzlich eine leere agents/__init__.py, damit
agents.research_agent als Package importiert. Starte den Agent,
dann fetch seine Card, um zu bestätigen, dass er ausliefert.
PYTHONPATH=. uv run uvicorn agents.research_agent:app --port 8001
# in a second terminal:
curl http://localhost:8001/.well-known/agent-card.jsonDu solltest JSON erhalten, das "name":"Research Agent",
den summarize_topic-Skill und "url":"http://localhost:8001"
enthält.
Der Writer Agent
Der Writer Agent ist strukturell identisch zum Research Agent. Genau das ist die Lektion: das Plumbing (Executor-Lifecycle, Routes, Card-Struktur) ist Boilerplate, das du kopierst, und der Wert steckt allein im Prompt und im beworbenen Skill. Exakt vier Dinge ändern sich.
1 — SYSTEM_PROMPT
rewrite statt summarize.
2 — AgentSkill id
rewrite_simple statt summarize_topic.
3 — Card-Identität
Writer Agent statt Research Agent.
4 — Port
8002 statt 8001.
Kopiere research_agent.py nach
agents/writer_agent.py, benenne _summarize in
_rewrite um, ändere die Status-Strings, dann tausche Prompt und Card.
SYSTEM_PROMPT = (
"You are an explainer. Rewrite the given text as a punchy, beginner-friendly "
"explanation a smart 15-year-old would enjoy. Keep it to 3-4 short sentences. "
"Use plain words and one concrete analogy. No preamble."
)
skill = AgentSkill(
id="rewrite_simple",
name="Rewrite Simply",
description="Rewrites any text into a punchy, beginner-friendly explanation.",
tags=["writing", "explainer"],
examples=["Rewrite this dense paragraph so a beginner gets it"],
)
return AgentCard(
name="Writer Agent",
description="Rewrites text into a punchy, beginner-friendly explanation using Groq.",
version="1.0.0",
default_input_modes=["text/plain"],
default_output_modes=["text/plain"],
capabilities=AgentCapabilities(streaming=False),
supported_interfaces=[
AgentInterface(url="http://localhost:8002", protocol_binding="JSONRPC"),
],
skills=[skill],
)PYTHONPATH=. uv run uvicorn agents.writer_agent:app --port 8002
curl http://localhost:8002/.well-known/agent-card.jsonDer Orchestrator
Der Orchestrator ist ein Client, kein Server. Er weiß nichts über die Agents außer ihren Base-URLs. Für jeden Agent resolved er die Card, baut aus ihr einen Client, sendet eine Message und liest das Artifact vom zurückgegebenen Task. Der Output des ersten Agents wird zum Input des zweiten.
Die A2A-Client-API verschob sich über die 1.x-Linie. Das Pattern unten folgt
dem offiziellen A2A v1.0 Client-Tutorial (A2ACardResolver +
create_client + send_message als async-Iterator).
Auf a2a-sdk==1.1.0 kann das Äquivalent
ClientFactory(ClientConfig(...)).create(card) oder
.create_from_url(url) sein. Bestätige die exakten Import-Namen
gegen deine installierte Version mit
uv run python -c "import a2a.client as c; print(dir(c))", bevor
du das als final behandelst.
"""Orchestrator — hires the Research Agent, then the Writer Agent, over A2A."""
import asyncio
import httpx
from a2a.client import A2ACardResolver, create_client
from a2a.client.client import ClientConfig
from a2a.helpers import new_text_message
from a2a.types import Role, SendMessageRequest, Task
RESEARCH_URL = "http://localhost:8001"
WRITER_URL = "http://localhost:8002"
def artifact_text(task: Task) -> str:
"""Pull the text out of the first artifact of a completed Task."""
for artifact in task.artifacts or []:
for part in artifact.parts:
if getattr(part, "text", None):
return part.text
return ""
async def call_agent(http: httpx.AsyncClient, base_url: str, text: str) -> str:
# 1. Resolve the card from /.well-known/agent-card.json — discovery, no hardcoding.
resolver = A2ACardResolver(httpx_client=http, base_url=base_url)
card = await resolver.get_agent_card()
# 2. Build a client from the card. Transport is read off the card, not assumed.
client = await create_client(agent=card, client_config=ClientConfig(streaming=False))
# 3. Send one message; these agents reply with a single completed Task.
request = SendMessageRequest(message=new_text_message(text, role=Role.ROLE_USER))
result = None
async for event in client.send_message(request):
result = event # last event is the final Task
return artifact_text(result)
async def main() -> None:
topic = "the transformer architecture in AI"
async with httpx.AsyncClient(timeout=60) as http:
summary = await call_agent(http, RESEARCH_URL, topic)
print("RESEARCH SUMMARY:\n", summary, "\n")
rewrite = await call_agent(http, WRITER_URL, summary)
print("WRITER REWRITE:\n", rewrite)
if __name__ == "__main__":
asyncio.run(main())Step 5 — Run It End to End
Drei Prozesse. Zwei Agent-Server bleiben laufen; der Orchestrator ist ein One-Shot-Skript. Das Topic fließt durch beide Agents und kommt als umgeschriebene Erklärung heraus.
# terminal 1
PYTHONPATH=. uv run uvicorn agents.research_agent:app --port 8001
# terminal 2
PYTHONPATH=. uv run uvicorn agents.writer_agent:app --port 8002
# terminal 3
PYTHONPATH=. uv run python orchestrator.pytopic → Research Agent :8001 → summary → Writer Agent :8002 → finale Erklärung. Der Orchestrator kettet beide, ohne je einen Endpoint hartzukodieren.
Den Draht verstehen
SDK weg — und das ist es, was über das Netzwerk geht, wenn der Orchestrator ein Topic sendet: ein gewöhnlicher HTTP POST, der eine JSON-RPC-2.0-Envelope trägt. Beachte die Protobuf-Fingerprints: camelCase-Felder, Enum-Strings in Großbuchstaben.
{
"jsonrpc": "2.0",
"id": "req-1",
"method": "message/send",
"params": {
"message": {
"role": "ROLE_USER",
"parts": [{ "text": "the transformer architecture in AI" }],
"messageId": "a1b2c3d4"
}
}
}{
"jsonrpc": "2.0",
"id": "req-1",
"result": {
"id": "task-9f8e",
"contextId": "ctx-7a6b",
"status": { "state": "TASK_STATE_COMPLETED" },
"artifacts": [{
"artifactId": "art-1",
"name": "result",
"parts": [{ "text": "Transformers are a neural-network design that..." }]
}]
}
}Die id auf der Response spiegelt die des Requests. Die
eigentliche Arbeit lebt unter result; ein error-Objekt
würde es bei Fehler ersetzen. Alles andere ist das Protobuf-Schema im JSON-Gewand.
Eine Failure lesen (falscher Port)
Zeig den Orchestrator auf einen Port, auf dem kein Agent lauscht, und die Failure ist präzise und früh — sie passiert bei der Discovery, bevor irgendeine Message gesendet wird. Nichts erreicht das LLM.
httpx.ConnectError: All connection attempts failed
... raised during A2ACardResolver.get_agent_card()
a2a.client.AgentCardResolutionError: agent card could not be resolved
at http://localhost:8003/.well-known/agent-card.jsonDer Fehler nennt den Discovery-Schritt, nicht den Message-Schritt — Beleg, dass A2A-Clients bei einer schlechten Adresse fail-fast sind, statt Arbeit ins Leere zu schicken. Fix die URL im Orchestrator und starte neu; kein Agent-Code ändert sich.
Was v1.0 für die Produktion ändert
Dieses Tutorial läuft auf localhost ohne Auth — richtig zum Lernen,
nicht für Deployment. A2A v1.0 (Anfang 2026) härtete die Spec entlang vier Achsen,
die direkt auf Enterprise-Anforderungen abbilden.
Signed Agent Cards
Eine kryptografische Signatur lässt einen empfangenden Agent verifizieren, dass die Card vom Domain-Inhaber ausgestellt wurde — die Verteidigung gegen Card-Forgery-Angriffe, die Agents auf einen bösartigen Endpoint umleiten. Dieses Trust-Modell macht dezentrale Discovery überhaupt erst tragfähig.
Multi-Tenancy
Ein einzelner Endpoint kann mehrere Agents hosten, sodass ein SaaS-Provider pro Tenant einen anderen Agent ausliefern kann, ohne pro Kunde neue Infrastruktur hochzuziehen.
Multi-Protocol Bindings
Derselbe logische Agent kann über JSON-RPC und gRPC exponiert werden — JSON-RPC für Reichweite und Debugbarkeit, gRPC für Durchsatz, ohne den Agent zu forken.
Version Negotiation
Spec-Level-Garantien für rückwärtskompatible Migration (v0.3 → v1.0), damit eine Flotte von Agents inkrementell upgraden kann statt in einem riskanten Big-Bang-Cutover.
Zum Ein-Jahres-Mark (April 2026) überschritt A2A 150+ unterstützende Organisationen, mit nativer Integration in Azure AI Foundry / Copilot Studio und Amazon Bedrock AgentCore. Die AP2-Extension ergänzt agentengetriebene Zahlungen. Für Inter-Agent-Integration gibt es heute praktisch keinen konkurrierenden Standard.
Was du als Nächstes baust
Swap in streaming
Setze streaming=True auf Capabilities und Client-Config, um
status_update- und artifact_update-Chunks zu erhalten,
sobald sie ankommen, statt eines finalen Task.
Add a third agent
Ein Fact-Checker zwischen Research und Writer. Die Orchestrator-Änderung ist eine
weitere call_agent-Zeile; die Agents bleiben unangetastet. Das ist
der Payoff Discovery-basierter Verdrahtung.
Persist tasks
Ersetze InMemoryTaskStore durch einen durablen Store, damit Tasks
Restarts überleben und per id gepollt werden können.
Harden for deploy
Signiere die Agent Cards, setze Auth vor die JSON-RPC-Route und geh weg von
localhost hinter echtes DNS/TLS.
Agent-Code verifiziert gegen a2a-sdk 1.1.0 · Client-Pattern & v1.0-Fakten aus a2a-protocol.org & Linux Foundation, Mai 2026.