1. Visão geral
Neste codelab, você vai executar um agente do ADK que reserva ingressos de cinema em várias lojas 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 de 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 autônomo
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 criar o agente, vamos entender os dois protocolos que tornam possível esse comércio seguro.
Protocolo de Comércio Universal (UCP)
O UCP padroniza a interação dos agentes de IA 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 nesse 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 as funcionalidades descobertas: pesquisa de produtos, criação de finalizações de compra e conclusão de compras.
Protocolo de Pagamentos de Agentes (AP2)
O AP2 padroniza como os pagamentos são autorizados por 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 de carrinho: quando um agente cria um checkout 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. - Autorização 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 vai continuar.
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 (JWT de divulgação seletiva) 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 seguintes comandos. 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 críticas 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: você pode 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 faz uma pausa para aguardar a confirmação de um humano.
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 decide 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 funciona 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, você usaria os SDKs oficiais assim que eles estivessem 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 mais adiante.
- 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 pagamentos 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 revisão dos dados 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 de exibição 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 o fluxo de compra. É aqui que o protocolo AP2 entra em ação para garantir transações seguras.
Criar finalização de compra
Essa ferramenta inicia uma sessão de finalização de compra em um cinema específico para um horário de exibição específico.
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 (simulação).
- 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: Full tools.py
Seu tools.py completo agora tem cinco funções de ferramenta e inicialização no nível do módulo para clientes da UCP e da AP2. Verifique se ele é parecido 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 pai (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 (bloqueio de preço assinado) do AP2complete_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 em 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. Este é 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 e 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