Home > Notes > AgenticAI > A2A

Detailed Notes on A2A Protocol

A2A was introduced by Google in April 2025 at Google Cloud Next, with contributions from over 50 partners like Atlassian, Cohere, and SAP. A2A emphasizes peer-to-peer agent interactions for complex, multi-agent workflows.

Definition

Components

A2A's design revolves around modular, extensible elements that enable flexible agent interactions. Key components include:

Architecture

A2A's architecture is decentralized and web-native, leveraging existing standards for broad compatibility. It's a client-server model where any agent can act as both client (initiating tasks) and server (fulfilling them), promoting peer-to-peer networks.

How It Works

  1. Discovery: Client fetches Agent Card (e.g., via GET /.well-known/agent-card.json) to learn capabilities and auth.
  2. Authentication: Client obtains tokens (e.g., OAuth) and authenticates requests.
  3. Task Creation: Client sends JSON-RPC request (e.g., "task/create") with Message params; server creates Task and assigns ID.
  4. Collaboration: Agents exchange Messages; server updates via SSE or webhooks.
  5. Completion: Server returns Artifact; Task state changes to "completed."

Python Code Demonstration: Simple A2A Client-Server Interaction

Adapted from GitHub repos like a2aproject/A2A and sap156/Agent-to-Agent-A2A-Protocol-Implementation. This uses FastAPI for the server and requests for the client, simulating task creation and SSE updates.

Server Code (a2a_server.py):

from fastapi import FastAPI, Response
from pydantic import BaseModel
import uvicorn
import json
import time
from sse_starlette.sse import EventSourceResponse

app = FastAPI(title="Simple A2A Server")

# Agent Card Endpoint
@app.get("/.well-known/agent-card.json")
def get_agent_card():
    return {
        "name": "Task Agent",
        "description": "Handles simple tasks",
        "url": "http://localhost:8000",
        "capabilities": {"streaming": True},
        "skills": [{"id": "process_task", "name": "Process Task", "description": "Processes a text task"}]
    }

class TaskRequest(BaseModel):
    message: dict

# Task Creation (JSON-RPC style)
@app.post("/task/create")
def create_task(request: dict):
    # Simulate task creation
    task_id = "task-123"
    return {"jsonrpc": "2.0", "result": {"id": task_id, "status": {"state": "submitted"}}}

# SSE for Updates
@app.get("/task/{task_id}/stream")
async def stream_task(task_id: str):
    async def event_generator():
        for state in ["running", "completed"]:
            yield json.dumps({"status": {"state": state}})
            time.sleep(2)
    return EventSourceResponse(event_generator())

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Client Code (a2a_client.py):

import requests
import sseclient

SERVER_URL = "http://localhost:8000"

# Fetch Agent Card
card_response = requests.get(f"{SERVER_URL}/.well-known/agent-card.json")
print("Agent Card:", card_response.json())

# Create Task
task_request = {
    "jsonrpc": "2.0",
    "id": 1,
    "method": "task/create",
    "params": {"message": {"role": "user", "parts": [{"kind": "text", "text": "Process this"}]}}
}
create_response = requests.post(f"{SERVER_URL}/task/create", json=task_request)
task_id = create_response.json()["result"]["id"]
print("Task Created:", task_id)

# Stream Updates
stream_response = requests.get(f"{SERVER_URL}/task/{task_id}/stream", stream=True)
client = sseclient.SSEClient(stream_response)
for event in client.events():
    print("Update:", event.data)

Run server first, then client to see discovery, task creation, and streaming.

Advantages

Disadvantages

Use Cases