A2A Protocol PoC: Build an Agent Server in Python

At Google Cloud Next 2025, Google introduced the Agent2Agent (A2A) protocol and donated it to the Linux Foundation less than three months later. By April 2026 — one year in — it has 150+ supporting organizations, 22,000+ GitHub stars, and active production deployments across finance, supply chain, and IT operations. Every major agentic framework (LangGraph, CrewAI, LlamaIndex Agents, Semantic Kernel, AutoGen) now ships native A2A support.

Effloow Lab ran a sandbox PoC with the official a2a-sdk 1.0.2 and the community python-a2a 0.5.10 library. This article shows what we found: working code, a non-obvious routing bug in the official SDK, and a workaround that gets a full Agent Card discovery + task round-trip running in under 30 lines of Python.

Lab notes are in data/lab-runs/a2a-poc.md.

Why A2A Matters: The Agent Interoperability Problem

Before A2A, if you built a LangGraph agent and your colleague built a CrewAI agent, there was no standard way for them to collaborate. You had to write custom glue code, agree on a message format, and hardcode each endpoint.

A2A solves this with three primitives:

1. Agent Card — a JSON document every A2A server publishes at /.well-known/agent-card.json. It describes what the agent can do, what communication modes it supports, and how to reach it.
2. Tasks — the unit of work. A client sends a message, the server creates a task, and the task progresses through a defined lifecycle.
3. Messages — the actual content exchanged within a task, made of typed Parts (text, file, structured data).

This is deliberately transport-agnostic. The spec uses HTTP + JSON-RPC 2.0 as its wire format, but nothing stops future bindings over gRPC or WebSocket.

A2A vs MCP: Different Layers, Not Competitors

Before diving into code, it helps to place A2A correctly in the stack.

In practice, a production agent system uses both. An orchestrator discovers sub-agents via A2A and discovers tools via MCP. The two protocols are complementary, not competing.

A2A Core Concepts

Agent Card

Every A2A server publishes a JSON document that other agents can fetch without any prior coordination:

{"name":"Echo Agent","description":"Echoes messages back.","version":"1.0.0","supportedInterfaces":[{"url":"http://localhost:9999/","protocolBinding":0}],"skills":[{"id":"echo","name":"Echo","description":"Echoes user text back with an [Echo] prefix.","tags":["echo","demo"],"examples":["Hello!","What is A2A?"]}],"capabilities":{"streaming":false,"pushNotifications":false}}

Skills function like an OpenAPI spec for agent capabilities — they tell downstream orchestrators what tasks are worth delegating.

Task Lifecycle

When a client calls /message:send, the server creates a Task. That task moves through defined states:

submitted → working → completed
                   ↘ failed
                   ↘ input-required  (agent needs more info)
                   ↘ canceled

The server emits TaskStatusUpdateEvent and TaskArtifactUpdateEvent events as the task progresses. For simple synchronous agents, you'll emit a single COMPLETED event with the reply message.

PoC: Echo Agent Server

Installation

pip install a2a-sdk uvicorn sse-starlette starlette  # official SDK
# OR
pip install python-a2a  # community library (simpler API)

Official SDK (a2a-sdk 1.0.2)

The official SDK uses protobuf types and a Starlette ASGI application. Here is a minimal echo server:

import asyncio, uuid, uvicorn
from starlette.applications import Starlette

from a2a.types import (
    AgentCard, AgentInterface,
    Message, Part, Role,
    TaskStatusUpdateEvent, TaskState,
)
from a2a.server.request_handlers import DefaultRequestHandlerV2
from a2a.server.request_handlers.default_request_handler_v2 import (
    AgentExecutor, RequestContext
)
from a2a.server.events.event_queue import EventQueue
from a2a.server.tasks.inmemory_task_store import InMemoryTaskStore
from a2a.server.events.in_memory_queue_manager import InMemoryQueueManager
from a2a.server.routes import create_rest_routes, create_agent_card_routes


class EchoAgentExecutor(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        user_text = ""
        if context.message and context.message.parts:
            for part in context.message.parts:
                if part.HasField("text"):
                    user_text = part.text
                    break

        reply = Message()
        reply.message_id = str(uuid.uuid4())
        reply.role = Role.ROLE_AGENT
        reply.parts.add().text = f"[Echo] {user_text or '(empty)'}"

        event = TaskStatusUpdateEvent()
        event.task_id = context.task_id or ""
        event.status.state = TaskState.TASK_STATE_COMPLETED
        event.status.message.CopyFrom(reply)
        await event_queue.enqueue_event(event)

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        pass


def build_agent_card(host="localhost", port=9999) -> AgentCard:
    card = AgentCard()
    card.name = "Echo Agent"
    card.description = "A minimal A2A PoC."
    card.version = "1.0.0"
    card.default_input_modes.extend(["text/plain"])
    card.default_output_modes.extend(["text/plain"])
    iface = card.supported_interfaces.add()
    iface.url = f"http://{host}:{port}/"
    skill = card.skills.add()
    skill.id = "echo"
    skill.name = "Echo"
    skill.tags.extend(["demo"])
    return card


def create_app():
    card = build_agent_card()
    handler = DefaultRequestHandlerV2(
        agent_executor=EchoAgentExecutor(),
        task_store=InMemoryTaskStore(),
        agent_card=card,
        queue_manager=InMemoryQueueManager(),
    )
    # IMPORTANT: agent card routes must come BEFORE create_rest_routes.
    # create_rest_routes registers a /{tenant} catch-all mount that would
    # shadow /.well-known/agent-card.json if registered first.
    routes = create_agent_card_routes(card) + create_rest_routes(handler)
    return Starlette(routes=routes)


if __name__ == "__main__":
    uvicorn.run(create_app(), host="localhost", port=9999)

Run it:

python echo_server.py
# GET http://localhost:9999/.well-known/agent-card.json  → 200 OK

Known issue (a2a-sdk 1.0.2 + protobuf ≥ 7.x): The Agent Card endpoint works correctly, but /message:send returns a 500 error. The validate_proto_required_fields() function inside the SDK calls field.label on a google._upb._message.FieldDescriptor object, which does not exist in protobuf 7.x's upb C extension. This is an upstream SDK bug. Track it on the a2aproject/a2a-python issues. Workaround: pin protobuf<5.0, or use the community library below.

Community Library (python-a2a 0.5.10) — Fully Working

python-a2a implements A2A v0.3.0 (the spec version before the 1.0 stable release). It is Flask-based and has a simpler API. Effloow Lab confirmed a full end-to-end echo round-trip with this library:

from python_a2a import A2AServer, AgentCard, AgentSkill, run_server
from python_a2a.models import Message, TextContent, MessageRole


class EchoServer(A2AServer):
    def __init__(self):
        card = AgentCard(
            name="Echo Agent",
            description="A minimal A2A PoC that echoes messages back.",
            url="http://localhost:9998",
            version="1.0.0",
            skills=[
                AgentSkill(
                    name="Echo",
                    description="Echoes user text back with an [Echo] prefix.",
                    examples=["Hello!", "What is A2A?"],
                )
            ],
        )
        super().__init__(agent_card=card)

    def handle_message(self, message: Message) -> Message:
        if hasattr(message.content, "text"):
            reply_text = f"[Echo] {message.content.text}"
        else:
            reply_text = "[Echo] (unsupported content type)"
        return Message(
            content=TextContent(text=reply_text),
            role=MessageRole.AGENT,
        )


if __name__ == "__main__":
    run_server(EchoServer(), host="localhost", port=9998)

Start this server, then in another terminal:

python -c "
from python_a2a import A2AClient
from python_a2a.models import Message, TextContent, MessageRole
client = A2AClient('http://localhost:9998')
print(client.agent_card.name)  # 'Echo Agent'
msg = Message(content=TextContent(text='Hello, A2A!'), role=MessageRole.USER)
r = client.send_message(msg)
print(r.content.text)  # '[Echo] Hello, A2A!'
"

PoC: Discovery Client

A proper A2A client separates the discovery step (fetching the Agent Card) from the call step. This lets orchestrators build routing tables at startup and reuse them across requests:

from python_a2a import A2AClient
from python_a2a.models import Message, TextContent, MessageRole


def discover_and_chat(server_url: str, text: str) -> str:
    client = A2AClient(server_url)

    # Step 1: Discover agent capabilities
    card = client.agent_card
    print(f"Agent: {card.name}")
    for skill in card.skills or []:
        print(f"  Skill: {skill.name} — {skill.description}")

    # Step 2: Delegate a task
    msg = Message(content=TextContent(text=text), role=MessageRole.USER)
    response = client.send_message(msg)
    return response.content.text


result = discover_and_chat("http://localhost:9998", "What is the A2A protocol?")
print(result)
# → [Echo] What is the A2A protocol?

Effloow Lab ran this exact code against the echo server and got the expected output:

[Discovered] Echo Agent: A minimal A2A PoC that echoes messages back.
  Skill: Echo
[Agent reply] [Echo] What is the A2A protocol?

What You Can Build With This Pattern

The echo server is trivial, but the pattern scales directly:

Specialist subagent — replace handle_message with a call to an LLM with a specialized system prompt (e.g., a code review agent, a SQL generation agent). The orchestrator routes tasks based on skills declared in the Agent Card without knowing anything about the agent's internals.

Framework adapter — wrap an existing LangGraph graph or CrewAI crew as an A2A server. Callers don't need to know the underlying framework.

Multi-agent pipeline — chain agents where each one's output becomes the next one's input message. Because every agent exposes a standard interface, you can swap implementations without rewriting the pipeline.

Common Mistakes

1. Registering routes in the wrong order (official SDK)

create_rest_routes() registers a /{tenant} catch-all Starlette mount. If you append create_agent_card_routes() after it, any GET request to /.well-known/agent-card.json matches /{tenant} first and returns 404. Always put the agent card routes first:

# Wrong
routes = create_rest_routes(handler) + create_agent_card_routes(card)

# Correct
routes = create_agent_card_routes(card) + create_rest_routes(handler)

2. Confusing the two Agent Card endpoints

The official SDK serves the card at /.well-known/agent-card.json (A2A v1.0 spec). The community library serves it at /.well-known/agent.json (A2A v0.3 spec). When mixing libraries or testing with curl, use the path that matches the spec version your server implements.

3. Sending role: "user" instead of role: "ROLE_USER" to the REST API

The official SDK's REST endpoint expects protobuf enum string names (ROLE_USER, ROLE_AGENT), not lowercase values. Sending "user" returns a 400 parse error.

4. Forgetting to emit a terminal event

If your AgentExecutor.execute() returns without enqueuing a TASK_STATE_COMPLETED, TASK_STATE_FAILED, or other terminal event, the calling client will wait indefinitely or timeout. Always end with a terminal TaskStatusUpdateEvent.

Framework Support

You don't have to write an A2A server from scratch. All major agentic frameworks now ship native A2A support:

FAQ

Q: Is A2A production-ready?

Yes. The v1.0 stable spec landed in March 2026. AWS (Amazon Bedrock AgentCore), Google Cloud (Vertex AI / Gemini Enterprise Agent Platform), and Microsoft Azure all support A2A natively. The Linux Foundation press release from April 2026 confirmed active production deployments in financial services, supply chain, and IT operations.

Q: Does A2A replace MCP?

No. MCP (Model Context Protocol) connects agents to tools and data sources — it's vertical. A2A connects agents to other agents — it's horizontal. Use both: MCP to ground individual agents in tools, A2A to let those agents collaborate.

Q: How do I handle authentication?

The A2A spec v1.0 introduces enterprise-grade security flows, including OAuth 2.0 and mutual TLS options declared in the Agent Card's securitySchemes field. For local development, no auth is required. For production, declare your security scheme in the Agent Card and validate tokens in your server middleware before the request reaches the AgentExecutor.

Q: What about streaming?

The official SDK supports Server-Sent Events (SSE) streaming via /message:stream. Set capabilities.streaming = True in your Agent Card, and use the EventQueue to emit partial TaskStatusUpdateEvent or TaskArtifactUpdateEvent objects as your agent processes the task. The client subscribes to the stream using GET /tasks/{id}:subscribe.

Q: Can I use A2A without the SDK?

Yes. The protocol is JSON-RPC 2.0 over HTTP. You can implement a minimal server with FastAPI or any HTTP framework — expose /.well-known/agent-card.json and handle POST /message:send. The SDK handles serialization and lifecycle bookkeeping, but it is not required.

Key Takeaways

A2A fills the interoperability gap that MCP leaves open: where MCP tells an agent what tools exist, A2A tells an agent what other agents exist and how to delegate tasks to them. The Agent Card / Task / Message triad is simple enough to implement from scratch but rich enough to support streaming, multi-tenancy, and enterprise auth flows.

Bottom Line

A2A is the standard for agent-to-agent communication in 2026 — 150+ organizations, native support in every major framework, and a stable v1.0 spec. The official Python SDK has a minor protobuf 7.x compatibility bug in its task endpoint, but the community library works end-to-end today. Start with python-a2a for rapid prototyping; switch to the official SDK once the upstream bug is fixed or your protobuf version allows.