1. Visão geral
Neste codelab, você vai executar um agente do ADK que reserva ingressos de cinema em vários comerciantes de teatro usando dois protocolos de comércio de código aberto:
- UCP (Protocolo de Comércio Universal): um padrão para agentes descobrirem comerciantes, pesquisarem catálogos e gerenciarem fluxos de finalização da compra.
- AP2 (Protocolo de Pagamentos com Agentes): um protocolo para autorização de pagamento segura e verificável usando mandatos assinados criptograficamente.
O app de demonstração, CineAgent, se conecta a duas lojas de teatro simuladas com recursos diferentes (seleção de assentos, formatos especializados e formas de pagamento) e organiza todo o fluxo de reserva, desde a pesquisa até o pagamento.
O que você vai aprender
- Como a descoberta de comerciantes da UCP funciona com os perfis do
/.well-known/ucp - Como um agente do ADK usa a UCP para pesquisar catálogos e criar finalizações de compra
- Como os mandatos do AP2 (CartMandate, PaymentMandate) protegem as transações
- Como os protocolos UCP e AP2 de ponta a ponta funcionam para proteger o e-commerce inteligente
O que é necessário
- Tenha um projeto do Google Cloud com o faturamento ativado.
- Um navegador da web, como o Chrome
- Python 3.11 ou mais recente
Este codelab é destinado a desenvolvedores intermediários que têm alguma familiaridade com Python e Google Cloud. Este codelab leva aproximadamente 15 minutos para ser concluído.
Os recursos criados neste codelab custam menos de US $5.
2. Noções básicas sobre os protocolos UCP e AP2
Antes de criarmos o agente, vamos entender os dois protocolos que tornam possível esse comércio agêntico seguro.
Protocolo de Comércio Universal (UCP)
O UCP padroniza a forma como os agentes de IA interagem com os comerciantes. Ele resolve o problema de os agentes precisarem aprender APIs personalizadas para cada loja ao introduzir um modelo de recurso padronizado.
Como funciona:
- Descoberta: todo comerciante em conformidade com a UCP expõe um perfil em um local padrão:
/.well-known/ucp. Exemplo: endpoint UCP da Everlane.Quando um agente lê esse perfil, ele procura:- Recursos: os recursos principais independentes que uma empresa oferece, como pesquisa de catálogo ou finalização da compra.
- Serviços: as camadas de comunicação de nível mais baixo usadas para trocar dados. Exemplos: API REST, MCP (Protocolo de Contexto de Modelo), A2A (Protocolo Agent2Agent).
- Extensões: se um comerciante precisar de um comportamento especializado, ele poderá definir Extensões personalizadas neste perfil.
- Operações: depois de descoberto, o agente usa o endpoint de serviço fornecido para realizar operações. Neste codelab, usamos o Protocolo de Contexto de Modelo (MCP) como o transporte de serviço. O agente faz chamadas JSON-RPC 2.0 para esse endpoint e invoca os recursos descobertos: pesquisa de produtos, criação de finalizações de compra e conclusão de compras.
Protocolo de Pagamentos com Agentes (AP2)
O AP2 padroniza a forma como os pagamentos são autorizados pelos agentes em nome dos usuários. Ele resolve o problema de segurança dos agentes que lidam com credenciais de pagamento sensíveis.
Como funciona:
- Autorização do carrinho: quando um agente cria um pagamento usando o protocolo UCP, o comerciante retorna um
CartMandate. É um objeto JSON que contém os detalhes do carrinho e uma assinatura criptográfica do comerciante. Ela funciona como uma garantia de bloqueio de preço. O comerciante não pode mudar o preço depois de emitir essa autorização. - Mandato de pagamento: depois de verificar o conteúdo do carrinho, o usuário (ou o agente em nome dele) cria um
PaymentMandatepara autorizar o pagamento. EssePaymentMandatefaz referência aoCartMandatee inclui a assinatura criptográfica (ou token de autorização) do usuário. - Verificação de assinatura dupla: o comerciante recebe os dois mandatos. Eles verificam a própria assinatura no
CartMandatee a assinatura do usuário noPaymentMandate. Se ambos forem válidos, a transação será concluída.
Esse sistema de "bloqueio duplo" garante que os comerciantes não possam cobrar demais e que os agentes não possam gastar sem autorização. Em produção, esses mandatos usam o SD-JWT (Selective Disclosure JWT) para proteger a privacidade do usuário.
3. Configurar o ambiente
Configuração do projeto do Google Cloud
Criar um projeto do Google Cloud
- No console do Google Cloud, na página do seletor de projetos, selecione ou crie um projeto na nuvem do Google Cloud.
- Verifique se o faturamento está ativado para seu projeto do Cloud. Saiba como verificar se o faturamento está ativado em um projeto.
Iniciar o Cloud Shell
O Cloud Shell é um ambiente de linha de comando executado no Google Cloud que vem pré-carregado com as ferramentas necessárias.
- Clique em Ativar o Cloud Shell na parte de cima do console do Google Cloud.
- Depois de se conectar ao Cloud Shell, verifique sua autenticação:
gcloud auth list - Confirme se o projeto está configurado:
gcloud config get project - Se o projeto não estiver definido como esperado, faça o seguinte:
export PROJECT_ID=<YOUR_PROJECT_ID> gcloud config set project $PROJECT_ID
Acessar modelos do Gemini
No ambiente shell do Cloud Shell, copie e cole os comandos a seguir. Isso vai permitir o acesso aos modelos do Gemini que o Cine Agent vai usar.
export GOOGLE_CLOUD_PROJECT=$PROJECT_ID
export GOOGLE_CLOUD_LOCATION=global
export GOOGLE_GENAI_USE_VERTEXAI=True
Configurar a estrutura de diretórios
Copie e cole os comandos a seguir para criar um diretório para o agente:
mkdir -p agent_payments
cd agent_payments
Instalar dependências
O Google Cloud Shell vem pré-instalado com o uv para gerenciar o ambiente e as dependências.
- Crie um arquivo
pyproject.tomlna raiz da pastaagent_paymentse adicione o seguinte conteúdo a ele. Esse arquivo define os metadados e as dependências do projeto.
[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",
]
- Execute o comando a seguir para criar o ambiente virtual e instalar todas as dependências:
uv sync
- Ative o ambiente virtual criado por
uv:
source .venv/bin/activate
4. Definir o agente
Antes de escrever qualquer lógica de ferramenta, vamos definir o próprio agente em um arquivo chamado agent.py. Esse agente vai atuar como o orquestrador do fluxo de reserva de filmes.
Crie 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),
],
)
Explicação de código
Vamos detalhar o que está acontecendo nessa definição de agente:
model="gemini-3.1-pro-preview": estamos usando o modelo de prévia mais recente do Gemini Pro para raciocínio complexo e uso de ferramentas.instruction: é o comando que orienta o comportamento do agente. Ele informa explicitamente ao agente para usar UCP e AP2, lista as ferramentas disponíveis e define regras importantes, como "Nunca invente dados" e "Os preços são em centavos".tools: é uma lista de funções Python (que vamos criar em seguida) que o agente pode chamar com base nas solicitações do usuário.require_confirmation: é possível encapsular qualquer ferramenta comFunctionTool(my_function,require_confirmation=True). Quando acionado, o agente pausa e aguarda uma aprovação simples de "sim" ou "não" antes de executar a ferramenta. Aqui, antes de executar a ferramentacomplete_purchase, o agente pausa para uma confirmação humana.
A lista de ferramentas
A definição do agente declara o que precisamos criar. Cada ferramenta é mapeada para uma operação específica no protocolo UCP ou AP2:
Ferramenta | O que faz? | Ação de protocolo |
| Encontrar comerciantes e recursos | Consultas |
| Pesquisar catálogos em várias lojas | Endpoint JSON-RPC para MCP |
| Ver os horários de exibição em uma loja específica | Endpoint JSON-RPC para MCP |
| Iniciar uma sessão de pagamento | Endpoint JSON-RPC para MCP |
| Autorizar o pagamento e concluir o pedido | Assina o mandato da AP2 e envia para o MCP |
O modelo do Gemini vai decidir quando chamar cada ferramenta com base na conversa. Em seguida, vamos implementar o que cada ferramenta faz em tools.py.
5. Criar as ferramentas do agente: descoberta e navegação
Agora vamos implementar as ferramentas que o agente vai usar para navegar e descobrir filmes. Cada ferramenta encapsula uma operação da UCP.
Crie um arquivo chamado tools.py e copie e cole o seguinte 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()
Como entender a configuração
Para manter o foco deste codelab na criação do agente, estamos usando duas classes auxiliares, UCPClient e AP2Handler, que vamos analisar em uma etapa posterior.
- O que são?: São classes auxiliares escritas à mão que criamos para este codelab e que simulam a interação com as lojas fictícias. Como os SDKs oficiais do UCP e do AP2 ainda não estão disponíveis, estamos usando esses helpers para preencher a lacuna. Em um ambiente de produção, use os SDKs oficiais assim que eles estiverem disponíveis.
- Por enquanto, trate-os como objetos auxiliares:
_ucp.discover(url): busca o perfil de um comerciante._ucp.mcp_call(url, method, params): envia uma solicitação JSON-RPC 2.0 ao endpoint do MCP do comerciante.
Descobrir teatros
Essa ferramenta é a primeira etapa do fluxo da UCP. Ele encontra os comerciantes que existem e o que eles oferecem.
Adicione 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)
O que isso faz:
- Ele itera uma lista de URLs de comerciantes fornecidos pelo servidor. Neste codelab, vamos configurar duas lojas fictícias na seção posterior.
- Para cada URL, ele chama
_ucp.discover(url), que acessa o endpoint/.well-known/ucp. - Ele coleta o nome, os recursos e os manipuladores de pagamento em uma lista de resumo.
- Ela retorna a lista como uma string JSON para o agente ler.
Pesquisar filmes
Essa ferramenta pesquisa em todos os comerciantes descobertos e mescla os resultados. Isso é crucial porque o mesmo filme pode estar sendo exibido em vários cinemas com formatos (IMAX, Dolby) e preços diferentes.
Adicione 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)
O que isso faz:
- Ele faz um loop por todos os cinemas descobertos.
- Ele envia uma solicitação JSON-RPC de catálogo de pesquisa (
_ucp.mcp_call(url, "search_catalog", {"query": query})) para o endpoint do MCP do comerciante. - Em seguida, ele faz uma limpeza para analisar os resultados e encontrar filmes e suas "variantes" (que representam horários e formatos específicos). Agrupa os filmes por ID para que o usuário não veja entradas duplicadas.
Receber detalhes do filme
Essa ferramenta recebe os detalhes completos do catálogo de um filme específico em um cinema específico.
Adicione 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)
O que isso faz:
- Ele chama o método
lookup_catalogno endpoint do MCP do comerciante, transmitindo omovie_idespecífico. Isso retorna informações detalhadas, como horários das sessões e disponibilidade de assentos para esse cinema específico.
6. Criar as ferramentas do agente: finalização da compra e pagamento
Essas ferramentas processam o fluxo de finalização da compra e compra. É aqui que o protocolo AP2 entra em ação para garantir transações seguras.
Criar a finalização de compra
Essa ferramenta inicia uma sessão de finalização de compra em um cinema e horário específicos.
Adicione 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)
O que isso faz:
- Ele chama o método
create_checkoutno endpoint do MCP do comerciante. - Ele transmite o
showtime_ide oquantitysolicitados pelo usuário. - O comerciante retorna um objeto JSON que contém um AP2 CartMandate.
Observação: se você analisar os dados da resposta, o ap2.cart_mandate vai conter um campo merchant_authorization. É a assinatura criptográfica do comerciante que bloqueia o preço cotado. Não é possível mudar depois.
Concluir a compra
Três coisas acontecem nessa ferramenta:
- Recebemos o CartMandate da finalização da compra e o verificamos (simulado).
- Crie e assine o PaymentMandate (simulado).
- Envie a autorização assinada ao comerciante para concluir a compra.
Adicione 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 que dois mandatos? O CartMandate garante que o comerciante não possa mudar o preço depois de informá-lo. O PaymentMandate garante que o agente não possa cobrar do usuário sem consentimento. O fluxo é:
Merchant locks price -> User authorizes charge -> Merchant verifies both -> Order completes.
Checkpoint: tools.py completo
Seu tools.py completo agora tem cinco funções de ferramenta e inicialização no nível do módulo para clientes UCP e AP2. Verifique se ele se parece com isto:
"""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. Tornar executável
Com comerciantes reais da UCP, você apontaria o agente para os URLs deles e pronto. Para este codelab, precisamos de duas coisas para testar localmente:
- Comerciantes simulados: servidores locais que simulam endpoints da UCP para que você tenha algo para testar.
- Helpers de protocolo: wrappers HTTP simples para UCP e AP2. Em produção, os SDKs oficiais substituem esses wrappers.
Observação: não é necessário ler esse código com atenção. Esses arquivos simulam o que a infraestrutura e os SDKs reais oferecem. Copie-os como estão.
Ajudantes de protocolo
A UCP e a AP2 ainda não têm SDKs de cliente. Esses dois arquivos processam a encanamento HTTP.
Crie 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()
Crie 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()
Comerciantes simulados
Crie 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
Isso cria dois servidores FastAPI, cada um com dois endpoints:
GET /.well-known/ucp: descoberta de UCP. Retorna as funcionalidades do comerciante, o URL do endpoint do MCP e as formas de pagamento aceitas.POST /mcp: operações do MCP (Protocolo de Contexto de Modelo). Processa chamadas JSON-RPC 2.0 para pesquisa no catálogo, finalização da compra, descontos e pagamento.
Inicie os comerciantes em um novo terminal. Eles precisam ficar em execução:
cd agent_payments
source .venv/bin/activate
python merchants.py
Você verá:
Merchants running: Meridian (:8081), StarLight (:8082)
De volta ao primeiro terminal, verifique a descoberta da UCP:
curl -s http://localhost:8081/.well-known/ucp | python -m json.tool
Você vai encontrar as funcionalidades do comerciante, o URL do endpoint do MCP e os manipuladores de pagamento.
8. Executar o agente com o ADK Web
Vamos usar a interface da Web integrada da CLI do ADK. Isso fornece uma interface de chat no navegador e processa automaticamente os comandos de confirmação da ferramenta.
Seu projeto vai ficar assim:
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
Testar
No terminal atual, navegue até o diretório principal (uma pasta acima de agent_payments) e inicie a interface da Web do ADK:
cd ../
adk web --allow_origins '*'
Você vai ver uma saída indicando que o servidor está em execução e um URL (geralmente http://localhost:8000 ou semelhante).
Fale com o agente
- Abra o URL fornecido por
adk webno seu navegador. - Uma interface de chat vai aparecer.
- Tente perguntar: "Que filmes estão passando?"
- O agente vai descobrir cinemas e pesquisar catálogos nos bastidores, agregando resultados dos dois comerciantes pela UCP.
- Peça para reservar ingressos: "Reserve dois ingressos para Oppenheimer às 19h".
- Quando o agente tentar chamar
complete_purchase, observe como a interface da Web do ADK mostra uma caixa de diálogo de confirmação ou um card. - Para autorizar a transação, responda no chat com esta string JSON exata:
{"confirmed": true}. - O agente vai concluir a compra e enviar a confirmação do pedido com os códigos dos ingressos.
Veja o que aconteceu nos bastidores:
create_checkout→ o comerciante retornou um CartMandate do AP2 (bloqueio de preço assinado)complete_purchase→ criou um PaymentMandate, assinou (SHA-256 simulado) e enviou os dois mandatos ao comerciante.- O comerciante verificou as duas assinaturas → emitiu ingressos (na simulação)
9. Limpar
Para evitar deixar servidores locais em execução, limpe os recursos:
- No terminal que executa
adk web, pressione Ctrl+C para interromper o servidor do agente. - No terminal que executa
python merchants.py, pressione Ctrl+C para interromper os comerciantes simulados. - Desative o ambiente virtual nos dois terminais executando:
deactivate
- (Opcional) Se você criou um projeto do Google Cloud para este codelab e quer excluí-lo, execute:
gcloud projects delete $GOOGLE_CLOUD_PROJECT
10. Parabéns! 🎉
Você criou um agente do ADK que descobre comerciantes, navega por catálogos e conclui compras usando UCP e AP2.
O que você aprendeu
Neste codelab, você criou um agente do ADK que processa fluxos de comércio seguros. Confira um resumo do que você criou e dos principais conceitos aplicados:
O que você criou:
- Cinco ferramentas de agente que envolvem operações de UCP e AP2: descoberta, pesquisa no catálogo, finalização da compra, pagamento.
- Assinatura do mandato AP2: CartMandate (bloqueio de preço do comerciante) + PaymentMandate (autorização do usuário).
- Pesquisa em várias lojas: um agente consulta vários cinemas e mescla os resultados.
Principais conceitos:
Protocolo | O que faz? | Como funciona |
Descoberta de UCP | O agente encontra comerciantes e os recursos deles |
|
UCP MCP | O agente navega pelos catálogos e cria finalizações de compra | Chamadas JSON-RPC 2.0 para o endpoint do MCP do comerciante |
AP2 CartMandate | O comerciante bloqueia o preço cotado | Assinado pelo comerciante, inclui total + validade |
AP2 PaymentMandate | O usuário autoriza a cobrança | Assinado pelo usuário, faz referência a "CartMandate". |
O que muda na produção?
Este codelab usa simulações. Em produção:
- A descoberta do UCP é resolvida em um registro, não em URLs de localhost fixados no código.
- Os endpoints do MCP são hospedados por comerciantes reais. Eles usam o mesmo protocolo JSON-RPC 2.0 e inventário real.
- As determinações da AP2 são assinadas com sd-jwt-vc, não com hashes SHA-256.
- A autorização de pagamento usa um SDK da Carteira AP2 com solicitações de consentimento do usuário.
- O front-end renderiza os resultados da ferramenta como uma interface avançada (grades de produtos, resumos de finalização de compra, cards de confirmação).
Próximas etapas
- Confira a especificação do protocolo AP2 e crie seu próprio agente com pagamentos ativados
- Implemente o UCP para seu comerciante e ative o comércio agêntico
- Conecte o AP2 ao A2A para fluxos de trabalho de comércio multiagente
- Saiba mais sobre a extensão AP2 x402 para pagamentos com criptomoedas