1. Présentation
Dans cet atelier de programmation, vous allez exécuter un agent ADK qui réserve des billets de cinéma auprès de plusieurs marchands de cinéma à l'aide de deux protocoles commerciaux Open Source :
- UCP (Universal Commerce Protocol) : norme permettant aux agents de découvrir des marchands, de rechercher des catalogues et de gérer les flux de paiement.
- AP2 (Agent Payments Protocol) : protocole permettant d'autoriser les paiements de manière sécurisée et vérifiable à l'aide de mandats signés de manière cryptographique.
L'application de démonstration, CineAgent, se connecte à deux faux marchands de cinéma dotés de différentes fonctionnalités (sélection de sièges, formats spécialisés et modes de paiement) et orchestre l'intégralité du processus de réservation, de la recherche au paiement.
Points abordés
- Fonctionnement de la découverte des marchands UCP via les profils
/.well-known/ucp - Comment un agent ADK utilise UCP pour rechercher des catalogues et créer des paiements
- Comment les mandats AP2 (CartMandate, PaymentMandate) sécurisent les transactions
- Comment les protocoles UCP et AP2 de bout en bout sécurisent le commerce électronique agentique
Prérequis
- Un projet Google Cloud avec facturation activée
- Un navigateur Web (par exemple, Chrome)
- Python 3.11+
Cet atelier de programmation s'adresse aux développeurs intermédiaires qui connaissent déjà Python et Google Cloud. Cet atelier de programmation prend environ 15 minutes.
Les ressources créées dans cet atelier de programmation devraient coûter moins de 5 $.
2. Comprendre les protocoles UCP et AP2
Avant de nous lancer dans la création de l'agent, examinons les deux protocoles qui rendent possible ce commerce agentique sécurisé.
Universal Commerce Protocol (UCP)
UCP standardise la façon dont les agents IA interagissent avec les marchands. Il résout le problème des agents qui doivent apprendre des API personnalisées pour chaque magasin en introduisant un modèle de ressources standardisé.
Fonctionnement :
- Découverte : chaque marchand conforme à UCP expose un profil à un emplacement standard :
/.well-known/ucp. Exemple : Point de terminaison UCP d'Everlane.Lorsqu'un agent lit ce profil, il recherche les éléments suivants :- Fonctionnalités : fonctionnalités de base autonomes proposées par une entreprise, comme la recherche dans le catalogue ou le règlement.
- Services : couches de communication de niveau inférieur utilisées pour échanger des données. Exemples : API REST, MCP (Model Context Protocol), A2A (Agent2Agent Protocol).
- Extensions : si un marchand a besoin d'un comportement spécialisé, il peut définir des extensions personnalisées dans ce profil.
- Opérations : une fois découvert, l'agent utilise le point de terminaison de service fourni pour effectuer des opérations. Dans cet atelier de programmation, nous utilisons le Model Context Protocol (MCP) comme transport de service. L'agent effectue des appels JSON-RPC 2.0 à ce point de terminaison pour invoquer les fonctionnalités découvertes : recherche de produits, création de paiements et finalisation d'achats.
Agent Payments Protocol (AP2)
AP2 normalise la façon dont les paiements sont autorisés par les agents au nom des utilisateurs. Il résout le problème de sécurité lié aux agents qui gèrent les identifiants de paiement sensibles.
Fonctionnement :
- Mandat de panier : lorsqu'un agent crée un règlement à l'aide du protocole UCP, le marchand renvoie un
CartMandate. Il s'agit d'un objet JSON contenant les détails du panier et une signature cryptographique du marchand. Elle fait office de garantie de blocage du prix. Le marchand ne peut pas modifier le prix après avoir émis ce mandat. - Mandat de paiement : après avoir vérifié le contenu du panier, l'utilisateur (ou l'agent en son nom) crée un
PaymentMandatepour autoriser le paiement. CePaymentMandatefait référence àCartMandateet inclut la signature cryptographique (ou le jeton d'autorisation) de l'utilisateur. - Vérification de la double signature : le marchand reçoit les deux mandats. Ils vérifient leur propre signature sur le
CartMandateet la signature de l'utilisateur sur lePaymentMandate. Si les deux sont valides, la transaction est effectuée.
Ce système de "double verrouillage" garantit que les marchands ne peuvent pas facturer un montant trop élevé et que les agents ne peuvent pas dépenser sans autorisation. En production, ces mandats utilisent SD-JWT (Selective Disclosure JWT) pour protéger la confidentialité des utilisateurs.
3. Configurer votre environnement
Configuration du projet Google Cloud
Créer un projet Google Cloud
- Dans la console Google Cloud, sur la page de sélection du 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.
Démarrer Cloud Shell
Cloud Shell est un environnement de ligne de commande exécuté dans Google Cloud et fourni avec les outils nécessaires.
- Cliquez sur Activer Cloud Shell en haut de la console Google Cloud.
- Une fois connecté à Cloud Shell, vérifiez votre authentification :
gcloud auth list - Vérifiez que votre projet est configuré :
gcloud config get project - Si votre projet n'est pas défini comme prévu, définissez-le :
export PROJECT_ID=<YOUR_PROJECT_ID> gcloud config set project $PROJECT_ID
Accéder aux modèles Gemini
Dans votre environnement Cloud Shell, copiez et collez les commandes suivantes. Cela permettra d'accéder aux modèles Gemini que l'agent Cine utilisera.
export GOOGLE_CLOUD_PROJECT=$PROJECT_ID
export GOOGLE_CLOUD_LOCATION=global
export GOOGLE_GENAI_USE_VERTEXAI=True
Configurer la structure des répertoires
Copiez et collez les commandes suivantes pour créer un répertoire pour l'agent :
mkdir -p agent_payments
cd agent_payments
Installer des dépendances
Google Cloud Shell est préinstallé avec uv pour gérer l'environnement et les dépendances.
- Créez un fichier
pyproject.tomlà la racine de votre dossieragent_paymentset ajoutez-y le contenu suivant. Ce fichier définit les métadonnées et les dépendances du projet.
[project]
name = "agent-payments-demo"
version = "0.1.0"
description = "CineAgent booking agent using UCP and AP2"
requires-python = ">=3.11"
dependencies = [
"google-adk>=1.29.0",
"google-genai>=1.27.0",
"fastapi>=0.115.0",
"uvicorn>=0.34.0",
"httpx>=0.28.0",
"ap2 @ git+https://github.com/google-agentic-commerce/AP2.git@main",
"ucp-sdk @ git+https://github.com/Universal-Commerce-Protocol/python-sdk.git@main",
]
- Exécutez la commande suivante pour créer l'environnement virtuel et installer toutes les dépendances :
uv sync
- Activez l'environnement virtuel créé par
uv:
source .venv/bin/activate
4. Définir l'agent
Avant d'écrire une logique d'outil, définissons l'agent lui-même dans un fichier nommé agent.py. Cet agent servira d'orchestrateur pour le flux de réservation de billets de cinéma.
Créez une agent.py :
"""CineAgent — movie ticket booking agent using UCP and AP2."""
from google.adk.agents import Agent
from google.adk.tools import FunctionTool
from agent_payments.tools import (
discover_theaters,
search_movies,
get_movie_detail,
create_checkout,
complete_purchase,
)
root_agent = Agent(
model="gemini-3.1-pro-preview",
name="cineagent",
description="Movie ticket booking agent using UCP and AP2.",
instruction="""You are CineAgent, a movie ticket booking assistant.
You help users find and book movie tickets across multiple theaters
using UCP (Universal Commerce Protocol) and AP2 (Agent Payments).
**Your tools:**
- discover_theaters: Find theaters and what they support
- search_movies: Search movies across all theaters
- get_movie_detail: Get showtimes at a specific theater
- create_checkout: Start a checkout session
- complete_purchase: Finalize with AP2 mandate signing
**Rules:**
- Always call discover_theaters first if you haven't yet
- Keep responses concise — summarize and suggest next steps
- Prices from tools are in cents (1500 = $15.00)
- Never invent data — only state what tools return
""",
tools=[
discover_theaters,
search_movies,
get_movie_detail,
create_checkout,
FunctionTool(complete_purchase, require_confirmation=True),
],
)
Explication du code
Examinons ce qui se passe dans cette définition d'agent :
model="gemini-3.1-pro-preview": nous utilisons le dernier modèle Gemini Pro en version Preview pour le raisonnement complexe et l'utilisation d'outils.instruction: il s'agit du prompt qui guide le comportement de l'agent. Il indique explicitement à l'agent d'utiliser UCP et AP2, liste les outils disponibles et définit des règles essentielles telles que "N'invente jamais de données" et "Les prix sont en centimes".tools: il s'agit d'une liste de fonctions Python (que nous allons créer ensuite) que l'agent peut choisir d'appeler en fonction des demandes de l'utilisateur.require_confirmation: vous pouvez encapsuler n'importe quel outil avecFunctionTool(my_function,require_confirmation=True). Lorsqu'il est déclenché, l'agent s'interrompt et attend une simple approbation ("oui" ou "non") avant d'exécuter l'outil. Ici, avant d'exécuter l'outilcomplete_purchase, l'agent s'interrompt pour obtenir une confirmation humaine.
Liste des outils
La définition de l'agent déclare ce que nous devons créer. Chaque outil correspond à une opération spécifique dans le protocole UCP ou AP2 :
Outil | Description | Action de protocole |
| Trouver des marchands et leurs fonctionnalités | Requêtes |
| Rechercher des catalogues de différents marchands | Point de terminaison JSON-RPC vers MCP |
| Obtenir les horaires des séances chez un marchand spécifique | Point de terminaison JSON-RPC vers MCP |
| Démarrer une session de paiement | Point de terminaison JSON-RPC vers MCP |
| Autoriser le paiement et finaliser la commande | Signe le mandat AP2 et l'envoie à MCP |
Le modèle Gemini décidera quand appeler chaque outil en fonction de la conversation. Notre prochaine tâche consiste à implémenter ce que fait chaque outil dans tools.py.
5. Créer les outils de l'agent : découverte et navigation
Implémentons maintenant les outils que l'agent utilisera pour parcourir et découvrir des films. Chaque outil encapsule une opération UCP.
Créez un fichier nommé tools.py et copiez-collez le code suivant :
"""Agent tools — each one wraps a UCP or AP2 operation."""
import asyncio
import json
from .ucp import UCPClient
from .ap2 import AP2Handler
# Initialize clients directly
_merchant_urls = ["http://localhost:8081", "http://localhost:8082"]
_ucp = UCPClient()
_ap2 = AP2Handler()
Comprendre la configuration
Pour que cet atelier de programmation reste axé sur la création de l'agent, nous utilisons deux classes d'assistance, UCPClient et AP2Handler, que nous examinerons dans une étape ultérieure.
- De quoi s'agit-il ? Il s'agit de classes d'assistance écrites à la main que nous avons créées pour cet atelier de programmation afin de simuler l'interaction avec les marchands fictifs. Étant donné que les SDK officiels UCP et AP2 ne sont pas encore disponibles, nous utilisons ces assistants pour combler le vide. Dans un environnement de production, vous utiliserez les SDK officiels une fois qu'ils seront disponibles.
- Pour l'instant, traitez-les comme des objets d'assistance :
_ucp.discover(url): récupère le profil d'un marchand._ucp.mcp_call(url, method, params): envoie une requête JSON-RPC 2.0 au point de terminaison MCP du marchand.
Découvrir les cinémas
Cet outil est la première étape du processus UCP. Il identifie les marchands existants et les services qu'ils proposent.
Ajoutez à tools.py :
async def discover_theaters() -> str:
"""Discover available theater merchants and their capabilities via UCP."""
theaters = []
for url in _merchant_urls:
info = await _ucp.discover(url)
theaters.append(
{
"url": url,
"name": info["name"],
"capabilities": info["capabilities"],
"payment_handlers": info["payment_handlers"],
}
)
return json.dumps(theaters, indent=2)
Fonction :
- Il parcourt une liste d'URL de marchands fournies par le serveur. Dans cet atelier de programmation, nous allons configurer deux marchands fictifs dans la section suivante.
- Pour chaque URL, il appelle
_ucp.discover(url), qui atteint le point de terminaison/.well-known/ucp. - Il collecte le nom, les fonctionnalités et les gestionnaires de paiement dans une liste récapitulative.
- Il renvoie la liste sous forme de chaîne JSON pour que l'agent puisse la lire.
Rechercher des films
Cet outil recherche tous les marchands découverts et fusionne les résultats. C'est essentiel, car le même film peut être projeté dans plusieurs cinémas avec des formats (IMAX, Dolby) et des prix différents.
Ajoutez à tools.py :
async def search_movies(query: str = "") -> str:
"""Search for movies across all theaters. Use '' to browse all."""
all_movies = {}
for url, merchant in _ucp.merchants.items():
result = await _ucp.mcp_call(url, "search_catalog", {"query": query})
for product in result.get("products", []):
mid = product["id"]
if mid not in all_movies:
all_movies[mid] = {
"id": mid,
"title": product["title"],
"categories": product.get("categories", []),
"theaters": {},
}
showtimes = []
for v in product.get("variants", []):
opts = {
o["name"]: o["value"]
for o in v.get("selected_options", [])
}
showtimes.append(
{
"id": v["id"],
"format": opts.get("format", "Standard"),
"time": opts.get("time", ""),
"price": v.get("price", {}),
"seats": v.get("availability", {}).get(
"seats_available", 0
),
}
)
all_movies[mid]["theaters"][url] = {
"name": merchant["name"],
"showtimes": showtimes,
}
return json.dumps(list(all_movies.values()), indent=2)
Fonction :
- Il parcourt tous les cinémas découverts.
- Il envoie une requête JSON-RPC de catalogue de recherche (
_ucp.mcp_call(url, "search_catalog", {"query": query})) au point de terminaison MCP du marchand. - Il effectue ensuite un peu de nettoyage pour analyser les résultats et trouver les films et leurs "variantes" (qui représentent des horaires et des formats spécifiques). Regroupe les films par ID afin que l'utilisateur ne voie pas d'entrées de films en double.
Obtenir les détails d'un film
Cet outil permet d'obtenir tous les détails du catalogue pour un film spécifique dans un cinéma spécifique.
Ajoutez à tools.py :
async def get_movie_detail(movie_id: str, merchant_url: str) -> str:
"""Get detailed showtimes for a movie at a specific theater."""
result = await _ucp.mcp_call(
merchant_url, "lookup_catalog", {"product_id": movie_id}
)
return json.dumps(result, indent=2)
Fonction :
- Il appelle la méthode
lookup_catalogsur le point de terminaison MCP du marchand, en transmettant lemovie_idspécifique. Vous obtiendrez des informations détaillées, comme les horaires des séances et les places disponibles pour ce cinéma en particulier.
6. Créer les outils de l'agent : paiement et règlement
Ces outils gèrent le parcours de paiement et d'achat. C'est là qu'intervient le protocole AP2 pour garantir la sécurité des transactions.
Créer une page de paiement
Cet outil lance une session de paiement dans un cinéma spécifique pour une séance spécifique.
Ajoutez à tools.py :
async def create_checkout(
merchant_url: str, showtime_id: str, quantity: int = 1
) -> str:
"""Create a checkout session for tickets at a theater."""
result = await _ucp.mcp_call(merchant_url, "create_checkout", {
"checkout": {
"line_items": [
{"item": {"id": showtime_id}, "quantity": quantity}
],
"context": {"country": "US", "currency": "USD"},
}
})
return json.dumps(result, indent=2)
Fonction :
- Il appelle la méthode
create_checkoutsur le point de terminaison MCP du marchand. - Il transmet les
showtime_idetquantitydemandés par l'utilisateur. - Le marchand renvoie un objet JSON contenant un AP2 CartMandate.
Remarque : Si vous examinez les données de réponse, ap2.cart_mandate contient un champ merchant_authorization. Il s'agit de la signature cryptographique du marchand qui bloque le prix indiqué. Ils ne pourront pas le modifier par la suite.
Finaliser l'achat
Trois choses se produisent dans cet outil :
- Nous obtenons le CartMandate à partir du paiement et le vérifions (simulé).
- Créez et signez le PaymentMandate (factice).
- Envoyez le mandat signé au marchand pour finaliser l'achat.
Ajoutez à tools.py :
async def complete_purchase(
checkout_id: str, merchant_url: str, payment_method: str = "card"
) -> str:
"""Complete purchase with AP2 payment authorization."""
# 1. Get the CartMandate from the checkout
checkout = await _ucp.mcp_call(
merchant_url, "get_checkout", {"checkout": {"id": checkout_id}}
)
cart_mandate = _ap2.process_cart_mandate(checkout)
if not cart_mandate:
return {"error": "No cart mandate — checkout may have expired"}
# 2-3. Create and sign the PaymentMandate
# In production, this call would trigger a user prompt (biometric or device auth)
# via the AP2 Wallet SDK. In this demo, it just computes a mock SHA-256 hash.
payment_mandate = _ap2.create_payment_mandate(cart_mandate, payment_method)
# 4. Send both mandates to complete the purchase
result = await _ucp.mcp_call(merchant_url, "complete_checkout", {
"checkout": {
"id": checkout_id,
"payment": {
"instruments": [{
"handler_id": f"card_{merchant_url.split(':')[-1]}",
"type": "card",
}],
},
"ap2": {"payment_mandate": payment_mandate},
}
})
return json.dumps(result, indent=2)
Pourquoi deux mandats ? CartMandate garantit que le marchand ne peut pas modifier le prix après l'avoir indiqué. PaymentMandate garantit que l'agent ne peut pas débiter l'utilisateur sans son consentement. Le flux est organisé ainsi :
Merchant locks price -> User authorizes charge -> Merchant verifies both -> Order completes.
Point de contrôle : tools.py complet
Votre tools.py complet doit maintenant comporter cinq fonctions d'outil et une initialisation au niveau du module pour les clients UCP et AP2. Vérifiez qu'il ressemble à ceci :
"""Agent tools — each one wraps a UCP or AP2 operation."""
import asyncio
import json
from ucp import UCPClient
from ap2 import AP2Handler
# Initialize clients directly
_merchant_urls = ["http://localhost:8081", "http://localhost:8082"]
_ucp = UCPClient()
_ap2 = AP2Handler()
async def discover_theaters() -> str:
"""Discover available theater merchants and their capabilities via UCP."""
theaters = []
for url in _merchant_urls:
info = await _ucp.discover(url)
theaters.append(
{
"url": url,
"name": info["name"],
"capabilities": info["capabilities"],
"payment_handlers": info["payment_handlers"],
}
)
return json.dumps(theaters, indent=2)
async def search_movies(query: str = "") -> str:
"""Search for movies across all theaters. Use '' to browse all."""
all_movies = {}
for url, merchant in _ucp.merchants.items():
result = await _ucp.mcp_call(url, "search_catalog", {"query": query})
for product in result.get("products", []):
mid = product["id"]
if mid not in all_movies:
all_movies[mid] = {
"id": mid,
"title": product["title"],
"categories": product.get("categories", []),
"theaters": {},
}
showtimes = []
for v in product.get("variants", []):
opts = {
o["name"]: o["value"]
for o in v.get("selected_options", [])
}
showtimes.append(
{
"id": v["id"],
"format": opts.get("format", "Standard"),
"time": opts.get("time", ""),
"price": v.get("price", {}),
"seats": v.get("availability", {}).get(
"seats_available", 0
),
}
)
all_movies[mid]["theaters"][url] = {
"name": merchant["name"],
"showtimes": showtimes,
}
return json.dumps(list(all_movies.values()), indent=2)
async def get_movie_detail(movie_id: str, merchant_url: str) -> str:
"""Get detailed showtimes for a movie at a specific theater."""
result = await _ucp.mcp_call(
merchant_url, "lookup_catalog", {"product_id": movie_id}
)
return json.dumps(result, indent=2)
async def create_checkout(
merchant_url: str, showtime_id: str, quantity: int = 1
) -> str:
"""Create a checkout session for tickets at a theater."""
result = await _ucp.mcp_call(merchant_url, "create_checkout", {
"checkout": {
"line_items": [
{"item": {"id": showtime_id}, "quantity": quantity}
],
"context": {"country": "US", "currency": "USD"},
}
})
return json.dumps(result, indent=2)
async def complete_purchase(
checkout_id: str, merchant_url: str, payment_method: str = "card"
) -> str:
"""Complete purchase with AP2 payment authorization."""
# 1. Get the CartMandate from the checkout
checkout = await _ucp.mcp_call(
merchant_url, "get_checkout", {"checkout": {"id": checkout_id}}
)
cart_mandate = _ap2.process_cart_mandate(checkout)
if not cart_mandate:
return {"error": "No cart mandate — checkout may have expired"}
# 2-3. Create and sign the PaymentMandate
# In production, this call would trigger a user prompt (biometric or device auth)
# via the AP2 Wallet SDK. In this demo, it just computes a mock SHA-256 hash.
payment_mandate = _ap2.create_payment_mandate(cart_mandate, payment_method)
# 4. Send both mandates to complete the purchase
result = await _ucp.mcp_call(merchant_url, "complete_checkout", {
"checkout": {
"id": checkout_id,
"payment": {
"instruments": [{
"handler_id": f"card_{merchant_url.split(':')[-1]}",
"type": "card",
}],
},
"ap2": {"payment_mandate": payment_mandate},
}
})
return json.dumps(result, indent=2)
7. Rendre le code exécutable
Avec de vrais marchands UCP, il vous suffirait d'indiquer les URL à l'agent. Pour cet atelier de programmation, nous avons besoin de deux éléments pour effectuer des tests en local :
- Marchands fictifs : serveurs locaux qui simulent des points de terminaison UCP pour vous permettre d'effectuer des tests
- Assistance pour les protocoles : wrappers HTTP fins pour UCP et AP2 (en production, les SDK officiels les remplacent)
Remarque : Vous n'avez pas besoin de lire attentivement ce code. Ces fichiers simulent ce que fourniraient une infrastructure et des SDK réels. Copiez-les tels quels.
Assistants de protocole
UCP et AP2 ne disposent pas encore de SDK client. Ces deux fichiers gèrent la plomberie HTTP.
Créez une ucp.py :
"""UCP client — discovers merchants and calls their MCP tools."""
import uuid
import httpx
class UCPClient:
def __init__(self):
self.client = httpx.AsyncClient(timeout=30)
self.merchants = {} # url -> merchant info dict
async def discover(self, merchant_url: str) -> dict:
"""Fetch a merchant's UCP profile from /.well-known/ucp."""
resp = await self.client.get(f"{merchant_url}/.well-known/ucp")
resp.raise_for_status()
profile = resp.json()
ucp = profile["ucp"]
info = {
"name": merchant_url.split("//")[-1],
"mcp_endpoint": ucp["services"]["dev.ucp.shopping"][0]["endpoint"],
"capabilities": list(ucp.get("capabilities", {}).keys()),
"payment_handlers": list(ucp.get("payment_handlers", {}).keys()),
}
self.merchants[merchant_url] = info
return info
async def mcp_call(
self, merchant_url: str, tool_name: str, arguments: dict
) -> dict:
"""Call a merchant's MCP tool via JSON-RPC 2.0."""
merchant = self.merchants[merchant_url]
resp = await self.client.post(
merchant["mcp_endpoint"],
json={
"jsonrpc": "2.0",
"id": uuid.uuid4().hex,
"method": "tools/call",
"params": {"name": tool_name, "arguments": arguments},
},
)
resp.raise_for_status()
data = resp.json()
if "error" in data:
raise Exception(f"MCP error: {data['error']}")
return data.get("result", {})
async def close(self):
await self.client.aclose()
Créez une ap2.py :
"""AP2 mandate handler — creates and signs payment mandates."""
import uuid
import hashlib
class AP2Handler:
def process_cart_mandate(self, checkout_response: dict) -> dict | None:
"""Extract the merchant-signed CartMandate from a checkout response.
The CartMandate is the merchant's cryptographic price guarantee —
it locks the total so it can't change between checkout and payment.
"""
return checkout_response.get("ap2", {}).get("cart_mandate")
def create_payment_mandate(
self, cart_mandate: dict, payment_method: str = "card"
) -> dict:
"""Create and sign a PaymentMandate authorizing payment.
References the merchant's CartMandate and adds user authorization.
Together they form a two-party agreement: merchant guarantees price,
user authorizes charge.
"""
contents = cart_mandate["contents"]
mandate_id = uuid.uuid4().hex
return {
"mandate_id": mandate_id,
"cart_reference": contents["id"],
"merchant": contents["merchant_name"],
"total": contents["total"],
"payment_method": payment_method,
"user_authorization": self._sign(mandate_id, contents["id"]),
}
def _sign(self, mandate_id: str, checkout_id: str) -> str:
"""Sign the mandate. Production uses real crypto (sd-jwt-vc)."""
payload = f"{mandate_id}:{checkout_id}"
return hashlib.sha256(payload.encode()).hexdigest()
Marchands fictifs
Créez une merchants.py :
"""Mock UCP merchant servers — two theaters with different capabilities."""
import uuid
import time
import multiprocessing
from datetime import datetime, timezone, timedelta
import uvicorn
from fastapi import FastAPI
# ── Theater data ────────────────────────────────────────────
THEATERS = {
8081: {
"name": "Meridian Cinemas",
"movies": [
{
"id": "opp",
"title": "Oppenheimer",
"categories": ["Drama", "History"],
"showtimes": [
{"id": "st_opp_7pm_imax", "format": "IMAX", "time": "7:00 PM", "price": 2200, "seats": 45},
{"id": "st_opp_930pm", "format": "Standard", "time": "9:30 PM", "price": 1500, "seats": 80},
],
},
{
"id": "dune3",
"title": "Dune: Part Three",
"categories": ["Sci-Fi", "Action"],
"showtimes": [
{"id": "st_dune_8pm_imax", "format": "IMAX", "time": "8:00 PM", "price": 2200, "seats": 30},
],
},
],
"discounts": {},
},
8082: {
"name": "StarLight Theaters",
"movies": [
{
"id": "opp",
"title": "Oppenheimer",
"categories": ["Drama", "History"],
"showtimes": [
{"id": "st_opp_6pm_atmos", "format": "Dolby Atmos", "time": "6:00 PM", "price": 1800, "seats": 60},
],
},
{
"id": "spider",
"title": "Spider-Verse",
"categories": ["Animation", "Action"],
"showtimes": [
{"id": "st_spider_4pm", "format": "Standard", "time": "4:00 PM", "price": 1200, "seats": 100},
],
},
],
"discounts": {},
},
}
def create_app(port):
theater = THEATERS[port]
app = FastAPI()
sessions = {}
# ── UCP Discovery endpoint ──────────────────────────────
@app.get("/.well-known/ucp")
def discovery():
caps = {
"dev.ucp.shopping.catalog.search": [{"version": "2026-01-15"}],
"dev.ucp.shopping.catalog.lookup": [{"version": "2026-01-15"}],
"dev.ucp.shopping.checkout": [{"version": "2026-01-15"}],
"dev.ucp.shopping.ap2_mandate": [{"version": "2026-01-15"}],
}
return {
"ucp": {
"version": "2026-01-15",
"services": {
"dev.ucp.shopping": [
{"version": "2026-01-15", "transport": "mcp",
"endpoint": f"http://localhost:{port}/mcp"}
]
},
"capabilities": caps,
"payment_handlers": {
"com.example.card": [
{"id": f"card_{port}", "version": "2026-01-15",
"available_instruments": [{"type": "card"}], "config": {}}
]
},
}
}
# ── MCP JSON-RPC endpoint ───────────────────────────────
@app.post("/mcp")
def mcp(body: dict):
tool = body["params"]["name"]
args = body["params"].get("arguments", {})
rid = body.get("id", "1")
if tool == "search_catalog":
q = args.get("query", "").lower()
hits = [m for m in theater["movies"]
if not q or q in m["title"].lower()
or any(q in c.lower() for c in m["categories"])]
return _ok(rid, {"products": [_product(m) for m in hits]})
if tool == "lookup_catalog":
mid = args.get("product_id") or (args.get("ids", [None])[0])
movie = next((m for m in theater["movies"] if m["id"] == mid), None)
if not movie:
return _err(rid, "Not found")
return _ok(rid, {"products": [_product(movie)]})
if tool == "create_checkout":
co = args.get("checkout", {})
sid = f"chk_{uuid.uuid4().hex[:12]}"
items, subtotal = [], 0
for li in co.get("line_items", []):
st = _find_showtime(li["item"]["id"])
if not st:
continue
mv = _find_movie(li["item"]["id"])
qty = li.get("quantity", 1)
amt = st["price"] * qty
subtotal += amt
items.append({
"id": f"li_{uuid.uuid4().hex[:8]}",
"item": {
"id": st["id"],
"title": f"{mv['title']} — {st['format']} {st['time']}",
"price": st["price"],
},
"quantity": qty,
"totals": [{"type": "subtotal", "amount": amt}],
})
tax = int(subtotal * 0.08)
total = subtotal + tax
session = {
"id": sid,
"status": "ready_for_complete",
"currency": "USD",
"line_items": items,
"totals": [
{"type": "subtotal", "display_text": "Subtotal", "amount": subtotal},
{"type": "tax", "display_text": "Tax", "amount": tax},
{"type": "total", "display_text": "Total", "amount": total},
],
"metadata": {"theater_name": theater["name"]},
"ap2": {
"cart_mandate": {
"contents": {
"id": sid,
"merchant_name": theater["name"],
"total": {
"label": "Total",
"amount": {"currency": "USD", "value": total / 100},
},
"cart_expiry": (
datetime.now(timezone.utc) + timedelta(minutes=10)
).isoformat(),
},
"merchant_authorization": f"mock_merchant_sig_{sid}",
}
},
}
sessions[sid] = session
return _ok(rid, session)
if tool == "get_checkout":
sid = args.get("checkout", {}).get("id") or args.get("id")
return _ok(rid, sessions.get(sid, {"error": "not_found"}))
if tool == "complete_checkout":
co = args.get("checkout", {})
sid = co.get("id")
session = sessions.get(sid)
if not session:
return _err(rid, "Not found")
session["status"] = "completed"
session["order"] = {
"id": f"ord_{uuid.uuid4().hex[:8]}",
"created_at": datetime.now(timezone.utc).isoformat(),
"tickets": [
{
"movie": li["item"]["title"],
"quantity": li["quantity"],
"ticket_code": uuid.uuid4().hex[:8].upper(),
}
for li in session["line_items"]
],
}
session["ap2"]["payment_mandate_verified"] = True
return _ok(rid, session)
return _err(rid, f"Unknown tool: {tool}")
def _ok(rid, result):
return {"jsonrpc": "2.0", "id": rid, "result": result}
def _err(rid, msg):
return {"jsonrpc": "2.0", "id": rid, "error": {"code": -32000, "message": msg}}
def _product(movie):
return {
"id": movie["id"],
"title": movie["title"],
"categories": movie["categories"],
"variants": [
{
"id": st["id"],
"selected_options": [
{"name": "format", "value": st["format"]},
{"name": "time", "value": st["time"]},
],
"price": {"amount": st["price"], "currency": "USD"},
"availability": {"available": True, "seats_available": st["seats"]},
}
for st in movie["showtimes"]
],
}
def _find_showtime(sid):
return next(
(st for m in theater["movies"] for st in m["showtimes"] if st["id"] == sid),
None,
)
def _find_movie(sid):
return next(
(m for m in theater["movies"] for st in m["showtimes"] if st["id"] == sid),
None,
)
return app
def _run(port):
uvicorn.run(create_app(port), host="0.0.0.0", port=port, log_level="warning")
if __name__ == "__main__":
for port in THEATERS:
multiprocessing.Process(target=_run, args=(port,), daemon=True).start()
print("Merchants running: Meridian (:8081), StarLight (:8082)")
print("Press Ctrl+C to stop")
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
pass
Cela crée deux serveurs FastAPI, chacun avec deux points de terminaison :
GET /.well-known/ucp: découverte UCP. Renvoie les capacités du marchand, l'URL du point de terminaison MCP et les modes de paiement acceptés.POST /mcp: opérations MCP (Model Context Protocol). Gère les appels JSON-RPC 2.0 pour la recherche dans le catalogue, le règlement, les remises et le paiement.
Démarrez les marchands dans un nouveau terminal. Ils doivent rester en cours d'exécution :
cd agent_payments
source .venv/bin/activate
python merchants.py
Vous devriez obtenir le résultat suivant :
Merchants running: Meridian (:8081), StarLight (:8082)
De retour dans votre premier terminal, vérifiez la découverte UCP :
curl -s http://localhost:8081/.well-known/ucp | python -m json.tool
Vous devriez voir les fonctionnalités du marchand, l'URL du point de terminaison MCP et les gestionnaires de paiement.
8. Exécuter l'agent avec ADK Web
Utilisons l'UI Web intégrée à l'ADK CLI. Il fournit une interface de chat dans le navigateur et gère automatiquement les invites de confirmation des outils.
Votre projet devrait maintenant se présenter comme suit :
agent_payments/
├── merchants.py # Mock UCP merchants
├── ucp.py # UCP client helper
├── ap2.py # AP2 mandate handler
├── tools.py # Agent tools
├── agent.py # Agent definition
└── pyproject.toml
Essayer
Dans le terminal actuel, accédez au répertoire parent (un dossier au-dessus de agent_payments) et démarrez l'UI Web ADK :
cd ../
adk web --allow_origins '*'
Vous devriez obtenir un résultat indiquant que le serveur est en cours d'exécution, avec une URL (généralement http://localhost:8000 ou similaire).
Parler à l'agent
- Ouvrez l'URL fournie par
adk webdans votre navigateur. - Une interface de chat s'affiche.
- Essayez de demander Quels sont les films à l'affiche ?
- L'agent découvre les cinémas et les catalogues de recherche en arrière-plan, en agrégeant les résultats des deux marchands via UCP.
- Demandez à réserver des billets : "Réserve deux billets pour Oppenheimer à 19h".
- Lorsque l'agent tente d'appeler
complete_purchase, notez comment l'interface utilisateur Web de l'ADK affiche une boîte de dialogue de confirmation ou une fiche. - Pour autoriser la transaction, répondez dans le chat avec cette chaîne JSON exacte :
{"confirmed": true}. - L'agent finalisera l'achat et vous enverra la confirmation de commande avec les codes des billets.
Voici ce qui s'est passé en coulisses :
create_checkout→ le marchand a renvoyé un CartMandate AP2 (prix bloqué signé)complete_purchase→ a créé un PaymentMandate, l'a signé (SHA-256 fictif) et a envoyé les deux mandats au marchand.- Le marchand a validé les deux signatures → émission des billets (dans la simulation)
9. Effectuer un nettoyage
Pour éviter de laisser des serveurs locaux en cours d'exécution, nettoyez les ressources :
- Dans le terminal exécutant
adk web, appuyez sur Ctrl+C pour arrêter le serveur de l'agent. - Dans le terminal exécutant
python merchants.py, appuyez sur Ctrl+C pour arrêter les marchands fictifs. - Désactivez l'environnement virtuel dans les deux terminaux en exécutant la commande suivante :
deactivate
- (Facultatif) Si vous avez créé un projet Google Cloud pour cet atelier de programmation et que vous souhaitez le supprimer, exécutez la commande suivante :
gcloud projects delete $GOOGLE_CLOUD_PROJECT
10. Félicitations ! 🎉
Vous avez créé un agent ADK qui découvre des marchands, parcourt des catalogues et effectue des achats à l'aide d'UCP et d'AP2.
Ce que vous avez appris
Dans cet atelier de programmation, vous avez créé un agent ADK qui gère les flux commerciaux sécurisés. Voici un résumé de ce que vous avez créé et des concepts clés que vous avez appliqués :
Ce que vous avez créé :
- Cinq outils d'agent encapsulant les opérations UCP et AP2 : découverte, recherche dans le catalogue, paiement.
- Signature du mandat AP2 : CartMandate (prix bloqué du marchand) + PaymentMandate (autorisation de l'utilisateur).
- Recherche multimarchands : un agent interroge plusieurs cinémas et fusionne les résultats.
Concepts clés :
Protocole | Description | Fonctionnement |
Découverte de l'UCP | L'agent trouve des marchands et leurs capacités |
|
UCP MCP | L'agent parcourt les catalogues et crée des paiements. | Appels JSON-RPC 2.0 au point de terminaison MCP du marchand |
AP2 CartMandate | Le marchand bloque le prix indiqué | Signée par le marchand, incluant le montant total et la date d'expiration |
AP2-PaymentMandate | L'utilisateur autorise le débit | Signé par l'utilisateur, référence CartMandate |
Quelles sont les différences en production ?
Cet atelier de programmation utilise des mocks. En production :
- La découverte UCP est résolue par rapport à un registre, et non à des URL localhost codées en dur.
- Les points de terminaison MCP sont hébergés par de vrais marchands. Ils utilisent le même protocole JSON-RPC 2.0 et proposent un inventaire réel.
- Les mandats AP2 sont signés avec sd-jwt-vc, et non avec des hachages SHA-256.
- L'autorisation de paiement utilise un SDK AP2 Wallet avec des invites de consentement de l'utilisateur.
- Le frontend affiche les résultats des outils sous forme d'UI enrichie (grilles de produits, récapitulatifs de paiement, cartes de confirmation).
Étapes suivantes
- Explorez les spécifications du protocole AP2 et créez votre propre agent compatible avec les paiements.
- Implémenter l'UCP pour votre marchand afin d'activer le commerce agentique
- Connecter AP2 à A2A pour les workflows commerciaux multi-agents
- En savoir plus sur l'extension AP2 x402 pour les paiements en cryptomonnaie