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.

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.

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

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.

​

Quick Start

1. Simple Agent Server

   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)
   
2. FastAPI Integration

   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:

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:

{
  "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:

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}

If the client disconnects before the stream finishes, the task is marked cancelled.

​

tasks/get

Retrieve an existing task by ID:

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:

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:

{"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:

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

A cancelled task is final — any background completion that arrives later is discarded, not applied.

---

​

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

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:

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.

When both agent and agents are provided, the single agent takes priority.

---

​

Streaming Events

​

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:

{
  "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.

A2A message/stream currently emits text-only artifacts. For live A2UI during streaming, prefer AG-UI or PraisonAIUI.

See A2UI Protocol for the tool contract and renderer options.

# 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:

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

Authenticated requests:

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", ...}'

The discovery endpoint (/.well-known/agent.json) remains public per A2A spec. Only POST /a2a requires authentication.

---

​

Task Lifecycle

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

---

​

Docker Deployment

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.

---

​

Related

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