Search⌘ K
AI Features
Home
Courses
Introduction to Google’s Agent2Agent (A2A) Protocol

Building A2A Compliant Client and Server

Explore how to create a fully A2A-compliant client and server from scratch. Learn agent discovery, JSON-RPC communication, structured message validation, and task lifecycle management using Python tools like FastAPI and Pydantic. This lesson helps you build foundational A2A agents capable of interoperable AI interactions.

We'll cover the following...

Now that we understand A2A’s core concepts, it’s time to build something real. From scratch, we’ll create a complete A2A-compliant system: a simple echo agent that receives messages from A2A clients over HTTP and responds with echoed content, paired with a client that discovers and communicates with it.

We’ll build incrementally, starting with the bare minimum and adding A2A compliance step-by-step. By the end, we’ll have a working agent that any A2A client can discover and use, and a client that can talk to any A2A-compliant agent. Along the way, we’ll explain each library, pattern, and design choice so you understand what to code and why.

What are we building?

Our goal is straightforward: create the simplest possible A2A interaction. We’ll build an echo agent that receives text messages and responds with “You said: [your message].” This might seem trivial, but it demonstrates all the essential A2A patterns:

  • Agent discovery through a standard endpoint.

  • JSON-RPC 2.0 communication protocol.

  • Task life cycle management (submitted to completed).

  • Structured message exchange with proper parts.

  • Error handling for malformed requests.

Think of this as the “hello world” of agent-to-agent communication, simple enough to understand completely, but comprehensive enough to serve as a foundation for more complex agents.

Before we start coding, ensure that we have the right tools. We’ll use Python with a few key libraries that handle the heavy lifting:

# Create a clean workspace
mkdir a2a-echo-example
cd a2a-echo-example
# Set up virtual environment
python3 -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate
# Install core dependencies
pip install fastapi uvicorn pydantic requests
Package installation

Here’s what each library does.

  • FastAPI: A modern Python web framework that makes building APIs easy. It handles HTTP routing, request validation, and automatic API documentation. We chose it because it’s beginner-friendly, has excellent type checking, and integrates seamlessly with A2A’s JSON-based communication.

  • Uvicorn: An ASGI server that runs our FastAPI application. Think of it as the engine that listens for HTTP requests and routes them to our agent code. It’s lightweight, fast, and perfect for development.

  • Pydantic: A data validation library which ensures that our A2A messages have the correct structure. It catches errors early and provides clear feedback when something goes wrong, which is essential for protocol compliance.

  • Requests: The standard Python library for making HTTP calls. Our client will use it to discover agents and send messages.

These libraries handle networking, validation, and HTTP complexity, leaving us free to focus on A2A-specific logic.

How to build the A2A server

We’ll start with the absolute minimum and build up to full A2A compliance. Create the server in a file named echo_agent.py.

Python 3.10.4
"""
Simple A2A Server Example - Step 1: Basic Web Server
"""
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def hello():
    return {"message": "Hello from our future A2A agent!"}
if __name__ == "__main__":
    import uvicorn
    print("🚀 Starting basic web server on http://localhost:8000")
    uvicorn.run(app, host="0.0.0.0", port=8000)

In the code above:

  • Lines 4–6: Loads the framework and creates the app instance that handles incoming HTTP requests.

  • Lines 8: Tell FastAPI to call the next function when it receives a GET request at /.

  • Lines 9–10: Returns a dictionary, which FastAPI automatically converts to JSON.

  • Line 12: Ensures this code runs only when executed directly, and not when imported.

  • Lines 13–15: Starts the HTTP server and listens on port 8000.

When we run it with the python echo_agent.py command, then visit http://localhost:8000. We should see:

{"message": "Hello from our future A2A agent!"}

FastAPI automatically parses the request, calls the function, converts the Python dictionary to JSON, sets the proper headers, and returns the response.

It’s not A2A-compliant yet, but it’s our foundation. Now, we’ll add A2A features step-by-step.

How to make the system A2A compliant

To make the agent discoverable, add the standard A2A discovery endpoint:

Python 3.10.4
# NEW: Add this endpoint for A2A discovery
@app.get("/.well-known/agent-card.json")
def agent_card():
    """A2A Agent Discovery Endpoint"""
    return {
        "name": "Echo Agent",
        "description": "A simple agent that echoes your messages back",
        "url": "http://localhost:8000",
        "version": "1.0.0",
        "protocolVersion": "0.3.0",
        "skills": [{
            "id": "echo",
            "name": "Echo Messages", 
            "description": "Repeats whatever you say",
            "examples": ["Hello", "How are you?"]
        }],
        "capabilities": {"streaming": False},
        "defaultInputModes": ["text/plain"],
        "defaultOutputModes": ["text/plain"]
    }
# UPDATE: Modify the hello message
@app.get("/")
def hello():
    return {"message": "Hello from our A2A agent!"}

In the code above:

  • Lines 2–9: Create the standard discovery endpoint at /.well-known/agent-card.json.

  • Lines 11–26: Return the agent’s metadata and capabilities.

  • Lines 13–17: Define basic identity information.

  • Lines 18–22: Provide skill definitions with examples.

  • Lines 23–25: List capabilities and data formats.

If we restart the server now, we can test both endpoints:

  • http://localhost:8000/.

  • http://localhost:8000/.well-known/agent-card.json.

Any A2A client can now discover our agent by fetching this endpoint.

How to add message structure validation

Now, we’ll add validation for A2A messages. Include these imports and models near the top:

Python 3.10.4
from pydantic import BaseModel
from typing import List
import uuid
from datetime import datetime
class Part(BaseModel):
    kind: str
    text: str = None
class Message(BaseModel):
    kind: str = "message"
    messageId: str
    role: str
    parts: List[Part]
class MessageParams(BaseModel):
    message: Message
class JSONRPCRequest(BaseModel):
    jsonrpc: str = "2.0"
    id: str
    method: str
    params: MessageParams

In the code above:

  • Lines 1–4: Import dependencies for data validation and unique IDs.

  • Lines 6–8: Define a Part model for message segments.

  • Lines 10–18: Define a Message model with ID, role, and content.

  • Lines 19–23: Wrap messages in a JSON-RPC 2.0 request.

Then, add this POST endpoint:

Python 3.10.4
@app.post("/")
def handle_message(request: JSONRPCRequest):
    """A2A Message Handler - Coming Next!"""
    return {"jsonrpc": "2.0", "id": request.id, "result": "Handler coming soon!"}

FastAPI now validates incoming messages automatically.

How to implement the agent logic

We can simply add a POST handler for our agent logic:

@app.post("/")
def handle_message(request: JSONRPCRequest):
    """A2A Message Handler"""
    # Validate the method
    if request.method != "message/send":
        return {
            "jsonrpc": "2.0", 
            "id": request.id, 
            "error": {
                "code": -32601, 
                "message": "Method not found"
            }
        }
    
    # Extract the user's message
    user_message = request.params.message
    user_text = user_message.parts[0].text if user_message.parts else "No text"
    
    # Create our A2A response as a completed task
    task = {
        "kind": "task",
        "id": str(uuid.uuid4()),
        "contextId": str(uuid.uuid4()),
        "status": {
            "state": "completed", 
            "timestamp": datetime.utcnow().isoformat() + "Z"
        },
        "history": [
            # Include the original user message
            user_message.dict(),
            # Add our agent response
            {
                "kind": "message",
                "messageId": str(uuid.uuid4()),
                "role": "agent",
                "parts": [{"kind": "text", "text": f"You said: '{user_text}'"}]
            }
        ]
    }
    
    return {"jsonrpc": "2.0", "id": request.id, "result": task}
POST handler for agent

In the code above:

  • Lines 5–13: Validate method is “message/send” and return JSON-RPC error (-32601) if not.

  • Lines 16–17: Extract user message from request parameters and safely get text content.

  • Line 20: Create a complete A2A task object with IDs, status, and conversation history.

  • Lines 21–24: Generate unique task and context identifiers using UUID4.

  • Lines 25–25: Mark task as “completed” with ISO 8601 timestamp.

  • Lines 28–38: Build conversation history with original user message and agent response.

  • Line 41: Return task wrapped in JSON-RPC 2.0 response envelope.

This is now a complete, A2A-compliant agent. The key insight is that A2A agents respond with task objects, not simple text. This provides structure, traceability, and room for complex interactions later.

Here’s the complete echo_agent.py after all incremental steps:

Python 3.10.4
"""
Simple A2A Server Example - Complete Implementation
"""
from fastapi import FastAPI
from pydantic import BaseModel
from typing import List
import uuid
from datetime import datetime
app = FastAPI()
# A2A Message Structure Models
class Part(BaseModel):
    kind: str
    text: str = None
class Message(BaseModel):
    kind: str = "message"
    messageId: str
    role: str
    parts: List[Part]
class MessageParams(BaseModel):
    message: Message
class JSONRPCRequest(BaseModel):
    jsonrpc: str = "2.0"
    id: str
    method: str
    params: MessageParams
@app.get("/.well-known/agent-card.json")
def agent_card():
    """A2A Agent Discovery Endpoint"""
    return {
        "name": "Echo Agent",
        "description": "A simple agent that echoes your messages back",
        "url": "http://localhost:8000",
        "version": "1.0.0",
        "protocolVersion": "0.3.0",
        "skills": [{
            "id": "echo",
            "name": "Echo Messages",
            "description": "Repeats whatever you say",
            "examples": ["Hello", "How are you?"]
        }],
        "capabilities": {"streaming": False},
        "defaultInputModes": ["text/plain"],
        "defaultOutputModes": ["text/plain"]
    }
@app.get("/")
def hello():
    return {"message": "Hello from our A2A agent! Visit /.well-known/agent-card.json to see my capabilities."}
@app.post("/")
def handle_message(request: JSONRPCRequest):
    """A2A Message Handler"""
    # Validate the method
    if request.method != "message/send":
        return {
            "jsonrpc": "2.0", 
            "id": request.id, 
            "error": {
                "code": -32601, 
                "message": "Method not found"
            }
        }
    
    # Extract the user's message
    user_message = request.params.message
    user_text = user_message.parts[0].text if user_message.parts else "No text"
    
    # Create our A2A response as a completed task
    task = {
        "kind": "task",
        "id": str(uuid.uuid4()),
        "contextId": str(uuid.uuid4()),
        "status": {
            "state": "completed", 
            "timestamp": datetime.utcnow().isoformat() + "Z"
        },
        "history": [
            # Include the original user message
            user_message.dict(),
            # Add our agent response
            {
                "kind": "message",
                "messageId": str(uuid.uuid4()),
                "role": "agent",
                "parts": [{"kind": "text", "text": f"You said: '{user_text}'"}]
            }
        ]
    }
    
    return {"jsonrpc": "2.0", "id": request.id, "result": task}
if __name__ == "__main__":
    import uvicorn
    print("🤖 Starting Simple A2A Echo Agent on http://localhost:8000")
    uvicorn.run(app, host="0.0.0.0", port=8000)

How to build the A2A client

Now, let’s create a client that can discover our agent and send it messages. We can create this in a file named simple_a2a_client.py:

Python 3.10.4
import requests
import uuid
import json
def main():
    # Step 1: Discover the Agent
    base_url = "http://localhost:8000"
    print("🔍 Discovering A2A agent...")
    
    try:
        agent_card = requests.get(f"{base_url}/.well-known/agent-card.json").json()
        print(f"✅ Found: {agent_card['name']} - {agent_card['description']}")
    except requests.RequestException as e:
        print(f"❌ Discovery failed: {e}")
        return
    
    # Step 2: Send a Message using A2A JSON-RPC
    print("\n💬 Sending message...")
    
    message = {
        "jsonrpc": "2.0",
        "id": str(uuid.uuid4()),
        "method": "message/send",
        "params": {
            "message": {
                "kind": "message",
                "messageId": str(uuid.uuid4()),
                "role": "user",
                "parts": [{"kind": "text", "text": "Hello A2A world!"}]
            }
        }
    }
    
    try:
        response = requests.post(base_url, json=message).json()
    except requests.RequestException as e:
        print(f"❌ Communication failed: {e}")
        return
    
    # Step 3: Handle the Response
    if "result" in response:
        task = response["result"] 
        print(f"📋 Task Status: {task['status']['state']}")
        
        # Find agent's reply in history
        for msg in task.get("history", []):
            if msg.get("role") == "agent":
                agent_reply = msg["parts"][0]["text"]
                print(f"🤖 Agent: {agent_reply}")
                break
    else:
        error = response.get("error", {})
        print(f"❌ Error: {error.get('message', 'Unknown error')}")
    
    print("\n✨ A2A communication complete!")
if __name__ == "__main__":
    main()

In the code above:

  • Lines 6–8: Import libraries for HTTP requests, unique IDs, and JSON handling.

  • Lines 12–18: Discover agent by fetching Agent Card from standard .well-known/agent-card.json endpoint.

  • Lines 21–34: Build complete JSON-RPC 2.0 request with nested A2A message structure.

  • Lines 25–26: Create outer JSON-RPC envelope with protocol version and unique request ID.

  • Lines 29–33: Create inner A2A message with user role, message ID, and text content.

  • Lines 36–40: Send JSON request to agent endpoint and parse response.

  • Lines 43–54: Extract task status and find agent’s reply in conversation history.

This client works with any A2A-compliant agent because it follows the standard discovery and communication protocols.

Testing our A2A system

Let’s see everything working together.

  1. Start the server in one terminal:

python echo_agent.py

  1. Run the client in another terminal:

cd app
python simple_a2a_client.py

You should see output like:

🔍 Discovering A2A agent...
✅ Found: Echo Agent - A simple agent that echoes your messages back
💬 Sending message...
📋 Task Status: completed
🤖 Agent: You said: 'Hello A2A world!'
✨ A2A communication complete!

You can also test discovery manually:

curl http://0.0.0.0:8000/.well-known/agent-card.json

Our simple echo agent demonstrates all the core A2A patterns, mentioned below.

  • Protocol compliance: We use JSON-RPC 2.0, standard discovery endpoints, and proper message structures.

  • Discoverability: Any A2A client can find and understand our agent through the Agent Card.

  • Structured communication: Messages have roles, unique IDs, and typed content parts.

  • Task management: Even simple responses are wrapped in task objects with proper life cycle tracking.

  • Error handling: We return standard JSON-RPC errors for invalid requests.

  • Extensibility: Our foundation can easily grow to support streaming, file handling, multi-turn conversations, and more complex skills.

Go ahead and observe the output yourself in the terminal below:

When the terminal connects you would see a “+” button right beside Terminal 1 tab, and you can run the second command in that tab to see the output of both of the commands yourself.

Terminal 1
Terminal
Loading...

As we build more sophisticated agents, keep the patterns below in mind.

  • Unique IDs everywhere: Tasks, messages, and contexts all need unique identifiers for tracking and debugging.

  • Fail gracefully: Return structured errors instead of crashing. Clients need predictable error formats.

  • Follow the spec: A2A’s value comes from consistency. Small deviations break interoperability.

  • Think in tasks: Even quick operations benefit from the task wrapper; it provides structure for logging, monitoring, and future extensions.

  • Validate early: Use Pydantic models to catch malformed messages before they reach your business logic.

You now have a complete, working A2A system. But this is just the beginning. The echo agent you built today will serve as the foundation for all these advanced topics. You’ve become familiar with the essential patterns, everything else builds on what you already know. More importantly, you’ve joined the A2A ecosystem. Your agent can now communicate with any other A2A-compliant agent, and your client can discover and use agents built by teams around the world. That’s the vision of A2A: a network of specialized, interoperable agents working together to solve complex problems.