1. Descripción general
En este codelab, ejecutarás un agente del ADK que reserva entradas de cine en varios comercios de cines con dos protocolos de comercio de código abierto:
- UCP (Universal Commerce Protocol): Es un estándar para que los agentes descubran comercios, busquen catálogos y administren flujos de confirmación de compra.
- AP2 (Agent Payments Protocol): Es un protocolo para la autorización de pagos seguros y verificables con mandatos firmados de forma criptográfica.
La app de demostración, CineAgent, se conecta a dos comercios de cine simulados con diferentes capacidades (selección de asientos, formatos especializados y formas de pago) y coordina el flujo de reserva completo, desde la búsqueda hasta el pago.
Qué aprenderás
- Cómo funciona el descubrimiento de comercios del UCP a través de los perfiles de
/.well-known/ucp - Cómo un agente de ADK usa UCP para buscar catálogos y crear pagos
- Cómo los mandatos de AP2 (CartMandate y PaymentMandate) protegen las transacciones
- Cómo funcionan los protocolos UCP y AP2 de extremo a extremo para proteger el comercio electrónico basado en agentes
Requisitos
- Un proyecto de Google Cloud con la facturación habilitada.
- Un navegador web, como Chrome
- Python 3.11 o versiones posteriores
Este codelab es para desarrolladores intermedios que tienen cierta familiaridad con Python y Google Cloud. Este codelab tarda aproximadamente 15 minutos en completarse.
Los recursos creados en este codelab deberían costar menos de USD 5.
2. Información sobre los protocolos UCP y AP2
Antes de comenzar a compilar el agente, comprendamos los dos protocolos que hacen posible este comercio seguro basado en agentes.
Universal Commerce Protocol (UCP)
El UCP estandariza la forma en que los agentes de IA interactúan con los comercios. Introduce un modelo de recursos estandarizado para resolver el problema de que los agentes tengan que aprender APIs personalizadas para cada tienda.
Cómo funciona:
- Descubrimiento: Todos los comercios que cumplen con los requisitos de la UCP exponen un perfil en una ubicación estándar:
/.well-known/ucp. Ejemplo: Extremo de UCP de Everlane.Cuando un agente lee este perfil, busca lo siguiente:- Capacidades: Son las funciones principales independientes que admite una empresa, como la búsqueda en el catálogo o la confirmación de compra.
- Servicios: Son las capas de comunicación de nivel inferior que se usan para intercambiar datos. Ejemplos: API de REST, MCP (Protocolo de contexto del modelo), A2A (Protocolo de Agent2Agent).
- Extensiones: Si un comercio necesita un comportamiento especializado, puede definir Extensiones personalizadas en este perfil.
- Operations: Una vez que se descubre, el agente usa el extremo de servicio proporcionado para realizar operaciones. En este codelab, usamos el Protocolo de contexto del modelo (MCP) como transporte de servicio. El agente realiza llamadas JSON-RPC 2.0 a este extremo para invocar las capacidades descubiertas: buscar productos, crear pagos y completar compras.
Agent Payments Protocol (AP2)
La AP2 estandariza la forma en que los agentes autorizan los pagos en nombre de los usuarios. Resuelve el problema de seguridad de los agentes que manejan credenciales de pago sensibles.
Cómo funciona:
- Mandato del carrito: Cuando un agente crea una confirmación de compra con el protocolo de UCP, el comercio devuelve un
CartMandate. Es un objeto JSON que contiene los detalles del carrito y una firma criptográfica del comercio. Actúa como una garantía de bloqueo de precios. El comercio no puede cambiar el precio después de emitir este mandato. - Mandato de pago: Después de verificar el contenido del carrito, el usuario (o el agente en nombre del usuario) crea un
PaymentMandatepara autorizar el pago. EstePaymentMandatehace referencia alCartMandatey contiene la firma criptográfica (o token de autorización) del usuario. - Verificación de doble firma: El comercio recibe ambos mandatos. Verifican su propia firma en el
CartMandatey la firma del usuario en elPaymentMandate. Si ambos son válidos, la transacción continúa.
Este sistema de "doble bloqueo" garantiza que los comercios no puedan cobrar de más y que los agentes no puedan gastar sin autorización. En producción, estos mandatos usan SD-JWT (JWT de divulgación selectiva) para proteger la privacidad del usuario.
3. Configura tu entorno
Configuración del proyecto de Google Cloud
Crea un proyecto de Google Cloud
- En la página del selector de proyectos de la consola de Google Cloud, selecciona o crea un proyecto de Google Cloud.
- Asegúrate de que la facturación esté habilitada para tu proyecto de Cloud. Obtén información para verificar si la facturación está habilitada en un proyecto.
Inicie Cloud Shell
Cloud Shell es un entorno de línea de comandos que se ejecuta en Google Cloud y que viene precargado con las herramientas necesarias.
- Haz clic en Activar Cloud Shell en la parte superior de la consola de Google Cloud.
- Una vez que te conectes a Cloud Shell, verifica tu autenticación:
gcloud auth list - Confirma que tu proyecto esté configurado:
gcloud config get project - Si tu proyecto no está configurado como se esperaba, configúralo:
export PROJECT_ID=<YOUR_PROJECT_ID> gcloud config set project $PROJECT_ID
Accede a los modelos de Gemini
En tu entorno de Cloud Shell, copia y pega los siguientes comandos. Esto habilitará el acceso a los modelos de Gemini que usará Cine Agent.
export GOOGLE_CLOUD_PROJECT=$PROJECT_ID
export GOOGLE_CLOUD_LOCATION=global
export GOOGLE_GENAI_USE_VERTEXAI=True
Configura la estructura de directorios
Copia y pega los siguientes comandos para crear un directorio nuevo para el agente:
mkdir -p agent_payments
cd agent_payments
Instala las dependencias
Google Cloud Shell viene preinstalado con uv para administrar el entorno y las dependencias.
- Crea un archivo
pyproject.tomlen la raíz de tu carpetaagent_paymentsy agrégale el siguiente contenido. Este archivo define los metadatos y las dependencias del proyecto.
[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",
]
- Ejecuta el siguiente comando para crear el entorno virtual y, luego, instalar todas las dependencias:
uv sync
- Activa el entorno virtual creado por
uv:
source .venv/bin/activate
4. Define el agente
Antes de escribir cualquier lógica de herramientas, definamos el agente en un archivo llamado agent.py. Este agente actuará como el organizador del flujo de reserva de películas.
Crea un 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),
],
)
Explicación del código
Analicemos lo que sucede en esta definición del agente:
model="gemini-3.1-pro-preview": Usamos el modelo de versión preliminar más reciente de Gemini Pro para el razonamiento complejo y el uso de herramientas.instruction: Es la instrucción que guía el comportamiento del agente. Le indica explícitamente al agente que use UCP y AP2, enumera las herramientas disponibles y establece reglas críticas, como "Nunca inventes datos" y "Los precios están en centavos".tools: Es una lista de funciones de Python (que compilaremos a continuación) que el agente puede elegir llamar según las solicitudes del usuario.require_confirmation: Puedes unir cualquier herramienta conFunctionTool(my_function,require_confirmation=True). Cuando se activa, el agente se pausa y espera una aprobación simple con un "sí" o un "no" antes de ejecutar la herramienta. Aquí, antes de ejecutar la herramientacomplete_purchase, el agente se detiene para obtener la confirmación de un humano.
La lista de herramientas
La definición del agente declara lo que necesitamos compilar. Cada herramienta se asigna a una operación específica en el protocolo de UCP o AP2:
Herramienta | Qué hace | Acción del protocolo |
| Cómo encontrar comercios y sus funciones | Consultas |
| Buscar catálogos en todos los comercios | Extremo de JSON-RPC a MCP |
| Obtener horarios de espectáculos en un comercio específico | Extremo de JSON-RPC a MCP |
| Inicia una sesión de confirmación de compra | Extremo de JSON-RPC a MCP |
| Autoriza el pago y completa el pedido | Firma el mandato de AP2 y lo envía al MCP |
El modelo de Gemini decidirá cuándo llamar a cada herramienta según la conversación. A continuación, debemos implementar lo que hace cada herramienta en tools.py.
5. Compila las herramientas del agente: Descubrimiento y navegación
Ahora implementemos las herramientas que usará el agente para explorar y descubrir películas. Cada herramienta encapsula una operación de UCP.
Crea un archivo nuevo llamado tools.py y copia y pega el siguiente código:
"""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()
Información sobre la configuración
Para que este codelab se centre en la creación del agente, usamos dos clases de ayuda, UCPClient y AP2Handler, que veremos en un paso posterior.
- ¿Qué son?: Son clases auxiliares escritas a mano que creamos para este codelab con el objetivo de simular la interacción con los comercios simulados. Como aún no están disponibles los SDKs oficiales de UCP y AP2, usamos estos asistentes para cerrar la brecha. En un entorno de producción, usarías los SDKs oficiales una vez que estén disponibles.
- Por el momento, trátalos como objetos auxiliares:
_ucp.discover(url): Recupera el perfil de un comercio._ucp.mcp_call(url, method, params): Envía una solicitud JSON-RPC 2.0 al extremo del MCP del comercio.
Descubre teatros
Esta herramienta es el primer paso en el flujo del UCP. Encuentra los comercios existentes y los que admiten la función.
Agregar 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)
Qué hace:
- Itera sobre una lista de URLs de comercios proporcionadas por el servidor. En este codelab, configuraremos dos comercios simulados en la sección posterior.
- Para cada URL, llama a
_ucp.discover(url), que alcanza el extremo/.well-known/ucp. - Recopila el nombre, las capacidades y los controladores de pagos en una lista de resumen.
- Devuelve la lista como una cadena JSON para que el agente la lea.
Buscar películas
Esta herramienta busca en todos los comercios descubiertos y combina los resultados. Esto es fundamental porque la misma película puede proyectarse en varios cines con diferentes formatos (IMAX, Dolby) y precios.
Agregar 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)
Qué hace:
- Itera por todos los cines que se descubrieron.
- Envía una solicitud JSON-RPC de catálogo de búsqueda (
_ucp.mcp_call(url, "search_catalog", {"query": query})) al extremo del MCP del comercio. - Luego, realiza una pequeña limpieza para analizar los resultados y encontrar películas y sus "variantes" (que representan horarios y formatos específicos). Agrupa las películas por ID para que el usuario no vea entradas duplicadas.
Obtener detalles de la película
Esta herramienta obtiene los detalles completos del catálogo de una película específica en un cine específico.
Agregar 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)
Qué hace:
- Llama al método
lookup_catalogen el extremo del MCP del comercio y pasa elmovie_idespecífico. Esto devuelve información detallada, como los horarios de las funciones y la disponibilidad de asientos para ese cine específico.
6. Crea las herramientas del agente: Checkout y Payment
Estas herramientas controlan el flujo de compra y confirmación de compra. Aquí es donde entra en juego el protocolo AP2 para garantizar transacciones seguras.
Crea la página de confirmación de compra
Esta herramienta inicia una sesión de confirmación de compra en un cine específico para un horario de función específico.
Agregar 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)
Qué hace:
- Llama al método
create_checkouten el extremo del MCP del comercio. - Pasa los parámetros
showtime_idyquantitysolicitados por el usuario. - El comercio devuelve un objeto JSON que contiene un AP2 CartMandate.
Nota: Si examinas los datos de la respuesta, ap2.cart_mandate contiene un campo merchant_authorization. Es la firma criptográfica del comercio que bloquea el precio cotizado. No podrán cambiarlo más adelante.
Completa la compra
En esta herramienta, suceden tres cosas:
- Obtenemos el objeto CartMandate de la confirmación de compra y lo verificamos (simulación).
- Crea y firma el PaymentMandate (simulado).
- Envía el mandato firmado al comercio para completar la compra.
Agregar 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)
¿Por qué dos mandatos? El objeto CartMandate garantiza que el comercio no pueda cambiar el precio después de cotizarlo. El objeto PaymentMandate garantiza que el agente no pueda cobrarle al usuario sin su consentimiento. El flujo es así:
Merchant locks price -> User authorizes charge -> Merchant verifies both -> Order completes.
Control: tools.py completo
Tu tools.py completo ahora debería tener 5 funciones de herramientas y una inicialización a nivel del módulo para los clientes de UCP y AP2. Verifica que se vea de la siguiente manera:
"""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. Cómo hacer que se pueda ejecutar
Con comercios reales del UCP, dirigirías al agente a sus URLs y listo. Para este codelab, necesitamos dos elementos para realizar pruebas a nivel local:
- Comercios simulados: Son servidores locales que simulan extremos de UCP para que tengas algo con qué realizar pruebas.
- Protocol helpers: Wrappers HTTP delgados para UCP y AP2 (en producción, los SDKs oficiales reemplazan estos wrappers)
Nota: No es necesario que leas este código con atención. Estos archivos simulan lo que proporcionarían la infraestructura y los SDKs reales. Cópialos tal como están.
Auxiliares de protocolos
UCP y AP2 aún no tienen SDKs de cliente. Estos dos archivos controlan la infraestructura de HTTP.
Crea un 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 un 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()
Comercios de simulación
Crea un 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
Esto crea dos servidores de FastAPI, cada uno con dos extremos:
GET /.well-known/ucp: Descubrimiento de UCP. Devuelve las capacidades del comercio, la URL del extremo del MCP y las formas de pago aceptadas.POST /mcp: Operaciones del MCP (Protocolo de contexto del modelo). Controla las llamadas JSON-RPC 2.0 para la búsqueda en el catálogo, la confirmación de compra, los descuentos y el pago.
Inicia los comercios en una terminal nueva, ya que deben seguir en ejecución:
cd agent_payments
source .venv/bin/activate
python merchants.py
Deberías ver lo siguiente:
Merchants running: Meridian (:8081), StarLight (:8082)
De vuelta en tu primera terminal, verifica el descubrimiento de UCP:
curl -s http://localhost:8081/.well-known/ucp | python -m json.tool
Deberías ver las capacidades del comercio, la URL del extremo del MCP y los controladores de pagos.
8. Ejecuta el agente con ADK Web
Usemos la IU web integrada de la CLI del ADK. Esto proporciona una interfaz de chat en el navegador y controla automáticamente las indicaciones de confirmación de herramientas.
Ahora, tu proyecto debería verse de la siguiente manera:
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
Probar
En la terminal actual, navega al directorio principal (una carpeta sobre agent_payments) y, luego, inicia la IU web del ADK:
cd ../
adk web --allow_origins '*'
Deberías ver un resultado que indique que el servidor se está ejecutando y que te proporcione una URL (por lo general, http://localhost:8000 o similar).
Hablar con el agente
- Abre la URL que proporciona
adk weben tu navegador. - Verás una interfaz de chat.
- Prueba preguntar: "¿Qué películas están en cartelera?"
- El agente descubrirá cines y buscará catálogos tras bambalinas, y agregará los resultados de ambos comercios a través de la UCP.
- Solicita reservar entradas: "Reserva 2 entradas para Oppenheimer a las 7 p.m.".
- Cuando el agente intenta llamar a
complete_purchase, observa cómo la IU web del ADK muestra un diálogo de confirmación o una tarjeta. - Para autorizar la transacción, responde en el chat con esta cadena JSON exacta:
{"confirmed": true}. - El agente completará la compra y te devolverá la confirmación del pedido con los códigos de las entradas.
Esto es lo que sucedió tras bambalinas:
create_checkout→ El comercio devolvió un CartMandate de AP2 (bloqueo de precio firmado).complete_purchase→ creó un PaymentMandate, lo firmó (SHA-256 simulado) y envió ambos mandatos al comercio.- El comercio verificó ambas firmas → se emitieron boletos (en la simulación)
9. Limpia
Para evitar dejar servidores locales en ejecución, limpia los recursos:
- En la terminal que ejecuta
adk web, presiona Ctrl + C para detener el servidor del agente. - En la terminal que ejecuta
python merchants.py, presiona Ctrl + C para detener los comercios simulados. - Ejecuta el siguiente comando para desactivar el entorno virtual en ambas terminales:
deactivate
- (Opcional) Si creaste un proyecto de Google Cloud nuevo para este codelab y deseas borrarlo, ejecuta el siguiente comando:
gcloud projects delete $GOOGLE_CLOUD_PROJECT
10. ¡Felicitaciones! 🎉
Creaste un agente de ADK que descubre comercios, explora catálogos y completa compras con UCP y AP2.
Qué aprendiste
En este codelab, compilaste un agente de ADK que controla flujos de comercio seguros. A continuación, se incluye un resumen de lo que creaste y los conceptos clave que aplicaste:
Qué compilaste:
- 5 herramientas de agentes que encapsulan las operaciones de UCP y AP2: descubrimiento, búsqueda en el catálogo, confirmación de compra y pago.
- Firma de mandato de AP2: CartMandate (bloqueo de precios del comercio) + PaymentMandate (autorización del usuario).
- Búsqueda en múltiples comercios: Un agente consulta varios cines y combina los resultados.
Conceptos clave:
Protocolo | Qué hace | Cómo funciona |
Descubrimiento de UCP | El agente encuentra comercios y sus capacidades |
|
UCP MCP | El agente explora catálogos y crea pagos | Llamadas JSON-RPC 2.0 al extremo del MCP del comercio |
AP2 CartMandate | El comercio bloquea el precio cotizado | Firmada por el comercio, incluye el total y la fecha de vencimiento |
AP2 PaymentMandate | El usuario autoriza el cargo | Firmado por el usuario y hace referencia a CartMandate |
¿Qué cambia en la producción?
En este codelab, se usan simulaciones. En producción:
- El descubrimiento de UCP se resuelve en un registro, no en URLs de localhost codificadas.
- Los extremos de MCP están alojados por comercios reales, con el mismo protocolo JSON-RPC 2.0 y el mismo inventario real.
- Las obligaciones de la AP2 se firman con sd-jwt-vc, no con hashes SHA-256.
- La autorización de pago usa un SDK de AP2 Wallet con mensajes de consentimiento del usuario
- El frontend renderiza los resultados de la herramienta como una IU enriquecida (cuadrículas de productos, resúmenes de confirmación de compra, tarjetas de confirmación).
Próximos pasos
- Explora la especificación del protocolo AP2 y crea tu propio agente habilitado para pagos
- Implementa el UCP para tu comercio y habilita el comercio basado en agentes
- Conecta AP2 con A2A para flujos de trabajo de comercio multiagente
- Obtén información sobre la extensión AP2 x402 para pagos con criptomonedas