Build an A2A Server in Python: Let Your AI Agent Talk to Other Agents

Most AI agents today are islands. They can use tools, call APIs, and talk to databases, but they cannot talk to each other. If you build a coding agent on LangGraph and someone else builds a research agent on CrewAI, there is no standard way for them to collaborate.

The Agent2Agent (A2A) protocol fixes this. It is an open standard, contributed by Google and now under the Linux Foundation, that lets agents built on different frameworks communicate over HTTP. Think of it as USB for AI agents: one cable, many devices.

This tutorial walks through building a working A2A server with the official Python SDK, then connecting to it with a client. By the end, you will have two agents talking to each other.

Prerequisites

• Python 3.10 or newer (check with python3 --version)
• uv recommended, or plain pip
• An HTTP client for testing (curl or a browser)
• No LLM API key required for the basics (the helloworld sample uses a simple echo logic)

What A2A Actually Is

A2A defines three core concepts:

1. Agent Card -- a JSON document that describes what an agent can do, what input/output it accepts, and where to reach it. Clients fetch this card to discover agents.
2. Task -- a unit of work. The client sends a message, the server creates a task, processes it, and returns results. Tasks can be synchronous (wait for result) or asynchronous (poll or get pushed).
3. Transport -- JSON-RPC 2.0 over HTTP(S). The SDK handles the protocol details so you focus on agent logic.

The protocol supports streaming via SSE and push notifications for long-running tasks, but this tutorial covers the basics: send a message, get a response.

Step 1: Install the SDK

# Create a project directory
mkdir a2a-tutorial && cd a2a-tutorial

# Create virtual environment
python3 -m venv .venv
source .venv/bin/activate

# Install the A2A SDK with HTTP server support
pip install "a2a-sdk[http-server]"

The [http-server] extra pulls in Starlette and uvicorn, which the SDK uses to serve your agent. As of August 2026, the latest version is 1.1.2.

Step 2: Define Your Agent

Create a file called agent.py. This is where your agent logic lives. For this tutorial, it echoes the user's message back with a twist.

class EchoAgent:
    """An agent that echoes user messages."""

    async def invoke(self, user_request: str) -> str:
        return f"Echo from A2A: you said '{user_request}'"

In a real project, this is where you would wire up your LLM call, tool use, or any other logic. The point of A2A is that the protocol does not care what happens inside invoke. It only cares about the message format going in and out.

Step 3: Build the Agent Executor

The SDK needs an AgentExecutor that bridges the A2A protocol to your agent. Create executor.py:

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

from agent import EchoAgent


class EchoAgentExecutor(AgentExecutor):
    def __init__(self):
        self.agent = EchoAgent()

    async def execute(self, context: RequestContext, event_queue: EventQueue):
        # Get or create the task
        if context.current_task:
            task = context.current_task
        else:
            task = new_task_from_user_message(context.message)
            await event_queue.enqueue_event(task)

        # Update status to "working"
        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..."),
        )

        # Run the agent
        query = get_message_text(context.message)
        result = await self.agent.invoke(user_request=query or "")

        # Return the result as an artifact
        await task_updater.add_artifact(
            parts=[new_text_part(text=result, media_type="text/plain")]
        )

        # Mark task complete
        await task_updater.update_status(
            state=TaskState.TASK_STATE_COMPLETED,
            message=new_text_message("Done."),
        )

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

The pattern is: receive request, create or reuse task, run your logic, return result, mark complete. The SDK handles all the JSON-RPC framing, task state management, and HTTP transport.

Step 4: Write the Server

Now wire everything together in server.py:

import uvicorn

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 executor import EchoAgentExecutor
from starlette.applications import Starlette


if __name__ == "__main__":
    # Define what this agent can do
    skill = AgentSkill(
        id="echo",
        name="Echo",
        description="Echoes the user's message back.",
        input_modes=["text/plain"],
        output_modes=["text/plain"],
        tags=["echo", "tutorial"],
        examples=["hello", "how are you"],
    )

    # The Agent Card -- clients fetch this to discover your agent
    agent_card = AgentCard(
        name="Echo Agent",
        description="A simple echo agent for learning A2A.",
        version="0.1.0",
        default_input_modes=["text/plain"],
        default_output_modes=["text/plain"],
        capabilities=AgentCapabilities(streaming=True),
        supported_interfaces=[
            AgentInterface(
                protocol_binding="JSONRPC",
                url="http://127.0.0.1:9999",
                protocol_version="1.0",
            )
        ],
        skills=[skill],
    )

    # Wire up the request handler
    request_handler = DefaultRequestHandler(
        agent_executor=EchoAgentExecutor(),
        task_store=InMemoryTaskStore(),
        agent_card=agent_card,
    )

    # Create routes
    routes = []
    routes.extend(create_agent_card_routes(agent_card))
    routes.extend(create_jsonrpc_routes(request_handler, "/"))

    app = Starlette(routes=routes)
    uvicorn.run(app, host="127.0.0.1", port=9999)

A few things to note:

• AgentCard is the discovery document. Clients hit .well-known/agent.json to fetch it.
• supported_interfaces tells clients how to reach this agent. We use JSON-RPC on port 9999.
• InMemoryTaskStore keeps task state in memory. Fine for development. In production, swap it for PostgreSQL or MySQL (the SDK supports both).

Step 5: Start the Server

python server.py

You should see uvicorn start up:

INFO:     Started server process [12345]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:9999

Step 6: Test the Agent Card

In a separate terminal:

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

This returns the Agent Card in JSON. You should see the agent name, description, skills, and connection info. This is how any A2A-compatible client discovers what your agent can do without reading documentation.

Step 7: Send a Message

The simplest test is a direct JSON-RPC call:

curl -X POST http://127.0.0.1:9999/ \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "message/send",
    "params": {
      "message": {
        "role": "user",
        "parts": [
          {"type": "text", "text": "Hello from curl"}
        ]
      }
    },
    "id": 1
  }'

The response contains the task with the agent's echo output. You should see something like:

{
  "jsonrpc": "2.0",
  "result": {
    "id": "...",
    "status": {"state": "completed"},
    "artifacts": [
      {
        "parts": [
          {"type": "text", "text": "Echo from A2A: you said 'Hello from curl'"}
        ]
      }
    ]
  }
}

Step 8: Build a Python Client

For a more realistic flow, write a client that discovers the agent, then sends a message. Create client.py:

import asyncio

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


async def main():
    # 1. Discover the agent
    async with httpx.AsyncClient() as http_client:
        resolver = A2ACardResolver(
            httpx_client=http_client,
            base_url="http://127.0.0.1:9999",
        )
        agent_card = await resolver.get_agent_card()
        print(f"Connected to: {agent_card.name}")
        print(f"Skills: {[s.name for s in agent_card.skills]}")

    # 2. Create a client and send a message
    config = ClientConfig(streaming=False)
    client = await create_client(agent=agent_card, client_config=config)

    message = new_text_message("Hello from Python client!", role=Role.ROLE_USER)
    request = SendMessageRequest(message=message)

    async for chunk in client.send_message(request):
        print(f"Response: {chunk}")

    await client.close()


asyncio.run(main())

Run it:

python client.py

This is the real A2A flow: discover via Agent Card, then communicate via the protocol. The client does not need to know anything about your server implementation. It only needs the Agent Card and the A2A SDK.

Step 9: Add Streaming Support

If you enabled streaming in the Agent Card (we did), the client can receive responses as they are generated. Change the client config:

config = ClientConfig(streaming=True)
client = await create_client(agent=agent_card, client_config=config)

Streaming is useful when your agent calls an LLM and you want to show tokens as they arrive rather than waiting for the full response.

When to Use A2A vs MCP

This comes up a lot, so here is the short version:

	A2A	MCP
Purpose	Agent-to-agent communication	Agent-to-tool communication
Analogy	Two people talking	A person using a screwdriver
Discovery	Agent Cards	Tool schemas
Use case	Your agent delegates research to a specialist agent	Your agent calls a search API or reads a file
Both needed?	Yes, they complement each other

In practice, an agent might use MCP to access its tools internally, and A2A to delegate work to other agents externally. They are not competing standards.

Production Considerations

A few things to handle before shipping:

1. Swap InMemoryTaskStore for a database. The SDK supports PostgreSQL, MySQL, and SQLite out of the box. In-memory loses everything on restart.

2. Add authentication. The Agent Card supports extended cards that are only visible after authentication. Use this to gate access to sensitive skills.

3. Run behind a reverse proxy. Put Caddy or Nginx in front for TLS termination, rate limiting, and logging.

4. Validate external Agent Cards. Any agent card from a third party is untrusted input. Sanitize fields like name and description before using them in LLM prompts to prevent prompt injection.

5. Add OpenTelemetry. The SDK has optional telemetry support. Install with pip install "a2a-sdk[telemetry]" and wire up tracing for debugging multi-agent workflows.

Where to Go Next

• Run the full samples from the a2a-samples repo to see more complex patterns
• Try the A2A Inspector to visually inspect your agent's capabilities
• Build an agent that uses both MCP (for tools) and A2A (for delegation to other agents)
• Check the DeepLearning.AI course on A2A for guided projects across multiple frameworks

References

• Agent2Agent Protocol Specification
• A2A Python SDK on PyPI
• A2A Samples Repository