A2A Protocol

Agent2Agent Protocol for inter-agent communication with JSON-RPC 2.0

PraisonAI supports the A2A Protocol for agent-to-agent communication, enabling your agents to be discovered and collaborate with other AI agents via JSON-RPC 2.0.

python

from praisonaiagents import Agent, A2A

agent = Agent(name="researcher", instructions="Answer research questions via A2A.")
agent.start("Summarise the latest on agent protocols.")

The user publishes an agent as an A2A server; other agents discover it via the agent card and collaborate over JSON-RPC.

Diagram

Note

Import cost: importing just the server side of A2A does not pull in the HTTP client stack.

python

from praisonaiagents.ui.a2a import A2A         # no aiohttp loaded
from praisonaiagents.ui.a2a import A2AClient   # aiohttp loaded on demand

A2AClient is lazy-loaded via the package’s __getattr__ — the first time you reference it, aiohttp is imported. Server-only deployments that never touch A2AClient never pay that cost. aiohttp remains a core dependency; this is an import-time optimisation, not a dependency change.

Diagram

Quick Start

Simple Agent Server

python

from praisonaiagents import Agent, A2A

def search_web(query: str) -> str:
    """Search the web for information."""
    return f"Results for: {query}"

agent = Agent(
    name="Research Assistant",
    role="Research Analyst", 
    goal="Help users research topics",
    tools=[search_web]
)

# Simple one-liner
a2a = A2A(agent=agent)
a2a.serve(port=8000)

FastAPI Integration

python

from fastapi import FastAPI
from praisonaiagents import A2A

app = FastAPI()
a2a = A2A(agent=agent, url="http://localhost:8000/a2a")
app.include_router(a2a.get_router())

JSON-RPC Methods

The A2A server exposes six JSON-RPC 2.0 methods for complete task lifecycle management:

Method	Description
`message/send`	Send a message, get task result
`message/stream`	Send a message, get SSE stream
`tasks/get`	Get task by ID
`tasks/list`	List tasks (optionally by contextId)
`tasks/cancel`	Cancel a running task (returns `-32002` if the task is already terminal)
`agent/getExtendedCard`	Get extended agent card (auth required)

message/send

Send a message and receive the complete task result:

bash

curl -X POST http://localhost:8000/a2a \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "message/send",
    "id": "1",
    "params": {
      "message": {
        "messageId": "msg-001",
        "role": "user",
        "parts": [{"text": "What are the latest AI trends?"}]
      }
    }
  }'

Response:

json

{
  "jsonrpc": "2.0",
  "id": "1",
  "result": {
    "id": "task-abc123",
    "contextId": "ctx-456",
    "status": {
      "state": "completed",
      "timestamp": "2024-01-15T10:30:00Z"
    },
    "history": [
      {
        "messageId": "msg-001",
        "role": "user",
        "parts": [{"text": "What are the latest AI trends?"}]
      },
      {
        "messageId": "msg-002",
        "role": "agent",
        "parts": [{"text": "Based on recent research..."}]
      }
    ],
    "artifacts": [
      {
        "artifactId": "art-789",
        "name": "AI Trends Report",
        "parts": [{"text": "Detailed analysis..."}]
      }
    ]
  }
}

message/stream

Send a message and receive real-time updates via Server-Sent Events:

bash

curl -X POST http://localhost:8000/a2a \
  -H "Content-Type: application/json" \
  -H "Accept: text/event-stream" \
  -d '{
    "jsonrpc": "2.0",
    "method": "message/stream",
    "id": "1",
    "params": {
      "message": {
        "role": "user",
        "parts": [{"text": "Explain quantum computing"}]
      }
    }
  }'

SSE Event Format:

event: task.status
data: {"taskId": "task-abc123", "status": {"state": "working"}}

event: task.artifact
data: {"taskId": "task-abc123", "artifact": {"parts": [{"text": "Quantum computing..."}]}}

event: task.status
data: {"taskId": "task-abc123", "status": {"state": "completed"}, "final": true}

event: done
data: {}

message/stream behaves like message/send: the task moves through working → completed | failed | cancelled in the task store, concurrent streams don’t block each other, AgentTeam is supported, and tasks/cancel reaches a running stream. After the stream ends, tasks/get for the same task ID returns the finished task with its history and artifact.

If the agent raises an error, the final failed event carries the reason in status.message:

event: task.status
data: {
  "taskId": "task-abc123",
  "status": {
    "state": "failed",
    "message": {
      "role": "agent",
      "parts": [{"text": "LLM provider returned 429 Too Many Requests"}]
    }
  },
  "final": true
}

If tasks/cancel lands while the stream is running, the stream ends with a final cancelled event:

event: task.status
data: {"taskId": "task-abc123", "status": {"state": "cancelled"}, "final": true}

tasks/get

Retrieve an existing task by ID:

bash

curl -X POST http://localhost:8000/a2a \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "tasks/get",
    "id": "2",
    "params": {"id": "task-abc123"}
  }'

tasks/cancel

Cancel a running task:

bash

curl -X POST http://localhost:8000/a2a \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "tasks/cancel", 
    "id": "3",
    "params": {"id": "task-abc123"}
  }'

Success response:

json

{"jsonrpc": "2.0", "id": "3", "result": {"id": "task-abc123", "status": {"state": "cancelled"}}}

If the task is already completed, failed, or cancelled, the state is left unchanged and an error is returned:

json

{"jsonrpc": "2.0", "id": "3", "error": {"code": -32002, "message": "Task not cancelable: already completed"}}

Configuration Options

Parameter	Type	Default	Description
`agent`	`Agent`	—	Single agent
`agents`	`AgentTeam`	—	Multi-agent team
`name`	`str`	agent name	Endpoint name
`description`	`str`	agent role	Description
`url`	`str`	`http://localhost:8000/a2a`	Endpoint URL
`version`	`str`	`"1.0.0"`	Version
`prefix`	`str`	`""`	URL prefix
`auth_token`	`str`	`None`	Bearer token auth
`extended_agent_card_callback`	`callable`	`None`	Extended card callback

python

a2a = A2A(
    agent=agent,
    name="Custom Name",
    description="Custom description", 
    url="http://localhost:8000/a2a",
    version="2.0.0",
    prefix="/api",
    auth_token="sk-secret-key",
)

Multi-Agent Support

Route messages to multi-agent workflows using AgentTeam:

python

from praisonaiagents import Agent, AgentTeam, Task

researcher = Agent(name="Researcher", role="Research Analyst", goal="Research topics")
writer = Agent(name="Writer", role="Content Writer", goal="Write content")

task1 = Task(description="Research AI trends", agent=researcher)
task2 = Task(description="Write a report", agent=writer)

team = AgentTeam(agents=[researcher, writer], tasks=[task1, task2])

a2a = A2A(agents=team, name="Research Team")
a2a.serve(port=8000)

Both message/send and message/stream work with AgentTeam — a team endpoint streams the team’s final result just like a single agent.

Streaming Events

Diagram

Event Types

Event Type	Description
`task.status`	Task state changes (`working`, `completed`, `failed`, `cancelled`). `failed` and `cancelled` carry an optional `status.message` with the reason.
`task.artifact`	Artifact outputs with content chunks. On completion the artifact is also stored on the task (visible via `tasks/get`).
`done`	Stream completion signal

Agent Card

The Agent Card is automatically generated from your agent’s configuration:

json

{
  "name": "Research Assistant",
  "url": "http://localhost:8000/a2a",
  "version": "1.0.0",
  "description": "Research Analyst. Help users research topics",
  "capabilities": {
    "streaming": true,
    "pushNotifications": false,
    "stateTransitionHistory": false
  },
  "skills": [
    {
      "id": "search_web",
      "name": "search_web", 
      "description": "Search the web for information.",
      "tags": ["tool"]
    }
  ],
  "provider": {
    "name": "PraisonAI"
  },
  "securitySchemes": {
    "bearerAuth": {
      "type": "http",
      "scheme": "bearer"
    }
  }
}

Access via: GET /.well-known/agent.json

Message Parts

A2A messages support multimodal content through different part types:

Part Type	A2A Field	Python Type
`TextPart`	`text`	`TextPart(text="...")`
`FilePart`	`file.uri`	`FilePart(file_uri="...")`
`DataPart`	`data`	`DataPart(data={...})`

A2UI payloads

When an agent uses send_a2ui_messages, sync message/send artifacts may include a DataPart with metadata.mimeType of application/json+a2ui. Use create_a2ui_part() / is_a2ui_part() from praisonaiagents.ui.a2ui for manual wrapping.

See A2UI Protocol for the tool contract and renderer options.

bash

# Multimodal message example
curl -X POST http://localhost:8000/a2a \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "message/send",
    "id": "1",
    "params": {
      "message": {
        "role": "user",
        "parts": [
          {"text": "Analyze this image"},
          {"fileWithUri": "https://example.com/chart.png", "mediaType": "image/png"},
          {"data": {"metadata": "analysis_request"}}
        ]
      }
    }
  }'

Authentication

Protect your A2A endpoint with bearer token authentication:

python

a2a = A2A(agent=agent, auth_token="sk-my-secret-key")
a2a.serve(port=8000)

Authenticated requests:

bash

curl -X POST http://localhost:8000/a2a \
  -H "Authorization: Bearer sk-my-secret-key" \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc": "2.0", "method": "message/send", ...}'

Task Lifecycle

Diagram

graph TB
    subgraph "Task States"
        Submitted[📝 submitted] --> Working[⚡ working]
        Working --> InputRequired[❓ input_required]
        Working --> Completed[✅ completed]
        Working --> Failed[❌ failed]
        Working --> Cancelled[🚫 cancelled]
        InputRequired --> Working

        AuthRequired[🔐 auth_required]
    end

    classDef submitted fill:#6366F1,stroke:#7C90A0,color:#fff
    classDef working fill:#F59E0B,stroke:#7C90A0,color:#fff
    classDef completed fill:#10B981,stroke:#7C90A0,color:#fff
    classDef failed fill:#EF4444,stroke:#7C90A0,color:#fff

    class Submitted submitted
    class Working,InputRequired working  
    class Completed completed
    class Failed,Cancelled failed
    class AuthRequired failed

Streamed tasks (message/stream) follow the same lifecycle as message/send tasks.

Docker Deployment

dockerfile

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

requirements.txt:

praisonaiagents
fastapi
uvicorn

Best Practices

Protect production endpoints

Set auth_token on any A2A server exposed beyond localhost. The discovery card stays public per spec — keep sensitive skills and tools out of the advertised card when auth is required.

Prefer streaming for long tasks

Use message/stream when responses may take more than a few seconds. Clients receive progress via SSE instead of blocking on a single JSON-RPC response. Streams run concurrently, so multiple clients can stream from the same server without waiting on each other.

Use AgentTeam for multi-step work

Route complex workflows through AgentTeam rather than chaining multiple single agents. One A2A endpoint can orchestrate monitor → act → feedback loops cleanly.

Publish accurate agent cards

Keep name, description, and url aligned with your deployment. External agents discover capabilities via GET /.well-known/agent.json — stale URLs break inter-agent handoffs.

Task store retention

Tasks live in memory, per server instance, and do not persist across restarts. The store holds up to ~10,000 tasks; beyond that, the oldest terminal tasks are evicted. Live tasks are never evicted.

A2UI Protocol

Declarative UI via send_a2ui_messages and DataPart artifacts

A2A Client

A2A client for connecting to other agents

MCP Lifecycle

Model Context Protocol for tool integration

A2A Protocol

Quick Start

JSON-RPC Methods

message/send

message/stream

tasks/get

tasks/cancel

Configuration Options

Multi-Agent Support

Streaming Events

Event Types

Agent Card

Message Parts

A2UI payloads

Authentication

Task Lifecycle

Docker Deployment

Best Practices

Related