Peyton Green

Posted on Apr 1

Build Your First A2A Agent Pair in Python (15 Minutes, No Cloud Required)

#python #ai #a2a #tutorial

Google's A2A protocol hit v1.0 on March 12, 2026. You've probably seen the announcement — enterprise framing, agent interoperability, Salesforce and SAP and ServiceNow on the logo slide.

What you haven't seen is a tutorial that just works locally, without a Google Cloud account, without paid APIs, without enterprise infrastructure. The official demos are GCP demos. They teach the platform, not the protocol.

This is the other tutorial. You'll build two Python agents that communicate over A2A, run them in two terminal windows, and watch a task travel from dispatcher to executor and back — in under 15 minutes, on localhost.

What A2A Is (30 Seconds)

A2A is a protocol for agent-to-agent task delegation. When your orchestrator agent needs to hand off work to a specialist agent, A2A is how they talk.

The mental model that clicked for me is the HTTP analogy. HTTP didn't create the web — it gave every computer a shared language. A2A does the same thing for agents: any A2A-compliant agent can call any other A2A-compliant agent, regardless of what framework built it or where it's hosted.

It's Apache 2.0 licensed and governed by the Agentic AI Foundation (same neutral body that governs MCP). It is not a Google product. It's an open standard that Google initiated.

A2A and MCP are not competitors. They operate at different layers:

┌─────────────────────────────────────────┐
│            Orchestrator Agent           │
├─────────────────────────────────────────┤
│  A2A: delegates tasks to other agents   │  ← what we're building today
├─────────────────────────────────────────┤
│  MCP: calls tools, reads files, APIs    │  ← what you already have
└─────────────────────────────────────────┘

MCP gives your agent access to external resources. A2A gives your agent access to other agents. Both. Different problems.

What We're Building

Two Python scripts:

stats_agent.py — an A2A-compliant agent server (port 9999) that accepts text and returns word count, average word length, and estimated reading time.
dispatcher.py — a client that discovers the stats agent via its Agent Card and submits a text analysis task.

Total code: ~110 lines. No GCP. No paid APIs. No Docker required.

Install

pip install "a2a-sdk[http-server]" uvicorn httpx

The [http-server] extra pulls in Starlette and sse-starlette for the server side. Requires Python 3.10+.

Versions tested: a2a-sdk==0.3.25, uvicorn==0.42.0, httpx==0.28.1.

Note on SDK versions: pip install a2a-sdk installs 0.3.25, the current stable release. A v1.0.0 alpha exists (1.0.0a0) with breaking changes: proto-based Part types replacing TextPart/FilePart/DataPart, SCREAMING_SNAKE_CASE enums, PascalCase operation names (e.g. SendMessage), gRPC transport added. Stable v1.0 SDK is estimated May–June 2026. This article targets 0.3.25 — if you're following along after stable v1.0 ships, pin to 0.3.25 first, then upgrade with the migration guide. The 0.3.x SDK is not deprecated.

Part 1: The Stats Agent (Server)

Every A2A agent announces itself via an Agent Card — a JSON document served at /.well-known/agent-card.json. This is how other agents discover what your agent can do before they call it.

Create stats_agent.py:

# stats_agent.py
import asyncio
import uvicorn
from a2a.server.apps import A2AStarletteApplication
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore, TaskUpdater
from a2a.types import AgentCapabilities, AgentCard, AgentSkill, TextPart
from a2a.utils import new_agent_text_message


def analyze_text(text: str) -> str:
    """Returns a plain-text analysis of the given text."""
    words = text.split()
    word_count = len(words)
    if word_count == 0:
        return "Empty input."
    cleaned = [w.strip(".,!?;:\"'") for w in words]
    avg_word_len = sum(len(w) for w in cleaned) / word_count
    reading_time_sec = (word_count / 238) * 60  # avg adult reading speed
    freq: dict[str, int] = {}
    for w in cleaned:
        key = w.lower()
        freq[key] = freq.get(key, 0) + 1
    most_common = max(freq, key=lambda w: freq[w])
    return (
        f"Word count: {word_count}\n"
        f"Average word length: {avg_word_len:.1f} characters\n"
        f"Estimated reading time: {reading_time_sec:.0f} seconds\n"
        f"Most frequent word: '{most_common}'"
    )


class StatsAgentExecutor(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        updater = TaskUpdater(
            event_queue=event_queue,
            task_id=context.task_id,
            context_id=context.context_id,
        )
        await updater.start_work()

        user_input = context.get_user_input()
        result = analyze_text(user_input)

        await event_queue.enqueue_event(new_agent_text_message(result))
        await updater.complete()

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        raise NotImplementedError("cancel not supported")


skill = AgentSkill(
    id="text_stats",
    name="Text Statistics",
    description="Analyzes text and returns word count, average word length, and reading time.",
    tags=["text", "analysis", "nlp"],
    examples=["Analyze this paragraph for me.", "How long is this text?"],
)

agent_card = AgentCard(
    name="Text Stats Agent",
    description="A specialist agent that computes statistics for any block of text.",
    url="http://localhost:9999/",
    version="1.0.0",
    default_input_modes=["text"],
    default_output_modes=["text"],
    capabilities=AgentCapabilities(streaming=False),
    skills=[skill],
)

app = A2AStarletteApplication(
    agent_card=agent_card,
    http_handler=DefaultRequestHandler(
        agent_executor=StatsAgentExecutor(),
        task_store=InMemoryTaskStore(),
    ),
)

if __name__ == "__main__":
    uvicorn.run(app.build(), host="0.0.0.0", port=9999)

Three things happening here:

Agent Card declares what this agent can do. Any other A2A agent can fetch this at http://localhost:9999/.well-known/agent-card.json and understand the agent's capabilities before sending a task.
StatsAgentExecutor handles the actual work. execute() is called once per task. TaskUpdater handles the lifecycle: start_work() transitions the task to working, complete() closes it out.
A2AStarletteApplication wires the card and the executor into an HTTP server with all the required A2A endpoints (/, /.well-known/agent-card.json, streaming if enabled).

Part 2: The Dispatcher (Client)

Create dispatcher.py:

# dispatcher.py
import asyncio
import httpx
from a2a.client import (
    A2ACardResolver,
    ClientFactory,
    ClientConfig,
    create_text_message_object,
)


SAMPLE_TEXT = """
The A2A protocol was released as v1.0 on March 12, 2026.
It is an open standard governed by the Agentic AI Foundation,
the same body that governs MCP. A2A handles agent-to-agent
communication; MCP handles agent-to-tool communication.
They are complementary protocols, not competitors.
"""


async def main():
    base_url = "http://localhost:9999"

    async with httpx.AsyncClient() as http:
        # Step 1: Discover the agent via its Agent Card
        resolver = A2ACardResolver(httpx_client=http, base_url=base_url)
        card = await resolver.get_agent_card()
        print(f"Discovered: {card.name}")
        print(f"Skills: {[s.name for s in card.skills]}\n")

        # Step 2: Create a client and send the task
        factory = ClientFactory(config=ClientConfig(httpx_client=http))
        client = factory.create(card)

        message = create_text_message_object(content=SAMPLE_TEXT.strip())

        # Step 3: Iterate response events (Message or (Task, Update) tuples)
        async for event in client.send_message(message):
            if hasattr(event, "parts"):
                # Direct Message response
                for part in event.parts:
                    if hasattr(part.root, "text"):
                        print("Result from Stats Agent:")
                        print(part.root.text)
            elif isinstance(event, tuple):
                task, _ = event
                # Task completed — look for agent reply in history
                if task.history:
                    for msg in task.history:
                        if msg.role.value == "agent":
                            for part in msg.parts:
                                if hasattr(part.root, "text"):
                                    print("Result from Stats Agent:")
                                    print(part.root.text)


asyncio.run(main())

The dispatcher doesn't know anything about the stats agent implementation. It discovers the agent's capabilities from the Agent Card, submits a task, and reads the response. Swap the base_url to point at any A2A-compliant agent and the same dispatcher code works.

Run It

Terminal 1 — start the stats agent:

python stats_agent.py

You should see uvicorn start on port 9999. Verify the agent card is being served:

curl http://localhost:9999/.well-known/agent-card.json

Terminal 2 — run the dispatcher:

python dispatcher.py

Expected output:

Discovered: Text Stats Agent
Skills: ['Text Statistics']

Result from Stats Agent:
Word count: 42
Average word length: 5.6 characters
Estimated reading time: 11 seconds
Most frequent word: 'the'

That's two agents communicating over A2A. The dispatcher found the stats agent by resolving its Agent Card, submitted a task using the standard A2A task schema, and the stats agent processed it through the full task lifecycle (submitted → working → completed).

Part 3: Add Streaming (For Long-Running Tasks)

For tasks that take more than a second or two — a long analysis, a web fetch, an LLM call — streaming lets the dispatcher receive progress updates instead of waiting for a single blocked response.

Change one thing:

In stats_agent.py, enable streaming on the agent card:

capabilities=AgentCapabilities(streaming=True),

The dispatcher code doesn't change. ClientFactory reads the agent card capabilities and automatically uses the streaming transport when the agent advertises it. The same async for event in client.send_message(message) loop handles both streaming and non-streaming responses — streaming delivers incremental (Task, Update) tuples as the task progresses, non-streaming delivers a single Message on completion.

With streaming enabled, your dispatcher gets status updates as the task moves from submitted → working → completed. For a long-running task (an LLM call, a web fetch, a database query), that loop is where you'd display progress to the user or forward partial results downstream.

What You've Built

Two A2A-compliant agents that communicate over an open standard. Nothing here is Google-specific. The stats agent could be replaced by any A2A-compliant agent from any vendor — the dispatcher code doesn't change.

That's the actual value of A2A: your orchestrator can call a Salesforce agent, a self-hosted RAG agent, or a neighbor's Python script — as long as they speak A2A, the handshake is the same.

Connecting to Your Existing Python Scripts

If you have existing Python automation scripts — Boto3 pipelines, data processing tools, Selenium automation — wrapping them as A2A agents takes about 10 additional lines:

class MyExistingScriptExecutor(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        await updater.start_work()

        params = context.get_user_input()      # parse JSON or plain text
        result = your_existing_function(params) # call your script

        await event_queue.enqueue_event(new_agent_text_message(str(result)))
        await updater.complete()

    async def cancel(self, context, event_queue):
        raise NotImplementedError

Your script is now callable by any A2A-compatible orchestrator — Claude Code, LangGraph, Google ADK, or your own dispatcher.

Going Further: A2A + LangChain (without the boilerplate)

If you're already using LangChain, langchain-adk 0.3.14 (released March 20, 2026) lets you wrap any LangChain agent as an A2A server in one import instead of the full executor/handler boilerplate above:

from langchain_adk.a2a import A2AServer
from langchain_adk.agents.llm_agent import LlmAgent

# Your existing LangChain agent becomes an A2A server
agent = LlmAgent(llm=llm, tools=your_tools)
server = A2AServer(agent=agent, name="My Agent", description="Does stuff")
server.run(port=9999)  # exposes /.well-known/agent-card.json automatically

Your LangChain orchestrator can then call other A2A-compliant agents — whether they're built with Google ADK, Crew.AI, or the raw a2a-sdk you used here — using A2AAgent as a standard LangChain tool.

This is where A2A gets interesting at scale: not "one agent calling one other agent" but an orchestrator that delegates to a dynamic pool of specialists, each discoverable via their Agent Card. langchain-adk uses plain asyncio for orchestration (not LangGraph), so there's no graph state overhead.

A full walkthrough (multi-agent delegation patterns, orchestrator + specialist setup) is in the companion article: LangChain's A2A Integration: Building Multi-Agent Systems in Python Without the Cloud Lock-In.

DEV Community