← Back to Blog
Oct 22, 2026 · 10 min read

Build an A2A Agent in Python: Another Agent Can Call Yours

A2AAgent ProtocolPythonAI AgentsTutorial

Your team has an agent that knows currency rates. Someone else's agent needs that answer. Right now the options are: they import your Python module and inherit your dependencies, or you hand-roll an HTTP endpoint plus a README describing your request shape. Both work until three agents are involved and nobody remembers what the fields are called.

A2A (Agent2Agent) is the standard that replaces the README. One agent describes itself in a machine-readable card, another agent discovers it, sends a task, and reads back the artifact. Google published A2A in April 2025 and handed stewardship to the Linux Foundation in June 2025. The spec reached v1.0 in March 2026, with a v1.0.1 bug-fix release in May 2026.

The one-line difference from MCP: MCP standardizes how an agent uses a tool, A2A standardizes how one agent delegates work to another. The agent you call stays opaque. You get its card, its task states, and its output, not its internals.

Prerequisites

  • Python 3.10 or newer
  • uv (or pip)
  • curl, for the smoke test

Step 1: install the SDK

uv add "a2a-sdk[http-server]"

The pip equivalent is pip install "a2a-sdk[http-server]". The SDK implements spec 1.0 with a compatibility mode for 0.3, and ships matching extras for gRPC, FastAPI, OpenTelemetry tracing, and SQL backends (Postgres, MySQL, SQLite).

Step 2: write the business logic and the executor

Start with agent_executor.py. The protocol glue has exact import paths, and getting them roughly right is the most common way people lose an afternoon:

from a2a.helpers import (
    get_message_text,
    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.tasks import TaskUpdater
from a2a.types import TaskState

RATES = {
    ("USD", "IDR"): 16250.0,
    ("IDR", "USD"): 1 / 16250.0,
    ("EUR", "USD"): 1.09,
}

class CurrencyAgent:
    """Converts an amount between a fixed set of currency pairs."""

    async def invoke(self, user_request: str) -> str:
        parts = user_request.upper().split()
        if len(parts) != 3:
            return "Send the request as: <amount> <from> <to>, for example 250 USD IDR"
        amount, src, dst = float(parts[0]), parts[1], parts[2]
        rate = RATES.get((src, dst))
        if rate is None:
            return f"No rate available for {src} to {dst}"
        return f"{amount:.2f} {src} = {amount * rate:.2f} {dst}"

That logic is deliberately boring. No LLM call, no API key, so you can see the protocol work without debugging a model at the same time.

Now the executor. Three things matter: the first event you enqueue must be the task, the working status tells the caller you started, and the artifact is where the answer goes.

class CurrencyAgentExecutor(AgentExecutor):
    def __init__(self) -> None:
        self.agent = CurrencyAgent()

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

        task_updater = TaskUpdater(
            event_queue=event_queue, task_id=task.id, context_id=task.context_id
        )
        await task_updater.update_status(
            state=TaskState.TASK_STATE_WORKING,
            message=new_text_message("Processing request..."),
        )

        query = get_message_text(context.message)
        result = (
            await self.agent.invoke(user_request=query)
            if query
            else "No text input was provided."
        )

        await task_updater.add_artifact(
            parts=[new_text_part(text=result, media_type="text/plain")]
        )
        await task_updater.update_status(
            state=TaskState.TASK_STATE_COMPLETED,
            message=new_text_message("Request is completed!"),
        )

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

Step 3: serve it

app.py defines what your agent can do (the skill), how others find it (the card), and where it listens.

import uvicorn
from starlette.applications import Starlette

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
from a2a.types import AgentCapabilities, AgentCard, AgentInterface, AgentSkill

from agent_executor import CurrencyAgentExecutor

skill = AgentSkill(
    id="convert_currency",
    name="Convert currency",
    description="Converts an amount between two currencies using a fixed rate table.",
    input_modes=["text/plain"],
    output_modes=["text/plain"],
    tags=["finance", "currency"],
    examples=["250 USD IDR", "100 EUR USD"],
)

card = AgentCard(
    name="Currency Agent",
    description="Answers currency conversion questions for other agents.",
    version="0.1.0",
    default_input_modes=["text/plain"],
    default_output_modes=["text/plain"],
    capabilities=AgentCapabilities(streaming=True, extended_agent_card=False),
    supported_interfaces=[
        AgentInterface(
            protocol_binding="JSONRPC",
            url="http://127.0.0.1:9999",
            protocol_version="1.0",
        )
    ],
    skills=[skill],
)

handler = DefaultRequestHandler(
    agent_executor=CurrencyAgentExecutor(),
    task_store=InMemoryTaskStore(),
    agent_card=card,
)

routes = create_agent_card_routes(card) + create_jsonrpc_routes(handler, "/")
app = Starlette(routes=routes)

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

The url inside supported_interfaces has to match where the service actually runs. If it lies, every client builds requests against the wrong address.

Step 4: check the card

python app.py

Then in a second shell:

curl -s http://127.0.0.1:9999/.well-known/agent-card.json | python3 -m json.tool

You should see the name, a supportedInterfaces entry with protocolBinding set to JSONRPC, and your skill. If this 404s, nothing downstream will work, because discovery is the first thing every client does. Python's snake_case field names come back as camelCase on the wire, and the path is /.well-known/agent-card.json. Guides written against 0.3 still tell you agent.json; that name was dropped in 0.3.0 and the new one was registered in the spec at 1.0.

Step 5: call it like another agent would

import asyncio
import httpx

from a2a.client import A2ACardResolver, ClientConfig, create_client
from a2a.helpers import display_agent_card, new_text_message
from a2a.types import Role, SendMessageRequest

BASE_URL = "http://127.0.0.1:9999"

async def main() -> None:
    async with httpx.AsyncClient() as http:
        card = await A2ACardResolver(
            httpx_client=http, base_url=BASE_URL
        ).get_agent_card()
    display_agent_card(card)

    client = await create_client(agent=card, client_config=ClientConfig(streaming=False))
    request = SendMessageRequest(
        message=new_text_message("250 USD IDR", role=Role.ROLE_USER)
    )
    async for chunk in client.send_message(request):
        print(chunk)
    await client.close()

asyncio.run(main())

The non-streaming iterator yields one completed task: status COMPLETED, an artifact holding the converted string, and the history of the exchange. Switch to ClientConfig(streaming=True) and the same call yields four events instead: the task, a WORKING status update, the artifact update, and COMPLETED. That is the difference between a blocking call and progress you can show a user while the work runs.

Step 6: put it on the internet

An A2A agent is an HTTP service, so treat it like one. The card endpoint stays public, because a client has to read it before it knows how to authenticate. Everything else gets a token check. Wire that up the normal Starlette way, with middleware around the JSON-RPC routes, and check authorization inside each executor call rather than assuming that a caller allowed to read the card is allowed to use it.

Two things that bite people in production:

  • InMemoryTaskStore forgets every task on restart. A task sitting in INPUT_REQUIRED when you deploy becomes TaskNotFoundError for the client that was mid-conversation. The SDK ships SQL task stores for Postgres, MySQL, and SQLite; use one.
  • Task IDs leak existence. If you answer "not found" for other people's tasks but "permission denied" for your own, you just confirmed which IDs are real. Scope every task-store query by caller and return the same not-found error in both cases.

A2A vs MCP: pick the right one

Usually you want both, because they answer different questions.

  • What is on the other end: MCP talks to a tool or resource, A2A talks to another agent.
  • What you can see: MCP tool calls are transparent, A2A exposes task states and artifacts only.
  • Unit of work: MCP has a tool call and a result, A2A has a task with a lifecycle.
  • Transports: MCP uses stdio and streamable HTTP, A2A uses JSON-RPC, gRPC, and HTTP+JSON.
  • Authorization: MCP leans on OAuth 2.1, A2A has signed agent cards through the AP2 extension.

If the thing you want to call is a function you could have imported, it is an MCP server. If it is a system with its own model, its own judgment, and its own idea of how long the work takes, it is an A2A agent.

Conclusion

You can now stand up an agent that other agents discover and call without reading your source. Next steps: replace the rate table with a real API call behind a timeout, install the [telemetry] extra and look at what the traces capture, and read the authorization section of the spec before you expose the port to anything outside your network.

References

Need Help Implementing This?

I help teams design and build scalable cloud infrastructure, DevOps pipelines, and production-grade systems.

Book a Free Consultation