1. Panoramica
In questo codelab, eseguirai un agente ADK che prenota biglietti del cinema presso più commercianti di cinema utilizzando due protocolli di commercio open source:
- UCP (Universal Commerce Protocol): uno standard che consente agli agenti di scoprire i commercianti, cercare cataloghi e gestire i flussi di pagamento.
- AP2 (Agent Payments Protocol): un protocollo per l'autorizzazione di pagamento sicura e verificabile tramite mandati firmati crittograficamente.
L'app demo CineAgent si connette a due commercianti di cinema fittizi con funzionalità diverse (selezione del posto, formati specializzati e metodi di pagamento) e orchestra l'intero flusso di prenotazione dalla ricerca al pagamento.
Obiettivi didattici
- Come funziona il rilevamento dei commercianti UCP tramite i profili
/.well-known/ucp - Come un agente ADK utilizza UCP per cercare cataloghi e creare acquisti
- In che modo i mandati AP2 (CartMandate, PaymentMandate) proteggono le transazioni
- Come funzionano i protocolli UCP ed AP2 end-to-end per proteggere l'e-commerce agentico
Che cosa ti serve
- Un progetto cloud Google Cloud con la fatturazione abilitata
- Un browser web come Chrome
- Python 3.11+
Questo codelab è destinato a sviluppatori di livello intermedio che hanno una certa familiarità con Python e Google Cloud. Il completamento di questo codelab richiede circa 15 minuti.
Le risorse create in questo codelab dovrebbero costare meno di 5 $.
2. Informazioni sui protocolli UCP e AP2
Prima di iniziare a creare l'agente, vediamo i due protocolli che rendono possibile questo commercio agentico sicuro.
Universal Commerce Protocol (UCP)
UCP standardizza il modo in cui gli agenti AI interagiscono con i commercianti. Risolve il problema degli agenti che devono apprendere API personalizzate per ogni singolo negozio introducendo un modello di risorse standardizzato.
Come funziona:
- Rilevamento: ogni commerciante conforme a UCP espone un profilo in una posizione standard:
/.well-known/ucp. Esempio: endpoint UCP di Everlane.Quando un agente legge questo profilo, cerca:- Funzionalità: le funzionalità di base autonome supportate da un'attività, come la ricerca nel catalogo o il pagamento.
- Servizi: i livelli di comunicazione di livello inferiore utilizzati per lo scambio di dati. Esempi: API REST, MCP (Model Context Protocol), A2A (Agent2Agent Protocol).
- Estensioni: se un commerciante ha bisogno di un comportamento specializzato, può definire Estensioni personalizzate in questo profilo.
- Operazioni: una volta scoperto, l'agente utilizza l'endpoint di servizio fornito per eseguire le operazioni. In questo codelab utilizziamo il Model Context Protocol (MCP) come trasporto del servizio. L'agente effettua chiamate JSON-RPC 2.0 a questo endpoint per richiamare le funzionalità rilevate: ricerca di prodotti, creazione di pagamenti e completamento degli acquisti.
Agent Payments Protocol (AP2)
AP2 standardizza il modo in cui i pagamenti vengono autorizzati dagli agenti per conto degli utenti. Risolve il problema di sicurezza degli agenti che gestiscono credenziali di pagamento sensibili.
Come funziona:
- Mandato del carrello: quando un agente crea un pagamento utilizzando il protocollo UCP, il commerciante restituisce un
CartMandate. Si tratta di un oggetto JSON contenente i dettagli del carrello e una firma crittografica del commerciante. Funge da garanzia di blocco del prezzo. Il commerciante non può modificare il prezzo dopo l'emissione di questo mandato. - Mandato di pagamento: dopo aver verificato i contenuti del carrello, l'utente (o l'agente per suo conto) crea un
PaymentMandateper autorizzare il pagamento. QuestoPaymentMandatefa riferimento aCartMandatee include la firma crittografica (o il token di autorizzazione) dell'utente. - Verifica della doppia firma: il commerciante riceve entrambi i mandati. Verificano la propria firma sul
CartMandatee la firma dell'utente sulPaymentMandate. Se entrambi sono validi, la transazione va a buon fine.
Questo sistema di "doppia chiusura" garantisce che i commercianti non possano addebitare costi eccessivi e che gli agenti non possano spendere senza autorizzazione. In produzione, questi mandati utilizzano SD-JWT (Selective Disclosure JWT) per proteggere la privacy degli utenti.
3. Configura l'ambiente
Configurazione del progetto Google Cloud
Crea un progetto Google Cloud
- Nella console Google Cloud, nella pagina di selezione del progetto, seleziona o crea un progetto Google Cloud.
- Verifica che la fatturazione sia attivata per il tuo progetto Cloud. Scopri come verificare se la fatturazione è abilitata per un progetto.
Avvia Cloud Shell
Cloud Shell è un ambiente a riga di comando in esecuzione in Google Cloud che viene precaricato con gli strumenti necessari.
- Fai clic su Attiva Cloud Shell nella parte superiore della console Google Cloud.
- Una volta connesso a Cloud Shell, verifica l'autenticazione:
gcloud auth list - Verifica che il progetto sia configurato:
gcloud config get project - Se il progetto non è impostato come previsto, impostalo:
export PROJECT_ID=<YOUR_PROJECT_ID> gcloud config set project $PROJECT_ID
Accedere ai modelli Gemini
Nell'ambiente Cloud Shell, copia e incolla i seguenti comandi. In questo modo verrà abilitato l'accesso ai modelli Gemini che verranno utilizzati da Cine Agent.
export GOOGLE_CLOUD_PROJECT=$PROJECT_ID
export GOOGLE_CLOUD_LOCATION=global
export GOOGLE_GENAI_USE_VERTEXAI=True
Configurare la struttura delle directory
Copia e incolla i seguenti comandi per creare una nuova directory per l'agente:
mkdir -p agent_payments
cd agent_payments
Installa le dipendenze
Google Cloud Shell è preinstallato con uv per gestire l'ambiente e le dipendenze.
- Crea un file
pyproject.tomlnella directory radice della cartellaagent_paymentse aggiungi il seguente contenuto. Questo file definisce i metadati e le dipendenze del progetto.
[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",
]
- Esegui questo comando per creare l'ambiente virtuale e installare tutte le dipendenze:
uv sync
- Attiva l'ambiente virtuale creato da
uv:
source .venv/bin/activate
4. Definisci l'agente
Prima di scrivere la logica di qualsiasi strumento, definiamo l'agente stesso in un file denominato agent.py. Questo agente fungerà da orchestratore per il flusso di prenotazione dei biglietti del cinema.
Crea 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),
],
)
Spiegazione del codice
Analizziamo nel dettaglio cosa succede in questa definizione di agente:
model="gemini-3.1-pro-preview": stiamo utilizzando l'ultima versione di anteprima del modello Gemini Pro per il ragionamento complesso e l'utilizzo degli strumenti.instruction: questo è il prompt che guida il comportamento dell'agente. Indica esplicitamente all'agente di utilizzare UCP e AP2, elenca gli strumenti disponibili e imposta regole fondamentali come "Non inventare mai dati" e "I prezzi sono in centesimi".tools: questo è un elenco di funzioni Python (che creeremo in seguito) che l'agente può scegliere di chiamare in base alle richieste dell'utente.require_confirmation: puoi racchiudere qualsiasi strumento conFunctionTool(my_function,require_confirmation=True). Quando viene attivato, l'agente si mette in pausa e attende una semplice approvazione con "sì" o "no" prima di eseguire lo strumento. Qui, prima di eseguire lo strumentocomplete_purchase, l'agente si mette in pausa per la conferma di una persona.
L'elenco degli strumenti
La definizione dell'agente dichiara ciò che dobbiamo creare. Ogni strumento corrisponde a un'operazione specifica nel protocollo UCP o AP2:
Strumento | Descrizione | Azione del protocollo |
| Trovare i commercianti e le loro funzionalità | Query |
| Cercare cataloghi tra i vari commercianti | Endpoint JSON-RPC a MCP |
| Visualizzare gli orari di programmazione presso un commerciante specifico | Endpoint JSON-RPC a MCP |
| Avvia una sessione di pagamento | Endpoint JSON-RPC a MCP |
| Autorizza il pagamento e completa l'ordine | Firma il mandato AP2 e lo invia a MCP |
Il modello Gemini deciderà quando chiamare ogni strumento in base alla conversazione. Il nostro compito successivo è implementare le funzionalità di ogni strumento in tools.py.
5. Crea gli strumenti dell'agente: scoperta e navigazione
Ora implementiamo gli strumenti che l'agente utilizzerà per navigare e scoprire film. Ogni strumento esegue il wrapping di un'operazione UCP.
Crea un nuovo file denominato tools.py e copia e incolla il seguente codice:
"""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()
Informazioni sulla configurazione
Per mantenere questo codelab incentrato sulla creazione dell'agente, utilizziamo due classi helper, UCPClient e AP2Handler, che esamineremo in un passaggio successivo.
- Che cosa sono? Si tratta di classi helper scritte a mano che abbiamo creato per questo codelab per simulare l'interazione con i commercianti simulati. Poiché gli SDK UCP e AP2 ufficiali non sono ancora disponibili, utilizziamo questi helper per colmare il divario. In un ambiente di produzione, utilizzerai gli SDK ufficiali non appena saranno disponibili.
- Per il momento, trattali come oggetti helper:
_ucp.discover(url): recupera il profilo di un commerciante._ucp.mcp_call(url, method, params): invia una richiesta JSON-RPC 2.0 all'endpoint MCP del commerciante.
Scopri i cinema
Questo strumento è il primo passaggio del flusso UCP. Trova i commercianti esistenti e i servizi che supportano.
Aggiungi a 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)
Cosa fa:
- Esegue l'iterazione su un elenco di URL del commerciante forniti dal server. In questo codelab, configureremo due commercianti simulati nella sezione successiva.
- Per ogni URL, chiama
_ucp.discover(url), che raggiunge l'endpoint/.well-known/ucp. - Raccoglie il nome, le funzionalità e i gestori dei pagamenti in un elenco riepilogativo.
- Restituisce l'elenco come stringa JSON da leggere per l'agente.
Cerca film
Questo strumento esegue ricerche in tutti i commercianti rilevati e unisce i risultati. Questo è fondamentale perché lo stesso film potrebbe essere proiettato in più cinema con formati diversi (IMAX, Dolby) e prezzi diversi.
Aggiungi a 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)
Cosa fa:
- Esegue un ciclo su tutti i cinema rilevati.
- Invia una richiesta JSON-RPC del catalogo di ricerca (
_ucp.mcp_call(url, "search_catalog", {"query": query})) all'endpoint MCP del commerciante. - Poi esegue una pulizia per analizzare i risultati e trovare i film e le relative "varianti" (che rappresentano orari e formati specifici). Raggruppa i film per ID in modo che l'utente non veda voci duplicate.
Ottieni dettagli film
Questo strumento recupera i dettagli completi del catalogo per un film specifico in un cinema specifico.
Aggiungi a 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)
Cosa fa:
- Chiama il metodo
lookup_catalogsull'endpoint MCP del commerciante, passando ilmovie_idspecifico. Vengono restituite informazioni dettagliate come gli orari degli spettacoli e la disponibilità dei posti per quel cinema specifico.
6. Crea gli strumenti dell'agente: Checkout e pagamento
Questi strumenti gestiscono il flusso di pagamento e acquisto. È qui che entra in gioco il protocollo AP2 per garantire transazioni sicure.
Crea Acquisto rapido
Questo strumento avvia una sessione di pagamento in un cinema specifico per un orario di proiezione specifico.
Aggiungi a 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)
Cosa fa:
- Chiama il metodo
create_checkoutsull'endpoint MCP del commerciante. - Trasmette
showtime_idequantityrichiesti dall'utente. - Il commerciante restituisce un oggetto JSON contenente un AP2 CartMandate.
Nota: se esamini i dati della risposta, ap2.cart_mandate contiene un campo merchant_authorization. Si tratta della firma crittografica del commerciante che blocca il prezzo indicato. Non potrà modificarlo in un secondo momento.
Completa l'acquisto
In questo strumento si verificano tre cose:
- Recuperiamo CartMandate dal pagamento e lo verifichiamo (mock).
- Crea e firma il PaymentMandate (mock).
- Invia il mandato firmato al commerciante per completare l'acquisto.
Aggiungi a 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)
Perché due mandati? Il mandato del carrello garantisce che il commerciante non possa modificare il prezzo dopo averlo indicato. Il mandato di pagamento garantisce che l'agente non possa addebitare costi all'utente senza il suo consenso. Il flusso è:
Merchant locks price -> User authorizes charge -> Merchant verifies both -> Order completes.
Posto di blocco: pieno tools.py
Il tuo tools.py completo ora dovrebbe avere 5 funzioni di strumenti e l'inizializzazione a livello di modulo per i client UCP e AP2. Verifica che abbia il seguente aspetto:
"""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. Rendere eseguibile un elemento
Con i commercianti UCP reali, indirizzeresti l'agente ai loro URL e il gioco è fatto. Per questo codelab, abbiamo bisogno di due cose per eseguire i test in locale:
- Commercianti simulati: server locali che simulano gli endpoint UCP in modo da avere qualcosa da testare
- Helper di protocollo: wrapper HTTP sottili per UCP e AP2 (in produzione, gli SDK ufficiali li sostituiscono)
Nota: non è necessario leggere attentamente questo codice. Questi file simulano ciò che fornirebbero infrastrutture e SDK reali. Copiali così come sono.
Helper di protocollo
UCP e AP2 non hanno ancora SDK client. Questi due file gestiscono l'infrastruttura HTTP.
Crea 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()
Crea 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()
Commercianti simulati
Crea 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
Vengono creati due server FastAPI, ciascuno con due endpoint:
GET /.well-known/ucp: discovery UCP. Restituisce le funzionalità del commerciante, l'URL dell'endpoint MCP e i metodi di pagamento accettati.POST /mcp: operazioni MCP (Model Context Protocol). Gestisce le chiamate JSON-RPC 2.0 per la ricerca nel catalogo, il checkout, gli sconti e il pagamento.
Avvia i commercianti in un nuovo terminale, che deve rimanere in esecuzione:
cd agent_payments
source .venv/bin/activate
python merchants.py
Dovresti vedere:
Merchants running: Meridian (:8081), StarLight (:8082)
Torna al primo terminale e verifica l'individuazione di UCP:
curl -s http://localhost:8081/.well-known/ucp | python -m json.tool
Dovresti visualizzare le funzionalità del commerciante, l'URL dell'endpoint MCP e i gestori dei pagamenti.
8. Esegui l'agente con ADK Web
Utilizziamo la UI web integrata di ADK CLI. In questo modo viene fornita un'interfaccia di chat nel browser e vengono gestite automaticamente le richieste di conferma degli strumenti.
Il tuo progetto dovrebbe ora essere simile a questo:
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
Prova
Nel terminale corrente, vai alla directory principale (una cartella sopra agent_payments) e avvia la GUI web dell'ADK:
cd ../
adk web --allow_origins '*'
Dovresti visualizzare un output che indica che il server è in esecuzione e che ti fornisce un URL (di solito http://localhost:8000 o simile).
Parla con l'agente
- Apri l'URL fornito da
adk webnel browser. - Vedrai un'interfaccia di chat.
- Prova a chiedere: "Che film ci sono?"
- L'agente scoprirà i cinema e cercherà i cataloghi dietro le quinte, aggregando i risultati di entrambi i commercianti tramite UCP.
- Chiedi di prenotare i biglietti: "Prenota due biglietti per Oppenheimer per le 19:00".
- Quando l'agente tenta di chiamare
complete_purchase, nota come l'interfaccia utente web dell'ADK visualizza una finestra di dialogo di conferma o una scheda. - Per autorizzare la transazione, rispondi nella chat con questa stringa JSON esatta:
{"confirmed": true}. - L'agente completerà l'acquisto e ti invierà la conferma dell'ordine con i codici dei biglietti.
Ecco cosa è successo dietro le quinte:
create_checkout→ il commerciante ha restituito un CartMandate (blocco del prezzo firmato) AP2complete_purchase→ ha creato un PaymentMandate, lo ha firmato (SHA-256 simulato) e ha inviato entrambi i mandati al commerciante- Il commerciante ha verificato entrambe le firme → biglietti emessi (nel mock)
9. Esegui la pulizia
Per evitare di lasciare in esecuzione i server locali, libera spazio nelle risorse:
- Nel terminale in cui è in esecuzione
adk web, premi Ctrl+C per arrestare il server dell'agente. - Nel terminale in cui è in esecuzione
python merchants.py, premi Ctrl+C per arrestare i commercianti simulati. - Disattiva l'ambiente virtuale in entrambi i terminali eseguendo:
deactivate
- (Facoltativo) Se hai creato un nuovo progetto Google Cloud per questo codelab e vuoi eliminarlo, esegui:
gcloud projects delete $GOOGLE_CLOUD_PROJECT
10. Complimenti! 🎉
Hai creato un agente ADK che scopre i commercianti, sfoglia i cataloghi e completa gli acquisti utilizzando UCP e AP2.
Che cosa hai imparato
In questo codelab hai creato un agente ADK che gestisce flussi di commercio sicuri. Ecco un riepilogo di ciò che hai creato e dei concetti chiave che hai applicato:
Cosa hai creato:
- 5 strumenti per gli agenti che eseguono il wrapping delle operazioni UCP e AP2: discovery, ricerca nel catalogo, pagamento.
- Firma del mandato AP2: CartMandate (blocco del prezzo del commerciante) + PaymentMandate (autorizzazione dell'utente).
- Ricerca multinegozio: un agente esegue query su più cinema e unisce i risultati.
Concetti chiave:
Protocollo | Descrizione | Come funziona |
UCP Discovery | L'agente trova i commercianti e le loro funzionalità |
|
UCP MCP | L'agente sfoglia i cataloghi e crea i pagamenti | Chiamate JSON-RPC 2.0 all'endpoint MCP del commerciante |
AP2 CartMandate | Il commerciante blocca il prezzo indicato | Firmata dal commerciante, include totale + scadenza |
AP2 PaymentMandate | L'utente autorizza l'addebito | Firmato dall'utente, fa riferimento a CartMandate |
Che cosa cambia nella produzione?
Questo codelab utilizza simulazioni. In produzione:
- Rilevamento UCP viene risolto in base a un registro, non a URL localhost hardcoded
- Gli endpoint MCP sono ospitati da commercianti reali: stesso protocollo JSON-RPC 2.0, inventario reale
- I mandati AP2 sono firmati con sd-jwt-vc, non con hash SHA-256
- L'autorizzazione di pagamento utilizza un SDK AP2 Wallet con richieste di consenso dell'utente
- Il frontend esegue il rendering dei risultati dello strumento come UI avanzata (griglie di prodotti, riepiloghi del pagamento, schede di conferma)
Passaggi successivi
- Esplora la specifica del protocollo AP2 e crea il tuo agente abilitato ai pagamenti
- Implementa l'UCP per il tuo commerciante per attivare il commercio agentico
- Collega AP2 ad A2A per workflow di commercio multi-agente
- Scopri di più sull'estensione AP2 x402 per i pagamenti in criptovaluta