A2A Python SDK: Build Your First A2A Agent in Python

Set Up the Project

Start by creating a new Python project and installing the SDK. A clean virtual environment is recommended to avoid dependency conflicts.

Run pip install a2a-sdk to install the core package. The SDK depends on httpx for HTTP client operations, uvicorn for the ASGI server, starlette for HTTP routing, and pydantic for data validation. These dependencies are installed automatically.

A typical project structure separates the server and client into different modules. Create a server.py for your A2A server and a client.py for testing. As your agent grows, you will likely add modules for the actual agent logic (LLM calls, tool use, data processing) separate from the A2A protocol layer.

The SDK's main imports are: AgentCard and AgentSkill for defining your agent's capabilities, A2AServer for running the server, A2AClient for connecting to other agents, and the message/task types (Task, Message, TextPart, DataPart, FilePart, Artifact) for constructing requests and parsing responses.

Build the Agent Card with the SDK

The SDK provides Pydantic models for constructing Agent Cards programmatically. Use the AgentCard class to build a type-safe, validated card instead of manually crafting JSON.

Create an AgentCard instance with the required fields. The name should be a clear, descriptive title. The description should explain what your agent does in enough detail for both humans and LLMs to evaluate its suitability. The url is the endpoint where the agent accepts A2A requests. For local development, use http://localhost:8000.

Define skills using the AgentSkill class. Each skill gets an id (lowercase with hyphens, like "analyze-data"), name (human-readable), and description (detailed capability explanation). Optionally add inputSchema and outputSchema as JSON Schema dictionaries for structured validation.

Set defaultInputModes and defaultOutputModes to declare content type support. Common values include text for natural language, data for structured JSON, and file for binary content. An agent that accepts text and returns text plus data would set input modes to ["text"] and output modes to ["text", "data"].

The SDK validates the Agent Card at construction time. If required fields are missing or field values are invalid, Pydantic raises a clear validation error before the server starts. This catches configuration mistakes early rather than at runtime when a client tries to fetch the card.

Implement the Server with a Task Handler

The server's core component is the task handler, a class or function that receives incoming tasks and produces results. The SDK defines a handler interface that your implementation must follow.

The standard approach is to subclass the SDK's AgentExecutor (or similar base class, depending on SDK version). Your subclass implements the execute method, which receives the incoming Task object and returns an updated task with results. Inside this method, you extract the client's message from the task, process it using your agent's logic, and construct artifacts containing the output.

Extracting input from the task follows a consistent pattern. Access task.messages to get the list of messages, take the last message (which is the most recent client input), and iterate over its parts to find the content. Check the part type: TextPart for text input, DataPart for structured JSON, FilePart for file content.

Constructing the response follows the reverse pattern. Create an Artifact with a list of parts containing your output. Wrap it in the task update along with the completed status. The SDK handles serializing this into the correct JSON-RPC response format.

For agents that need clarification, return the task with input-required status and a message asking the client what you need to know. When the client responds, the SDK calls your handler again with the updated task containing the new message. Check the message history to distinguish initial requests from follow-up responses.

Wire your handler to the server by passing it to the A2AServer constructor along with the Agent Card. Call server.start() to launch. The server automatically configures routes for the Agent Card endpoint and the A2A JSON-RPC endpoint.

Build the Client

The SDK's A2AClient class simplifies connecting to A2A servers. Initialize it with the server's base URL, and the client handles Agent Card fetching, JSON-RPC request formatting, and response parsing.

Fetching the Agent Card is the first step. Call client.get_agent_card() to retrieve and parse the server's Agent Card. Examine the returned AgentCard object to verify the server has the skills you need. In orchestration scenarios, you might fetch cards from multiple servers and choose the best one based on skill descriptions.

Submitting a task requires constructing a Task or SendTaskRequest with a unique task ID and a message containing your instructions. The SDK provides helper functions for generating task IDs. Add parts to your message: a TextPart with natural language instructions, a DataPart with structured parameters, or a FilePart with file content. Call client.send_task() to submit and wait for the response.

Processing the response means examining the returned task object. Check the status field: completed means results are in the artifacts, input-required means the server needs more information, and failed means something went wrong. For completed tasks, iterate over task.artifacts and extract the parts from each artifact.

The SDK's client is async-first, using Python's asyncio. All network operations return coroutines that must be awaited. For simple scripts, wrap your client code in an async def main() function and run it with asyncio.run(main()).

Add Streaming and Advanced Features

Streaming replaces the single-response pattern with a continuous event stream. On the client side, use client.send_task_subscribe() instead of send_task(). This returns an async iterator that yields events as the server produces them.

Each event has a type. TaskStatusUpdateEvent carries a new task status and optional message (useful for progress updates). TaskArtifactUpdateEvent carries a new or updated artifact (useful for partial results). Your client code iterates over these events and handles each type appropriately.

On the server side, streaming requires modifying your handler to yield events instead of returning a single result. The SDK provides a streaming handler interface where your implementation is an async generator. Yield status updates as you work, yield partial artifacts when portions of the output are ready, and yield the final completed status with the full artifact when done.

For long-running tasks, implement push notification support. The SDK's server automatically handles the tasks/pushNotification/set and tasks/pushNotification/get methods. When a client registers a webhook URL, the server stores it and sends HTTP POST notifications to that URL whenever the task state changes. Your handler code does not need to manage notifications directly, the SDK handles them based on your handler's status updates and artifact yields.

Additional SDK features include: middleware support for adding logging, metrics, or authentication checks to all incoming requests; error handling utilities for constructing protocol-compliant error responses; and task storage adapters for persisting tasks to databases instead of the default in-memory store.