The Agent2Agent (A2A) Protocol: Learning v1.0 by Building a Small Agent

By · · AI Engineering

There's a reason I held off on writing about A2A until now.

I'd been watching the Agent2Agent (A2A) Protocol since it first showed up, originally out of Google and eventually handed off to the Linux Foundation. The idea was always right: a neutral wire protocol so agents built on different frameworks could talk to each other. But every time I sat down to try it, the spec was still moving. v0.1, v0.2, v0.3, breaking changes between minor releases. Agent Cards weren't cryptographically signed, so discovering a new agent was basically a handshake on faith. One endpoint, one agent, which ruled out every SaaS shape I cared about. JSON-RPC was the only binding, which locked out teams standardized on gRPC. And the one that really kept me out: no version negotiation, so any upgrade meant a flag-day cutover across every agent in a deployment.

Promising protocol, but not one you could bet on yet. So I waited.

Then on March 12, 2026, v1.0 shipped. The four headline changes happen to be the four things that had been keeping me out:

The v1.0 announcement keeps using the word maturity rather than reinvention, and I think that's the right word. This isn't a redesign. It's the point where I stopped flinching every time a new version dropped.

So I built a small agent on top of the official a2a-sdk to actually use the thing.

One caveat up front: the Python SDK is still on 0.3.x. It accepts the v1.0 card fields (my sample declares protocol_version="1.0"), but the v1.0-specific features -- signed cards, the gRPC binding, version negotiation -- aren't wired up in the SDK yet. What works today is the core flow: discovery, message/send, streaming, cooperative cancellation, input-required multi-turn, and push-notification webhooks. That's most of what you need to write a working agent.

Repo: github.com/encryptedtouhid/a2a_protocol_sample -- a small A2A reference agent in Python, built on the official a2a-sdk.

This post is what I wish someone had handed me when I started.

So what is A2A, really?

The elevator pitch: A2A is to agent-to-agent communication what HTTP is to browser-to-server communication. A neutral wire protocol that two systems, built by teams who don't know each other and don't share a stack, can both agree to speak.

That's the whole value proposition.

Underneath, it's JSON-RPC 2.0 with a Server-Sent Events layer for streaming and optional webhooks for push notifications. Nothing exotic on the wire -- you can pretty-print any message and read it with your eyes.

The five objects you need to get comfortable with:

The task state machine is small and pleasant to look at:

submitted -> working -> completed
                     \-> canceled / failed / rejected
                     \-> input-required (resumable)
                     \-> auth-required (resumable)

Notice the two resumable states. That's where A2A gets more interesting than a plain request/response protocol, and I'll get to that.

Only five RPC methods actually matter

The spec has a handful of methods. If you squint, five of them are doing all the real work:

Method What it does
message/send Send a message, get a task back. Plain request/response.
message/stream Same idea, but the response is an SSE stream of events.
tasks/get Fetch a task by id. Optionally trim the history.
tasks/cancel Cooperatively ask a task to stop.
tasks/pushNotificationConfig/* Register a webhook so the agent can call you.

The rest -- tasks/resubscribe and the authenticated extended card -- is quality of life. Nice to have, not what makes A2A work.

Inside the sample repo

The actual layout is much smaller than you'd guess:

src/a2a_sample/
  __init__.py    # re-exports, 3 lines
  auth.py        # bearer-token Starlette middleware, 31 lines
  executor.py    # AgentExecutor: routes the first word to a skill, 61 lines
  server.py      # agent card + A2AStarletteApplication wiring, 143 lines
  skills.py      # echo / summarize / count / form / debug, 118 lines

demos/
  run_server.py
  webhook_receiver.py
  _common.py
  01_discovery.py                # fetch the well-known card
  02_send_message.py             # plain message/send
  03_streaming.py                # message/stream over SSE
  04_multiturn_input_required.py # input-required, multi-turn
  05_cancel.py                   # tasks/cancel mid-stream
  06_push_notifications.py       # register a webhook and watch it fire
  07_extended_card.py            # authenticated extended card

About 356 lines of application code across five files. Three dependencies:

a2a-sdk[http-server]>=0.3.26
uvicorn[standard]>=0.30
httpx>=0.28

The SDK does most of the heavy lifting. Once you see what's already in the box, the protocol gets a lot easier to reason about:

What's left for you to write:

That's the entire pattern. The SDK handles protocol correctness; you handle agent behavior.

The five skills

The sample exposes four public skills and one auth-gated skill, and each one pokes a different corner of the protocol:

The dispatcher in executor.py picks a skill based on the first word of the user's first message:

initial_text = skills.initial_user_text(task, context.message).strip()
head = initial_text.split(None, 1)[0].lower() if initial_text else ""

try:
    if head == "summarize":
        await skills.skill_summarize(initial_text, updater)
    elif head == "count":
        await skills.skill_count(initial_text, updater)
    elif head == "form":
        await skills.skill_form(updater, task, context.message)
    elif head == "debug":
        await skills.skill_debug(updater)
    else:
        await skills.skill_echo(initial_text, updater)

A str.split() and an if/elif. For a reference sample, that's the right amount of code.

Discovery: the well-known agent card

The first thing any A2A client does is ask the agent to introduce itself. A2A borrows RFC 8615's well-known URI -- you fetch a JSON document from a fixed path and read off everything you need. The SDK's A2ACardResolver does the work:

async def fetch_public_card(http: httpx.AsyncClient) -> AgentCard:
    return await A2ACardResolver(http, base_url=BASE_URL).get_agent_card()

Run demo 01 and you get back the agent's skills, its preferred transport, its security schemes, whether it supports streaming, all of it. No docs to read, no Slack message to ask "hey what's the API look like" -- the agent tells you itself.

It's a boring design choice, which is usually a good sign in protocol work.

Sending a message

Everything JSON-RPC goes through the SDK's client, built from the agent card plus an auth interceptor:

def build_client(http, card, *, streaming: bool = False) -> Client:
    factory = ClientFactory(
        ClientConfig(
            httpx_client=http,
            streaming=streaming,
            supported_transports=[TransportProtocol.jsonrpc],
        )
    )
    return factory.create(
        card,
        interceptors=[AuthInterceptor(StaticCredentialService(BEARER_TOKEN))],
    )

Sending a message is a one-liner on top:

async for event in client.send_message(_user("echo hello A2A")):
    if isinstance(event, Message):
        continue
    task, update = event
    if update is None:
        print(f"  task id={task.id} state={task.status.state.value}")

The task comes back already in the completed state because echo is instant. For anything slower, you'd want streaming.

Where streaming starts to earn its keep

Request/response is fine for echo. It's not fine for an agent that spends fifteen seconds thinking and produces output incrementally. You don't want to hold an HTTP connection open for fifteen seconds and dump everything at the end -- the user would assume it crashed.

A2A's answer is message/stream. Same message shape, but the response is an SSE stream of JSON-RPC envelopes. The client side dispatches on event type:

async for event in client.send_message(msg):
    if isinstance(event, Message):
        continue
    task, update = event
    if isinstance(update, TaskArtifactUpdateEvent):
        chunk = update.artifact.parts[0].root
        text = getattr(chunk, "text", "")
        print(f"[artifact] append={update.append} last={update.last_chunk} chunk={text!r}")
    elif isinstance(update, TaskStatusUpdateEvent):
        print(f"[status] state={update.status.state.value} final={update.final}")

The count skill on the server side emits each number as an appended chunk of the same artifact. append=True on chunks 2 through N, last_chunk=True on the last one:

async def skill_count(text: str, updater: TaskUpdater) -> None:
    target = max(1, min(int(match.group()) if match else 5, 100))
    artifact_id = f"count-{updater.task_id}"
    for i in range(1, target + 1):
        await updater.add_artifact(
            [Part(root=TextPart(text=f"{i}\n"))],
            artifact_id=artifact_id,
            name="count",
            append=i > 1,
            last_chunk=i == target,
        )
        await asyncio.sleep(0.3)
    await updater.complete()

That append=True, last_chunk=False pattern is how A2A says "glue these together until you see last_chunk=True." Same mental model as HTTP chunked transfer, just lifted up to the agent's output instead of the HTTP body.

First time I ran this, I left it running and just watched the chunks arrive. There's something satisfying about watching a protocol do its thing in real time.

Cancellation done right: cooperative, not forced

This one I actually have opinions about.

A2A cancellation is cooperative. When you call tasks/cancel, the server doesn't hard-kill a coroutine -- the SDK raises asyncio.CancelledError inside the running skill, and the executor is expected to emit a terminal canceled status and return. It sounds like more work than a hard kill, but it's the right call. Forced termination is how you end up with half-written rows, partial files, and state that corrupts the next run.

The executor handles it with a try/except around the dispatch:

try:
    if head == "count":
        await skills.skill_count(initial_text, updater)
    # ... other skills
except asyncio.CancelledError:
    await updater.update_status(TaskState.canceled, final=True)
    raise

The cancel demo fires the cancel mid-stream:

if isinstance(update, TaskArtifactUpdateEvent):
    text = update.artifact.parts[0].root.text.strip()
    print(f"[chunk]   {text}")
    if not canceled and text == "3":
        await client.cancel_task(TaskIdParams(id=task_id))
        canceled = True

Run demo 05 and you see it: three chunks arrive, the cancel request goes out, one or two more chunks land in flight, then the stream closes with state=canceled final=True. No orphaned state.

The multi-turn trick: input-required

Okay, this is the feature I didn't know I wanted until I used it.

In A2A, a task can pause and wait for more input without losing its place. The agent transitions to input-required, the client comes back with another message carrying the same task_id and context_id, and the task picks up where it left off. Same task id. Same artifacts. Same context. Just... paused.

The form skill demonstrates it. Three turns: ask for name, ask for email, complete with a contact record.

task = await _send(client, _user("form"))
print(f"[turn 1] state={task.status.state.value}")  # input-required

task = await _send(client, _user("Ada Lovelace", task.id, task.context_id))
print(f"[turn 2] state={task.status.state.value}")  # input-required

task = await _send(client, _user("[email protected]", task.id, task.context_id))
print(f"[turn 3] state={task.status.state.value}")  # completed

Where the skill stashes the collected name between turns is the nice trick. It doesn't need a database; it writes the data into the agent message's metadata field on turn 2, then walks the task history on turn 3 to read it back:

if prior_agent_turns == 1:
    await updater.requires_input(
        updater.new_agent_message(
            [Part(root=TextPart(text=FORM_QUESTIONS[1]))],
            metadata={"name": incoming_text},
        ),
        final=True,
    )
    return

# later, on turn 3:
for m in task.history or []:
    if m.role == Role.agent and m.metadata and "name" in m.metadata:
        name = str(m.metadata["name"])
        break

The form demo looks trivial, but the same state machine handles any "agent waits for something external" case -- a human approval, an auth challenge (via the defined-but-not-demoed auth-required state), a slow third-party service. One mental model covers all of them.

This is the bit I keep coming back to when I think about why A2A exists. Anyone can design a request/response protocol. Designing one that pauses cleanly and resumes without losing its mind -- that's the hard part, and A2A got it right.

Push notifications for the really long-running stuff

SSE is wonderful when the client can stay connected. But sometimes the task runs for an hour and the client is a mobile app that gets backgrounded, or a serverless function that already returned, or just a process that might die and come back. For that, A2A lets you register a webhook:

config = TaskPushNotificationConfig(
    task_id=task_id,
    push_notification_config=PushNotificationConfig(
        url=WEBHOOK_URL,
        token=SHARED_SECRET,
    ),
)
result = await client.set_task_callback(config)

The SDK's BasePushNotificationSender POSTs every event to your webhook with an X-A2A-Notification-Token header carrying your shared secret. You can register multiple webhooks per task, list them, delete them -- standard CRUD.

Server-side wiring is about four lines:

push_store = InMemoryPushNotificationConfigStore()
push_sender = BasePushNotificationSender(
    httpx.AsyncClient(timeout=10.0),
    config_store=push_store,
)

For demo 06 you have to start a little webhook receiver in a third terminal. Watching the events land in the receiver while the task is still running is when the async design actually clicks.

One thing to call out: v1.0 also defines JWT-signed push payloads via the a2a-sdk[encryption] extra, but the sample uses only the shared-secret token. Fine for a demo on localhost, not fine for production cross-org setups. Don't ship this exact pattern past your own infrastructure without upgrading the verification.

Auth: a tiny Starlette middleware

The sample accepts one bearer token, demo-secret-token, declared in the card's securitySchemes and required on every non-discovery path. The middleware fits in about a dozen lines:

class BearerAuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        if request.url.path in PUBLIC_PATHS:
            return await call_next(request)
        header = request.headers.get("authorization", "")
        scheme, _, token = header.partition(" ")
        if scheme.lower() != "bearer" or token != BEARER_TOKEN:
            return JSONResponse(
                {"error": "invalid or missing bearer token"},
                status_code=401,
            )
        return await call_next(request)

Real production agents would wire in OAuth2, OIDC, or mTLS -- the card supports declaring any of those. This is just the simplest thing that works so the rest of the demos have something to authenticate against.

What this sample deliberately doesn't do

v1.0 shipped, but not all of it is in the SDK yet, and the sample doesn't reach past what the SDK does. So here's what's missing:

The core message flow is solid. The security and cross-stack features still need the ecosystem to catch up.

Running it yourself

Clone, set up, run the server in one terminal, fire demos from another:

git clone https://github.com/encryptedtouhid/a2a_protocol_sample.git
cd a2a_protocol_sample
./run.sh setup

# terminal 1
./run.sh server

# terminal 2
./run.sh demo 01   # fetch the agent card
./run.sh demo 03   # watch the streaming counter
./run.sh demo 04   # play with input-required

For push notifications (demo 06) you also need a webhook receiver in a third terminal:

./run.sh webhook

The run.sh script handles the venv and works on Linux, macOS, and Windows (Git Bash / WSL). ./run.sh list shows every demo. Go through them in order on the first pass; each one introduces one new concept and builds on the last.

Why I care about this one

We spent a decade learning that microservices only work when services agree on a wire protocol. Nobody ships a new microservice in 2026 and says "let's invent our own transport." Agents are walking the same path, just five years behind.

A2A is the lingua franca that lets a LangGraph orchestrator delegate to a Semantic Kernel agent, which hands off to a CrewAI team, without any of them importing each other's SDKs. No adapters. No shared deployment. No "we need to agree on Python versions" meeting. Just JSON over HTTP with shapes everyone can parse.

And it pairs neatly with MCP:

They're both still young, and they'll both change. The hard part of any protocol ecosystem is getting enough people to agree on one, and right now that's A2A.

If you want to go deeper

Honestly, the fastest way to learn a protocol is to build something on top of it. The repo's ~360 lines are a reasonable starting sample if you'd rather read first and build second.

Happy to hear what you think. If something in the code confuses you, open an issue -- that probably means I need to either fix the code or write better comments, and either way I'd want to know.