A2A for Beginners, Part 3: Building an Agent the Smart Way

In Part 2, we built an A2A agent from scratch with raw Flask and JSON-RPC. We wrote every request handler, serialized every response, and managed every task ID ourselves. That was useful for understanding the protocol, but you wouldn't want to build production agents that way.

This time we're doing it the smart way. We'll use the official a2a-sdk to build two agents, then make one delegate work to the other over A2A. Two agents. Two separate processes. Real agent-to-agent communication.

By the end of this post, you'll have:

1. A Currency Agent that converts between major currencies
2. A Travel Agent that knows travel prices and delegates currency conversion to the Currency Agent
3. A client that talks to the Travel Agent and sees the whole chain in action

Why Use an SDK?

In Part 2, our Flask server was about 200 lines of code. Most of that wasn't conversion logic - it was protocol plumbing: parsing JSON-RPC envelopes, generating task IDs, managing task state, serving the agent card, building response objects.

The official SDK handles all of that. Here's what you get for free:

• Agent card hosting - the SDK serves /.well-known/agent-card.json automatically
• JSON-RPC routing - message/send, tasks/get, and tasks/cancel are wired up for you
• Task lifecycle - task creation, state transitions, and storage are managed by the framework
• Streaming - built-in support for message/stream with no extra code
• Type safety - Pydantic models for every A2A type (messages, tasks, artifacts, events)
• Client library - discover agents, send messages, and read responses with a few lines

You focus on what your agent actually does. The SDK handles how it speaks A2A.

Prerequisites

You need Python 3.10+ and pip.

mkdir a2a-travel && cd a2a-travel
python -m venv venv
source venv/bin/activate    # On Windows: venv\Scripts\activate
pip install "a2a-sdk[http-server]" httpx uvicorn

The http-server extra pulls in Starlette and the HTTP transport layer. httpx is for the async HTTP client we'll use for agent-to-agent communication.

Step 1: The Currency Agent

Let's start with the simpler agent. It takes a message like "100 USD to EUR" and returns the converted amount.

The Agent Card

In Part 2, we hand-crafted a JSON dict for our agent card. With the SDK, it's a Pydantic model:

from a2a.types import (
    AgentCapabilities,
    AgentCard,
    AgentSkill,
)

skill = AgentSkill(
    id="currency_conversion",
    name="Currency Conversion",
    description="Converts an amount from one currency to another.",
    tags=["currency", "finance", "conversion"],
    examples=[
        "100 USD to EUR",
        "5000 JPY to GBP",
        "250 CHF to USD",
    ],
)

agent_card = AgentCard(
    name="Currency Converter",
    description=(
        "Converts between major currencies (USD, EUR, GBP, JPY, CHF) "
        "using current exchange rates."
    ),
    url="http://localhost:5001",
    version="1.0.0",
    defaultInputModes=["text"],
    defaultOutputModes=["text"],
    capabilities=AgentCapabilities(streaming=True),
    skills=[skill],
)

No raw dicts. No worrying about whether you spelled protocolVersion correctly - the model fills in the default (0.3.0). The SDK validates everything at construction time.

The Executor

This is the core concept. Instead of writing Flask route handlers, you implement a single class with two methods:

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue


class CurrencyExecutor(AgentExecutor):

    async def execute(
        self, context: RequestContext, event_queue: EventQueue
    ) -> None:
        # Your agent logic goes here
        ...

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

execute() is called whenever a message arrives. The context gives you the incoming message, the task ID, and the context ID. The event_queue is where you push status updates and results.

Here's the full executor for the Currency Agent:

from a2a.types import (
    Artifact,
    TaskArtifactUpdateEvent,
    TaskState,
    TaskStatus,
    TaskStatusUpdateEvent,
    TextPart,
)
from a2a.utils.message import new_agent_text_message
from a2a.utils.task import new_task


class CurrencyExecutor(AgentExecutor):

    async def execute(
        self, context: RequestContext, event_queue: EventQueue
    ) -> None:
        # Create or reuse a task
        task = context.current_task or new_task(context.message)
        await event_queue.enqueue_event(task)

        # Tell the caller we're working on it
        await event_queue.enqueue_event(
            TaskStatusUpdateEvent(
                taskId=context.task_id,
                contextId=context.context_id,
                final=False,
                status=TaskStatus(
                    state=TaskState.working,
                    message=new_agent_text_message(
                        "Converting currency..."
                    ),
                ),
            )
        )

        # Extract the user's text
        user_text = ""
        for part in context.message.parts:
            if hasattr(part, "root"):
                part = part.root
            if isinstance(part, TextPart):
                user_text = part.text
                break

        # Do the conversion
        result = convert(user_text)

        # Send the result as an artifact
        await event_queue.enqueue_event(
            TaskArtifactUpdateEvent(
                taskId=context.task_id,
                contextId=context.context_id,
                artifact=Artifact(
                    artifactId=f"{context.task_id}-result",
                    name="conversion_result",
                    parts=[TextPart(text=result)],
                ),
            )
        )

        # Mark the task as done
        await event_queue.enqueue_event(
            TaskStatusUpdateEvent(
                taskId=context.task_id,
                contextId=context.context_id,
                final=True,
                status=TaskStatus(state=TaskState.completed),
            )
        )

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

Compare this with Part 2's handle_message_send() function. We're not touching JSON-RPC. We're not generating UUIDs for tasks. We're not building response dicts. We push events to a queue and the SDK turns them into proper A2A responses.

A few things to notice:

• new_task(context.message) creates a task from the incoming message. The SDK generates the task ID.
• TaskStatusUpdateEvent updates the task's state. The final=True on the last event tells the SDK this is the terminal state.
• TaskArtifactUpdateEvent adds an artifact (the result) to the task.
• TextPart wraps text content. Parts are the atomic unit of content in A2A.

The Conversion Logic

This is the boring part, and that's the point. The protocol doesn't care how your agent thinks.

import re

RATES = {
    ("USD", "EUR"): 0.92, ("USD", "GBP"): 0.79,
    ("USD", "JPY"): 149.50, ("USD", "CHF"): 0.88,
    ("EUR", "USD"): 1.09, ("EUR", "GBP"): 0.86,
    ("EUR", "JPY"): 162.50, ("EUR", "CHF"): 0.96,
    ("GBP", "USD"): 1.27, ("GBP", "EUR"): 1.16,
    ("GBP", "JPY"): 189.50, ("GBP", "CHF"): 1.12,
    ("JPY", "USD"): 0.0067, ("JPY", "EUR"): 0.0062,
    ("JPY", "GBP"): 0.0053, ("JPY", "CHF"): 0.0059,
    ("CHF", "USD"): 1.14, ("CHF", "EUR"): 1.04,
    ("CHF", "GBP"): 0.89, ("CHF", "JPY"): 170.0,
}

PATTERN = re.compile(
    r"(\d+(?:\.\d+)?)\s*([A-Za-z]{3})\s+(?:to\s+)?([A-Za-z]{3})",
    re.IGNORECASE,
)

def convert(text):
    match = PATTERN.search(text)
    if not match:
        return "I didn't understand that. Try: 100 USD to EUR"

    amount = float(match.group(1))
    source = match.group(2).upper()
    target = match.group(3).upper()

    rate = RATES.get((source, target))
    if rate is None:
        return f"No rate available for {source} -> {target}"

    return f"{amount * rate:,.2f} {target}"

Wiring It Up

Three lines to create a running A2A server:

import uvicorn
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore

request_handler = DefaultRequestHandler(
    agent_executor=CurrencyExecutor(),
    task_store=InMemoryTaskStore(),
)

app = A2AStarletteApplication(
    agent_card=agent_card,
    http_handler=request_handler,
)

uvicorn.run(app.build(), host="127.0.0.1", port=5001)

DefaultRequestHandler connects your executor to the A2A request flow. InMemoryTaskStore keeps tasks in memory (swap it for a database-backed store in production). A2AStarletteApplication builds a Starlette app with all the A2A endpoints wired up.

Start it:

python currency_agent.py

Test it with curl:

# Discovery
curl http://localhost:5001/.well-known/agent-card.json | python -m json.tool

# Conversion
curl -X POST http://localhost:5001 \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": "req-1",
    "method": "message/send",
    "params": {
      "message": {
        "messageId": "msg-1",
        "role": "user",
        "parts": [{"kind": "text", "text": "100 USD to EUR"}]
      }
    }
  }'

Response:

{
  "jsonrpc": "2.0",
  "id": "req-1",
  "result": {
    "id": "...",
    "kind": "task",
    "status": { "state": "completed" },
    "artifacts": [
      {
        "artifactId": "...",
        "name": "conversion_result",
        "parts": [{ "kind": "text", "text": "92.00 EUR" }]
      }
    ]
  }
}

The SDK generated the task, managed its lifecycle, and formatted the JSON-RPC response. We just had to push events.

Step 2: The Travel Agent

Here's where it gets interesting. The Travel Agent knows typical travel costs in local currencies, but when a user asks for a price in a different currency, it delegates the conversion to the Currency Agent over A2A.

The Knowledge Base

Nothing fancy - a dict mapping travel items to prices in local currencies:

TRAVEL_ITEMS = {
    "flight to paris": (420, "EUR"),
    "flight to tokyo": (185_000, "JPY"),
    "flight to london": (380, "GBP"),
    "flight to zurich": (450, "CHF"),
    "hotel in paris": (160, "EUR"),
    "hotel in tokyo": (18_000, "JPY"),
    "hotel in london": (195, "GBP"),
    "hotel in zurich": (280, "CHF"),
    "dinner in paris": (55, "EUR"),
    "dinner in tokyo": (3_500, "JPY"),
    "dinner in london": (40, "GBP"),
    "dinner in zurich": (65, "CHF"),
    "museum in paris": (17, "EUR"),
    "temple in tokyo": (500, "JPY"),
    "tour in london": (28, "GBP"),
    "boat on lake zurich": (15, "CHF"),
}

Delegating to the Currency Agent

This is the core of agent-to-agent communication. The Travel Agent uses the SDK's client library to:

1. Discover the Currency Agent by fetching its agent card
2. Create a client from the card
3. Send a message and read the response

import httpx
from uuid import uuid4
from a2a.client import A2ACardResolver
from a2a.client.client import ClientConfig
from a2a.client.client_factory import ClientFactory
from a2a.types import Message, Role, TextPart

CURRENCY_AGENT_URL = "http://localhost:5001"


async def delegate_conversion(amount, source, target):
    # Step 1: Discover
    async with httpx.AsyncClient() as httpx_client:
        resolver = A2ACardResolver(
            httpx_client=httpx_client,
            base_url=CURRENCY_AGENT_URL,
        )
        card = await resolver.get_agent_card()

    # Step 2: Create client
    factory = ClientFactory(config=ClientConfig(streaming=False))
    client = factory.create(card)

    try:
        # Step 3: Send message
        message = Message(
            role=Role.user,
            parts=[TextPart(text=f"{amount:.2f} {source} to {target}")],
            messageId=uuid4().hex,
        )
        response = client.send_message(message)

        # Step 4: Read result
        result_text = None
        async for chunk in response:
            delegate_task, _ = chunk
            for artifact in delegate_task.artifacts:
                for part in artifact.parts:
                    if hasattr(part, "root"):
                        part = part.root
                    if isinstance(part, TextPart):
                        result_text = part.text
        return result_text
    finally:
        await client.close()

Look at what's happening here. The Travel Agent is acting as an A2A client, the same way a human user's client would. It discovers the Currency Agent by fetching its agent card, creates a client from the card, sends a message, and reads the result.

This is the promise of A2A in practice: agent-to-agent communication using the exact same protocol as agent-to-human communication. The Currency Agent doesn't know (or care) whether it's being called by a person or by another agent.

The Travel Executor

The Travel Agent's executor parses the user's request, looks up the travel item, and delegates if a currency conversion is needed:

class TravelExecutor(AgentExecutor):

    async def execute(
        self, context: RequestContext, event_queue: EventQueue
    ) -> None:
        task = context.current_task or new_task(context.message)
        await event_queue.enqueue_event(task)

        user_text = ""
        for part in context.message.parts:
            if hasattr(part, "root"):
                part = part.root
            if isinstance(part, TextPart):
                user_text = part.text
                break

        # Parse the request: "flight to paris in USD"
        item_key, target_currency = parse_request(user_text)

        if item_key is None:
            await self._complete(
                context, event_queue,
                "I don't recognize that item.",
            )
            return

        price, local_currency = TRAVEL_ITEMS[item_key]
        item_label = item_key.title()

        # No conversion needed
        if not target_currency or target_currency == local_currency:
            await self._complete(
                context, event_queue,
                f"{item_label}: {price:,.0f} {local_currency}",
            )
            return

        # Tell the user we're delegating
        await event_queue.enqueue_event(
            TaskStatusUpdateEvent(
                taskId=context.task_id,
                contextId=context.context_id,
                final=False,
                status=TaskStatus(
                    state=TaskState.working,
                    message=new_agent_text_message(
                        f"Asking the Currency Agent to convert "
                        f"{price:,.0f} {local_currency} to "
                        f"{target_currency}..."
                    ),
                ),
            )
        )

        # Delegate to the Currency Agent
        try:
            converted = await delegate_conversion(
                price, local_currency, target_currency
            )
        except Exception as exc:
            await self._complete(
                context, event_queue,
                f"{item_label}: {price:,.0f} {local_currency}\n"
                f"(Could not convert - is the Currency Agent "
                f"running? Error: {exc})",
            )
            return

        await self._complete(
            context, event_queue,
            f"{item_label}: {price:,.0f} {local_currency} "
            f"(~{converted})",
        )

    async def _complete(self, context, event_queue, text):
        await event_queue.enqueue_event(
            TaskArtifactUpdateEvent(
                taskId=context.task_id,
                contextId=context.context_id,
                artifact=Artifact(
                    artifactId=f"{context.task_id}-result",
                    name="travel_result",
                    parts=[TextPart(text=text)],
                ),
            )
        )
        await event_queue.enqueue_event(
            TaskStatusUpdateEvent(
                taskId=context.task_id,
                contextId=context.context_id,
                final=True,
                status=TaskStatus(state=TaskState.completed),
            )
        )

    async def cancel(self, context, event_queue):
        raise Exception("cancel not supported")

The server setup is identical to the Currency Agent, just with a different card and port:

agent_card = AgentCard(
    name="Travel Planner",
    description=(
        "Knows typical travel costs in cities worldwide and converts "
        "prices to your preferred currency using the Currency Agent."
    ),
    url="http://localhost:5002",
    version="1.0.0",
    defaultInputModes=["text"],
    defaultOutputModes=["text"],
    capabilities=AgentCapabilities(streaming=True),
    skills=[skill],
)

# ... same DefaultRequestHandler + A2AStarletteApplication pattern

uvicorn.run(app.build(), host="127.0.0.1", port=5002)

Step 3: The Client

The client discovers the Travel Agent and sends queries. The SDK's client library makes this straightforward:

import asyncio
from uuid import uuid4

import httpx
from a2a.client import A2ACardResolver
from a2a.client.client import ClientConfig
from a2a.client.client_factory import ClientFactory
from a2a.types import Message, Role, TaskState, TextPart


async def main():
    # Discover the Travel Agent
    async with httpx.AsyncClient() as httpx_client:
        resolver = A2ACardResolver(
            httpx_client=httpx_client,
            base_url="http://localhost:5002",
        )
        card = await resolver.get_agent_card()

    print(f"Connected to: {card.name}")

    # Create a client from the card
    factory = ClientFactory(config=ClientConfig(streaming=False))
    client = factory.create(card)

    # Send a message
    message = Message(
        role=Role.user,
        parts=[TextPart(text="hotel in tokyo in USD")],
        messageId=uuid4().hex,
    )
    response = client.send_message(message)

    # Read the response
    async for chunk in response:
        task, _ = chunk
        if task.status.state == TaskState.completed:
            for artifact in task.artifacts:
                for part in artifact.parts:
                    if hasattr(part, "root"):
                        part = part.root
                    if isinstance(part, TextPart):
                        print(part.text)

    await client.close()


asyncio.run(main())

Running the Demo

Open three terminals:

# Terminal 1: Start the Currency Agent
python currency_agent.py
# Output: Currency Agent starting on http://localhost:5001

# Terminal 2: Start the Travel Agent
python travel_agent.py
# Output: Travel Agent starting on http://localhost:5002

# Terminal 3: Run the client
python client.py --demo

Here's what happens:

Discovering agent at http://localhost:5002...
Connected to: Travel Planner
Description:  Knows typical travel costs in cities worldwide and
              converts prices to your preferred currency using
              the Currency Agent.

> list
  Here's what I can look up for you:

  - boat on lake zurich: 15 CHF
  - dinner in london: 40 GBP
  - dinner in paris: 55 EUR
  - dinner in tokyo: 3,500 JPY
  - flight to london: 380 GBP
  - flight to paris: 420 EUR
  - flight to tokyo: 185,000 JPY
  - hotel in paris: 160 EUR
  - hotel in tokyo: 18,000 JPY
  - museum in paris: 17 EUR
  - temple in tokyo: 500 JPY
  - tour in london: 28 GBP
  ...

> flight to paris
  Flight To Paris: 420 EUR

> flight to paris in USD
  Flight To Paris: 420 EUR (~457.80 USD)

> hotel in tokyo in USD
  Hotel In Tokyo: 18,000 JPY (~120.60 USD)

> dinner in london in EUR
  Dinner In London: 40 GBP (~46.40 EUR)

> boat on lake zurich in JPY
  Boat On Lake Zurich: 15 CHF (~2,550.00 JPY)

When you ask "flight to paris in USD", here's the full chain:

1. Client discovers the Travel Agent at localhost:5002
2. Client sends "flight to paris in USD" to the Travel Agent
3. Travel Agent looks up the price: 420 EUR
4. Travel Agent discovers the Currency Agent at localhost:5001
5. Travel Agent sends "420.00 EUR to USD" to the Currency Agent
6. Currency Agent computes: 420 * 1.09 = 457.80
7. Currency Agent returns "457.80 USD" to the Travel Agent
8. Travel Agent combines: "Flight To Paris: 420 EUR (~457.80 USD)"
9. Travel Agent returns the combined result to the Client

Two separate processes. Two separate agent cards. One protocol.

What the SDK Handles

Let's compare the raw approach from Part 2 with the SDK approach:

Concern	Part 2 (Raw Flask)	Part 3 (SDK)
Agent card endpoint	Manual Flask route	Automatic
JSON-RPC parsing	Manual validation	Automatic
Task ID generation	uuid.uuid4() by hand	new_task()
Task state management	Manual dict updates	Event queue
Response formatting	Build dicts by hand	Pydantic serialization
Error codes	Remember -32001, -32002...	Built-in error types
Streaming	Not implemented	Built-in
Client library	Manual requests calls	ClientFactory + card resolver
Type safety	None (raw dicts)	Full Pydantic validation

The SDK doesn't just reduce boilerplate - it eliminates entire categories of bugs. You can't accidentally serve an invalid agent card. You can't forget to include jsonrpc: "2.0" in a response. You can't mix up task states.

What We Didn't Build

Even with the SDK, we kept things simple:

• No LLM - both agents use deterministic logic. Swap in an LLM call and nothing about the A2A layer changes.
• No authentication - in production, you'd add securitySchemes to the agent card and validate credentials.
• No persistent storage - InMemoryTaskStore loses everything on restart. The SDK supports database-backed stores.
• No streaming requests in the demo client - for simplicity, our client code uses non-streaming requests, but the SDK has streaming support ready to go.
• No deployment - both agents run locally. In the real world, they could be on different servers, different clouds, different continents.

But the shape is right. The Travel Agent discovers a remote agent, delegates work, and returns a combined result. That's agent-to-agent communication. Everything else is configuration.

If you want the finished code, it's on GitHub: waggle-examples/tutorials/a2a-with-sdk.