Shipping a single LLM agent is easy; coordinating several that trust each other’s output is not. The agent2agent multi-agent collaboration pattern solves this by standardizing how agents advertise capabilities and delegate work over a common protocol. This guide walks through a concrete implementation path using the open A2A specification, from capability discovery to resilient task execution.
1. Publish an Agent Card for Discovery
Every agent in an agent2agent multi-agent collaboration topology needs a machine-readable description of what it does and how to reach it. The A2A spec calls this an Agent Card. It is a static JSON document served at a well-known path or returned from a registry.
{
"name": "invoice-extractor",
"description": "Extracts line items from PDF invoices",
"endpoint": "https://agents.example.com/invoice",
"capabilities": ["extract.invoice", "ocr.pdf"],
"auth": "bearer",
"version": "1.0.0"
}Do not treat the card as documentation. Code should fetch it at startup and validate the schema. A stale card that points to a decommissioned endpoint will silently break delegation chains.
A common pitfall is embedding business logic versions in the capability strings without a real semantic version. Use explicit version fields and reject mismatched majors at the client. If you support multiple capability sets, list them as discrete entries rather than overloading one string.
2. Speak JSON-RPC Over HTTP
A2A transports messages as JSON-RPC 2.0 over HTTPS. This is deliberately boring. Below is a minimal Python receiver using aiohttp that accepts a tasks/send method and returns a result.
from aiohttp import web
import json
async def handle(request):
payload = await request.json()
if payload.get("method") == "tasks/send":
task = payload["params"]["task"]
# dummy processing
return web.json_response({
"jsonrpc": "2.0",
"id": payload["id"],
"result": {"status": "completed", "output": {"items": 3}}
})
return web.json_response({
"jsonrpc": "2.0",
"id": payload.get("id"),
"error": {"code": -32601, "message": "method not found"}
})
app = web.Application()
app.router.add_post("/invoice", handle)
web.run_app(app, port=8080)The client side is equally simple. Use an HTTP session with a timeout and retry wrapper.
import aiohttp, asyncio
async def send_task(endpoint, task):
async with aiohttp.ClientSession() as s:
async with s.post(endpoint, json={
"jsonrpc": "2.0",
"id": "req-1",
"method": "tasks/send",
"params": {"task": task}
}) as r:
return await r.json()The tradeoff: JSON-RPC gives you no opinion on auth or pagination. You must define those yourself in the agent card and enforce them consistently. For bearer auth, pull the token from your secrets manager and attach it on every call; do not hardcode it in the card.
3. Model Tasks as Stateful Units
A task in agent2agent multi-agent collaboration is not a single request-response. It has an ID, a lifecycle (submitted, working, completed, failed), and optionally artifacts. Design your agents to persist task state in a store keyed by task ID.
tasks = {}
async def handle(request):
payload = await request.json()
if payload["method"] == "tasks/send":
task_id = payload["params"]["task"]["id"]
tasks[task_id] = {"state": "working"}
# ... process ...
tasks[task_id] = {"state": "completed", "output": {"ok": True}}
return web.json_response({
"jsonrpc": "2.0", "id": payload["id"],
"result": tasks[task_id]
})Avoid blocking the HTTP response while the task runs. Return state: working and let the caller poll tasks/get or register a webhook. Long-running agents that hold connections open will exhaust your worker pool under concurrent load.
Idempotency Is Non-Negotiable
If a network flap causes a retry, the same task ID may arrive twice. Key processing on the ID and make the handler return the existing result if already terminal. Skipping this leads to double-charged API calls and corrupted artifacts.
Artifact Handling
Large outputs (parsed PDFs, images) should not sit in the JSON response. Store them in object storage and return a URI. The agent card can declare artifact_types to signal accepted formats.
4. Delegate Without Losing Control
The orchestrator agent holds the conversation context and decides which specialist to call. It reads agent cards, picks a capability match, and delegates.
async def delegate(session, card, task):
resp = await session.post(card["endpoint"], json={
"jsonrpc": "2.0", "id": task["id"],
"method": "tasks/send",
"params": {"task": task}
})
data = await resp.json()
if "error" in data:
raise RuntimeError(data["error"]["message"])
return data["result"]Keep the orchestrator’s prompts focused on routing, not on solving the subtask. A frequent mistake is stuffing the specialist’s input with the full chat history. Send only the minimal payload the capability requires.
Cascading failure is the biggest risk in agent2agent multi-agent collaboration. If the invoice extractor times out, the orchestrator should fall back to a second agent or degrade gracefully, not bubble a 500 to the user. Implement a simple priority list:
async def delegate_with_fallback(session, cards, task):
last_err = None
for card in cards:
try:
return await delegate(session, card, task)
except Exception as e:
last_err = e
raise last_err5. Make Model Calls Resilient
Most agents ultimately call an LLM. When you run multi-agent systems in production, provider outages are a matter of when, not if. Route model traffic through a layer that honors client routing directives and provides automatic fallback when a provider is rate-limited or degraded. For instance, n4n.ai exposes one OpenAI-compatible endpoint addressing 240+ models, so an agent can switch models via a header without rewriting its A2A client or its tool-calling logic.
# agent uses OpenAI client pointed at gateway
from openai import OpenAI
client = OpenAI(base_url="https://api.n4n.ai/v1", api_key="sk-...")
client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role":"user","content":"summarize"}],
extra_headers={"x-model-fallback": "claude-3-haiku"}
)This keeps the agent’s internal logic stable while the gateway handles provider negotiation. Without such a layer, you will write bespoke retry code for every model SDK. Also forward provider cache-control hints where your gateway supports them; prompt caching cuts latency and cost on repeated orchestrator calls.
6. Meter and Observe Every Hop
Per-token usage metering is mandatory when agents call each other and then call models. Attach a correlation ID to each A2A task and propagate it into model calls. Log the agent card name, task ID, latency, and token count at each boundary.
{
"correlation_id": "trace-123",
"agent": "invoice-extractor",
"task_id": "t-001",
"model_tokens": 420,
"duration_ms": 850
}Without this, you cannot tell whether a cost spike came from a chatty orchestrator or a loop where two agents repeatedly re-delegate.
Tracebacks Across Agents
Standardize error shapes in the JSON-RPC error object. Include the failing capability and the upstream task ID. A generic -32603 with “internal error” will cost you hours in a distributed setup.
Health Checks
Expose a health method in your agent card. The orchestrator should skip agents that return degraded rather than queue behind them.
Common Tradeoffs
Latency vs. autonomy. Fine-grained delegation improves reuse but adds network hops. Batch where possible.
Strict schema vs. flexibility. A rigid agent card prevents surprise inputs but forces version churn. Use JSON Schema for params and allow additionalProperties: false only after the capability is stable.
Central registry vs. peer discovery. A registry simplifies lookup but is a single point of failure. Cache cards aggressively with a short TTL.
Synchronous vs. asynchronous tasks. Synchronous is easier to debug; asynchronous scales better. Adopt asynchronous early if you expect long-running subtasks.
The agent2agent multi-agent collaboration model is not magic; it is disciplined interfacing. Build the card, speak JSON-RPC, track tasks, delegate with fallbacks, and meter everything. Do that and your swarm of agents will behave like a system instead of a chat room.