Ce este protocolul A2A și de ce ar trebui să-ți pese?
Dacă ai citit deja articolul nostru despre MCP (Model Context Protocol), știi cum un agent AI poate accesa unelte externe — API-uri, baze de date, fișiere. Dar ce faci când ai mai mulți agenți și trebuie cumva să colaboreze între ei? Păi, exact aici intră în scenă protocolul A2A (Agent-to-Agent).
A2A e un standard deschis creat de Google în aprilie 2025 și donat Fundației Linux în iunie 2025. Scopul lui sună simplu, dar e destul de ambițios: să permită agenților AI — construiți pe framework-uri diferite, de companii diferite, pe servere diferite — să se descopere reciproc, să comunice și să colaboreze. Totul fără a-și expune logica internă, memoria sau instrumentele.
Gândește-te la A2A ca la HTTP-ul pentru agenții AI. La fel cum HTTP a permis oricărui browser să acceseze orice server web, A2A permite oricărui agent AI să comunice cu orice alt agent, indiferent de tehnologia din spate. E o analogie simplificată, dar surprinzător de precisă.
Până în martie 2026, peste 100 de companii au aderat la ecosistemul A2A, iar Agentic AI Foundation (AAIF) — co-fondată de OpenAI, Anthropic, Google, Microsoft, AWS și Block — găzduiește atât A2A, cât și MCP sub aceeași umbrelă. Practic, cele două protocoale au devenit stiva standard pentru comunicarea AI.
A2A vs MCP: două protocoale complementare
Înainte să intrăm în cod, hai să clarificăm relația dintre A2A și MCP. Sunt concepte pe care mulți le confundă (și sincer, e ușor de înțeles de ce).
MCP — agentul vorbește cu uneltele
- Creator: Anthropic (2024)
- Scop: Conectează un agent AI la unelte, baze de date și servicii externe
- Analogie: USB-C pentru AI — un conector universal pentru instrumente
- Exemplu: Un agent care citește din baza de date, trimite email-uri sau accesează un API REST
A2A — agentul vorbește cu alți agenți
- Creator: Google (2025), acum sub Linux Foundation
- Scop: Permite agenților AI să se descopere și să colaboreze ca parteneri egali
- Analogie: HTTP pentru AI — un protocol universal de comunicare agent-la-agent
- Exemplu: Un agent de planificare care delegă cercetarea către un agent de research și analiza către un agent de date
Pe scurt: MCP se ocupă de integrarea cu uneltele, iar A2A de orchestrarea colaborativă între agenți. Într-un sistem de producție serios, ai nevoie de ambele. Agentul tău folosește MCP ca să-și acceseze instrumentele și A2A ca să vorbească cu alți agenți.
Conceptele fundamentale ale protocolului A2A
OK, înainte de a scrie cod, trebuie să înțelegi câteva concepte-cheie. N-o să dureze mult, promit. Protocolul folosește JSON-RPC 2.0 peste HTTP(S) și suportă comunicare sincronă, streaming prin SSE (Server-Sent Events) și notificări push asincrone.
Agent Card — cartea de vizită a agentului
Fiecare agent A2A publică un Agent Card — un document JSON disponibil la endpoint-ul /.well-known/agent.json. Agent Card-ul descrie identitatea agentului, capabilitățile sale, protocoalele suportate și modul de autentificare. Nimic din logica internă nu e expus.
{
"name": "Agent de Cercetare",
"description": "Cercetează subiecte și returnează rapoarte structurate",
"version": "1.0.0",
"url": "https://research-agent.example.com",
"capabilities": {
"streaming": true,
"pushNotifications": false
},
"authentication": {
"schemes": ["bearer"]
},
"defaultInputModes": ["text/plain"],
"defaultOutputModes": ["text/plain", "application/json"],
"skills": [
{
"id": "research",
"name": "Cercetare Web",
"description": "Cercetează un subiect și generează un raport detaliat",
"tags": ["research", "analysis", "report"]
}
]
}Când un agent client vrea să afle ce poate face un agent server, pur și simplu îi citește Agent Card-ul. Gândește-te la el ca la un fișier openapi.json — dar pentru agenți AI, nu pentru API-uri REST.
Task — unitatea de lucru
Când un client trimite o cerere către un server A2A, aceasta devine un Task. Fiecare task are un ID unic și parcurge un ciclu de viață clar definit:
- submitted — task-ul a fost primit
- working — agentul lucrează la el
- input-required — agentul are nevoie de informații suplimentare de la client
- completed — task-ul s-a finalizat cu succes
- failed — task-ul a eșuat
- canceled — task-ul a fost anulat
Această natură stateful e o diferență majoră față de MCP. A2A e proiectat pentru task-uri care pot dura minute sau chiar ore, cu mai multe schimburi de mesaje, pauze și reluări. Un task de tipul „cercetează piața imobiliară din București și generează un raport" poate implica zeci de iterații între agenți.
Și asta e foarte tare, sincer.
Artifact — rezultatul concret
Un Artifact e produsul generat de agent ca răspuns la task. Poate fi un document text, o imagine, un fișier JSON, un spreadsheet sau orice alt tip de deliverable. Artefactele pot fi trimise incremental prin streaming, ceea ce e util mai ales pentru rezultate mari.
Implementare practică: serverul A2A în Python
Bun, hai să trecem la cod. Vom construi un agent A2A funcțional folosind SDK-ul oficial a2a-python. Începem cu instalarea dependențelor.
# Instalare SDK-ul oficial A2A
pip install a2a-python
# Dependențe suplimentare
pip install uvicorn httpxPasul 1: Definirea Agent Card-ului
Primul pas e să definești identitatea agentului — ce poate face și cum poate fi contactat. Practic, îi faci „pașaportul".
from a2a.types import (
AgentCard,
AgentSkill,
AgentCapabilities,
)
agent_card = AgentCard(
name="Agent de Sumarizare",
description="Primește texte și returnează rezumate concise",
version="1.0.0",
url="http://localhost:9999",
capabilities=AgentCapabilities(
streaming=True,
pushNotifications=False,
),
defaultInputModes=["text/plain"],
defaultOutputModes=["text/plain"],
skills=[
AgentSkill(
id="summarize",
name="Sumarizare Text",
description="Rezumă un text lung în puncte cheie",
tags=["summarization", "nlp", "text"],
)
],
)Pasul 2: Implementarea AgentExecutor
Aici e partea cea mai interesantă — logica reală a agentului. AgentExecutor e clasa abstractă pe care o implementezi pentru a defini ce face agentul când primește cereri.
from a2a.server.agent_execution import AgentExecutor
from a2a.server.events import EventQueue
from a2a.server.request_handlers import RequestContext
from a2a.types import (
TaskState,
TaskStatus,
TaskStatusUpdateEvent,
TaskArtifactUpdateEvent,
Artifact,
Part,
TextPart,
Message,
)
import google.generativeai as genai
class SummarizationAgentExecutor(AgentExecutor):
"""Agent care primește text și returnează un rezumat."""
def __init__(self):
self.model = genai.GenerativeModel("gemini-2.0-flash")
async def execute(
self,
context: RequestContext,
event_queue: EventQueue,
) -> None:
# Extrage textul din mesajul primit
user_message = context.get_user_message()
input_text = ""
if user_message and user_message.parts:
for part in user_message.parts:
if hasattr(part, "text"):
input_text += part.text
if not input_text:
await event_queue.enqueue_event(
TaskStatusUpdateEvent(
taskId=context.task_id,
contextId=context.context_id,
final=True,
status=TaskStatus(
state=TaskState.failed,
message=Message(
role="agent",
parts=[TextPart(text="Nu am primit text.")],
),
),
)
)
return
# Actualizăm starea: agentul lucrează
await event_queue.enqueue_event(
TaskStatusUpdateEvent(
taskId=context.task_id,
contextId=context.context_id,
final=False,
status=TaskStatus(state=TaskState.working),
)
)
# Generăm rezumatul cu LLM
prompt = f"Rezumă următorul text în 3-5 puncte cheie:
{input_text}"
response = self.model.generate_content(prompt)
summary = response.text
# Trimitem artefactul rezultat
await event_queue.enqueue_event(
TaskArtifactUpdateEvent(
taskId=context.task_id,
contextId=context.context_id,
artifact=Artifact(
parts=[TextPart(text=summary)],
),
)
)
# Marcăm task-ul ca finalizat
await event_queue.enqueue_event(
TaskStatusUpdateEvent(
taskId=context.task_id,
contextId=context.context_id,
final=True,
status=TaskStatus(state=TaskState.completed),
)
)
async def cancel(
self,
context: RequestContext,
event_queue: EventQueue,
) -> None:
await event_queue.enqueue_event(
TaskStatusUpdateEvent(
taskId=context.task_id,
contextId=context.context_id,
final=True,
status=TaskStatus(state=TaskState.canceled),
)
)Pasul 3: Pornirea serverului A2A
Acum conectăm totul și pornim serverul. E surprinzător de puțin cod pentru cât de mult face.
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
import uvicorn
# Configurarea handler-ului de cereri
request_handler = DefaultRequestHandler(
agent_executor=SummarizationAgentExecutor(),
task_store=InMemoryTaskStore(),
)
# Construirea aplicației A2A
server_app = A2AStarletteApplication(
agent_card=agent_card,
http_handler=request_handler,
)
# Pornirea serverului
if __name__ == "__main__":
uvicorn.run(
server_app.build(),
host="0.0.0.0",
port=9999,
)Serverul expune automat Agent Card-ul la http://localhost:9999/.well-known/agent.json și acceptă cereri A2A pe endpoint-urile standard. Nu trebuie să configurezi nimic suplimentar.
Clientul A2A: cum trimiți cereri unui agent
Odată ce serverul rulează, poți interacționa cu el programatic folosind clasa A2AClient din SDK. Să vedem cum arată în practică.
import httpx
from a2a.client import A2AClient
from a2a.types import (
MessageSendParams,
SendMessageRequest,
Message,
TextPart,
)
import asyncio
import uuid
async def main():
async with httpx.AsyncClient() as httpx_client:
# Descoperirea agentului — citește Agent Card-ul
client = await A2AClient.get_client_from_agent_card_url(
httpx_client,
"http://localhost:9999",
)
# Afișează capabilitățile agentului
print(f"Agent: {client.agent_card.name}")
print(f"Skills: {[s.name for s in client.agent_card.skills]}")
# Trimite un task
task_id = str(uuid.uuid4())
context_id = str(uuid.uuid4())
request = SendMessageRequest(
id=str(uuid.uuid4()),
params=MessageSendParams(
message=Message(
role="user",
parts=[
TextPart(
text="Inteligența artificială generativă "
"transformă modul în care companiile "
"dezvoltă produse software. Prin utilizarea "
"modelelor lingvistice mari, echipele pot "
"genera cod, documentație și teste automat. "
"Aceasta reduce semnificativ timpul de "
"dezvoltare și permite inginerilor să se "
"concentreze pe arhitectură și design."
)
],
messageId=str(uuid.uuid4()),
),
taskId=task_id,
contextId=context_id,
),
)
response = await client.send_message(request)
print(f"Răspuns: {response}")
asyncio.run(main())Integrarea A2A cu framework-uri existente
Vestea bună e că nu trebuie să construiești totul de la zero. SDK-ul A2A se integrează nativ cu cele mai populare framework-uri agentice. Hai să vedem cele mai folosite opțiuni.
Google ADK — funcția to_a2a()
Dacă deja folosești Google Agent Development Kit, poți expune orice agent ca server A2A cu o singură linie de cod. Da, serios — o singură linie.
from google.adk import Agent
from google.adk.a2a import to_a2a
# Definirea agentului ADK
root_agent = Agent(
name="analyst",
model="gemini-2.0-flash",
instruction="Analizezi date financiare și oferi rapoarte.",
)
# Expunere ca server A2A — generează Agent Card automat
app = to_a2a(root_agent)
# Rulare: uvicorn main:app --host 0.0.0.0 --port 8000Funcția to_a2a() generează automat un Agent Card pe baza configurației agentului — numele, descrierea și skill-urile sunt extrase direct din cod. Nu mai trebuie să le definești manual.
Pydantic AI — FastA2A
FastA2A e o implementare independentă de framework, construită pe Starlette, care funcționează cu orice agent Pydantic AI. La fel de simplă ca varianta ADK.
from pydantic_ai import Agent
agent = Agent(
"openai:gpt-4o",
instructions="Ești un asistent de traducere. "
"Traduci texte din engleză în română.",
)
# O singură linie — agentul devine server A2A
app = agent.to_a2a()
# Rulare: uvicorn translation_agent:app --host 0.0.0.0 --port 8001LangGraph — agent cu stare complexă
Pentru agenți care necesită grafuri de execuție și stare complexă, LangGraph oferă o integrare elegantă. Poți construi un agent LangGraph cu ReAct pattern și apoi îl expui ca server A2A prin adaptarea unui AgentExecutor custom care folosește graful LangGraph ca motor de raționament. E ceva mai multă muncă decât variantele de mai sus, dar flexibilitatea merită.
Orchestrarea multi-agent: un exemplu complet
Acum ajungem la partea cu adevărat fascinantă. Puterea reală a protocolului A2A se vede când orchestrezi mai mulți agenți într-un workflow complex. Să zicem că ai nevoie de un sistem care:
- Primește o întrebare de la utilizator
- O trimite unui agent de cercetare (A2A Server pe portul 9001)
- Trimite rezultatul unui agent de sumarizare (A2A Server pe portul 9002)
- Returnează rezumatul final
import httpx
from a2a.client import A2AClient
from a2a.types import (
MessageSendParams,
SendMessageRequest,
Message,
TextPart,
)
import uuid
import asyncio
async def orchestrate(question: str) -> str:
async with httpx.AsyncClient() as httpx_client:
# Descoperirea agenților
research_client = await A2AClient.get_client_from_agent_card_url(
httpx_client, "http://localhost:9001"
)
summary_client = await A2AClient.get_client_from_agent_card_url(
httpx_client, "http://localhost:9002"
)
# Pasul 1: Trimite întrebarea la agentul de cercetare
research_response = await research_client.send_message(
SendMessageRequest(
id=str(uuid.uuid4()),
params=MessageSendParams(
message=Message(
role="user",
parts=[TextPart(text=question)],
messageId=str(uuid.uuid4()),
),
taskId=str(uuid.uuid4()),
contextId=str(uuid.uuid4()),
),
)
)
# Extrage rezultatul cercetării
research_text = extract_text(research_response)
# Pasul 2: Trimite cercetarea la agentul de sumarizare
summary_response = await summary_client.send_message(
SendMessageRequest(
id=str(uuid.uuid4()),
params=MessageSendParams(
message=Message(
role="user",
parts=[TextPart(text=research_text)],
messageId=str(uuid.uuid4()),
),
taskId=str(uuid.uuid4()),
contextId=str(uuid.uuid4()),
),
)
)
return extract_text(summary_response)
def extract_text(response) -> str:
"""Extrage textul din răspunsul A2A."""
if hasattr(response, "result"):
task = response.result
if task.artifacts:
for artifact in task.artifacts:
for part in artifact.parts:
if hasattr(part, "text"):
return part.text
return ""
# Rulare
result = asyncio.run(
orchestrate("Care sunt tendințele AI în 2026?")
)
print(result)Acest pattern — un orchestrator care descoperă agenți prin Agent Cards și le trimite task-uri secvențial — e fundamentul sistemelor multi-agent de producție. Fiecare agent poate fi dezvoltat, testat și deployat independent, pe servere diferite, de echipe diferite. Și asta e exact frumusețea standardizării.
Securitate în protocolul A2A
Să fim serioși: securitatea nu e opțională într-un sistem unde agenți AI comunică între ei. E absolut critică. Protocolul A2A suportă scheme de autentificare aliniate cu specificația OpenAPI.
Autentificare și autorizare
- API Keys: pentru comunicarea internă între agenți în aceeași rețea
- OAuth 2.0: pentru scenarii cross-organizaționale unde agenții aparțin unor companii diferite
- OpenID Connect: pentru identitate federalizată și single sign-on
Schema de autentificare e declarată în Agent Card, iar clientul trebuie să se autentifice înainte de a trimite task-uri. Odată autentificat, serverul se ocupă de autorizare și control al accesului.
Regula de aur: tratează agenții externi ca surse nesigure
Un principiu pe care-l repet mereu: orice agent care operează în afara controlului tău direct trebuie tratat ca o entitate potențial nesigură. Toate datele primite — Agent Card, mesaje, artefacte, statusuri — trebuie validate și sanitizate. Nu executa niciodată cod primit de la un agent extern fără sandboxing. Asta nu e paranoia, e bun simț ingineresc.
Provocări și bune practici
Implementarea A2A la scară vine cu provocări specifice. N-o să pretind că e totul simplu — iată ce trebuie să ai în vedere.
Conectivitate N-pătrat
Într-un sistem cu N agenți care comunică direct între ei, numărul de conexiuni crește pătratic (N×N). La 10 agenți ai 100 de conexiuni potențiale. La 50 de agenți? 2.500.
Soluția: folosește un agent orchestrator central (pattern-ul „puppeteer") care coordonează comunicarea și reduce complexitatea la N conexiuni. E mult mai gestionabil.
Gestionarea erorilor
Într-un lanț de agenți, un singur eșec poate bloca întregul workflow. Aici n-ai voie să improvizezi. Implementează:
- Retry cu backoff exponențial pentru erori temporare
- Circuit breaker pentru agenți care nu răspund
- Fallback agents — agenți de rezervă cu capabilități similare
- Timeout-uri explicite pe fiecare task
Observabilitate
Adaugă logging și tracing (OpenTelemetry) pe fiecare interacțiune A2A. Când ceva nu funcționează într-un lanț de 5 agenți, trebuie să identifici rapid care agent a cauzat problema. Folosește taskId și contextId ca identificatori de corelație în log-uri — te vor salva de multe ore de debugging.
Întrebări frecvente (FAQ)
Ce problemă rezolvă protocolul A2A?
A2A rezolvă problema interoperabilității între agenți AI. Fără un protocol standard, fiecare pereche de agenți necesită cod de integrare custom, ceea ce devine impracticabil pe măsură ce numărul de agenți crește. A2A oferă o „limbă comună" pentru toți agenții, indiferent de framework sau vendor.
Pot folosi A2A și MCP în același proiect?
Da, și chiar e recomandat. MCP și A2A sunt complementare: MCP conectează un agent la uneltele sale (baze de date, API-uri, fișiere), iar A2A permite agenților să colaboreze între ei. Un agent poate folosi MCP intern pentru a accesa resurse și A2A extern pentru a comunica cu alți agenți. Practic, se completează perfect.
Ce framework-uri suportă A2A nativ?
În martie 2026, suport nativ A2A oferă Google Agent Development Kit (ADK), Pydantic AI (prin FastA2A), Microsoft Agent Framework, BeeAI, Strands Agents SDK și LangGraph (prin adaptoare). SDK-ul oficial a2a-python funcționează cu orice framework, fiind construit pe Starlette și compatibil cu orice server ASGI.
A2A e pregătit pentru producție?
SDK-ul oficial implementează specificația A2A v0.3.0 și e folosit în producție de companii precum Google, IBM și Microsoft. Totuși — și asta e important — protocolul e încă în evoluție activă, deci trebuie să fii pregătit pentru schimbări în API între versiuni minore. Pentru proiecte critice, fixează versiunea SDK-ului și monitorizează changelog-ul oficial.
Cum gestionez securitatea într-un sistem A2A cu mulți agenți?
Folosește OAuth 2.0 pentru autentificarea cross-organizațională, validează toate datele primite de la agenți externi, implementează rate limiting pe server și monitorizează interacțiunile cu logging structurat. Și regula de bază rămâne aceeași: tratează orice agent din afara controlului tău direct ca o sursă potențial nesigură.