aiagent.org aiagent.org A2A
Agentic Interoperability

Agent-to-Agent (A2A) Protocols

Establish standardized formats and protocols for asynchronous message passing, target handoffs, task delegation, and execution synchronization between autonomous agents.

Core Concepts Setup & Installation Routing Architectures Message Envelopes Simulation Code FAQ

Core A2A Concepts

Foundational principles of agent-to-agent communication layers.

Asynchronous Messages

Unlike linear function execution, A2A relies on async message passing. Agents drop envelopes in queues and proceed with other work, waking up when they receive a return task token.

Explicit Handoffs

The pattern where Agent A delegates control of a thread to Agent B because B is specialized. The handoff payload contains the current conversational state and task description.

Transport Layer

The system routing the physical packets. Can be local queues (memory), socket connections, or distributed message brokers like Kafka, RabbitMQ, or HTTP webhooks.

Task Negotiation

How agents request help. Inspired by classic FIPA protocols, Agent A requests bids from Agent B/C, evaluates their capacity, and issues a formal task assignment.

JSON Event Envelopes

Standard envelope schemas wrapping agent payloads. Envelopes must declare metadata: sender ID, receiver ID, transaction correlation ID, timestamp, and message type.

State Synchronization

Keeping track of progress across multiple running agents. Uses shared stores (like Redis or Cosmos DB) with locks to prevent write conflicts.

Setup & Installation

Configure communication libraries and messaging infrastructure for Agent-to-Agent protocols.

1. Conceptual Nature of A2A

Unlike Anthropic's proprietary MCP, Agent-to-Agent (A2A) is an architectural communication pattern rather than a single standalone software library. Developers implement it using standard message serializers (like JSON/Protobuf) sent over communication networks (like HTTP, WebSockets, or AMQP/RabbitMQ).

2. Standard Framework Libraries

If you wish to use prebuilt open-source frameworks implementing standard A2A protocols, install the following:

# 1. Agent Protocol (standard API wrapper by AutoGPT/e2b)
pip install agent-protocol

# 2. SPADE (Smart Python Agent Development Environment)
pip install spade

3. Install Transport & Verification Packages

For building custom message routing envelopes over distributed queues:

# Install Pydantic (envelope serialization & schema checking)
pip install pydantic

# Install AMQP and Redis drivers
pip install aio-pika redis[hiredis]

2. Message Broker Setup (RabbitMQ / Redis)

Docker (Recommended)

# Run RabbitMQ container
docker run -d \
  --name rabbitmq-dev \
  -p 5672:5672 \
  -p 15672:15672 \
  rabbitmq:3-management

# Run Redis container
docker run -d --name redis-dev -p 6379:6379 redis:alpine

macOS

# Install RabbitMQ & Redis
brew install rabbitmq redis

# Start services
brew services start rabbitmq
brew services start redis

# Check broker status
rabbitmqctl status

Ubuntu/Debian

# Install from apt
sudo apt update
sudo apt install -y rabbitmq-server redis-server

# Start and enable RabbitMQ
sudo systemctl enable rabbitmq-server
sudo systemctl start rabbitmq-server

Routing Topologies

Common routing configurations in production multi-agent systems.

1. Linear / Handoff Chain

Dynamic routing where each agent decides which agent to call next based on output state (e.g. Writer -> Editor -> Publisher). Highly efficient but lacks centralized verification.

2. Central Supervisor

Hub-and-spoke setup. The central supervisor agent breaks the main goal into tasks, forwards sub-tasks to specialist agents (spokes), and aggregates the final results securely.

3. Event-Driven Mesh

Fully decoupled. Agents monitor a shared message broker topic (like RabbitMQ) for specific events. Useful for parallel execution, triggers, and highly distributed deployments.

Communicative Acts & Schema

FIPA communicative act intents mapped directly to standard JSON payloads.

Communicative Acts (Intents)

Intent Description
REQUEST Ask receiver to execute a specific task.
PROPOSE Offer to perform action under conditions.
INFORM Assert a statement of fact or result.
ACCEPT / REFUSE Acknowledge or decline task proposal.
FAILURE Report error executing target action.

ACL Envelope Example

{ "envelope_version": "1.0", "message_id": "msg_f829c91", "correlation_id": "tx_99182a", // Track conversation chain "sender": "planner-agent", "receiver": "database-agent", "intent": "REQUEST", "payload": { "task": "fetch_logs", "params": { "limit": 5, "filter": "warnings" } }, "timestamp": 1784650500 }

Agent Simulation System

An asynchronous python framework simulating dynamic handoffs and message loops.

import asyncio
import uuid

class MessageBroker:
  def __init__(self):
    self.queues = {}

  def register_agent(self, agent_name):
    self.queues[agent_name] = asyncio.Queue()

  async def send_message(self, receiver, envelope):
    await self.queues[receiver].put(envelope)

class BaseAgent:
  def __init__(self, name, broker):
    self.name = name
    self.broker = broker
    self.broker.register_agent(name)

  async def start(self):
    while True:
      msg = await self.broker.queues[self.name].get()
      await self.process(msg)

  async def process(self, envelope):
    raise NotImplementedError()

class SupervisorAgent(BaseAgent):
  async def process(self, envelope):
    intent = envelope["intent"]
    print(f"[{self.name}] Received envelope {envelope['message_id']} ({intent})")
    if intent == "REQUEST":
      # Delegate task to worker agent
      delegated_msg = {
        "message_id": str(uuid.uuid4())[:8],
        "correlation_id": envelope["message_id"],
        "sender": self.name,
        "receiver": "worker-agent",
        "intent": "REQUEST",
        "payload": envelope["payload"]
      }
      await self.broker.send_message("worker-agent", delegated_msg)
    elif intent == "INFORM":
      print(f"[{self.name}] Final Result received: {envelope['payload']['result']}")

class WorkerAgent(BaseAgent):
  async def process(self, envelope):
    print(f"[{self.name}] Processing task: {envelope['payload']['task']}")
    reply = {
      "message_id": str(uuid.uuid4())[:8],
      "correlation_id": envelope["message_id"],
      "sender": self.name,
      "receiver": envelope["sender"],
      "intent": "INFORM",
      "payload": {"result": "Optimization task complete"}
    }
    await self.broker.send_message(envelope["sender"], reply)

Frequently Asked Questions

What is FIPA-ACL?▶
FIPA-ACL (Agent Communication Language) is a standard created in the late 90s by the Foundation for Intelligent Physical Agents. It defines communicative acts (inform, request, propose, reject, agree) that model social interactions between software systems. Modern A2A protocols adapt FIPA structures into JSON payloads.
Should I use HTTP or WebSockets for A2A?▶
HTTP REST APIs are good for synchronous request-response routing (e.g. querying a search agent). WebSockets or message brokers (RabbitMQ/Kafka) are required for async, long-running agent tasks where workers push status reports as they process data blocks.
How do we prevent concurrent write conflicts to shared state?▶
State synchronization is maintained by wrapping memory writes in distributed locks (e.g., Redis Redlock or database optimistic locks). Each transaction must declare a state version token; if an agent tries to write using an outdated state token, the transaction fails and the agent must pull the latest version and retry.