Step-by-Step · Wire-Level · A2A v1.0

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.

2
Agents, die sich nie gesehen haben
1
Orchestrator, der beide beauftragt
JSON-RPC
Transport über plain HTTP POST
v1.0
Protocol, Stand 2026
AT
Aria Tan
Agent Architect · Multi-Agent Systems

"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.

Lernpfad

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 Begriffen

2 — 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 bauen

3 — 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 lesen
Concepts · 00

Die 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.

Architecture · 01

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.

Der Datenfluss in einem Satz.

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.

Setup · 02

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

terminal · scaffold
uv init a2a-agents --python 3.13
cd a2a-agents

Das 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.

terminal · dependencies
uv add "a2a-sdk[http-server]==1.1.0" "groq==1.4.0" uvicorn httpx python-dotenv

Lege eine .env mit deinem Key an, dazu ein committetes .env.example-Template. Stelle sicher, dass .gitignore .env, .venv/ und __pycache__/ ausschließt.

.env
# .env
GROQ_API_KEY=gsk_your_actual_key_here

Step 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.

test_groq.py
"""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.

terminal
PYTHONPATH=. uv run python test_groq.py
Failure modes.

Ein Authentication-Error heißt, der Key ist falsch. GROQ_API_KEY not found heißt, deine .env fehlt oder load_dotenv() wurde übersprungen.

Step 2 · 04

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.

agents/research_agent.py · top
"""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.

agents/research_agent.py · executor
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 None

In 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.

agents/research_agent.py · app
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()
Package note.

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.

terminal · run + verify
PYTHONPATH=. uv run uvicorn agents.research_agent:app --port 8001
# in a second terminal:
curl http://localhost:8001/.well-known/agent-card.json

Du solltest JSON erhalten, das "name":"Research Agent", den summarize_topic-Skill und "url":"http://localhost:8001" enthält.

Step 3 · 05

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.

agents/writer_agent.py · the four changes
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],
)
terminal · run + verify (second terminal)
PYTHONPATH=. uv run uvicorn agents.writer_agent:app --port 8002
curl http://localhost:8002/.well-known/agent-card.json
Step 4 · 06

Der 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.

SDK surface — verify before running.

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.py
"""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.

three terminals
# 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.py
Die Pipeline in einem Bild.

topic → Research Agent :8001 → summary → Writer Agent :8002 → finale Erklärung. Der Orchestrator kettet beide, ohne je einen Endpoint hartzukodieren.

The Wire · 08

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.

request · POST http://localhost:8001/
{
  "jsonrpc": "2.0",
  "id": "req-1",
  "method": "message/send",
  "params": {
    "message": {
      "role": "ROLE_USER",
      "parts": [{ "text": "the transformer architecture in AI" }],
      "messageId": "a1b2c3d4"
    }
  }
}
response · the completed Task
{
  "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.

traceback · abridged
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.json
So liest du es.

Der 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.

v1.0 & Production · 10

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.

Adoption signal.

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.