SK-11-A2A : le protocole Agent2Agent à côté de MCP

Navigation : Index | << 10b-NotebookMaker-batch | 08-MCP — le volet outils

08-SemanticKernel-MCP a montré comment un agent se branche à des outils via MCP. Mais que se passe-t-il quand un agent doit parler à un autre agent — pas à une fonction, mais à un programme qui a ses propres compétences, sa propre file de travail, son propre cycle de vie ? C’est le rôle d’A2A (Agent2Agent) : un protocole ouvert, publié par Google en 2025 puis confié à la Linux Foundation, qui normalise la façon dont deux agents se découvrent (la carte d’agent), se parlent (messages JSON-RPC) et suivent l’avancement d’un travail (le cycle de vie des tâches).

La distinction à retenir avant tout :

MCP relie un agent à un outil — l’agent appelant garde tout le raisonnement, le serveur expose des fonctions.
A2A relie un agent à un agent — le serveur EST un agent, qui raisonne de son côté et rend un résultat.

Dans ce notebook, nous montons deux agents locaux qui échangent réellement via A2A : un agent serveur (exposé en HTTP sur 127.0.0.1) et un agent client, en utilisant le SDK Python de référence a2a-sdk (version 1.1.5, installée ci-dessous). Aucune clé API n’est nécessaire : les agents de démonstration sont déterministes, et c’est bien le protocole que nous observons.

1. Deux protocoles pour deux liens différents

Un système agentique complet a besoin des deux liens :

flowchart LR
    U([Utilisateur]) --> A[Agent orchestrateur]
    A -- "MCP : appelle des outils" --> O1[Serveur MCP<br/>météo, fichiers, recherche]
    A -- "A2A : délègue à un pair" --> B[Agent distant<br/>ses propres compétences]
    B -- "MCP (ses outils à lui)" --> O2[Serveurs MCP distants]

Avec MCP, l’orchestrateur choisit un outil, l’appelle, interprète le résultat : l’intelligence reste chez l’appelant.
Avec A2A, l’orchestrateur remet une tâche à un agent pair, qui la traite avec sa propre intelligence et rend un résultat : l’intelligence est des deux côtés.

Le tableau de décision complet est en section 6 ; observons d’abord la brique centrale d’A2A : la carte d’agent.

# Installation / verification de l'environnement
import importlib.util
import subprocess
import sys

if importlib.util.find_spec("a2a") is None:
    print("a2a-sdk absent : installation...")
    subprocess.run([sys.executable, "-m", "pip", "install", "--quiet", "a2a-sdk"], check=True)

import a2a  # noqa: F401
from a2a.types import AgentCard  # noqa: F401

import httpx  # noqa: F401
import uvicorn  # noqa: F401
from fastapi import FastAPI  # noqa: F401

print("a2a-sdk importe, protocole utilisable (SDK de reference 1.x)")
print("httpx et uvicorn disponibles pour le serveur et le client locaux")

a2a-sdk importe, protocole utilisable (SDK de reference 1.x)
httpx et uvicorn disponibles pour le serveur et le client locaux

2. La carte d’agent (Agent Card)

La carte d’agent est le contrat de découverte d’A2A : un document JSON que tout agent serveur publie à l’URL bien connue /.well-known/agent-card.json. Elle déclare le nom de l’agent, sa version, ses compétences (skills), ses modes d’entrée/sortie, ses interfaces supportées (JSON-RPC, REST…) et ses capacités (streaming, push notifications). C’est l’équivalent, côté agent-à-agent, de la liste d’outils tools/list côté MCP — mais pour des agents entiers.

Construisons la carte de notre agent serveur de démonstration : un traducteur majuscule.

from a2a.types import AgentCapabilities, AgentCard, AgentInterface, AgentSkill

PORT = 7433
BASE_URL = f"http://127.0.0.1:{PORT}"

skill_traduction = AgentSkill(
    id="majuscule",
    name="Traduction en majuscules",
    description="Recoit un texte libre, le renvoie en majuscules avec un prefixe d'agent distant.",
    tags=["texte", "demonstration"],
    input_modes=["text/plain"],
    output_modes=["text/plain"],
)

carte_agent = AgentCard(
    name="agent-traducteur",
    description="Agent de demonstration : traduit toute phrase en MAJUSCULES.",
    version="1.0.0",
    supported_interfaces=[
        AgentInterface(url=BASE_URL, protocol_binding="JSONRPC")
    ],
    capabilities=AgentCapabilities(streaming=False),
    default_input_modes=["text/plain"],
    default_output_modes=["text/plain"],
    skills=[skill_traduction],
)

print(carte_agent)

name: "agent-traducteur"
description: "Agent de demonstration : traduit toute phrase en MAJUSCULES."
supported_interfaces {
  url: "http://127.0.0.1:7433"
  protocol_binding: "JSONRPC"
}
version: "1.0.0"
capabilities {
  streaming: false
}
default_input_modes: "text/plain"
default_output_modes: "text/plain"
skills {
  id: "majuscule"
  name: "Traduction en majuscules"
  description: "Recoit un texte libre, le renvoie en majuscules avec un prefixe d\'agent distant."
  tags: "texte"
  tags: "demonstration"
  input_modes: "text/plain"
  output_modes: "text/plain"
}

Lecture du résultat — la carte est un contrat

La carte affichée ci-dessus porte tout ce qu’un client a besoin de savoir avant de parler à l’agent : identité (name, version), adresse et protocole (supported_interfaces : notre agent parle JSON-RPC sur 127.0.0.1:7433), capacités (pas de streaming ici), et la compétence majuscule avec ses modes. Dans le SDK 1.x, la carte est un message protobuf — d’où l’affichage en format texte protobuf, avec ses accolades imbriquées : c’est la représentation sérialisable réelle du protocole, pas un simple dictionnaire Python.

Point sécurité : la carte est servie par le serveur lui-même. Un client qui fait confiance à une carte non signée fait confiance à… ce que le serveur veut bien déclarer. La carte porte d’ailleurs un champ signatures (vide ici) : la spécification prévoit la signature des cartes, et le notebook de la série Security Tooling-MCP-Attack-Surface traite exactement cette famille de risques côté MCP.

3. Un serveur A2A minimal, en local

Monter un agent A2A avec le SDK de référence demande quatre pièces :

un exécuteur (AgentExecutor) — la logique de l’agent : reçoit le message, produit des événements ;
un gestionnaire de requêtes (DefaultRequestHandlerV2) — traduit les appels JSON-RPC en invocations de l’exécuteur, avec un magasin de tâches en mémoire ;
les routes FastAPI — la carte (/.well-known/agent-card.json) et le point d’entrée JSON-RPC (/) ;
un serveur HTTP — ici uvicorn dans un fil d’arrière-plan, pour rester dans le notebook.

Nous démarrons le serveur dans un fil daemon : il mourra avec le noyau du notebook, ce qui est exactement ce que l’on veut pour une démonstration locale.

import asyncio
import threading
import time

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events.event_queue import EventQueue
from a2a.server.request_handlers.default_request_handler_v2 import (
    DefaultRequestHandlerV2,
)
from a2a.server.routes import add_a2a_routes_to_fastapi
from a2a.server.routes.agent_card_routes import create_agent_card_routes
from a2a.server.routes.jsonrpc_routes import create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore
from a2a.helpers.proto_helpers import new_text_message


class TraducteurExecutor(AgentExecutor):
    """L'agent serveur : sa competence est de mettre en majuscules."""

    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        texte_recu = context.get_user_input() or ""
        reponse = f"[agent-traducteur] {texte_recu.upper()}"
        await event_queue.enqueue_event(new_text_message(reponse))

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        raise RuntimeError("annulation non supportee par l'agent de demonstration")


handler = DefaultRequestHandlerV2(
    agent_executor=TraducteurExecutor(),
    task_store=InMemoryTaskStore(),
    agent_card=carte_agent,
)

app_a2a = FastAPI(title="a2a-traducteur")
add_a2a_routes_to_fastapi(
    app_a2a,
    agent_card_routes=create_agent_card_routes(carte_agent),
    jsonrpc_routes=create_jsonrpc_routes(handler, rpc_url="/"),
)

_fil_serveur = threading.Thread(
    target=lambda: uvicorn.run(app_a2a, host="127.0.0.1", port=PORT, log_level="error"),
    daemon=True,
)
_fil_serveur.start()
time.sleep(1.5)  # laisse le serveur ouvrir son socket

reponse_carte = httpx.get(f"{BASE_URL}/.well-known/agent-card.json", timeout=5.0)
print("statut /.well-known/agent-card.json :", reponse_carte.status_code)
print("nom declare par la carte servie :", reponse_carte.json()["name"])

statut /.well-known/agent-card.json : 200
nom declare par la carte servie : agent-traducteur

Lecture du résultat — le serveur est en ligne

Le code HTTP 200 et le nom agent-traducteur confirment que la carte est servie à l’URL bien connue : n’importe quel client A2A qui connaît l’adresse peut maintenant découvrir cet agent sans documentation préalable. C’est la première moitié du protocole ; la seconde est l’échange lui-même.

4. Le cycle de vie d’une tâche A2A

A2A ne suppose pas qu’un appel = une réponse immédiate : une tâche confiée à un agent peut durer, demander une entrée supplémentaire, échouer. Le protocole définit donc des états de tâche normalisés, et le client peut les suivre. Voici les états que le SDK expose :

from a2a.types import TaskState

etats = [v.name for v in TaskState.DESCRIPTOR.values_by_name.values() if v.number != 0]
print("Etats de tache A2A :", etats)

# Transitions pedagogiques (illustration, pas une enum du protocole)
transitions_legales = {
    "TASK_STATE_SUBMITTED": ["TASK_STATE_WORKING", "TASK_STATE_COMPLETED",
                             "TASK_STATE_FAILED", "TASK_STATE_CANCELED"],
    "TASK_STATE_WORKING": ["TASK_STATE_INPUT_REQUIRED", "TASK_STATE_COMPLETED",
                           "TASK_STATE_FAILED", "TASK_STATE_CANCELED"],
    "TASK_STATE_INPUT_REQUIRED": ["TASK_STATE_WORKING", "TASK_STATE_CANCELED",
                                  "TASK_STATE_REJECTED"],
}
for depart, arrives in transitions_legales.items():
    print(f"{depart:28s} -> {', '.join(arrives)}")

Etats de tache A2A : ['TASK_STATE_SUBMITTED', 'TASK_STATE_WORKING', 'TASK_STATE_COMPLETED', 'TASK_STATE_FAILED', 'TASK_STATE_CANCELED', 'TASK_STATE_INPUT_REQUIRED', 'TASK_STATE_REJECTED', 'TASK_STATE_AUTH_REQUIRED']
TASK_STATE_SUBMITTED         -> TASK_STATE_WORKING, TASK_STATE_COMPLETED, TASK_STATE_FAILED, TASK_STATE_CANCELED
TASK_STATE_WORKING           -> TASK_STATE_INPUT_REQUIRED, TASK_STATE_COMPLETED, TASK_STATE_FAILED, TASK_STATE_CANCELED
TASK_STATE_INPUT_REQUIRED    -> TASK_STATE_WORKING, TASK_STATE_CANCELED, TASK_STATE_REJECTED

Interprétation — un cycle de vie, pas une fonction

Deux états méritent l’arrêt : TASK_STATE_INPUT_REQUIRED (l’agent redemande la parole — il lui manque une information pour continuer ; c’est la différence la plus visible avec un simple appel d’outil MCP, qui ne peut pas « poser une question ») et TASK_STATE_REJECTED (l’agent refuse la tâche — un serveur d’outils ne refuse pas, un agent peut). Notre exécuteur de démonstration est synchrone et simple : il publie directement un message de réponse final, sans passer par les états intermédiaires. Un agent réel (génération longue, streaming, validation humaine) émettrait des mises à jour WORKING puis COMPLETED, via la même file d’événements.

5. L’échange client : découverte puis message

Côté client, le SDK fabrique un client à partir de l’URL seule : il va chercher la carte lui-même (découverte), négocie l’interface, puis envoie le message. La requête explicite est un SendMessageRequest (message + configuration) ; la réponse arrive comme un flux d’événements (StreamResponse), dont on extrait le texte.

from a2a.client import ClientConfig, ClientFactory
from a2a.helpers.proto_helpers import get_message_text, new_text_message
from a2a.types import Role, SendMessageConfiguration, SendMessageRequest


async def echange_a2a(texte: str) -> str:
    async with httpx.AsyncClient(timeout=10.0) as http:
        factory = ClientFactory(
            ClientConfig(streaming=False, polling=False, httpx_client=http)
        )
        client = await factory.create_from_url(BASE_URL)
        requete = SendMessageRequest(
            message=new_text_message(texte, role=Role.ROLE_USER),
            configuration=SendMessageConfiguration(return_immediately=True),
        )
        reponse_texte = None
        async for mise_a_jour in client.send_message(requete):
            message = getattr(mise_a_jour, "message", None)
            if message is not None:
                reponse_texte = get_message_text(message)
        return reponse_texte


print("envoye   : bonjour a2a, ici l'agent orchestrateur")
# await top-level : le kernel Jupyter fournit deja une boucle asyncio
_reponse_a2a = await echange_a2a("bonjour a2a, ici l'agent orchestrateur")
print("recu     :", _reponse_a2a)

envoye   : bonjour a2a, ici l'agent orchestrateur
recu     : [agent-traducteur] BONJOUR A2A, ICI L'AGENT ORCHESTRATEUR

Lecture du résultat — ce qui a traversé le fil

Le client a : (1) récupéré la carte à /.well-known/agent-card.json, (2) ouvert une session JSON-RPC sur /, (3) envoyé un message au rôle ROLE_USER, (4) reçu la réponse de l’agent (préfixe [agent-traducteur] + texte en majuscules) dans un événement de type message. Aucune intelligence n’a été dupliquée côté client : la mise en majuscules vit dans l’agent serveur — c’est toute la différence avec un appel d’outil MCP, où la logique de l’« outil » serait une simple fonction sans état ni cycle de vie.

Note d’implémentation honnête : dans le SDK 1.x, send_message rend un itérateur asynchrone même en mode non-streaming, et la requête doit porter une SendMessageConfiguration (le client y écrit ses préférences) — deux points où l’API 1.x diffère des tutoriels écrits pour la série 0.x du SDK (A2AStarletteApplication). Ce notebook est écrit et exécuté contre la version installée localement (1.1.5).

6. A2A ou MCP : comment choisir

Critère	MCP	A2A
Qui raisonne côté serveur ?	personne — des fonctions	un agent complet
Granularité	l’outil (météo, requête SQL, fichier)	la tâche déléguée
Découverte	`tools/list` (liste d’outils)	carte d’agent (`/.well-known/agent-card.json`)
Durée de vie	un appel, une réponse	une tâche avec états, pouvant durer
Interaction possible	non — l’outil ne parle pas	oui — l’agent peut redemander une entrée
Cas d’usage typique	brancher un LLM à des capacités locales	faire coopérer deux systèmes agentiques

En pratique, les deux se composent : un agent A2A peut exposer ses propres outils via MCP (le schéma Mermaid de la section 1 le montre). La fonction ci-dessous encode la règle de décision en trois lignes — simplifiée à dessein, l’exercice 1 vous demande de l’étendre.

def recommander_protocole(besoin: str) -> str:
    """Regle de decision simplifiee : 'outil' vs 'agent distant'."""
    besoin_norm = besoin.lower()
    if any(mot in besoin_norm for mot in ("agent", "deleguer", "pair", "negocier")):
        return "A2A : le besoin suppose une intelligence distante qui raisonne"
    return "MCP : le besoin est une capacite locale, sans etat ni dialogue"


for besoin in [
    "acceder a la meteo de Paris",
    "deleguer la redaction d'un rapport a un agent redacteur",
    "lire un fichier du disque",
    "negocier un planning avec l'agent d'un collegue",
]:
    print(f"{besoin:52s} -> {recommander_protocole(besoin)}")

acceder a la meteo de Paris                          -> MCP : le besoin est une capacite locale, sans etat ni dialogue
deleguer la redaction d'un rapport a un agent redacteur -> A2A : le besoin suppose une intelligence distante qui raisonne
lire un fichier du disque                            -> MCP : le besoin est une capacite locale, sans etat ni dialogue
negocier un planning avec l'agent d'un collegue      -> A2A : le besoin suppose une intelligence distante qui raisonne

7. Exercices

Trois exercices, du plus mécanique au plus conceptuel. Les stubs sont volontairement minimaux : à vous de les compléter (la cellule doit s’exécuter sans erreur même non complétée).

Exercice 1 : aiguilleur d’agents

Un orchestrateur reçoit une demande et une liste de cartes d’agents (dictionnaires simplifiés). Écrire choisir_agent qui renvoie le nom de l’agent dont une compétence correspond le mieux au besoin (mots-clés en commun), ou None si aucun ne convient.

cartes_simplifiees = [
    {
        "nom": "agent-traducteur",
        "skills": ["traduction, majuscules, texte"],
    },
    {
        "nom": "agent-meteo",
        "skills": ["meteo, previsions, temperature"],
    },
    {
        "nom": "agent-planning",
        "skills": ["agenda, reunion, creneau, negocier"],
    },
]


def choisir_agent(besoin: str, cartes: list) -> str | None:
    """
    TODO etudiant : renvoyer le nom de l'agent le mieux place pour ce besoin.
    Indice : decouper le besoin en mots, compter les intersections avec chaque
    chaine skills, renvoyer le nom du maximum (ou None si zero partout).
    """
    # Etape 1 : normaliser le besoin en mots minuscules
    # Etape 2 : pour chaque carte, compter les mots du besoin presents dans skills
    # Etape 3 : renvoyer le nom de la meilleure carte, ou None
    return None  # TODO etudiant


print("Exercice a completer : choisir_agent('je veux la meteo de demain', cartes)")
resultat_ex1 = choisir_agent("je veux la meteo de demain", cartes_simplifiees)
print("resultat actuel :", resultat_ex1)

Exercice a completer : choisir_agent('je veux la meteo de demain', cartes)
resultat actuel : None

Exercice 2 : un exécuteur qui compte

Écrire CompteurExecutor, un AgentExecutor dont la compétence est de renvoyer le nombre de mots du message reçu (format libre, par exemple [compteur] 7 mots). Le squelette est donné ; seule la méthode execute est à compléter — en vous inspirant de TraducteurExecutor (section 3).

class CompteurExecutor(AgentExecutor):
    """Competence : compter les mots du message recu."""

    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        """
        TODO etudiant : recuperer le texte (context.get_user_input()), compter
        ses mots, envoyer new_text_message(f"[compteur] {n} mots").
        """
        # Etape 1 : texte_recu = context.get_user_input() or ""
        # Etape 2 : n_mots = len(texte_recu.split())
        # Etape 3 : await event_queue.enqueue_event(new_text_message(...))
        pass  # TODO etudiant


print("Exercice a completer : CompteurExecutor.execute")
print("verification attendue une fois complete : '[compteur] 5 mots' pour")
print("  'combien de mots dans cette phrase'")

Exercice a completer : CompteurExecutor.execute
verification attendue une fois complete : '[compteur] 5 mots' pour
  'combien de mots dans cette phrase'

Exercice 3 : détecter une compétence manquante

Avant de déléguer, un bon client A2A vérifie que la carte de l’agent distant déclare bien la compétence voulue. Écrire skill_disponible qui, given la carte sérialisée (dictionnaire JSON, comme celle servie par /.well-known/agent-card.json) et un identifiant de compétence, renvoie True/False.

carte_json = {
    "name": "agent-traducteur",
    "version": "1.0.0",
    "skills": [
        {"id": "majuscule", "name": "Traduction en majuscules"},
        {"id": "minuscule", "name": "Traduction en minuscules"},
    ],
}


def skill_disponible(carte: dict, identifiant: str) -> bool:
    """
    TODO etudiant : renvoyer True si la carte declare un skill avec cet id.
    Indice : la carte serialisee porte une liste "skills" de dictionnaires "id".
    """
    # Etape 1 : parcourir carte.get("skills", [])
    # Etape 2 : comparer chaque skill.get("id") a l'identifiant
    return False  # TODO etudiant


print("Exercice a completer : skill_disponible(carte_json, 'majuscule')")
print("resultat actuel :", skill_disponible(carte_json, "majuscule"))

Exercice a completer : skill_disponible(carte_json, 'majuscule')
resultat actuel : False

8. Conclusion

Ce notebook a fait coopérer deux agents locaux via le protocole A2A de référence : carte d’agent publiée à l’URL bien connue, serveur JSON-RPC monté avec a2a-sdk 1.1.5, client qui découvre puis délègue, cycle de vie des tâches passé en revue. Le verdict SOTA est SOTA-OK : le SDK de référence est installé et utilisé (pas de réimplémentation jouet du protocole), et toutes les sorties affichées viennent d’exécutions réelles.

Ce qui n’est pas montré ici, volontairement : le streaming d’événements (SSE), les notifications push, l’authentification (cartes signées, security_schemes), et la rencontre avec un agent propulsé par un LLM. Ce sont des extensions naturelles : la structure AgentExecutor reçoit n’importe quelle logique — remplacer la mise en majuscules par un appel à un modèle de la série Texte ne change rien au protocole, ce qui est précisément la promesse d’A2A.

Pour le volet sécurité des protocoles d’agents — descriptions d’outils qui injectent des instructions, définitions qui changent après approbation, collisions de noms entre serveurs — voir le notebook jumeau de la série Security : Tooling-MCP-Attack-Surface.

Retour au sommet