A2A Architecture

Architecture and data flow diagrams for the A2A protocol

A2A Protocol Architecture

Overview

The A2A (Agent-to-Agent) protocol enables PraisonAI agents to communicate with other A2A-compatible systems using JSON-RPC 2.0 over HTTP.

Architecture Diagram

Diagram

PraisonAI A2A Server

A2A Client

PraisonAI Agent

Core Modules

JSON-RPC Handlers

FastAPI Router

Discovery

Health

JSON-RPC

Route

Route

Route

Route

Client Code

GET /.well-known/agent.json

GET /status

POST /a2a (JSON-RPC 2.0)

message/send

message/stream

tasks/get

tasks/cancel

AgentCard Generator

TaskStore

Message Converter

SSE Streaming

agent.chat()

Data Flow: message/send

Diagram
AgentConverterTaskStorePOST /a2aA2A ClientAgentConverterTaskStorePOST /a2aA2A Clientmethod: “message/send”JSON-RPC requestValidate jsonrpc == “2.0”create_task(message)Task (submitted)update_status(working)extract_user_input(message)“user text”agent.chat(“user text”)response textpraisonai_to_a2a_message()create_artifact()add_artifact + add_to_historyupdate_status(completed)Task (completed)JSON-RPC response with Task

Data Flow: message/stream

Diagram
AgentSSE StreamerTaskStorePOST /a2aA2A ClientAgentSSE StreamerTaskStorePOST /a2aA2A Clientmethod: “message/stream”JSON-RPC requestcreate_task(message)Task (submitted)stream_agent_response()SSE: event:task.status (working)agent.chat(“user text”)response textSSE: event:task.artifact (response)SSE: event:task.status (completed)SSE: event:done

Task Lifecycle

Diagram

message/send or message/stream

Agent starts processing

Agent responds successfully

Agent error

tasks/cancel

Agent needs more info

User sends follow-up

submitted

working

completed

failed

cancelled

input_required

Module Structure

Diagram

praisonaiagents.ui.a2a

a2a.py
A2A class + router

types.py
Pydantic models

task_store.py
Task CRUD

streaming.py
SSE encoder

conversion.py
Message converter

agent_card.py
Card generator

Quick Setup

python
from praisonaiagents import Agent, A2A
from fastapi import FastAPI
import uvicorn

agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant",
    tools=[search, calculate]
)

# 2 lines to expose as A2A server
a2a = A2A(agent=agent, url="http://localhost:8000/a2a")

app = FastAPI()
app.include_router(a2a.get_router())
uvicorn.run(app, port=8000)

JSON-RPC Methods

MethodDescriptionRequired Params
message/sendSend message, get responseparams.message
message/streamStream response as SSEparams.message
tasks/getGet task by IDparams.id
tasks/cancelCancel task by IDparams.id

Error Codes

CodeMeaning
-32700Parse error (invalid JSON)
-32600Invalid Request (bad jsonrpc)
-32601Method not found
-32602Invalid params
-32603Internal error
-32000Task not found