1. Übersicht
In diesem Codelab führen Sie einen ADK-Agenten aus, der Kinokarten bei mehreren Kinobetreibern über zwei Open-Source-Commerce-Protokolle bucht:
- UCP (Universal Commerce Protocol): Ein Standard für Agenten, um Händler zu finden, Kataloge zu durchsuchen und Kaufabwicklungen zu verwalten.
- AP2 (Agent Payments Protocol): Ein Protokoll für die sichere, überprüfbare Zahlungsautorisierung mit kryptografisch signierten Mandaten.
Die Demo-App CineAgent stellt eine Verbindung zu zwei fiktiven Kinohändlern mit unterschiedlichen Funktionen (Sitzplatzauswahl, spezielle Formate und Zahlungsmethoden) her und orchestriert den gesamten Buchungsprozess von der Suche bis zur Zahlung.
Lerninhalte
- So funktioniert die Händlersuche über
/.well-known/ucp-Profile in UCP - So verwendet ein ADK-Agent UCP, um Kataloge zu durchsuchen und Check-outs zu erstellen
- Wie AP2-Mandate (CartMandate, PaymentMandate) Transaktionen sichern
- So schützen die Ende-zu-Ende-Protokolle UCP und AP2 den agentischen E-Commerce
Voraussetzungen
- Ein Google Cloud-Projekt mit aktivierter Abrechnung
- Ein Webbrowser wie Chrome
- Python 3.11 und höher
Dieses Codelab richtet sich an fortgeschrittene Entwickler, die mit Python und Google Cloud vertraut sind. Dieses Codelab dauert etwa 15 Minuten.
Die in diesem Codelab erstellten Ressourcen sollten weniger als 5 $kosten.
2. UCP- und AP2-Protokolle
Bevor wir uns mit der Entwicklung des Agents befassen, wollen wir uns die beiden Protokolle ansehen, die diesen sicheren Agentenhandel ermöglichen.
Universal Commerce Protocol (UCP)
UCP standardisiert die Interaktion von KI-Agents mit Händlern. Es löst das Problem, dass Agents für jeden einzelnen Shop benutzerdefinierte APIs lernen müssen, indem ein standardisiertes Ressourcenmodell eingeführt wird.
Funktionsweise:
- Auffindbarkeit: Jeder UCP-konforme Händler stellt ein Profil an einem Standardort bereit:
/.well-known/ucp. Beispiel: UCP-Endpunkt von Everlane.Wenn ein Agent dieses Profil liest, sucht er nach:- Funktionen: Die eigenständigen Kernfunktionen, die ein Unternehmen unterstützt, z. B. die Katalogsuche oder die Kaufabwicklung.
- Dienste: Die Kommunikationsschichten auf niedrigerer Ebene, die zum Austausch von Daten verwendet werden. Beispiele: REST API, MCP (Model Context Protocol), A2A (Agent2Agent Protocol).
- Erweiterungen: Wenn ein Händler ein spezielles Verhalten benötigt, kann er in diesem Profil benutzerdefinierte Erweiterungen definieren.
- Vorgänge: Sobald der Agent den Dienstendpunkt ermittelt hat, verwendet er ihn, um Vorgänge auszuführen. In diesem Codelab verwenden wir das Model Context Protocol (MCP) als Diensttransport. Der Agent führt JSON-RPC 2.0-Aufrufe an diesen Endpunkt aus, um die erkannten Funktionen aufzurufen: Produkte suchen, Check-outs erstellen und Käufe abschließen.
Agent Payments Protocol (AP2)
AP2 standardisiert, wie Zahlungen von Kundenservicemitarbeitern im Namen von Nutzern autorisiert werden. So wird das Sicherheitsproblem gelöst, dass Kundenservicemitarbeiter vertrauliche Zahlungsanmeldedaten verarbeiten.
Funktionsweise:
- Warenkorb-Mandat: Wenn ein Agent einen Checkout mit dem UCP-Protokoll erstellt, gibt der Händler eine
CartMandatezurück. Dies ist ein JSON-Objekt mit den Warenkorbdetails und einer kryptografischen Signatur des Händlers. Sie fungiert als Garantie für den Preis. Der Händler kann den Preis nach der Ausstellung dieses Mandats nicht mehr ändern. - Zahlungsauftrag: Nachdem der Nutzer (oder der Kundenservicemitarbeiter in seinem Namen) den Inhalt des Einkaufswagens überprüft hat, erstellt er eine
PaymentMandate, um die Zahlung zu autorisieren. DiesePaymentMandateverweist auf dieCartMandateund enthält die kryptografische Signatur (oder das Autorisierungstoken) des Nutzers. - Bestätigung mit doppelter Unterschrift: Der Händler erhält beide Mandate. Sie überprüfen ihre eigene Signatur auf dem
CartMandateund die Signatur des Nutzers auf demPaymentMandate. Wenn beide gültig sind, wird die Transaktion fortgesetzt.
Dieses „Doppelschloss“-System sorgt dafür, dass Händler nicht zu viel berechnen und Kundenservicemitarbeiter nicht ohne Autorisierung Geld ausgeben können. In der Produktion wird für diese Mandate SD-JWT (Selective Disclosure JWT) verwendet, um den Datenschutz der Nutzer zu schützen.
3. Umgebung einrichten
Google Cloud-Projekt einrichten
Google Cloud-Projekt erstellen
- Wählen Sie in der Google Cloud Console auf der Seite zur Projektauswahl ein Google Cloud-Projekt aus oder erstellen Sie eines.
- Die Abrechnung für das Cloud-Projekt muss aktiviert sein. So prüfen Sie, ob die Abrechnung für ein Projekt aktiviert ist.
Cloud Shell starten
Cloud Shell ist eine Befehlszeilenumgebung, die in Google Cloud ausgeführt wird und mit den erforderlichen Tools vorinstalliert ist.
- Klicken Sie oben in der Google Cloud Console auf Cloud Shell aktivieren.
- Prüfen Sie nach der Verbindung mit Cloud Shell Ihre Authentifizierung:
gcloud auth list - Prüfen Sie, ob Ihr Projekt konfiguriert ist:
gcloud config get project - Wenn Ihr Projekt nicht wie erwartet festgelegt ist, legen Sie es fest:
export PROJECT_ID=<YOUR_PROJECT_ID> gcloud config set project $PROJECT_ID
Auf Gemini-Modelle zugreifen
Kopieren Sie die folgenden Befehle in Ihre Cloud Shell-Umgebung und fügen Sie sie dort ein. Dadurch wird der Zugriff auf die Gemini-Modelle ermöglicht, die vom Cine-Agent verwendet werden.
export GOOGLE_CLOUD_PROJECT=$PROJECT_ID
export GOOGLE_CLOUD_LOCATION=global
export GOOGLE_GENAI_USE_VERTEXAI=True
Verzeichnisstruktur einrichten
Kopieren Sie die folgenden Befehle und fügen Sie sie ein, um ein neues Verzeichnis für den Agent zu erstellen:
mkdir -p agent_payments
cd agent_payments
Abhängigkeiten installieren
In Google Cloud Shell ist uv vorinstalliert, um die Umgebung und Abhängigkeiten zu verwalten.
- Erstellen Sie im Stammverzeichnis Ihres
agent_payments-Ordners eine Datei mit dem Namenpyproject.tomlund fügen Sie ihr folgenden Inhalt hinzu. In dieser Datei werden die Projektmetadaten und ‑abhängigkeiten definiert.
[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",
]
- Führen Sie den folgenden Befehl aus, um die virtuelle Umgebung zu erstellen und alle Abhängigkeiten zu installieren:
uv sync
- Aktivieren Sie die von
uverstellte virtuelle Umgebung:
source .venv/bin/activate
4. Agent definieren
Bevor wir die Tool-Logik schreiben, definieren wir den Agenten selbst in einer Datei namens agent.py. Dieser Agent fungiert als Orchestrator für den Filmreservierungsprozess.
agent.py erstellen:
"""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),
],
)
Erläuterung zum Code:
Sehen wir uns an, was in dieser Agent-Definition passiert:
model="gemini-3.1-pro-preview": Wir verwenden das neueste Gemini Pro-Vorschaumodell für komplexe Schlussfolgerungen und die Verwendung von Tools.instruction: Diese Eingabeaufforderung steuert das Verhalten des Agenten. Es wird dem Agenten ausdrücklich mitgeteilt, dass er UCP und AP2 verwenden soll. Außerdem werden die verfügbaren Tools aufgeführt und wichtige Regeln wie „Niemals Daten erfinden“ und „Preise sind in Cent angegeben“ festgelegt.tools: Dies ist eine Liste von Python-Funktionen (die wir als Nächstes erstellen), die der Agent basierend auf den Anfragen des Nutzers aufrufen kann.require_confirmation: Sie können jedes Tool mitFunctionTool(my_function,require_confirmation=True)umschließen. Wenn der Agent ausgelöst wird, pausiert er und wartet auf eine einfache Bestätigung mit „Ja“ oder „Nein“, bevor er das Tool ausführt. Hier pausiert der KI-Agent vor der Ausführung des Toolscomplete_purchase, um eine Bestätigung durch den Nutzer einzuholen.
Die Tool-Liste
In der Agent-Definition wird deklariert, was wir erstellen müssen. Jedes Tool entspricht einem bestimmten Vorgang im UCP- oder AP2-Protokoll:
Tool | Funktion | Protokollaktion |
| Händler und ihre Funktionen finden | Abfragen |
| Kataloge verschiedener Händler durchsuchen | JSON-RPC-zu-MCP-Endpunkt |
| Vorführzeiten bei einem bestimmten Händler abrufen | JSON-RPC-zu-MCP-Endpunkt |
| Checkout-Sitzung starten | JSON-RPC-zu-MCP-Endpunkt |
| Zahlung autorisieren und Bestellung abschließen | Unterzeichnet das AP2-Mandat und sendet es an MCP |
Das Gemini-Modell entscheidet anhand des Gesprächs, wann die einzelnen Tools aufgerufen werden sollen. Als Nächstes müssen wir implementieren, was die einzelnen Tools in tools.py tun.
5. Agent-Tools erstellen: Discovery und Browsing
Implementieren wir nun die Tools, mit denen der Agent Filme durchsuchen und finden kann. Jedes Tool umschließt einen UCP-Vorgang.
Erstellen Sie eine neue Datei mit dem Namen tools.py und kopieren Sie den folgenden Code:
"""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()
Einrichtung
Damit sich dieses Codelab auf die Erstellung des Agenten konzentriert, verwenden wir zwei Hilfsklassen, UCPClient und AP2Handler, die wir uns in einem späteren Schritt ansehen.
- Was ist das?: Es handelt sich um handgeschriebene Hilfsklassen, die wir für dieses Codelab erstellt haben, um die Interaktion mit den Mock-Händlern zu simulieren. Da offizielle UCP- und AP2-SDKs noch nicht verfügbar sind, verwenden wir diese Hilfsprogramme, um die Lücke zu schließen. In einer Produktionsumgebung würden Sie die offiziellen SDKs verwenden, sobald sie verfügbar sind.
- Behandeln Sie sie vorerst als Hilfsobjekte:
_ucp.discover(url): Ruft das Profil eines Händlers ab._ucp.mcp_call(url, method, params): Sendet eine JSON-RPC 2.0-Anfrage an den MCP-Endpunkt des Händlers.
Theater entdecken
Dieses Tool ist der erste Schritt im UCP-Ablauf. Es wird ermittelt, welche Händler es gibt und was sie unterstützen.
Zu tools.py hinzufügen:
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)
Funktionsweise:
- Es wird eine vom Server bereitgestellte Liste von Händler-URLs durchlaufen. In diesem Codelab richten wir im späteren Abschnitt zwei Test-Händler ein.
- Für jede URL wird
_ucp.discover(url)aufgerufen, wodurch der/.well-known/ucp-Endpunkt erreicht wird. - Darin werden Name, Funktionen und Zahlungshandler in einer Zusammenfassungsliste erfasst.
- Die Liste wird als JSON-String zurückgegeben, damit der Agent sie lesen kann.
Filme suchen
Mit diesem Tool wird bei allen erkannten Händlern gesucht und die Ergebnisse werden zusammengeführt. Das ist wichtig, da derselbe Film möglicherweise in mehreren Kinos mit unterschiedlichen Formaten (IMAX, Dolby) und unterschiedlichen Preisen läuft.
Zu tools.py hinzufügen:
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)
Funktionsweise:
- Es wird eine Schleife durch alle gefundenen Kinos durchlaufen.
- Es wird eine JSON-RPC-Anfrage für den Suchkatalog (
_ucp.mcp_call(url, "search_catalog", {"query": query})) an den MCP-Endpunkt des Händlers gesendet. - Anschließend werden die Ergebnisse bereinigt, um Filme und ihre „Varianten“ (die bestimmte Vorstellungszeiten und Formate darstellen) zu finden. Gruppiert die Filme nach ID, damit der Nutzer keine doppelten Filmeinträge sieht.
Filmdetails abrufen
Mit diesem Tool werden die vollständigen Katalogdetails für einen bestimmten Film in einem bestimmten Kino abgerufen.
Zu tools.py hinzufügen:
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)
Funktionsweise:
- Dabei wird die Methode
lookup_catalogam MCP-Endpunkt des Händlers aufgerufen und die spezifischemovie_idübergeben. Daraufhin werden detaillierte Informationen wie Spielzeiten und verfügbare Plätze für das jeweilige Kino angezeigt.
6. Agent-Tools erstellen: Checkout und Zahlung
Diese Tools übernehmen den Checkout- und Kaufvorgang. Hier kommt das AP2-Protokoll ins Spiel, um sichere Transaktionen zu gewährleisten.
Direktkauf erstellen
Mit diesem Tool wird eine Abrechnungssitzung in einem bestimmten Kino für eine bestimmte Vorstellung gestartet.
Zu tools.py hinzufügen:
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)
Funktionsweise:
- Dabei wird die Methode
create_checkoutam MCP-Endpunkt des Händlers aufgerufen. - Sie übergibt die vom Nutzer angeforderten
showtime_idundquantity. - Der Händler gibt ein JSON-Objekt mit einem AP2 CartMandate zurück.
Hinweis: Wenn Sie die Antwortdaten untersuchen, enthält ap2.cart_mandate das Feld merchant_authorization. Dies ist die kryptografische Signatur des Händlers, mit der der angegebene Preis gesichert wird. Sie können sie später nicht mehr ändern.
Einkauf abschließen
In diesem Tool passieren drei Dinge:
- Wir erhalten das CartMandate von der Kasse und bestätigen es (mock).
- Erstellen und unterzeichnen Sie das PaymentMandate (Mock).
- Senden Sie das unterschriebene Mandat an den Händler, um den Kauf abzuschließen.
Zu tools.py hinzufügen:
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)
Warum zwei Mandate? Das CartMandate sorgt dafür, dass der Händler den Preis nach der Angabe nicht mehr ändern kann. Das Zahlungsauftragsmandat sorgt dafür, dass der Kundenservicemitarbeiter den Nutzer nicht ohne Einwilligung belasten kann. Der Ablauf ist:
Merchant locks price -> User authorizes charge -> Merchant verifies both -> Order completes.
Checkpoint: Vollständige tools.py
Ihr vollständiges tools.py sollte jetzt fünf Tool-Funktionen und eine Initialisierung auf Modulebene für UCP- und AP2-Clients haben. Prüfen Sie, ob es so aussieht:
"""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. Ausführbar machen
Bei echten UCP-Händlern würden Sie den Agenten auf ihre URLs verweisen und das war es dann auch schon. Für dieses Codelab benötigen wir zwei Dinge, um lokal zu testen:
- Simulierte Händler: Lokale Server, die UCP-Endpunkte simulieren, damit Sie etwas zum Testen haben
- Protokoll-Helfer: schlanke HTTP-Wrapper für UCP und AP2 (in der Produktion werden diese durch offizielle SDKs ersetzt)
Hinweis: Sie müssen diesen Code nicht sorgfältig lesen. Diese Dateien simulieren die Funktionen, die echte Infrastruktur und SDKs bieten würden. Kopieren Sie sie unverändert.
Protokoll-Helper
Für UCP und AP2 gibt es noch keine Client-SDKs. Diese beiden Dateien kümmern sich um die HTTP-Verbindung.
ucp.py erstellen:
"""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()
ap2.py erstellen:
"""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()
Mock-Händler
merchants.py erstellen:
"""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
Dadurch werden zwei FastAPI-Server mit jeweils zwei Endpunkten erstellt:
GET /.well-known/ucp– UCP-Erkennung Gibt die Funktionen des Händlers, die MCP-Endpunkt-URL und die akzeptierten Zahlungsmethoden zurück.POST /mcp: MCP-Vorgänge (Model Context Protocol). Verarbeitet JSON-RPC 2.0-Aufrufe für die Katalogsuche, den Check-out, Rabatte und die Zahlung.
Starten Sie die Händler in einem neuen Terminal. Sie müssen weiter ausgeführt werden:
cd agent_payments
source .venv/bin/activate
python merchants.py
Hier sollten Sie dies sehen:
Merchants running: Meridian (:8081), StarLight (:8082)
Kehren Sie zu Ihrem ersten Terminal zurück und prüfen Sie die UCP-Erkennung:
curl -s http://localhost:8081/.well-known/ucp | python -m json.tool
Sie sollten die Funktionen des Händlers, die MCP-Endpunkt-URL und die Zahlungshandler sehen.
8. Agenten mit ADK Web ausführen
Verwenden wir die integrierte Web-UI der ADK CLI. Dadurch wird eine Chatoberfläche im Browser bereitgestellt und Tool-Bestätigungsaufforderungen werden automatisch verarbeitet.
Ihr Projekt sollte nun so aussehen:
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
Jetzt ausprobieren
Wechseln Sie im aktuellen Terminal zum übergeordneten Verzeichnis (ein Ordner über agent_payments) und starten Sie die ADK-Web-UI:
cd ../
adk web --allow_origins '*'
Die Ausgabe sollte darauf hinweisen, dass der Server ausgeführt wird, und eine URL (normalerweise http://localhost:8000 oder ähnlich) enthalten.
Mit dem Kundenservicemitarbeiter sprechen
- Öffnen Sie die von
adk webbereitgestellte URL in Ihrem Browser. - Es wird eine Chat-Oberfläche angezeigt.
- Fragen Sie zum Beispiel: „Welche Filme laufen im Kino?“
- Der Agent sucht im Hintergrund nach Kinos und Katalogen und fasst die Ergebnisse beider Händler über UCP zusammen.
- Tickets buchen: „Buche 2 Tickets für Oppenheimer um 19:00 Uhr.“
- Wenn der Agent versucht,
complete_purchaseaufzurufen, wird in der ADK-Web-UI ein Bestätigungsdialogfeld oder eine Karte angezeigt. - Um die Transaktion zu autorisieren, antworten Sie im Chat mit diesem genauen JSON-String:
{"confirmed": true}. - Der Kundenservicemitarbeiter schließt den Kauf ab und sendet Ihnen die Bestellbestätigung mit den Ticketcodes zurück.
So lief es hinter den Kulissen ab:
create_checkout→ Der Händler hat ein AP2-CartMandate (signierte Preisgarantie) zurückgegeben.complete_purchase→ hat ein PaymentMandate erstellt, es signiert (Mock-SHA-256) und beide Mandate an den Händler gesendet- Der Händler hat beide Signaturen bestätigt → Tickets wurden ausgestellt (im Mock)
9. Bereinigen
So vermeiden Sie, dass lokale Server weiter ausgeführt werden:
- Drücken Sie im Terminal, in dem
adk webausgeführt wird, Strg + C, um den Agentenserver zu beenden. - Drücken Sie im Terminal, in dem
python merchants.pyausgeführt wird, Strg+C, um die Test-Händler zu beenden. - Deaktivieren Sie die virtuelle Umgebung in beiden Terminals mit dem folgenden Befehl:
deactivate
- (Optional) Wenn Sie für dieses Codelab ein neues Google Cloud-Projekt erstellt haben und es löschen möchten, führen Sie Folgendes aus:
gcloud projects delete $GOOGLE_CLOUD_PROJECT
10. Glückwunsch! 🎉
Sie haben einen ADK-Agenten erstellt, der Händler findet, Kataloge durchsucht und Käufe mit UCP und AP2 abschließt.
Das haben Sie gelernt
In diesem Codelab haben Sie einen ADK-Agenten erstellt, der sichere Commerce-Abläufe verarbeitet. Hier eine Zusammenfassung der erstellten Inhalte und der angewendeten Schlüsselkonzepte:
Was Sie erstellt haben:
- 5 Agent-Tools, die UCP- und AP2-Vorgänge umfassen: Erkennung, Katalogsuche, Direktkauf und Zahlung.
- AP2-Mandatsunterzeichnung: CartMandate (Händlerpreisbindung) + PaymentMandate (Nutzerautorisierung).
- Suche bei mehreren Händlern: Ein Agent fragt mehrere Kinos ab und führt die Ergebnisse zusammen.
Wichtige Konzepte:
Protokoll | Funktion | Funktionsweise |
UCP Discovery | Der Agent findet Händler und ihre Funktionen. |
|
UCP MCP | Der Agent durchsucht Kataloge und erstellt Check-outs. | JSON-RPC 2.0-Aufrufe an den MCP-Endpunkt des Händlers |
AP2 CartMandate | Der Händler fixiert den angegebenen Preis. | Vom Händler unterzeichnet, enthält Gesamtbetrag und Ablaufdatum |
AP2 PaymentMandate | Nutzer autorisiert die Belastung | Vom Nutzer unterzeichnet, verweist auf CartMandate |
Was ist bei der Produktion anders?
In diesem Codelab werden Mocks verwendet. In Produktion:
- Die UCP-Erkennung wird anhand einer Registry aufgelöst, nicht anhand von hartcodierten localhost-URLs.
- MCP-Endpunkte werden von echten Händlern gehostet – dasselbe JSON-RPC 2.0-Protokoll, echtes Inventar
- AP2-Mandate werden mit sd-jwt-vc signiert, nicht mit SHA-256-Hashes.
- Zahlungsautorisierung verwendet ein AP2 Wallet SDK mit Nutzereinwilligungsaufforderungen
- Im Frontend werden Tool-Ergebnisse als Rich UI gerendert (Produktübersichten, Zusammenfassungen an der Kasse, Bestätigungskarten).
Nächste Schritte
- AP2-Protokollspezifikation ansehen und eigenen Agenten mit Zahlungsfunktion erstellen
- UCP für Ihren Händler implementieren, um den agentischen Handel zu ermöglichen
- AP2 mit A2A für Multi-Agent-Commerce-Workflows verbinden
- Weitere Informationen zur AP2 x402-Erweiterung für Kryptowährungszahlungen