1. Przegląd
W tym ćwiczeniu uruchomisz agenta ADK, który rezerwuje bilety do kina u różnych sprzedawców, korzystając z 2 protokołów handlowych typu open source:
- UCP (Universal Commerce Protocol): standard umożliwiający agentom odkrywanie sprzedawców, przeszukiwanie katalogów i zarządzanie procesami płatności.
- AP2 (Agent Payments Protocol): protokół bezpiecznej, weryfikowalnej autoryzacji płatności za pomocą podpisanych kryptograficznie upoważnień.
Aplikacja w wersji demonstracyjnej CineAgent łączy się z 2 fikcyjnymi sprzedawcami biletów do kina o różnych możliwościach (wybór miejsc, specjalne formaty i formy płatności) i koordynuje cały proces rezerwacji od wyszukiwania po płatność.
Czego się nauczysz
- Jak działa odkrywanie sprzedawców UCP za pomocą profili
/.well-known/ucp - Jak agent ADK używa UCP do wyszukiwania katalogów i tworzenia płatności
- Jak wymagania AP2 (CartMandate, PaymentMandate) zabezpieczają transakcje
- Jak protokoły UCP i AP2 działają w celu zabezpieczenia handlu elektronicznego opartego na agentach
Czego potrzebujesz
- projekt Google Cloud z włączonymi płatnościami;
- przeglądarka, np. Chrome;
- Python 3.11 lub nowszy
To ćwiczenie jest przeznaczone dla średnio zaawansowanych programistów, którzy znają już Pythona i Google Cloud. Wykonanie tego laboratorium zajmie około 15 minut.
Zasoby utworzone w tym laboratorium powinny kosztować mniej niż 5 USD.
2. Omówienie protokołów UCP i AP2
Zanim przejdziemy do tworzenia agenta, poznajmy 2 protokoły, które umożliwiają bezpieczny handel oparty na agentach.
Protokół Universal Commerce Protocol (UCP)
UCP ujednolica sposób, w jaki agenci AI wchodzą w interakcje ze sprzedawcami. Rozwiązuje problem związany z koniecznością uczenia się przez agentów niestandardowych interfejsów API dla każdego sklepu, wprowadzając standardowy model zasobów.
Jak to działa:
- Wykrywanie: każdy sprzedawca zgodny z UCP udostępnia profil w standardowej lokalizacji:
/.well-known/ucp. Przykład: punkt końcowy UCP Everlane.Gdy agent odczytuje ten profil, szuka:- Funkcje: samodzielne podstawowe funkcje obsługiwane przez firmę, takie jak wyszukiwanie w katalogu lub płatność.
- Usługi: niższe warstwy komunikacji używane do wymiany danych. Przykłady: REST API, MCP (Model Context Protocol), A2A (Agent2Agent Protocol).
- Rozszerzenia: jeśli sprzedawca potrzebuje specjalistycznych funkcji, może zdefiniować w tym profilu niestandardowe rozszerzenia.
- Operacje: po wykryciu agent używa podanego punktu końcowego usługi do wykonywania operacji. W tym ćwiczeniu używamy protokołu Model Context Protocol (MCP) jako transportu usługi. Agent wykonuje wywołania JSON-RPC 2.0 do tego punktu końcowego, aby wywołać wykryte funkcje: wyszukiwanie produktów, tworzenie płatności i dokonywanie zakupów.
Protokół płatności dla agentów (AP2)
AP2 ujednolica sposób autoryzacji płatności przez agentów w imieniu użytkowników. Rozwiązuje problem z bezpieczeństwem związany z obsługą poufnych danych logowania do płatności przez pracowników.
Jak to działa:
- Cart Mandate: gdy agent tworzy płatność za pomocą protokołu UCP, sprzedawca zwraca wartość
CartMandate. Jest to obiekt JSON zawierający szczegóły koszyka i podpis kryptograficzny sprzedawcy. Działa jako gwarancja ceny. Po wydaniu tego polecenia zapłaty sprzedawca nie może zmienić ceny. - Upoważnienie do płatności: po sprawdzeniu zawartości koszyka użytkownik (lub agent w jego imieniu) tworzy
PaymentMandate, aby autoryzować płatność. TenPaymentMandateodwołuje się doCartMandatei zawiera podpis kryptograficzny użytkownika (lub token autoryzacji). - Weryfikacja podwójnego podpisu: sprzedawca otrzymuje oba mandaty. Weryfikują swój podpis na dokumencie
CartMandatei podpis użytkownika na dokumenciePaymentMandate. Jeśli oba są prawidłowe, transakcja zostanie zrealizowana.
Ten system „podwójnego zabezpieczenia” gwarantuje, że sprzedawcy nie mogą naliczać zbyt wysokich opłat, a agenci nie mogą wydawać środków bez autoryzacji. W środowisku produkcyjnym te wymagania są realizowane za pomocą SD-JWT (Selective Disclosure JWT), aby chronić prywatność użytkowników.
3. Konfigurowanie środowiska
Konfiguracja projektu Google Cloud
Tworzenie projektu Google Cloud
- W konsoli Google Cloud na stronie selektora projektu wybierz lub utwórz projekt w chmurze Google.
- Sprawdź, czy w projekcie Cloud włączone są płatności. Dowiedz się, jak sprawdzić, czy w projekcie są włączone płatności.
Uruchamianie Cloud Shell
Cloud Shell to środowisko wiersza poleceń działające w Google Cloud, które zawiera niezbędne narzędzia.
- Kliknij Aktywuj Cloud Shell u góry konsoli Google Cloud.
- Po połączeniu z Cloud Shell sprawdź uwierzytelnianie:
gcloud auth list - Sprawdź, czy projekt jest skonfigurowany:
gcloud config get project - Jeśli projekt nie jest ustawiony zgodnie z oczekiwaniami, ustaw go:
export PROJECT_ID=<YOUR_PROJECT_ID> gcloud config set project $PROJECT_ID
Dostęp do modeli Gemini
W środowisku Cloud Shell skopiuj i wklej te polecenia. Umożliwi to dostęp do modeli Gemini, z których będzie korzystać agent Cine.
export GOOGLE_CLOUD_PROJECT=$PROJECT_ID
export GOOGLE_CLOUD_LOCATION=global
export GOOGLE_GENAI_USE_VERTEXAI=True
Konfigurowanie struktury katalogów
Skopiuj i wklej te polecenia, aby utworzyć nowy katalog agenta:
mkdir -p agent_payments
cd agent_payments
Instalowanie zależności
Google Cloud Shell ma wstępnie zainstalowany uv do zarządzania środowiskiem i zależnościami.
- Utwórz plik
pyproject.tomlw katalogu głównym folderuagent_paymentsi dodaj do niego tę treść. Ten plik określa metadane i zależności projektu.
[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",
]
- Aby utworzyć środowisko wirtualne i zainstalować wszystkie zależności, uruchom to polecenie:
uv sync
- Aktywuj środowisko wirtualne utworzone przez
uv:
source .venv/bin/activate
4. Określanie agenta
Zanim napiszemy logikę narzędzia, zdefiniujmy samego agenta w pliku o nazwie agent.py. Ten agent będzie pełnił rolę aranżera procesu rezerwacji biletów do kina.
Utwórz 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),
],
)
Wyjaśnienie kodu
Przyjrzyjmy się, co się dzieje w tej definicji agenta:
model="gemini-3.1-pro-preview": do złożonego rozumowania i korzystania z narzędzi używamy najnowszego modelu Gemini Pro w wersji testowej.instruction: to prompt, który określa zachowanie agenta. Wyraźnie informuje agenta o konieczności używania UCP i AP2, zawiera listę dostępnych narzędzi i określa kluczowe reguły, takie jak „Nigdy nie wymyślaj danych” i „Ceny są podawane w centach”.tools: to lista funkcji Pythona (które utworzymy w następnym kroku), które agent może wywołać na podstawie żądań użytkownika.require_confirmation: możesz opakować dowolne narzędzie za pomocąFunctionTool(my_function,require_confirmation=True). Gdy zostanie wywołane, narzędzie wstrzymuje działanie i czeka na proste potwierdzenie „tak” lub „nie”, zanim zostanie wykonane. W tym przypadku przed wykonaniem narzędziacomplete_purchaseagent wstrzymuje działanie, aby uzyskać potwierdzenie od człowieka.
Lista narzędzi
Definicja agenta określa, co musimy utworzyć. Każde narzędzie jest powiązane z określoną operacją w protokole UCP lub AP2:
Narzędzie | Działanie | Działanie protokołu |
| Znajdowanie sprzedawców i ich rozwiązań | Zapytania |
| Przeszukiwanie katalogów różnych sprzedawców | Punkt końcowy JSON-RPC do MCP |
| Sprawdzanie godzin seansów u konkretnego sprzedawcy | Punkt końcowy JSON-RPC do MCP |
| Rozpoczęcie sesji płatności | Punkt końcowy JSON-RPC do MCP |
| Autoryzuj płatność i dokończ zamówienie | Podpisuje mandat AP2 i wysyła go do MCP |
Model Gemini zdecyduje, kiedy wywołać poszczególne narzędzia na podstawie rozmowy. Następnie musimy zaimplementować w tools.py działanie każdego narzędzia.
5. Tworzenie narzędzi agenta: odkrywanie i przeglądanie
Teraz wdróżmy narzędzia, których agent będzie używać do przeglądania i odkrywania filmów. Każde narzędzie obejmuje operację UCP.
Utwórz nowy plik o nazwie tools.py i wklej do niego ten kod:
"""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()
Opis konfiguracji
Aby skupić się na tworzeniu agenta, używamy 2 klas pomocniczych: UCPClient i AP2Handler. Przyjrzymy się im w dalszej części tego ćwiczenia.
- Co to jest: Są to napisane ręcznie klasy pomocnicze, które stworzyliśmy na potrzeby tego ćwiczenia, aby symulować interakcję z przykładowymi sprzedawcami. Oficjalne pakiety SDK UCP i AP2 nie są jeszcze dostępne, dlatego używamy tych narzędzi, aby wypełnić tę lukę. W środowisku produkcyjnym należy używać oficjalnych pakietów SDK, gdy staną się dostępne.
- Na razie traktuj je jako obiekty pomocnicze:
_ucp.discover(url): pobiera profil sprzedawcy._ucp.mcp_call(url, method, params): wysyła żądanie JSON-RPC 2.0 do punktu końcowego MCP sprzedawcy.
Odkrywanie kin
To narzędzie jest pierwszym krokiem w procesie UCP. Określa, którzy sprzedawcy są dostępni i co obsługują.
Dodaj do: 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)
Co to robi:
- Iteruje po liście adresów URL sprzedawców podanych przez serwer. W dalszej części tego laboratorium skonfigurujemy 2 przykładowych sprzedawców.
- W przypadku każdego adresu URL wywołuje funkcję
_ucp.discover(url), która wysyła żądanie do punktu końcowego/.well-known/ucp. - Zbiera nazwę, możliwości i programy obsługi płatności na liście podsumowującej.
- Zwraca listę w postaci ciągu JSON, który może odczytać agent.
Szukaj filmów
To narzędzie przeszukuje wszystkich wykrytych sprzedawców i łączy wyniki. Jest to kluczowe, ponieważ ten sam film może być wyświetlany w wielu kinach w różnych formatach (IMAX, Dolby) i w różnych cenach.
Dodaj do: 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)
Co to robi:
- Przechodzi przez wszystkie wykryte kina.
- Wysyła żądanie JSON-RPC wyszukiwania katalogu (
_ucp.mcp_call(url, "search_catalog", {"query": query})) do punktu końcowego MCP sprzedawcy. - Następnie porządkuje wyniki, aby znaleźć filmy i ich „wersje” (czyli konkretne godziny seansów i formaty). Grupuje filmy według identyfikatora, aby użytkownik nie widział zduplikowanych pozycji.
Pobieranie szczegółów filmu
To narzędzie pobiera pełne szczegóły katalogu dotyczące konkretnego filmu w konkretnym kinie.
Dodaj do: 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)
Co to robi:
- Wywołuje metodę
lookup_catalogw punkcie końcowym MCP sprzedawcy, przekazując konkretnymovie_id. Wyświetlą się szczegółowe informacje, takie jak godziny seansów i dostępność miejsc w danym kinie.
6. Tworzenie narzędzi agenta: płatność
Te narzędzia obsługują proces płatności i zakupu. W tym miejscu protokół AP2 zapewnia bezpieczeństwo transakcji.
Tworzenie procesu płatności
To narzędzie rozpoczyna sesję płatności w konkretnym kinie o określonej godzinie seansu.
Dodaj do: 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)
Co to robi:
- Wywołuje metodę
create_checkoutw punkcie końcowym MCP sprzedawcy. - Przekazuje wartości
showtime_idiquantity, o które prosi użytkownik. - Sprzedawca zwraca obiekt JSON zawierający AP2 CartMandate.
Uwaga: jeśli sprawdzisz dane odpowiedzi, zobaczysz, że ap2.cart_mandate zawiera pole merchant_authorization. Jest to podpis kryptograficzny sprzedawcy blokujący podaną cenę. Nie mogą go później zmienić.
Dokończ zakup
W tym narzędziu dzieją się 3 rzeczy:
- Pobieramy CartMandate z płatności i weryfikujemy je (mock).
- Utwórz i podpisz PaymentMandate (mock).
- Wyślij podpisane upoważnienie do sprzedawcy, aby dokończyć zakup.
Dodaj do: 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)
Dlaczego potrzebne są 2 mandaty? Zasada CartMandate zapewnia, że sprzedawca nie może zmienić ceny po jej podaniu. Dokument PaymentMandate zapewnia, że agent nie może obciążyć użytkownika bez jego zgody. Proces wygląda tak:
Merchant locks price -> User authorizes charge -> Merchant verifies both -> Order completes.
Punkt kontrolny: pełny tools.py
Twój kompletny tools.py powinien teraz zawierać 5 funkcji narzędziowych i inicjowanie na poziomie modułu dla klientów UCP i AP2. Sprawdź, czy wygląda tak:
"""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. Uruchamianie kodu
W przypadku prawdziwych sprzedawców korzystających z UCP wystarczy wskazać agentowi ich adresy URL. Aby przeprowadzić test lokalny, potrzebujemy 2 rzeczy:
- Sprzedawcy testowi – serwery lokalne, które symulują punkty końcowe UCP, dzięki czemu możesz przeprowadzać testy.
- Pomocnicze protokoły – cienkie otoczki HTTP dla UCP i AP2 (w wersji produkcyjnej zastępują je oficjalne pakiety SDK).
Uwaga: nie musisz uważnie czytać tego kodu. Te pliki symulują to, co zapewniają prawdziwa infrastruktura i zestawy SDK. Skopiuj je w niezmienionej formie.
Pomocnicy protokołu
UCP i AP2 nie mają jeszcze pakietów SDK klienta – te 2 pliki obsługują infrastrukturę HTTP.
Utwórz 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()
Utwórz 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()
Sprzedawcy testowi
Utwórz 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
Spowoduje to utworzenie 2 serwerów FastAPI, z których każdy będzie miał 2 punkty końcowe:
GET /.well-known/ucp– UCP discovery. Zwraca możliwości sprzedawcy, adres URL punktu końcowego MCP i akceptowane formy płatności.POST /mcp– operacje MCP (Model Context Protocol). Obsługuje wywołania JSON-RPC 2.0 dotyczące wyszukiwania katalogu, realizacji transakcji, rabatów i płatności.
Uruchom sprzedawców w nowym terminalu – muszą oni pozostać aktywni:
cd agent_payments
source .venv/bin/activate
python merchants.py
Zobaczysz, że:
Merchants running: Meridian (:8081), StarLight (:8082)
Wróć do pierwszego terminala i sprawdź wykrywanie UCP:
curl -s http://localhost:8081/.well-known/ucp | python -m json.tool
Powinny być widoczne możliwości sprzedawcy, adres URL punktu końcowego MCP i programy obsługi płatności.
8. Uruchamianie agenta za pomocą ADK Web
Skorzystajmy z wbudowanego interfejsu internetowego interfejsu wiersza poleceń ADK. Zapewnia interfejs czatu w przeglądarce i automatycznie obsługuje prośby o potwierdzenie narzędzia.
Projekt powinien teraz wyglądać tak:
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
Wypróbuj
W bieżącym terminalu przejdź do katalogu nadrzędnego (folder wyżej niż agent_payments) i uruchom interfejs ADK Web UI:
cd ../
adk web --allow_origins '*'
Powinny pojawić się dane wyjściowe wskazujące, że serwer działa, oraz adres URL (zwykle http://localhost:8000 lub podobny).
Porozmawiaj z pracownikiem obsługi klienta
- Otwórz w przeglądarce adres URL podany przez
adk web. - Zobaczysz interfejs czatu.
- Spróbuj zapytać: „Jakie filmy są teraz wyświetlane?”
- Agent odkryje kina i wyszuka katalogi za kulisami, agregując wyniki od obu sprzedawców za pomocą UCP.
- Poproś o zarezerwowanie biletów: „Zarezerwuj 2 bilety na film Oppenheimer na godzinę 19:00”.
- Gdy agent spróbuje zadzwonić pod numer
complete_purchase, zobaczysz, jak w interfejsie internetowym ADK pojawi się okno dialogowe z potwierdzeniem lub karta. - Aby autoryzować transakcję, odpowiedz na czacie, podając ten dokładny ciąg znaków JSON:
{"confirmed": true}. - Przedstawiciel sfinalizuje zakup i prześle Ci potwierdzenie zamówienia z kodami biletów.
Oto, co się działo za kulisami:
create_checkout→ sprzedawca zwrócił AP2 CartMandate (podpisana blokada ceny)complete_purchase→ utworzył PaymentMandate, podpisał go (mock SHA-256) i wysłał oba mandaty do sprzedawcy.- Sprzedawca zweryfikował oba podpisy → wystawiono bilety (w wersji demonstracyjnej)
9. Czyszczenie danych
Aby uniknąć pozostawienia uruchomionych serwerów lokalnych, zwolnij miejsce na zasoby:
- W terminalu, w którym działa
adk web, naciśnij Ctrl+C, aby zatrzymać serwer agenta. - W terminalu, w którym działa
python merchants.py, naciśnij Ctrl+C, aby zatrzymać fikcyjnych sprzedawców. - Dezaktywuj środowisko wirtualne w obu terminalach, uruchamiając polecenie:
deactivate
- (Opcjonalnie) Jeśli na potrzeby tego ćwiczenia w Codelabs został przez Ciebie utworzony nowy projekt Google Cloud, który chcesz usunąć, uruchom to polecenie:
gcloud projects delete $GOOGLE_CLOUD_PROJECT
10. Gratulacje! 🎉
Utworzono agenta ADK, który wykrywa sprzedawców, przegląda katalogi i dokonuje zakupów za pomocą UCP i AP2.
Czego się dowiedziałeś
W tym ćwiczeniu utworzyliśmy agenta ADK, który obsługuje bezpieczne przepływy handlowe. Oto podsumowanie tego, co udało Ci się stworzyć, i najważniejszych zastosowanych przez Ciebie pojęć:
Co udało Ci się stworzyć:
- 5 narzędzi agentowych obejmujących operacje UCP i AP2: odkrywanie, przeszukiwanie katalogu, płatność.
- Podpisanie upoważnienia AP2 – CartMandate (blokada ceny sprzedawcy) + PaymentMandate (autoryzacja użytkownika).
- Wyszukiwanie u wielu sprzedawców – jeden agent wysyła zapytania do wielu kin i łączy wyniki.
Kluczowe pojęcia:
Protokół | Działanie | Jak to działa |
UCP Discovery | Agent znajduje sprzedawców i ich możliwości |
|
UCP MCP | Agent przegląda katalogi i tworzy płatności | Wywołania JSON-RPC 2.0 do punktu końcowego MCP sprzedawcy |
AP2 CartMandate | Sprzedawca blokuje podaną cenę | Podpisany przez sprzedawcę, zawiera łączną kwotę i datę ważności |
AP2 PaymentMandate | Użytkownik autoryzuje obciążenie | Podpisany przez użytkownika, odwołuje się do CartMandate |
Co się zmieniło w produkcji?
W tym laboratorium używane są makiety. W wersji produkcyjnej:
- Wykrywanie UCP odbywa się na podstawie rejestru, a nie adresów URL localhost zakodowanych na stałe.
- Punkty końcowe MCP są hostowane przez prawdziwych sprzedawców – ten sam protokół JSON-RPC 2.0, prawdziwe zasoby reklamowe.
- Dokumenty AP2 są podpisywane za pomocą sd-jwt-vc, a nie funkcji skrótu SHA-256.
- Autoryzacja płatności korzysta z pakietu SDK portfela AP2 z prośbami o zgodę użytkownika.
- Frontend renderuje wyniki narzędzia jako bogaty interfejs użytkownika (siatki produktów, podsumowania płatności, karty potwierdzenia).
Dalsze kroki
- Zapoznaj się ze specyfikacją protokołu AP2 i stwórz własnego agenta obsługującego płatności.
- Wdróż UCP dla sprzedawcy, aby włączyć handel agentowy
- Łączenie AP2 z A2A w przypadku wieloagentowych przepływów pracy związanych z handlem
- Dowiedz się więcej o rozszerzeniu AP2 x402 w przypadku płatności kryptowalutami