1. Introduction
Le protocole agent-to-agent (A2A) est conçu pour standardiser la communication entre les agents d'IA, en particulier ceux déployés dans des systèmes externes. Auparavant, de tels protocoles étaient établis pour les outils appelés Model Context Protocol (MCP), qui est une norme émergente permettant de connecter les LLM aux données et aux ressources. A2A tente de compléter le protocole MCP. Alors que le protocole MCP vise à réduire la complexité pour connecter les agents aux outils et aux données, A2A se concentre sur la façon de permettre aux agents de collaborer dans leurs modalités naturelles. Il permet aux agents de communiquer en tant qu'agents (ou en tant qu'utilisateurs) plutôt qu'en tant qu'outils. Par exemple, il permet une communication bidirectionnelle lorsque vous souhaitez commander quelque chose.
A2A est conçu pour compléter MCP. Dans la documentation officielle, il est recommandé aux applications d'utiliser MCP pour les outils et A2A pour les agents, représentés par AgentCard ( nous en parlerons plus tard). Les frameworks peuvent ensuite utiliser A2A pour communiquer avec leur utilisateur, les agents à distance et d'autres agents.
Dans cette démonstration, nous allons commencer par implémenter A2A à l'aide de son SDK Python. Nous allons explorer un cas d'utilisation dans lequel nous disposons d'un concierge d'achat personnel qui peut nous aider à communiquer avec les agents des vendeurs de burgers et de pizzas pour gérer notre commande.
A2A utilise le principe client-serveur. Voici le flux A2A typique que nous attendons dans cette démonstration.
- Le client A2A effectue d'abord une découverte sur toutes les fiches d'agent de serveur A2A accessibles et utilise ses informations pour créer un client de connexion.
- Si nécessaire, le client A2A enverra un message au serveur A2A, qui l'évaluera comme une tâche à accomplir. Si l'URL du récepteur de notifications push est configurée sur le client A2A et prise en charge par le serveur A2A, le serveur pourra également publier l'état de la progression de la tâche sur le point de terminaison de réception du client.
- Une fois la tâche terminée, le serveur A2A envoie l'artefact de réponse au client A2A.
Dans cet atelier de programmation, vous allez suivre une approche par étapes :
- Préparez votre projet Google Cloud et activez toutes les API requises.
- Configurer l'espace de travail pour votre environnement de programmation
- Préparer les variables d'environnement pour les services d'agents de burgers et de pizzas, puis les tester localement
- Déployer l'agent de burgers et de pizzas sur Cloud Run
- Inspecter les détails de l'établissement du serveur A2A
- Préparer les variables d'environnement pour le service de conciergerie d'achat et tester localement
- Déployer un concierge d'achat dans Agent Engine
- Se connecter au moteur de l'agent via une interface locale
- Examiner les détails de l'établissement du client A2A et de sa modélisation des données
- Inspecter le payload et l'interaction entre le client et le serveur A2A
Présentation de l'architecture
Nous allons déployer l'architecture de service suivante.
Nous allons déployer deux services qui serviront de serveur A2A : l'agent Burger ( basé sur le framework d'agent CrewAI) et l'agent Pizza ( basé sur le framework d'agent Langgraph). L'utilisateur n'interagira directement qu'avec le concierge d'achat, qui sera exécuté à l'aide du framework Agent Development Kit (ADK) et qui servira de client A2A.
Chacun de ces agents disposera de son propre environnement et de son propre déploiement.
Prérequis
- Vous êtes à l'aise avec Python.
- Comprendre l'architecture full stack de base à l'aide du service HTTP
Points abordés
- Structure de base du serveur A2A
- Structure principale du client A2A
- Déployer le service d'agent sur Cloud Run
- Déployer un service d'agent dans Agent Engine
- Comment le client A2A se connecte au serveur A2A
- Structure des requêtes et des réponses sur une connexion sans flux
Prérequis
- Navigateur Web Chrome
- Un compte Gmail
- Un projet Cloud pour lequel la facturation est activée
Cet atelier de programmation, conçu pour les développeurs de tous niveaux (y compris les débutants), utilise Python dans son exemple d'application. Toutefois, vous n'avez pas besoin de connaître Python pour comprendre les concepts présentés.
2. Avant de commencer
Sélectionner le projet actif dans la console Cloud
Cet atelier de programmation suppose que vous disposez déjà d'un projet Google Cloud pour lequel la facturation est activée. Si vous ne l'avez pas encore, vous pouvez suivre les instructions ci-dessous pour commencer.
- Dans la console Google Cloud, sur la page du sélecteur de projet, sélectionnez ou créez un projet Google Cloud.
- Assurez-vous que la facturation est activée pour votre projet Cloud. Découvrez comment vérifier si la facturation est activée sur un projet.
Configurer un projet Cloud dans le terminal Cloud Shell
- Vous allez utiliser Cloud Shell, un environnement de ligne de commande exécuté dans Google Cloud et fourni avec bq. Cliquez sur "Activer Cloud Shell" en haut de la console Google Cloud. Si vous êtes invité à autoriser, cliquez sur Autoriser.
- Une fois connecté à Cloud Shell, vérifiez que vous êtes déjà authentifié et que le projet est défini sur votre ID de projet à l'aide de la commande suivante :
gcloud auth list
- Exécutez la commande suivante dans Cloud Shell pour vérifier que la commande gcloud connaît votre projet.
gcloud config list project
- Si votre projet n'est pas défini, utilisez la commande suivante pour le définir :
gcloud config set project <YOUR_PROJECT_ID>
Vous pouvez également voir l'ID PROJECT_ID dans la console.
Cliquez dessus pour afficher tous vos projets et leur ID sur la droite.
- Activez les API requises à l'aide de la commande ci-dessous. Cette opération peut prendre quelques minutes. Veuillez patienter.
gcloud services enable aiplatform.googleapis.com \
run.googleapis.com \
cloudbuild.googleapis.com \
cloudresourcemanager.googleapis.com
Si la commande s'exécute correctement, un message semblable à celui ci-dessous s'affiche :
Operation "operations/..." finished successfully.
Vous pouvez également utiliser la console au lieu de la commande gcloud. Pour ce faire, recherchez chaque produit ou utilisez ce lien.
Si vous oubliez d'activer une API, vous pourrez toujours le faire au cours de l'implémentation.
Consultez la documentation pour connaître les commandes gcloud ainsi que leur utilisation.
Accéder à l'éditeur Cloud Shell et configurer le répertoire de travail de l'application
Nous pouvons maintenant configurer notre éditeur de code pour effectuer certaines tâches de codage. Pour cela, nous allons utiliser l'éditeur Cloud Shell.
- Cliquez sur le bouton "Ouvrir l'éditeur" pour ouvrir un éditeur Cloud Shell dans lequel vous pourrez écrire votre code.
- Assurez-vous que le projet Cloud Code est défini en bas à gauche (barre d'état) de l'éditeur Cloud Shell, comme indiqué dans l'image ci-dessous, et qu'il est défini sur le projet Google Cloud actif pour lequel la facturation est activée. Cliquez sur Autoriser si vous y êtes invité. Si vous avez déjà suivi la commande précédente, le bouton peut également pointer directement vers votre projet activé au lieu du bouton de connexion.
- Ensuite, clonons le répertoire de travail du modèle pour cet atelier de programmation à partir de GitHub en exécutant la commande suivante. Le répertoire de travail sera créé dans le répertoire purchasing-concierge-a2a.
git clone https://github.com/alphinside/purchasing-concierge-intro-a2a-codelab-starter.git purchasing-concierge-a2a
- Ensuite, accédez à la section supérieure de l'éditeur Cloud Shell et cliquez sur File > Open Folder (Fichier > Ouvrir le dossier), recherchez votre répertoire nom d'utilisateur, puis le répertoire purchasing-concierge-a2a, puis cliquez sur le bouton OK. Le répertoire choisi deviendra le répertoire de travail principal. Dans cet exemple, le nom d'utilisateur est alvinprayuda. Le chemin d'accès au répertoire est donc indiqué ci-dessous.
Votre éditeur Cloud Shell devrait maintenant se présenter comme suit :
Configurer l'environnement
L'étape suivante consiste à préparer l'environnement de développement. Votre terminal actif actuel doit se trouver dans le répertoire de travail purchasing-concierge-a2a. Dans cet atelier de programmation, nous utiliserons Python 3.12 et le gestionnaire de projets Python uv pour simplifier la création et la gestion de la version Python et de l'environnement virtuel.
- Si vous n'avez pas encore ouvert le terminal, ouvrez-le en cliquant sur Terminal > Nouveau terminal ou en utilisant le raccourci clavier Ctrl+Maj+C. Une fenêtre de terminal s'ouvre en bas du navigateur.
- Initialisons maintenant l'environnement virtuel du concierge d'achat à l'aide de
uv(déjà préinstallé sur le terminal Cloud). Exécutez cette commande :
uv sync --frozen
Cela créera le répertoire .venv et installera les dépendances. Un petit aperçu rapide du fichier pyproject.toml vous donnera des informations sur les dépendances affichées comme ceci :
dependencies = [
"a2a-sdk>=0.2.16",
"google-adk>=1.8.0",
"gradio>=5.38.2",
]
- Pour tester l'environnement virtuel, créez un fichier main.py et copiez-y le code suivant :
def main():
print("Hello from purchasing-concierge-a2a!")
if __name__ == "__main__":
main()
- Exécutez ensuite la commande suivante :
uv run main.py
Vous obtiendrez un résultat semblable à celui ci-dessous.
Using CPython 3.12 Creating virtual environment at: .venv Hello from purchasing-concierge-a2a!
Cela montre que le projet Python est correctement configuré.
Nous pouvons maintenant passer à l'étape suivante : configurer et déployer notre agent de vendeur à distance.
3. Déployer l'agent de vendeur à distance (serveur A2A) sur Cloud Run
Dans cette étape, nous allons déployer ces deux agents de vendeurs à distance, marqués par le cadre rouge. L'agent de burger sera optimisé par le framework d'agent CrewAI et l'agent de pizza sera optimisé par l'agent Langgraph. Les deux seront optimisés par le modèle Gemini Flash 2.0.
Déployer l'agent Remote Burger
Le code source de l'agent Burger se trouve dans le répertoire remote_seller_agents/burger_agent. L'initialisation de l'agent peut être inspectée dans le script agent.py. Voici l'extrait de code de l'agent initialisé.
from crewai import Agent, Crew, LLM, Task, Process
from crewai.tools import tool
...
model = LLM(
model="vertex_ai/gemini-2.5-flash-lite", # Use base model name without provider prefix
)
burger_agent = Agent(
role="Burger Seller Agent",
goal=(
"Help user to understand what is available on burger menu and price also handle order creation."
),
backstory=("You are an expert and helpful burger seller agent."),
verbose=False,
allow_delegation=False,
tools=[create_burger_order],
llm=model,
)
agent_task = Task(
description=self.TaskInstruction,
agent=burger_agent,
expected_output="Response to the user in friendly and helpful manner",
)
crew = Crew(
tasks=[agent_task],
agents=[burger_agent],
verbose=False,
process=Process.sequential,
)
inputs = {"user_prompt": query, "session_id": sessionId}
response = crew.kickoff(inputs)
return response
...
Tous les fichiers qui se trouvent dans le répertoire remote_seller_agents/burger_agent suffisent déjà à déployer notre agent sur Cloud Run afin qu'il soit accessible en tant que service. Nous y reviendrons plus tard. Exécutez la commande suivante pour le déployer :
gcloud run deploy burger-agent \
--source remote_seller_agents/burger_agent \
--port=8080 \
--allow-unauthenticated \
--min 1 \
--region us-central1 \
--update-env-vars GOOGLE_CLOUD_LOCATION=us-central1 \
--update-env-vars GOOGLE_CLOUD_PROJECT={your-project-id}
Si vous êtes invité à créer un dépôt de conteneurs pour le déploiement à partir de la source, répondez Y. Cela ne se produisait que si vous n'aviez jamais déployé de code source sur Cloud Run auparavant. Une fois le déploiement réussi, un journal semblable à celui-ci s'affiche.
Service [burger-agent] revision [burger-agent-xxxxx-xxx] has been deployed and is serving 100 percent of traffic. Service URL: https://burger-agent-xxxxxxxxx.us-central1.run.app
La partie xxxx sera un identifiant unique lorsque nous déploierons le service. Essayons maintenant d'accéder à la route https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json de ces services d'agent de burger déployés via le navigateur. Il s'agit de l'URL permettant d'accéder à la fiche de l'agent serveur A2A déployé.
Si le déploiement a réussi, la réponse suivante s'affiche dans votre navigateur lorsque vous accédez à https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json :
Il s'agit des informations de la fiche de l'agent Burger qui doivent être accessibles à des fins de découverte. Nous y reviendrons plus tard. Notez que la valeur url est toujours définie sur http://0.0.0.0:8080/ ici. Cette valeur url doit être l'information principale pour le client A2A afin d'envoyer des messages depuis le monde extérieur. Elle n'est pas configurée correctement. Pour cette démonstration, nous devrons remplacer cette valeur par l'URL de notre service d'agent de burgers en ajoutant une variable d'environnement supplémentaire HOST_OVERRIDE.
Mettre à jour la valeur de l'URL de l'agent Burger sur la fiche de l'agent à l'aide d'une variable d'environnement
Pour ajouter HOST_OVERRIDE au service d'agent Burger, procédez comme suit :
- Recherchez Cloud Run dans la barre de recherche en haut de la console Cloud.
- Cliquez sur le service Cloud Run burger-agent précédemment déployé.
- Copiez l'URL de burger-service, puis cliquez sur Modifier et déployer la nouvelle révision.
- Cliquez ensuite sur la section Variables et secrets.
- Cliquez ensuite sur Ajouter une variable et définissez la valeur
HOST_OVERRIDEsur l'URL du service ( celle avec le modèlehttps://burger-agent-xxxxxxxxx.us-central1.run.app).
- Enfin, cliquez sur le bouton deploy (déployer) pour redéployer votre service.
Désormais, lorsque vous accédez à nouveau à la fiche de l'agent burger-agent dans le navigateur https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json , la valeur url est déjà correctement configurée.
Déployer l'agent Remote Pizza
De même, le code source de l'agent de pizza se trouve dans le répertoire remote_seller_agents/pizza_agent. L'initialisation de l'agent peut être inspectée dans le script agent.py. Voici l'extrait de code de l'agent initialisé.
from langchain_google_vertexai import ChatVertexAI
from langgraph.prebuilt import create_react_agent
...
self.model = ChatVertexAI(
model="gemini-2.5-flash-lite",
location=os.getenv("GOOGLE_CLOUD_LOCATION"),
project=os.getenv("GOOGLE_CLOUD_PROJECT"),
)
self.tools = [create_pizza_order]
self.graph = create_react_agent(
self.model,
tools=self.tools,
checkpointer=memory,
prompt=self.SYSTEM_INSTRUCTION,
)
...
Comme pour l'étape de déploiement de l'agent de burger précédente, tous les fichiers qui existent dans le répertoire remote_seller_agents/pizza_agent sont déjà suffisants pour déployer notre agent sur Cloud Run afin qu'il soit accessible en tant que service. Exécutez la commande suivante pour le déployer :
gcloud run deploy pizza-agent \
--source remote_seller_agents/pizza_agent \
--port=8080 \
--allow-unauthenticated \
--min 1 \
--region us-central1 \
--update-env-vars GOOGLE_CLOUD_LOCATION=us-central1 \
--update-env-vars GOOGLE_CLOUD_PROJECT={your-project-id}
Une fois le déploiement réussi, un journal semblable à celui-ci s'affiche.
Service [pizza-agent] revision [pizza-agent-xxxxx-xxx] has been deployed and is serving 100 percent of traffic. Service URL: https://pizza-agent-xxxxxxxxx.us-central1.run.app
La partie xxxx sera un identifiant unique lorsque nous déploierons le service. Il en va de même pour l'agent de burger. Lorsque vous essayez d'accéder à la route https://pizza-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json de ces services d'agent de pizza déployés via le navigateur pour accéder à la fiche de l'agent serveur A2A, la valeur url de l'agent de pizza sur sa fiche d'agent n'est pas encore correctement configurée. Nous devons également ajouter HOST_OVERRIDE à sa variable d'environnement.
Mettre à jour la valeur de l'URL de l'agent Pizza sur la fiche de l'agent à l'aide d'une variable d'environnement
Pour ajouter HOST_OVERRIDE au service d'agent de pizza, procédez comme suit :
- Recherchez Cloud Run dans la barre de recherche en haut de la console Cloud.
- Cliquez sur le service Cloud Run pizza-agent déployé précédemment.
- Cliquez sur Modifier et déployer la nouvelle révision.
- Copiez l'URL du service de pizza, puis cliquez sur la section Variables et secrets.
- Cliquez ensuite sur Ajouter une variable et définissez la valeur
HOST_OVERRIDEsur l'URL du service ( celle avec le modèlehttps://pizza-agent-xxxxxxxxx.us-central1.run.app).
- Enfin, cliquez sur le bouton deploy (déployer) pour redéployer votre service.
Désormais, lorsque vous accéderez à nouveau à la fiche de l'agent pizza-agent dans le navigateur https://pizza-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json, la valeur url sera déjà correctement configurée.
À ce stade, nous avons déjà déployé avec succès les services de burgers et de pizzas sur Cloud Run. Abordons à présent les composants principaux du serveur A2A.
4. Composants principaux du serveur A2A
Nous allons maintenant aborder le concept et les composants de base du serveur A2A.
Fiche d'agent
Chaque serveur A2A doit disposer d'une fiche d'agent accessible sur la ressource /.well-known/agent.json. Cela permet de prendre en charge la phase de découverte sur le client A2A, qui doit fournir des informations et des contextes complets sur la façon d'accéder à l'agent et de connaître toutes ses capacités. C'est un peu comme une documentation d'API bien documentée avec Swagger ou Postman.
Contenu de la fiche de l'agent Burger déployé
{
"capabilities": {
"streaming": true
},
"defaultInputModes": [
"text",
"text/plain"
],
"defaultOutputModes": [
"text",
"text/plain"
],
"description": "Helps with creating burger orders",
"name": "burger_seller_agent",
"protocolVersion": "0.2.6",
"skills": [
{
"description": "Helps with creating burger orders",
"examples": [
"I want to order 2 classic cheeseburgers"
],
"id": "create_burger_order",
"name": "Burger Order Creation Tool",
"tags": [
"burger order creation"
]
}
],
"url": "https://burger-agent-109790610330.us-central1.run.app",
"version": "1.0.0"
}
Ces fiches d'agent mettent en avant de nombreux composants importants, tels que les compétences de l'agent, les capacités de streaming, les modalités acceptées, la version du protocole et d'autres éléments.
Toutes ces informations peuvent être utilisées pour développer un mécanisme de communication approprié afin que le client A2A puisse communiquer correctement. La modalité et le mécanisme d'authentification compatibles garantissent que la communication peut être établie correctement et que les informations de l'agent skills peuvent être intégrées à l'invite système du client A2A pour donner à l'agent du client un contexte sur les capacités et les compétences de l'agent distant à invoquer. Pour en savoir plus sur les champs de cette fiche d'agent, consultez cette documentation.
Dans notre code, l'implémentation de la fiche d'agent est établie à l'aide du SDK Python A2A. Consultez l'extrait remote_seller_agents/burger_agent/main.py ci-dessous pour l'implémentation.
...
capabilities = AgentCapabilities(streaming=True)
skill = AgentSkill(
id="create_burger_order",
name="Burger Order Creation Tool",
description="Helps with creating burger orders",
tags=["burger order creation"],
examples=["I want to order 2 classic cheeseburgers"],
)
agent_host_url = (
os.getenv("HOST_OVERRIDE")
if os.getenv("HOST_OVERRIDE")
else f"http://{host}:{port}/"
)
agent_card = AgentCard(
name="burger_seller_agent",
description="Helps with creating burger orders",
url=agent_host_url,
version="1.0.0",
defaultInputModes=BurgerSellerAgent.SUPPORTED_CONTENT_TYPES,
defaultOutputModes=BurgerSellerAgent.SUPPORTED_CONTENT_TYPES,
capabilities=capabilities,
skills=[skill],
)
...
Vous y trouverez plusieurs champs, comme :
AgentCapabilities: déclaration de fonctions facultatives supplémentaires prises en charge par le service d'agent,comme la possibilité de diffuser du contenu en streaming et/ou la prise en charge des notifications pushAgentSkill: outils ou fonctions compatibles avec l'agentInput/OutputModes: modalité de type d'entrée/sortie acceptéeUrl: adresse à utiliser pour communiquer avec l'agent
Dans cette configuration, nous fournissons une création dynamique de l'URL de l'hôte de l'agent, ce qui facilite le passage des tests locaux au déploiement dans le cloud. C'est pourquoi nous devons ajouter la variable HOST_OVERRIDE à l'étape précédente.
File d'attente de tâches et AgentExecutor
Un serveur A2A peut traiter les requêtes de différents agents ou utilisateurs et isoler parfaitement chaque tâche. Pour mieux visualiser les contextes de ces éléments, vous pouvez examiner l'image ci-dessous.
Ainsi, chaque serveur A2A doit être en mesure de suivre les tâches entrantes et de stocker les informations appropriées à leur sujet. Le SDK A2A fournit des modules pour relever ce défi dans le serveur A2A. Tout d'abord, nous pouvons instancier la logique de traitement de la requête entrante. En héritant de la classe abstraite AgentExecutor, nous pouvons contrôler la manière dont nous souhaitons gérer l'exécution et l'annulation des tâches. Cette implémentation d'exemple peut être inspectée au niveau du module remote_seller_agents/burger_agent/agent_executor.py ( chemin d'accès similaire pour le cas du vendeur de pizzas).
...
class BurgerSellerAgentExecutor(AgentExecutor):
"""Burger Seller AgentExecutor."""
def __init__(self):
self.agent = BurgerSellerAgent()
async def execute(
self,
context: RequestContext,
event_queue: EventQueue,
) -> None:
query = context.get_user_input()
try:
result = self.agent.invoke(query, context.context_id)
print(f"Final Result ===> {result}")
parts = [Part(root=TextPart(text=str(result)))]
await event_queue.enqueue_event(
completed_task(
context.task_id,
context.context_id,
[new_artifact(parts, f"burger_{context.task_id}")],
[context.message],
)
)
except Exception as e:
print("Error invoking agent: %s", e)
raise ServerError(error=ValueError(f"Error invoking agent: {e}")) from e
async def cancel(
self, request: RequestContext, event_queue: EventQueue
) -> Task | None:
raise ServerError(error=UnsupportedOperationError())
...
Dans le code ci-dessus, nous implémentons un schéma de traitement de base dans lequel l'agent sera directement appelé lorsque la requête entrante et les événements de tâche terminée seront envoyés une fois l'appel terminé. Toutefois, nous n'avons pas implémenté la méthode d'annulation ici, car il s'agissait d'une opération de courte durée.
Une fois l'exécuteur créé, nous pouvons utiliser directement les DefaultRequestHandler, InMemoryTaskStore et A2AStarletteApplication intégrés pour lancer le serveur HTTP. Cette implémentation peut être inspectée dans remote_seller_agents/burger_agent/__main__.py.
...
request_handler = DefaultRequestHandler(
agent_executor=BurgerSellerAgentExecutor(),
task_store=InMemoryTaskStore(),
)
server = A2AStarletteApplication(
agent_card=agent_card, http_handler=request_handler
)
uvicorn.run(server.build(), host=host, port=port)
...
Ce module vous fournira l'implémentation de la route /.well-known/agent.json pour accéder à la fiche de l'agent, ainsi que le point de terminaison POST pour prendre en charge le protocole A2A.
Résumé
En résumé, notre serveur A2A déployé à l'aide du SDK Python peut prendre en charge les deux fonctionnalités ci-dessous :
- Publication de la fiche de l'agent sur le parcours
/.well-known/agent.json - Gérer les requêtes JSON-RPC avec la mise en file d'attente des tâches en mémoire
Le point d'entrée pour démarrer ces fonctionnalités peut être inspecté dans le script __main__.py ( sur remote_seller_agents/burger_agent ou remote_seller_agents/pizza_agent) .
5. Déployer le concierge d'achat : client A2A vers l'agent moteur
Au cours de cette étape, nous allons déployer l'agent de conciergerie d'achat. C'est avec cet agent que nous allons interagir.
Le code source de notre agent concierge d'achat se trouve dans le répertoire purchasing_concierge. L'initialisation de l'agent peut être inspectée dans le script purchasing_agent.py. Voici l'extrait de code de l'agent initialisé.
from google.adk import Agent
...
def create_agent(self) -> Agent:
return Agent(
model="gemini-2.5-flash-lite",
name="purchasing_agent",
instruction=self.root_instruction,
before_model_callback=self.before_model_callback,
before_agent_callback=self.before_agent_callback,
description=(
"This purchasing agent orchestrates the decomposition of the user purchase request into"
" tasks that can be performed by the seller agents."
),
tools=[
self.send_task,
],
)
...
Nous allons déployer cet agent dans le moteur d'agent. Vertex AI Agent Engine est un ensemble de services qui permet aux développeurs de déployer, de gérer et de faire évoluer des agents d'IA en production. Il gère l'infrastructure pour faire évoluer les agents en production, ce qui nous permet de nous concentrer sur la création d'applications. Pour en savoir plus, consultez ce document . Auparavant, nous devions préparer les fichiers nécessaires au déploiement de notre service d'agent (comme le script de serveur main et le Dockerfile). Dans ce cas, nous pouvons déployer notre agent directement à partir du script Python sans avoir à développer notre propre service de backend en utilisant une combinaison d'ADK et d'Agent Engine. Pour le déployer, procédez comme suit :
- Tout d'abord, nous devons créer notre espace de stockage intermédiaire dans Cloud Storage.
gcloud storage buckets create gs://purchasing-concierge-{your-project-id} --location=us-central1
- Maintenant, nous devons d'abord préparer la variable .env. Copions .env.example dans le fichier .env.
cp .env.example .env
- Ouvrez maintenant le fichier .env. Vous verrez le contenu suivant :
GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT={your-project-id}
GOOGLE_CLOUD_LOCATION=us-central1
STAGING_BUCKET=gs://purchasing-concierge-{your-project-id}
PIZZA_SELLER_AGENT_URL={your-pizza-agent-url}
BURGER_SELLER_AGENT_URL={your-burger-agent-url}
AGENT_ENGINE_RESOURCE_NAME={your-agent-engine-resource-name}
Cet agent communiquera avec les agents de burgers et de pizzas. Nous devons donc fournir les identifiants appropriés pour chacun d'eux. Nous devrons mettre à jour PIZZA_SELLER_AGENT_URL et BURGER_SELLER_AGENT_URL avec l'URL Cloud Run des étapes précédentes.
Si vous l'oubliez, accédez à la console Cloud Run. Saisissez "Cloud Run" dans la barre de recherche en haut de la console, puis effectuez un clic droit sur l'icône Cloud Run pour l'ouvrir dans un nouvel onglet.
Vous devriez voir les services d'agent de vendeur à distance que vous avez déployés précédemment, comme indiqué ci-dessous.
Pour afficher l'URL publique de ces services, cliquez sur l'un d'eux. Vous serez redirigé vers la page "Informations sur le service". Vous pouvez voir l'URL en haut de la page, juste à côté des informations sur la région.
La variable d'environnement finale devrait ressembler à celle-ci :
GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT={your-project-id}
GOOGLE_CLOUD_LOCATION=us-central1
STAGING_BUCKET=gs://purchasing-concierge-{your-project-id}
PIZZA_SELLER_AGENT_URL=https://pizza-agent-xxxxx.us-central1.run.app
BURGER_SELLER_AGENT_URL=https://burger-agent-xxxxx.us-central1.run.app
AGENT_ENGINE_RESOURCE_NAME={your-agent-engine-resource-name}
- Nous sommes maintenant prêts à déployer notre agent de concierge d'achat. Dans cette démonstration, nous allons effectuer le déploiement à l'aide du script
deploy_to_agent_engine.pydont le contenu est indiqué ci-dessous.
"""
Copyright 2025 Google LLC
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import vertexai
from vertexai.preview import reasoning_engines
from vertexai import agent_engines
from dotenv import load_dotenv
import os
from purchasing_concierge.agent import root_agent
load_dotenv()
PROJECT_ID = os.getenv("GOOGLE_CLOUD_PROJECT")
LOCATION = os.getenv("GOOGLE_CLOUD_LOCATION")
STAGING_BUCKET = os.getenv("STAGING_BUCKET")
vertexai.init(
project=PROJECT_ID,
location=LOCATION,
staging_bucket=STAGING_BUCKET,
)
adk_app = reasoning_engines.AdkApp(
agent=root_agent,
)
remote_app = agent_engines.create(
agent_engine=adk_app,
display_name="purchasing-concierge",
requirements=[
"google-cloud-aiplatform[adk,agent_engines]",
"a2a-sdk==0.2.16",
],
extra_packages=[
"./purchasing_concierge",
],
env_vars={
"GOOGLE_GENAI_USE_VERTEXAI": os.environ["GOOGLE_GENAI_USE_VERTEXAI"],
"PIZZA_SELLER_AGENT_URL": os.environ["PIZZA_SELLER_AGENT_URL"],
"BURGER_SELLER_AGENT_URL": os.environ["BURGER_SELLER_AGENT_URL"],
},
)
print(f"Deployed remote app resource: {remote_app.resource_name}")
Voici les étapes nécessaires pour déployer notre agent ADK dans le moteur d'agent. Nous devons d'abord créer un objet AdkApp à partir de notre root_agent ADK. Nous pouvons ensuite exécuter la méthode agent_engines.create en fournissant l'objet adk_app, en spécifiant les exigences dans le champ requirements, en spécifiant le chemin d'accès au répertoire de l'agent dans extra_packages et en fournissant les variables d'environnement nécessaires.
Nous pouvons le déployer en exécutant le script :
uv run deploy_to_agent_engine.py
Une fois le déploiement réussi, un journal semblable à celui-ci s'affiche. Notez que xxxx est l'ID de votre projet et que yyyy est l'ID de ressource de votre moteur d'agent.
AgentEngine created. Resource name: projects/xxxx/locations/us-central1/reasoningEngines/yyyy
To use this AgentEngine in another session:
agent_engine = vertexai.agent_engines.get('projects/xxxx/locations/us-central1/reasoningEngines/yyyy)
Deployed remote app resource: projects/xxxx/locations/us-central1/reasoningEngines/xxxx
Lorsque nous l'inspectons dans le tableau de bord du moteur d'agent (recherchez "moteur d'agent" dans la barre de recherche), notre déploiement précédent s'affiche.
Tester l'agent déployé sur Agent Engine
Vous pouvez interagir avec le moteur d'agent à l'aide de la commande curl et du SDK. Par exemple, exécutez la commande suivante pour interagir avec l'agent déployé.
Vous pouvez essayer d'envoyer cette requête pour vérifier si l'agent a bien été déployé.
curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://us-central1-aiplatform.googleapis.com/v1/projects/{YOUR_PROJECT_ID}/locations/us-central1/reasoningEngines/{YOUR_AGENT_ENGINE_RESOURCE_ID}:streamQuery?alt=sse -d '{
"class_method": "stream_query",
"input": {
"user_id": "user_123",
"message": "List available burger menu please",
}
}'
Si l'opération réussit, plusieurs événements de réponse s'affichent dans la console, comme ceci :
{
"content": {
"parts": [
{
"text": "Here is our burger menu:\n- Classic Cheeseburger: IDR 85K\n- Double Cheeseburger: IDR 110K\n- Spicy Chicken Burger: IDR 80K\n- Spicy Cajun Burger: IDR 85K"
}
],
"role": "model"
},
"usage_metadata": {
"candidates_token_count": 51,
"candidates_tokens_details": [
{
"modality": "TEXT",
"token_count": 51
}
],
"prompt_token_count": 907,
"prompt_tokens_details": [
{
"modality": "TEXT",
"token_count": 907
}
],
"total_token_count": 958,
"traffic_type": "ON_DEMAND"
},
"invocation_id": "e-14679918-af68-45f1-b942-cf014368a733",
"author": "purchasing_agent",
"actions": {
"state_delta": {},
"artifact_delta": {},
"requested_auth_configs": {}
},
"id": "dbe7fc43-b82a-4f3e-82aa-dd97afa8f15b",
"timestamp": 1754287348.941454
}
Nous essaierons d'utiliser l'UI à l'étape suivante. Cependant, discutons d'abord des composants principaux et du flux typique des clients A2A.
6. Composants principaux du client A2A
L'image ci-dessus illustre le flux typique des interactions A2A :
- Le client essaiera de trouver une fiche d'agent publiée dans l'URL de l'agent distant fournie dans la route
/.well-known/agent.json. - Ensuite, si nécessaire, il enverra un message à cet agent avec le message et les paramètres de métadonnées nécessaires ( par exemple, l'ID de session, le contexte historique, etc.). Le serveur percevra ce message comme une tâche à accomplir.
- Le serveur A2A traite la requête. S'il prend en charge les notifications push, il peut également en publier certaines tout au long du traitement de la tâche ( cette fonctionnalité ne fait pas partie de cet atelier de programmation).
- Une fois l'opération terminée, le serveur A2A renvoie l'artefact de réponse au client.
Voici certains des objets principaux pour les interactions ci-dessus (pour en savoir plus, cliquez ici) :
- Message : tour de communication entre un client et un agent à distance
- Tâche : unité de travail fondamentale gérée par A2A, identifiée par un ID unique
- Artefact : résultat (par exemple, un document, une image ou des données structurées) généré par l'agent à la suite d'une tâche, composé de parties
- Part : plus petite unité de contenu dans un message ou un artefact. Une partie peut être un texte, une image, une vidéo, un fichier, etc.
Découverte de cartes
Lorsque le service client A2A est en cours de démarrage, la procédure habituelle consiste à essayer d'obtenir les informations de la fiche de l'agent et à les stocker pour y accéder facilement en cas de besoin. Dans cet atelier de programmation, nous l'implémentons sur before_agent_callback. Vous pouvez voir l'implémentation dans purchasing_concierge/purchasing_agent.py. Consultez l'extrait de code ci-dessous.
...
async def before_agent_callback(self, callback_context: CallbackContext):
if not self.a2a_client_init_status:
httpx_client = httpx.AsyncClient(timeout=httpx.Timeout(timeout=30))
for address in self.remote_agent_addresses:
card_resolver = A2ACardResolver(
base_url=address, httpx_client=httpx_client
)
try:
card = await card_resolver.get_agent_card()
remote_connection = RemoteAgentConnections(
agent_card=card, agent_url=card.url
)
self.remote_agent_connections[card.name] = remote_connection
self.cards[card.name] = card
except httpx.ConnectError:
print(f"ERROR: Failed to get agent card from : {address}")
agent_info = []
for ra in self.list_remote_agents():
agent_info.append(json.dumps(ra))
self.agents = "\n".join(agent_info)
...
Ici, nous essayons d'accéder à toutes les fiches d'agents disponibles à l'aide du module client A2A intégré A2ACardResolver. Nous collectons ensuite la connexion nécessaire pour envoyer un message à l'agent. Après cela, nous devons également lister tous les agents disponibles et leurs spécifications dans la requête afin que notre agent sache qu'il peut communiquer avec ces agents.
Outil de requête et d'envoi de tâches
Voici la requête et l'outil que nous fournissons à notre agent ADK.
...
def root_instruction(self, context: ReadonlyContext) -> str:
current_agent = self.check_active_agent(context)
return f"""You are an expert purchasing delegator that can delegate the user product inquiry and purchase request to the
appropriate seller remote agents.
Execution:
- For actionable tasks, you can use `send_task` to assign tasks to remote agents to perform.
- When the remote agent is repeatedly asking for user confirmation, assume that the remote agent doesn't have access to user's conversation context.
So improve the task description to include all the necessary information related to that agent
- Never ask user permission when you want to connect with remote agents. If you need to make connection with multiple remote agents, directly
connect with them without asking user permission or asking user preference
- Always show the detailed response information from the seller agent and propagate it properly to the user.
- If the remote seller is asking for confirmation, rely the confirmation question to the user if the user haven't do so.
- If the user already confirmed the related order in the past conversation history, you can confirm on behalf of the user
- Do not give irrelevant context to remote seller agent. For example, ordered pizza item is not relevant for the burger seller agent
- Never ask order confirmation to the remote seller agent
Please rely on tools to address the request, and don't make up the response. If you are not sure, please ask the user for more details.
Focus on the most recent parts of the conversation primarily.
If there is an active agent, send the request to that agent with the update task tool.
Agents:
{self.agents}
Current active seller agent: {current_agent["active_agent"]}
"""
...
async def send_task(self, agent_name: str, task: str, tool_context: ToolContext):
"""Sends a task to remote seller agent
This will send a message to the remote agent named agent_name.
Args:
agent_name: The name of the agent to send the task to.
task: The comprehensive conversation context summary
and goal to be achieved regarding user inquiry and purchase request.
tool_context: The tool context this method runs in.
Yields:
A dictionary of JSON data.
"""
if agent_name not in self.remote_agent_connections:
raise ValueError(f"Agent {agent_name} not found")
state = tool_context.state
state["active_agent"] = agent_name
client = self.remote_agent_connections[agent_name]
if not client:
raise ValueError(f"Client not available for {agent_name}")
session_id = state["session_id"]
task: Task
message_id = ""
metadata = {}
if "input_message_metadata" in state:
metadata.update(**state["input_message_metadata"])
if "message_id" in state["input_message_metadata"]:
message_id = state["input_message_metadata"]["message_id"]
if not message_id:
message_id = str(uuid.uuid4())
payload = {
"message": {
"role": "user",
"parts": [
{"type": "text", "text": task}
], # Use the 'task' argument here
"messageId": message_id,
"contextId": session_id,
},
}
message_request = SendMessageRequest(
id=message_id, params=MessageSendParams.model_validate(payload)
)
send_response: SendMessageResponse = await client.send_message(
message_request=message_request
)
print(
"send_response",
send_response.model_dump_json(exclude_none=True, indent=2),
)
if not isinstance(send_response.root, SendMessageSuccessResponse):
print("received non-success response. Aborting get task ")
return None
if not isinstance(send_response.root.result, Task):
print("received non-task response. Aborting get task ")
return None
return send_response.root.result
...
Dans l'invite, nous fournissons à notre agent de conciergerie d'achat le nom et la description de tous les agents à distance disponibles. Dans l'outil self.send_task, nous fournissons un mécanisme permettant de récupérer le client approprié pour se connecter à l'agent et d'envoyer les métadonnées requises à l'aide de l'objet SendMessageRequest.
Protocoles de communication
La définition Task est un domaine appartenant au serveur A2A. Toutefois, du point de vue du client A2A, il s'agit d'un message envoyé au serveur. C'est au serveur de définir comment les messages entrants du client sont définis en tant que tâche et si l'exécution de la tâche nécessite l'interaction du client. Pour en savoir plus sur le cycle de vie des tâches, consultez cette documentation. Le concept de niveau supérieur peut être visualisé ci-dessous :
Cet échange de message > tâche est implémenté à l'aide du format de charge utile au-dessus de la norme JSON-RPC, comme illustré dans l'exemple de protocole message/send ci-dessous :
{
# identifier for this request
"id": "abc123",
# version of JSON-RPC protocol
"jsonrpc": "2.0",
# method name
"method": "message/send",
# parameters/arguments of the method
"params": {
"message": "hi, what can you help me with?"
}
}
Différentes méthodes sont disponibles, par exemple pour prendre en charge différents types de communication (synchronisée, par flux, asynchrone) ou pour configurer les notifications concernant l'état de la tâche. Un serveur A2A peut être configuré de manière flexible pour gérer ces normes de définition de tâches. Vous trouverez des informations détaillées sur ces méthodes dans ce document.
7. Tests d'intégration et inspection des charges utiles
Inspectons maintenant notre concierge d'achat avec l'interaction de l'agent distant à l'aide d'une interface utilisateur Web.
Tout d'abord, nous devons mettre à jour le AGENT_ENGINE_RESOURCE_NAME dans le .Fichier env. Assurez-vous de fournir le nom de ressource correct pour le moteur de l'agent. Votre fichier .env devrait se présenter comme suit :
GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT={your-project-id}
GOOGLE_CLOUD_LOCATION=us-central1
STAGING_BUCKET=gs://purchasing-concierge-{your-project-id}
PIZZA_SELLER_AGENT_URL=https://pizza-agent-xxxxx.us-central1.run.app
BURGER_SELLER_AGENT_URL=https://burger-agent-xxxxx.us-central1.run.app
AGENT_ENGINE_RESOURCE_NAME=projects/xxxx/locations/us-central1/reasoningEngines/yyyy
Exécutez ensuite la commande suivante pour déployer une application Gradio.
uv run purchasing_concierge_ui.py
Si l'opération réussit, le résultat suivant s'affiche :
* Running on local URL: http://0.0.0.0:8080 * To create a public link, set `share=True` in `launch()`.
Ensuite, Ctrl+cliquez sur l'URL http://0.0.0.0:8080 dans le terminal ou cliquez sur le bouton d'aperçu sur le Web pour ouvrir l'interface utilisateur Web.
Essayez d'avoir une conversation comme celle-ci :
- Montre-moi le menu des burgers et des pizzas
- Je veux commander une pizza au poulet grillé et un burger cajun épicé.
et poursuivez la conversation jusqu'à ce que vous ayez terminé la commande. Examinez le déroulement de l'interaction, ainsi que l'appel d'outil et la réponse. L'image suivante est un exemple de résultat d'interaction.
Nous pouvons constater que la communication avec deux agents différents entraîne deux comportements différents, et qu'A2A peut bien gérer cela. L'agent du vendeur de pizzas accepte directement notre demande d'agent d'achat, tandis que l'agent de burgers a besoin de notre confirmation avant de traiter notre demande. Une fois que nous avons confirmé, l'agent peut transmettre la confirmation à l'agent de burgers.
Maintenant que nous avons abordé les concepts de base de l'A2A, voyons comment il est implémenté en tant qu'architecture client/serveur.
8. Défi
Maintenant, pouvez-vous préparer le fichier nécessaire et déployer l'application Gradio dans Cloud Run par vous-même ? Il est temps de relever le défi !
9. Effectuer un nettoyage
Pour éviter que les ressources utilisées dans cet atelier de programmation soient facturées sur votre compte Google Cloud :
- Dans la console Google Cloud, accédez à la page Gérer les ressources.
- Dans la liste des projets, sélectionnez le projet que vous souhaitez supprimer, puis cliquez sur Supprimer.
- Dans la boîte de dialogue, saisissez l'ID du projet, puis cliquez sur Arrêter pour supprimer le projet.
- Vous pouvez également accéder à Cloud Run dans la console, sélectionner le service que vous venez de déployer, puis le supprimer.