Python A2A Tutorial 20250513

Welcome to the Agent2Agent (A2A) Python Quickstart Tutorial!

In this tutorial, you will explore a simple "echo" A2A server using the Python SDK. This will introduce you to the fundamental concepts and components of an A2A server. You will then look at a more advanced example that integrates a Large Language Model (LLM).

This hands-on guide will help you understand:

The basic concepts behind the A2A protocol.
How to set up a Python environment for A2A development using the SDK.
How Agent Skills and Agent Cards describe an agent.
How an A2A server handles tasks.
How to interact with an A2A server using a client.
How streaming capabilities and multi-turn interactions work.
How an LLM can be integrated into an A2A agent.

By the end of this tutorial, you will have a functional understanding of A2A agents and a solid foundation for building or integrating A2A-compliant applications.

Introduction
Setup Your Environment
Agent Skills & Agent Card
The Agent Executor
Starting the Server
Interacting with the Server
Streaming & Multi-Turn Interactions
Next Steps

Introduction

The A2A protocol provides a standardized way for AI agents to discover each other and communicate. In this tutorial, we'll build a simple A2A agent using Python, then see how it can be extended with more advanced features.

Our tutorial will focus on:

Building a basic "Hello World" agent that responds with a simple message
Setting up the A2A server and client to facilitate communication
Exploring more advanced features like streaming and multi-turn interactions
Integrating a real LLM using LangGraph to power a more sophisticated agent

Setup Your Environment

Prerequisites

Python 3.10 or higher.
Access to a terminal or command prompt.
Git, for cloning the repository.
A code editor (e.g., VS Code) is recommended.

Clone the Repository

If you haven't already, clone the A2A repository and navigate to the Python SDK directory:

git clone https://github.com/google/A2A.git -b main --depth 1
cd A2A/a2a-python-sdk

Python Environment & SDK Installation

We recommend using a virtual environment for Python projects. The A2A Python SDK uses uv for dependency management, but you can use pip with venv as well.

Create and activate a virtual environment:

Using venv (standard library):

Mac/Linux

python -m venv .venv
source .venv/bin/activate

Windows

python -m venv .venv
.venv\Scripts\activate

Install the A2A SDK and its dependencies:

The a2a-python-sdk directory contains the SDK source code. To make it and its dependencies available in your environment, run:
```
pip install -e .[dev]
```
This command installs the SDK in "editable" mode (-e), meaning changes to the SDK source code are immediately available. It also installs development dependencies specified in pyproject.toml.

Verify Installation

After installation, you should be able to import the a2a package in a Python interpreter:

python -c "import a2a; print('A2A SDK imported successfully')"

If this command runs without error and prints the success message, your environment is set up correctly.

Agent Skills & Agent Card

Before an A2A agent can do anything, it needs to define what it can do (its skills) and how other agents or clients can find out about these capabilities (its Agent Card).

We'll use the helloworld example located in a2a-python-sdk/examples/helloworld/.

Agent Skills

An Agent Skill describes a specific capability or function the agent can perform. It's a building block that tells clients what kinds of tasks the agent is good for.

Key attributes of an AgentSkill (defined in a2a.types):

id: A unique identifier for the skill.
name: A human-readable name.
description: A more detailed explanation of what the skill does.
tags: Keywords for categorization and discovery.
examples: Sample prompts or use cases.
inputModes / outputModes: Supported MIME types for input and output (e.g., "text/plain", "application/json").

In examples/helloworld/__main__.py, you can see how a skill for the Helloworld agent is defined:

# examples/helloworld/__main__.py
# ...
    skill = AgentSkill(
        id='hello_world',
        name='Returns hello world',
        description='just returns hello world',
        tags=['hello world'],
        examples=['hi', 'hello world'],
    )
# ...

This skill is very simple: it's named "Returns hello world" and primarily deals with text.

Agent Card

The Agent Card is a JSON document that an A2A Server makes available, typically at a .well-known/agent.json endpoint. It's like a digital business card for the agent.

Key attributes of an AgentCard (defined in a2a.types):

name, description, version: Basic identity information.
url: The endpoint where the A2A service can be reached.
capabilities: Specifies supported A2A features like streaming or pushNotifications.
authentication: Details on how clients should authenticate.
defaultInputModes / defaultOutputModes: Default MIME types for the agent.
skills: A list of AgentSkill objects that the agent offers.

The helloworld example defines its Agent Card like this:

# examples/helloworld/__main__.py
# ...
    agent_card = AgentCard(
        name='Hello World Agent',
        description='Just a hello world agent',
        url='http://localhost:9999/', # Agent will run here
        version='1.0.0',
        defaultInputModes=['text'],
        defaultOutputModes=['text'],
        capabilities=AgentCapabilities(), # Basic capabilities
        skills=[skill], # Includes the skill defined above
        authentication=AgentAuthentication(schemes=['public']), # No auth needed
    )
# ...

This card tells us the agent is named "Hello World Agent", runs at http://localhost:9999/, supports text interactions, and has the hello_world skill. It also indicates public authentication, meaning no specific credentials are required.

Understanding the Agent Card is crucial because it's how a client discovers an agent and learns how to interact with it.

The Agent Executor

The core logic of how an A2A agent processes requests and generates responses is handled by an Agent Executor. The a2a-python-sdk provides an abstract base class a2a.server.AgentExecutor that you implement.

`AgentExecutor` Interface

The AgentExecutor class defines methods that correspond to different A2A RPC calls:

async def on_message_send(...): Handles standard request/response messages (message/send).
async def on_message_stream(...): Handles requests that expect a streaming response (message/sendStream).
async def on_cancel(...): Handles requests to cancel a task (tasks/cancel).
async def on_resubscribe(...): Handles requests to resubscribe to a task's stream (tasks/resubscribe).

Helloworld Agent Executor

Let's look at examples/helloworld/agent_executor.py. It defines HelloWorldAgentExecutor.

The Agent (HelloWorldAgent): This is a simple helper class that encapsulates the actual "business logic".

# examples/helloworld/agent_executor.py
class HelloWorldAgent:
    async def invoke(self):
        return 'Hello World'

    async def stream(self) -> AsyncGenerator[dict[str, Any], None]:
        yield {'content': 'Hello ', 'done': False}
        await asyncio.sleep(2) # Simulate work
        yield {'content': 'World', 'done': True}

It has an invoke method for single responses and a stream method for generating multiple chunks.

The Executor (HelloWorldAgentExecutor): This class implements the AgentExecutor interface.

__init__:

# examples/helloworld/agent_executor.py
class HelloWorldAgentExecutor(AgentExecutor):
    def __init__(self):
        self.agent = HelloWorldAgent()

It instantiates the HelloWorldAgent.

on_message_send:

# examples/helloworld/agent_executor.py
async def on_message_send(
    self, request: SendMessageRequest, task: Task | None
) -> SendMessageResponse:
    result = await self.agent.invoke() # Calls the agent's logic

    message: Message = Message( # Constructs the A2A Message
        role=Role.agent,
        parts=[Part(root=TextPart(text=result))],
        messageId=str(uuid4()),
    )
    # Wraps it in a success response
    return SendMessageResponse(
        root=SendMessageSuccessResponse(id=request.id, result=message)
    )

When a non-streaming message/send request comes in:

It calls self.agent.invoke() to get the "Hello World" string.
It constructs an A2A Message object with the agent's role and the result as a TextPart.
It wraps this Message in a SendMessageSuccessResponse.

on_message_stream:

# examples/helloworld/agent_executor.py
async def on_message_stream( # type: ignore
    self, request: SendMessageStreamingRequest, task: Task | None
) -> AsyncGenerator[SendMessageStreamingResponse, None]:
    async for chunk in self.agent.stream(): # Iterates over agent's stream
        message: Message = Message(
            role=Role.agent,
            parts=[Part(root=TextPart(text=chunk['content']))],
            messageId=str(uuid4()),
            final=chunk['done'], # Indicates if this is the last chunk
        )
        # Yields each chunk as a streaming success response
        yield SendMessageStreamingResponse(
            root=SendMessageStreamingSuccessResponse(
                id=request.id, result=message
            )
        )

When a streaming message/sendStream request is received:

It iterates through the chunks produced by self.agent.stream().
For each chunk, it creates an A2A Message. The final attribute of the message is important for streaming; it tells the client if more chunks are coming.
Each message is yielded, wrapped in a SendMessageStreamingSuccessResponse. This is how SSE (Server-Sent Events) are generated by the SDK.

on_cancel and on_resubscribe: The Helloworld example marks these as UnsupportedOperationError because it doesn't implement task cancellation or resubscription.

# examples/helloworld/agent_executor.py
# ...
async def on_cancel(
    self, request: CancelTaskRequest, task: Task
) -> CancelTaskResponse:
    return CancelTaskResponse(
        root=JSONRPCErrorResponse(
            id=request.id, error=UnsupportedOperationError()
        )
    )
# ... similar for on_resubscribe

The AgentExecutor acts as the bridge between the raw A2A protocol requests/responses and your agent's specific logic.

Starting the Server

Now that we have an Agent Card and an Agent Executor, we can set up and start the A2A server.

The a2a-python-sdk provides an A2AServer class that simplifies running an A2A-compliant HTTP server. It uses Starlette and Uvicorn under the hood.

Server Setup in Helloworld

Let's look at examples/helloworld/__main__.py again to see how the server is initialized and started.

# examples/helloworld/__main__.py
from agent_executor import HelloWorldAgentExecutor

from a2a.server import A2AServer, DefaultA2ARequestHandler
from a2a.types import (
    # ... other imports ...
    AgentCard,
    # ...
)

if __name__ == '__main__':
    # ... AgentSkill and AgentCard definition from previous steps ...
    skill = AgentSkill(...)
    agent_card = AgentCard(...)

    # 1. Request Handler
    request_handler = DefaultA2ARequestHandler(
        agent_executor=HelloWorldAgentExecutor()
    )

    # 2. A2A Server
    server = A2AServer(agent_card=agent_card, request_handler=request_handler)

    # 3. Start Server
    server.start(host='0.0.0.0', port=9999)

Let's break this down:

DefaultA2ARequestHandler:
- The SDK provides DefaultA2ARequestHandler. This handler takes your AgentExecutor implementation (here, HelloWorldAgentExecutor) and routes incoming A2A RPC calls to the appropriate methods (on_message_send, on_message_stream, etc.) on your executor.
- It also manages task persistence if a TaskStore is provided (which Helloworld doesn't use explicitly, so an in-memory one is used by default within the handler for streaming contexts).
A2AServer:
- The A2AServer class is instantiated with the agent_card and the request_handler.
- The agent_card is crucial because the server will expose it at the /.well-known/agent.json endpoint.
- The request_handler is responsible for processing all incoming A2A method calls.
server.start():
- This method starts the Uvicorn server, making your agent accessible over HTTP.
- host='0.0.0.0' makes the server accessible on all network interfaces on your machine.
- port=9999 specifies the port to listen on. This matches the url in the AgentCard.

Running the Helloworld Server

Navigate to the a2a-python-sdk directory in your terminal (if you're not already there) and ensure your virtual environment is activated.

To run the Helloworld server:

python examples/helloworld/__main__.py

You should see output similar to this, indicating the server is running:

INFO:     Started server process [xxxxx]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:9999 (Press CTRL+C to quit)

Your A2A Helloworld agent is now live and listening for requests! In the next step, we'll interact with it.

Interacting with the Server

With the Helloworld A2A server running, let's send some requests to it. The SDK includes a client (A2AClient) that simplifies these interactions.

The Helloworld Test Client

The examples/helloworld/test_client.py script demonstrates how to:

Fetch the Agent Card from the server.
Create an A2AClient instance.
Send both non-streaming (message/send) and streaming (message/sendStream) requests.
Handle task-related operations like get_task and cancel_task (though Helloworld doesn't fully support these).

Open a new terminal window, activate your virtual environment, and navigate to the a2a-python-sdk directory.

Activate virtual environment (Be sure to do this in the same directory where you created the virtual environment):

Mac/Linux

source .venv/bin/activate

Windows

.venv\Scripts\activate

Run the test client:

python examples/helloworld/test_client.py

Understanding the Client Code

Let's look at key parts of examples/helloworld/test_client.py:

Fetching the Agent Card & Initializing the Client:
```
# examples/helloworld/test_client.py
async with httpx.AsyncClient() as httpx_client:
    client = await A2AClient.get_client_from_agent_card_url(
        httpx_client, 'http://localhost:9999'
    )
```
The A2AClient.get_client_from_agent_card_url class method is a convenience. It first fetches the AgentCard from the server's /.well-known/agent.json endpoint (based on the provided base URL) and then initializes the client with it.

Sending a Non-Streaming Message (send_message):

# examples/helloworld/test_client.py
send_message_payload: dict[str, Any] = {
    'message': {
        'role': 'user',
        'parts': [{'type': 'text', 'text': 'how much is 10 USD in INR?'}], # Content doesn't matter for Helloworld
        'messageId': uuid4().hex,
    },
    # 'id' for the task can also be provided here. If not, the server/handler might generate one.
}

response = await client.send_message(payload=send_message_payload)
print(response.model_dump(mode='json', exclude_none=True))

The payload constructs the params for the message/send RPC call.
It includes a message object with the role set to "user" and the content in parts.
The Helloworld agent will simply echo "Hello World" back.
The response will be a SendMessageResponse object, which contains either a SendMessageSuccessResponse (with the agent's Message as the result) or a JSONRPCErrorResponse.

Handling Task IDs (Illustrative): The Helloworld client attempts to demonstrate get_task and cancel_task.

# examples/helloworld/test_client.py
# ... (after send_message)
if isinstance(response.root, SendMessageSuccessResponse) and isinstance(
    response.root.result, Task # If the agent returned a Task object
):
    task_id: str = response.root.result.id
    # ... client.get_task(...) and client.cancel_task(...) ...
else:
    # Helloworld send_message returns a direct Message, not a Task object,
    # so this branch will be taken.
    print(
        'Received an instance of Message, getTask and cancelTask are not applicable for invocation'
    )

Important Note: The Helloworld HelloWorldAgentExecutor.on_message_send returns a direct Message as the result, not a Task object. More complex agents that manage long-running operations would typically return a Task object, whose id could then be used for get_task or cancel_task. The langgraph example demonstrates this.

Sending a Streaming Message (send_message_streaming):
```
# examples/helloworld/test_client.py
stream_response = client.send_message_streaming(
    payload=send_message_payload # Same payload can be used
)
async for chunk in stream_response:
    print(chunk.model_dump(mode='json', exclude_none=True))
```
- This method calls the agent's message/sendStream endpoint.
- It returns an AsyncGenerator. As the server streams SSE events, the client receives them as SendMessageStreamingResponse objects.
- The Helloworld agent will stream "Hello " and then "World".

Expected Output

When you run test_client.py, you'll see JSON outputs for:

The non-streaming response (a single "Hello World" message).
A message indicating that get_task and cancel_task are not applicable because the non-streaming Helloworld returns a direct message.
The streaming responses (two chunks: "Hello " and then "World", each wrapped in an A2A message structure).

{'id': '67f37deb381f4cae8c26e78e4d95bdb3', 'jsonrpc': '2.0', 'result': {'messageId': '7dd074d5-aff5-41c6-828a-f3a38325c46b', 'parts': [{'text': 'Hello World', 'type': 'text'}], 'role': 'agent', 'type': 'message'}}
Received an instance of Message, getTask and cancelTask are not applicable for invocation
{'id': '39aacacac4914ba4ac75a00e0a870615', 'jsonrpc': '2.0', 'result': {'final': False, 'messageId': '9a95f6e6-9577-46d7-b814-31a61efbd1d7', 'parts': [{'text': 'Hello ', 'type': 'text'}], 'role': 'agent', 'type': 'message'}}
{'id': '39aacacac4914ba4ac75a00e0a870615', 'jsonrpc': '2.0', 'result': {'final': True, 'messageId': 'a55c44da-dcda-47ac-a255-0f314a5de8c9', 'parts': [{'text': 'World', 'type': 'text'}], 'role': 'agent', 'type': 'message'}

This confirms your server is correctly handling basic A2A interactions!

Now you can shut down the server by typing Ctrl+C in the terminal window.

Streaming & Multi-Turn Interactions (LangGraph Example)

The Helloworld example demonstrates the basic mechanics of A2A. For more advanced features like robust streaming, task state management, and multi-turn conversations powered by an LLM, we'll turn to the langgraph example located in a2a-python-sdk/examples/langgraph/.

This example features a "Currency Agent" that uses the Gemini model via LangChain and LangGraph to answer currency conversion questions.

Setting up the LangGraph Example

Create a Gemini API Key, if you don't already have one.
Environment Variable:

Create a .env file in the a2a-python-sdk/examples/langgraph/ directory:
```
# In a2a-python-sdk/examples/langgraph/
echo "GOOGLE_API_KEY=YOUR_API_KEY_HERE" > .env
```
Replace YOUR_API_KEY_HERE with your actual Gemini API key.
Install Dependencies (if not already covered): The langgraph example has its own pyproject.toml which includes dependencies like langchain-google-genai and langgraph. When you installed the SDK, these should have been installed as part of the langgraph-example extras. If you encounter import errors, you might need to install them explicitly from within the a2a-python-sdk/examples/langgraph/ directory:
```
# From a2a-python-sdk/examples/langgraph/
pip install -e .[dev]
```
Typically, the top-level SDK install should cover this.

Running the LangGraph Server

Navigate to the a2a-python-sdk/examples/langgraph/ directory in your terminal and ensure your virtual environment (from the SDK root) is activated.

Start the LangGraph agent server:

# From a2a-python-sdk/examples/langgraph/
python __main__.py

This will start the server, usually on http://localhost:10000.

Interacting with the LangGraph Agent

Open a new terminal window, activate your virtual environment, and navigate to a2a-python-sdk/examples/langgraph/.

Run its test client:

# From a2a-python-sdk/examples/langgraph/
python test_client.py

Now, you can shut down the server by typing Ctrl+C in the terminal window.

Key Features Demonstrated

The langgraph example showcases several important A2A concepts:

LLM Integration:
- examples/langgraph/agent.py defines CurrencyAgent. It uses ChatGoogleGenerativeAI and LangGraph's create_react_agent to process user queries.
- This demonstrates how a real LLM can power the agent's logic.
Task State Management:
- examples/langgraph/__main__.py initializes an InMemoryTaskStore.
- The CurrencyAgentExecutor (in examples/langgraph/agent_executor.py) uses this task_store to save and retrieve Task objects.
- Unlike Helloworld, on_message_send in CurrencyAgentExecutor returns a full Task object. The id of this task can be used for subsequent interactions.
Streaming with TaskStatusUpdateEvent and TaskArtifactUpdateEvent:
- The on_message_stream method in CurrencyAgentExecutor demonstrates a more realistic streaming scenario.
- As the LangGraph agent processes the request (which might involve calling tools like get_exchange_rate), it yields intermediate updates.
- examples/langgraph/helpers.py (process_streaming_agent_response) shows how these agent steps are converted into A2A TaskStatusUpdateEvent (e.g., "Looking up exchange rates...") and TaskArtifactUpdateEvent (when the final answer is ready).
- The test_client.py's run_streaming_test function will print these individual chunks.
Multi-Turn Conversation (TaskState.INPUT_REQUIRED):
- The CurrencyAgent can ask for clarification if a query is ambiguous (e.g., "USD to what?").
- When this happens, the agent_response in CurrencyAgentExecutor will indicate require_user_input: True.
- The Task status will be set to TaskState.input_required.
- The test_client.py run_multi_turn_test function demonstrates this:
  - It sends an initial ambiguous query ("how much is 100 USD?").
  - The agent responds with TaskState.input_required and a message asking for the target currency.
  - The client then sends a second message with the same sessionId (derived from contextId) to provide the missing information ("in GBP").

Exploring the Code

Take some time to look through these files in examples/langgraph/:

__main__.py: Server setup, Agent Card definition (note capabilities.streaming=True).
agent.py: The CurrencyAgent with LangGraph and tool definitions.
agent_executor.py: The CurrencyAgentExecutor implementing A2A methods, managing task state, and handling streaming.
helpers.py: Utility functions for creating and updating task objects and processing agent responses for streaming.
test_client.py: Demonstrates various interaction patterns.

This example provides a much richer illustration of how A2A facilitates complex, stateful, and asynchronous interactions between agents.

Next Steps

Congratulations on completing the A2A Python SDK Tutorial! You've learned how to:

Set up your environment for A2A development.
Define Agent Skills and Agent Cards using the SDK's types.
Implement a basic HelloWorld A2A server and client.
Understand and implement streaming capabilities.
Integrate a more complex agent using LangGraph, demonstrating task state management and tool use.

You now have a solid foundation for building and integrating your own A2A-compliant agents.

Understanding A2A Protocol - A detailed guide to the fundamentals of the A2A protocol
A2A vs MCP: A Comprehensive Comparison - An in-depth comparison between A2A and MCP protocols
Python A2A Tutorial with Source Code - Hands-on implementation guide for A2A protocol in Python
A2A Sample Methods and JSON Responses - Reference guide for A2A protocol methods and response formats

Go A2A

Python A2A Tutorial 20250513

Table of Contents

Introduction

Setup Your Environment

Prerequisites

Clone the Repository

Python Environment & SDK Installation

Verify Installation

Agent Skills & Agent Card

Agent Skills

Agent Card

The Agent Executor

`AgentExecutor` Interface

Helloworld Agent Executor

Starting the Server

Server Setup in Helloworld

Running the Helloworld Server

Interacting with the Server

The Helloworld Test Client

Understanding the Client Code

Expected Output

Streaming & Multi-Turn Interactions (LangGraph Example)

Setting up the LangGraph Example

Running the LangGraph Server

Interacting with the LangGraph Agent

Key Features Demonstrated

Exploring the Code

Next Steps

Related Articles

Python A2A Tutorial 20250513

Table of Contents

Introduction

Setup Your Environment

Prerequisites

Clone the Repository

Python Environment & SDK Installation

Verify Installation

Agent Skills & Agent Card

Agent Skills

Agent Card

The Agent Executor

AgentExecutor Interface

Helloworld Agent Executor

Starting the Server

Server Setup in Helloworld

Running the Helloworld Server

Interacting with the Server

The Helloworld Test Client

Understanding the Client Code

Expected Output

Streaming & Multi-Turn Interactions (LangGraph Example)

Setting up the LangGraph Example

Running the LangGraph Server

Interacting with the LangGraph Agent

Key Features Demonstrated

Exploring the Code

Next Steps

Related Articles

`AgentExecutor` Interface