Mengamankan Perdagangan Agen dengan AP2 dan UCP

1. Ringkasan

Dalam codelab ini, Anda akan menjalankan agen ADK yang memesan tiket bioskop di beberapa penjual tiket bioskop menggunakan dua protokol perdagangan open source:

UCP (Universal Commerce Protocol): Standar bagi agen untuk menemukan penjual, menelusuri katalog, dan mengelola alur checkout.
AP2 (Agent Payments Protocol): Protokol untuk otorisasi pembayaran yang aman dan dapat diverifikasi menggunakan mandat yang ditandatangani secara kriptografis.

Aplikasi demo, CineAgent, terhubung ke dua penjual bioskop tiruan dengan kemampuan yang berbeda (pemilihan tempat duduk, format khusus, dan metode pembayaran) serta mengatur alur pemesanan lengkap dari penelusuran hingga pembayaran.

Yang akan Anda pelajari

Cara kerja penemuan penjual UCP melalui profil /.well-known/ucp
Cara agen ADK menggunakan UCP untuk menelusuri katalog dan membuat checkout
Cara AP2 mewajibkan (CartMandate, PaymentMandate) transaksi yang aman
Cara kerja protokol UCP dan AP2 menyeluruh untuk mengamankan e-commerce berbasis agen

Yang Anda butuhkan

Project Google Cloud yang mengaktifkan penagihan
Browser web seperti Chrome
Python 3.11+

Codelab ini ditujukan bagi developer tingkat menengah yang sudah memahami Python dan Google Cloud. Codelab ini membutuhkan waktu sekitar 15 menit untuk diselesaikan.

Resource yang dibuat dalam codelab ini seharusnya berbiaya kurang dari $5.

2. Memahami Protokol UCP dan AP2

Sebelum kita mulai membangun agen, mari kita pahami dua protokol yang memungkinkan perdagangan berbasis agen yang aman ini.

Universal Commerce Protocol (UCP)

UCP menstandardisasi cara agen AI berinteraksi dengan penjual. Solusi ini mengatasi masalah agen yang harus mempelajari API kustom untuk setiap toko dengan memperkenalkan model resource standar.

Cara kerjanya:

Penemuan: Setiap penjual yang mematuhi UCP mengekspos profil di lokasi standar: /.well-known/ucp. Contoh: Endpoint UCP Everlane.Saat agen membaca profil ini, agen akan mencari:
- Kemampuan: Fitur inti mandiri yang didukung bisnis, seperti penelusuran katalog atau checkout.
- Layanan: Lapisan komunikasi tingkat bawah yang digunakan untuk bertukar data. Contoh: REST API, MCP (Model Context Protocol), A2A (Agent2Agent Protocol).
- Ekstensi: Jika penjual memerlukan perilaku khusus, mereka dapat menentukan Ekstensi kustom dalam profil ini.
Operasi: Setelah ditemukan, agen menggunakan endpoint layanan yang disediakan untuk melakukan operasi. Dalam codelab ini, kita akan menggunakan Model Context Protocol (MCP) sebagai transpor layanan. Agen membuat panggilan JSON-RPC 2.0 ke endpoint ini untuk memanggil kemampuan yang ditemukan: menelusuri produk, membuat checkout, dan menyelesaikan pembelian.

Alur Kerja UCP

Agent Payments Protocol (AP2)

AP2 menstandardisasi cara pembayaran diotorisasi oleh agen atas nama pengguna. Cara ini mengatasi masalah keamanan terkait agen yang menangani kredensial pembayaran sensitif.

Cara kerjanya:

Mandat Keranjang: Saat agen membuat checkout menggunakan UCP Protocol, penjual akan menampilkan CartMandate. Ini adalah objek JSON yang berisi detail keranjang dan tanda tangan kriptografi dari penjual. Fitur ini berfungsi sebagai jaminan penguncian harga. Penjual tidak dapat mengubah harga setelah mengeluarkan mandat ini.
Perintah Pembayaran: Setelah memverifikasi isi keranjang, pengguna (atau agen atas nama pengguna) membuat PaymentMandate untuk mengizinkan pembayaran. PaymentMandate ini mereferensikan CartMandate dan menyertakan tanda tangan kriptografi (atau token otorisasi) pengguna.
Verifikasi Tanda Tangan Ganda: Penjual menerima kedua surat kuasa. Mereka memverifikasi tanda tangan mereka sendiri pada CartMandate dan tanda tangan pengguna pada PaymentMandate. Jika keduanya valid, transaksi akan dilanjutkan.

Sistem "kunci ganda" ini memastikan bahwa penjual tidak dapat menagih terlalu mahal dan agen tidak dapat berbelanja tanpa otorisasi. Dalam produksi, mandat ini menggunakan SD-JWT (Selective Disclosure JWT) untuk melindungi privasi pengguna.

Alur Kerja AP2

3. Menyiapkan lingkungan Anda

Penyiapan Project Google Cloud

Buat Project Google Cloud

Di Konsol Google Cloud, di halaman pemilih project, pilih atau buat project Google Cloud.
Pastikan penagihan diaktifkan untuk project Cloud Anda. Pelajari cara memeriksa apakah penagihan telah diaktifkan pada suatu project.

Mulai Cloud Shell

Cloud Shell adalah lingkungan command line yang berjalan di Google Cloud yang telah dilengkapi dengan alat yang diperlukan.

Klik Activate Cloud Shell di bagian atas konsol Google Cloud.
Setelah terhubung ke Cloud Shell, verifikasi autentikasi Anda:
```
gcloud auth list
```
Pastikan project Anda dikonfigurasi:
```
gcloud config get project
```

Jika project Anda tidak ditetapkan seperti yang diharapkan, tetapkan project:

export PROJECT_ID=<YOUR_PROJECT_ID>
gcloud config set project $PROJECT_ID

Mengakses model Gemini

Di lingkungan Cloud Shell Anda, salin dan tempel perintah berikut. Tindakan ini akan mengaktifkan akses ke model Gemini yang akan digunakan Cine Agent.

export GOOGLE_CLOUD_PROJECT=$PROJECT_ID
export GOOGLE_CLOUD_LOCATION=global
export GOOGLE_GENAI_USE_VERTEXAI=True

Menyiapkan Struktur Direktori

Salin dan tempel perintah berikut untuk membuat direktori baru bagi agen:

mkdir -p agent_payments
cd agent_payments

Menginstal Dependensi

Google Cloud Shell telah diinstal sebelumnya dengan uv untuk mengelola lingkungan dan dependensi.

Buat file pyproject.toml di root folder agent_payments Anda, lalu tambahkan konten berikut ke dalamnya. File ini menentukan metadata dan dependensi project.

[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",
]

Jalankan perintah berikut untuk membuat lingkungan virtual dan menginstal semua dependensi:

uv sync

Aktifkan lingkungan virtual yang dibuat oleh uv:

source .venv/bin/activate

4. Menentukan agen

Sebelum menulis logika alat apa pun, mari kita tentukan agen itu sendiri dalam file bernama agent.py. Agen ini akan bertindak sebagai pengelola alur pemesanan tiket bioskop.

Buat 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),
    ],
)

Penjelasan Kode

Mari kita uraikan apa yang terjadi dalam definisi agen ini:

model="gemini-3.1-pro-preview": Kami menggunakan model pratinjau Gemini Pro terbaru untuk penalaran yang kompleks dan penggunaan alat.
instruction: Ini adalah perintah yang memandu perilaku agen. Agen ini secara eksplisit memberi tahu agen untuk menggunakan UCP dan AP2, mencantumkan alat yang tersedia, dan menetapkan aturan penting seperti "Jangan mengarang data" dan "Harga dalam sen".
tools: Ini adalah daftar fungsi Python (yang akan kita buat berikutnya) yang dapat dipilih agen untuk dipanggil berdasarkan permintaan pengguna.
require_confirmation: Anda dapat menggabungkan alat apa pun dengan FunctionTool(my_function,require_confirmation=True). Saat dipicu, agen akan menjeda dan menunggu persetujuan "ya" atau "tidak" sederhana sebelum menjalankan alat. Di sini, sebelum menjalankan alat complete_purchase, agen akan berhenti sejenak untuk menunggu konfirmasi dari manusia.

Daftar Alat

Definisi agen menyatakan apa yang perlu kita bangun. Setiap alat dipetakan ke operasi tertentu dalam protokol UCP atau AP2:

Alat	Fungsinya	Tindakan Protokol
`discover_theaters`	Menemukan penjual dan kemampuannya	Kueri `/.well-known/ucp`
`search_movies`	Menelusuri katalog di seluruh penjual	Endpoint JSON-RPC ke MCP
`get_movie_detail`	Mendapatkan jadwal tayang di penjual tertentu	Endpoint JSON-RPC ke MCP
`create_checkout`	Mulai sesi checkout	Endpoint JSON-RPC ke MCP
`complete_purchase`	Otorisasi pembayaran dan selesaikan pesanan	Menandatangani Mandat AP2 & mengirimkannya ke MCP

Model Gemini akan memutuskan kapan harus memanggil setiap alat berdasarkan percakapan. Tugas kita selanjutnya adalah menerapkan fungsi setiap alat di tools.py.

5. Membangun alat agen: Penemuan & Penjelajahan

Sekarang, mari kita terapkan alat yang akan digunakan agen untuk menjelajahi dan menemukan film. Setiap alat membungkus operasi UCP.

Buat file baru bernama tools.py, lalu salin dan tempel kode berikut:

"""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()

Memahami Penyiapan

Agar codelab ini tetap berfokus pada pembuatan agen, kita menggunakan dua class helper, UCPClient dan AP2Handler, yang akan kita lihat di langkah selanjutnya.

Apa saja?: Ini adalah class helper yang ditulis tangan dan kami buat untuk codelab ini guna menyimulasikan interaksi dengan penjual tiruan. Karena SDK UCP dan AP2 resmi belum tersedia, kami menggunakan helper ini untuk menjembatani kesenjangan tersebut. Di lingkungan produksi, Anda akan menggunakan SDK resmi setelah tersedia.
Untuk saat ini, perlakukan sebagai objek pembantu:
- _ucp.discover(url): Mengambil profil penjual.
- _ucp.mcp_call(url, method, params): Mengirim permintaan JSON-RPC 2.0 ke endpoint MCP penjual.

Menemukan Teater

Alat ini adalah langkah pertama dalam alur UCP. API ini menemukan penjual yang ada dan apa yang mereka dukung.

Tambahkan ke 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)

Fungsinya:

Iterasi dilakukan pada daftar URL penjual yang disediakan oleh server. Dalam codelab ini, kita akan menyiapkan dua penjual tiruan di bagian selanjutnya.
Untuk setiap URL, _ucp.discover(url) akan dipanggil, yang akan memanggil endpoint /.well-known/ucp.
Layanan ini mengumpulkan nama, kemampuan, dan pengendali pembayaran ke dalam daftar ringkasan.
API ini menampilkan daftar sebagai string JSON agar dapat dibaca oleh agen.

Telusuri Film

Alat ini menelusuri semua penjual yang ditemukan dan menggabungkan hasilnya. Hal ini sangat penting karena film yang sama dapat diputar di beberapa bioskop dengan format yang berbeda (IMAX, Dolby) dan harga yang berbeda.

Tambahkan ke 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)

Fungsinya:

Fungsi ini melakukan loop melalui semua bioskop yang ditemukan.
Aplikasi ini mengirim permintaan JSON-RPC katalog penelusuran (_ucp.mcp_call(url, "search_catalog", {"query": query})) ke endpoint MCP penjual.
Kemudian, kode tersebut melakukan sedikit pembersihan untuk mengurai hasil guna menemukan film dan "varian"nya (yang merepresentasikan waktu tayang dan format tertentu). Mengelompokkan film menurut ID sehingga pengguna tidak melihat entri film duplikat.

Mendapatkan Detail Film

Alat ini mendapatkan detail katalog lengkap untuk film tertentu di bioskop tertentu.

Tambahkan ke 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)

Fungsinya:

Metode ini memanggil metode lookup_catalog di endpoint MCP penjual, dengan meneruskan movie_id tertentu. Tindakan ini akan menampilkan informasi mendetail seperti jadwal pertunjukan dan ketersediaan kursi untuk bioskop tertentu tersebut.

6. Membangun alat agen: Checkout & Pembayaran

Alat ini menangani alur checkout dan pembelian. Di sinilah protokol AP2 berperan untuk memastikan transaksi yang aman.

Buat Checkout

Alat ini memulai sesi checkout di bioskop tertentu untuk waktu pertunjukan tertentu.

Tambahkan ke 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)

Fungsinya:

API ini memanggil metode create_checkout di endpoint MCP penjual.
Aplikasi ini meneruskan showtime_id dan quantity yang diminta oleh pengguna.
Penjual menampilkan objek JSON yang berisi AP2 CartMandate.

Catatan: Jika Anda memeriksa data respons, ap2.cart_mandate berisi kolom merchant_authorization. Ini adalah tanda tangan kriptografi penjual yang mengunci harga yang dikutip. Mereka tidak dapat mengubahnya nanti.

Selesaikan Pembelian

Ada tiga hal yang terjadi di alat ini:

Kita mendapatkan CartMandate dari checkout dan memverifikasinya (mock).
Buat dan tandatangani PaymentMandate (tiruan).
Kirimkan Surat Perintah yang telah ditandatangani kepada Penjual untuk menyelesaikan pembelian.

Tambahkan ke 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)

Mengapa ada dua mandat? CartMandate memastikan penjual tidak dapat mengubah harga setelah mengutipnya. PaymentMandate memastikan agen tidak dapat menagih pengguna tanpa izin. Alurnya adalah:

Merchant locks price -> User authorizes charge -> Merchant verifies both -> Order completes.

Checkpoint: Penuh `tools.py`

tools.py lengkap Anda sekarang akan memiliki 5 fungsi alat dan inisialisasi tingkat modul untuk klien UCP dan AP2. Pastikan tampilannya seperti ini:

"""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. Jadikan dapat dijalankan

Dengan penjual UCP yang sebenarnya, Anda akan mengarahkan agen ke URL mereka dan selesai. Untuk codelab ini, kita memerlukan dua hal untuk diuji secara lokal:

Penjual tiruan — server lokal yang menyimulasikan endpoint UCP sehingga Anda memiliki sesuatu untuk diuji
Helper protokol — wrapper HTTP tipis untuk UCP dan AP2 (dalam produksi, SDK resmi menggantikan wrapper ini)

Catatan: Anda tidak perlu membaca kode ini dengan cermat. File ini menyimulasikan apa yang akan disediakan oleh infrastruktur dan SDK sebenarnya. Salin apa adanya.

Helper protokol

UCP dan AP2 belum memiliki SDK klien — kedua file ini menangani HTTP plumbing.

Buat 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()

Buat 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()

Penjual tiruan

Buat 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

Tindakan ini akan membuat dua server FastAPI, masing-masing dengan dua endpoint:

GET /.well-known/ucp — Penemuan UCP. Menampilkan kemampuan penjual, URL endpoint MCP, dan metode pembayaran yang diterima.
POST /mcp — Operasi MCP (Model Context Protocol). Menangani panggilan JSON-RPC 2.0 untuk penelusuran katalog, checkout, diskon, dan pembayaran.

Mulai penjual di terminal baru — mereka harus tetap berjalan:

cd agent_payments
source .venv/bin/activate
python merchants.py

Anda akan melihat:

Merchants running: Meridian (:8081), StarLight (:8082)

Kembali di terminal pertama, verifikasi penemuan UCP:

curl -s http://localhost:8081/.well-known/ucp | python -m json.tool

Anda akan melihat kemampuan penjual, URL endpoint MCP, dan handler pembayaran.

8. Menjalankan Agen dengan Web ADK

Mari kita gunakan UI Web bawaan ADK CLI. Hal ini menyediakan antarmuka chat di browser dan secara otomatis menangani perintah konfirmasi alat.

Project Anda sekarang akan terlihat seperti ini:

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

Coba

Di terminal saat ini, buka direktori induk (satu folder di atas agent_payments) dan mulai UI Web ADK:

cd ../
adk web --allow_origins '*'

Anda akan melihat output yang menunjukkan bahwa server sedang berjalan, dan output tersebut akan memberi Anda URL (biasanya http://localhost:8000 atau yang serupa).

Berbicara dengan agen

Buka URL yang diberikan oleh adk web di browser Anda.
Anda akan melihat antarmuka chat.
Coba tanyakan: "Film apa yang sedang diputar?"
Agen akan menemukan bioskop dan menelusuri katalog di balik layar, menggabungkan hasil dari kedua penjual melalui UCP.
Minta untuk memesan tiket: "Pesan 2 tiket untuk Oppenheimer pukul 19.00".
Saat agen mencoba memanggil complete_purchase, perhatikan bagaimana UI Web ADK memunculkan dialog konfirmasi atau kartu.
Untuk mengizinkan transaksi, balas dalam chat dengan string JSON persis seperti ini: {"confirmed": true}.
Agen akan menyelesaikan pembelian dan mengembalikan konfirmasi pesanan Anda dengan kode tiket.

Berikut yang terjadi di balik layar:

create_checkout → penjual menampilkan CartMandate AP2 (penguncian harga yang ditandatangani)
complete_purchase → membuat PaymentMandate, menandatanganinya (SHA-256 tiruan), mengirimkan kedua mandat kepada penjual
Penjual memverifikasi kedua tanda tangan → mengeluarkan tiket (dalam tiruan)

9. Pembersihan

Untuk menghindari server lokal tetap berjalan, bersihkan resource:

Di terminal yang menjalankan adk web, tekan Ctrl+C untuk menghentikan server agen.
Di terminal yang menjalankan python merchants.py, tekan Ctrl+C untuk menghentikan penjual tiruan.
Nonaktifkan lingkungan virtual di kedua terminal dengan menjalankan:

deactivate

(Opsional) Jika Anda membuat project Google Cloud baru untuk codelab ini dan ingin menghapusnya, jalankan:

gcloud projects delete $GOOGLE_CLOUD_PROJECT

10. Selamat! 🎉

Anda telah membangun agen ADK yang menemukan penjual, menjelajahi katalog, dan menyelesaikan pembelian menggunakan UCP dan AP2.

Yang telah Anda pelajari

Dalam codelab ini, Anda telah membuat agen ADK yang menangani alur perdagangan yang aman. Berikut adalah ringkasan tentang apa yang telah Anda bangun dan konsep utama yang Anda terapkan:

Yang Anda bangun:

5 alat agen yang membungkus operasi UCP dan AP2 — penemuan, penelusuran katalog, checkout, pembayaran.
Penandatanganan mandat AP2 — CartMandate (penguncian harga penjual) + PaymentMandate (otorisasi pengguna).
Penelusuran multi-penjual — satu agen yang membuat kueri beberapa bioskop, menggabungkan hasil.

Konsep Utama:

Protokol	Fungsinya	Cara kerjanya
Penemuan UCP	Agen menemukan penjual dan kemampuannya	`/.well-known/ucp` → kemampuan, endpoint MCP, metode pembayaran
UCP MCP	Agen menjelajahi katalog, membuat checkout	Panggilan JSON-RPC 2.0 ke endpoint MCP penjual
AP2 CartMandate	Penjual mengunci harga penawaran	Ditandatangani oleh penjual, mencakup total + masa berlaku
AP2 PaymentMandate	Pengguna mengizinkan penagihan	Ditandatangani oleh pengguna, mereferensikan CartMandate

Apa yang berbeda dalam produksi?

Codelab ini menggunakan tiruan. Dalam produksi:

Penemuan UCP diselesaikan terhadap registry, bukan URL localhost yang di-hardcode
Endpoint MCP dihosting oleh penjual sungguhan — protokol JSON-RPC 2.0 yang sama, inventaris sungguhan
Mandat AP2 ditandatangani dengan sd-jwt-vc, bukan hash SHA-256
Otorisasi pembayaran menggunakan AP2 Wallet SDK dengan perintah izin pengguna
Frontend merender hasil alat sebagai UI yang kaya (petak produk, ringkasan checkout, kartu konfirmasi)

Langkah berikutnya

Pelajari spesifikasi protokol AP2 dan buat agen yang mendukung pembayaran Anda sendiri
Terapkan UCP untuk penjual Anda guna mengaktifkan perdagangan berbasis agen
Menghubungkan AP2 dengan A2A untuk alur kerja perdagangan multi-agen
Pelajari ekstensi AP2 x402 untuk pembayaran mata uang kripto

Mengamankan Perdagangan Agen dengan AP2 dan UCP

1. Ringkasan

Yang akan Anda pelajari

Yang Anda butuhkan

2. Memahami Protokol UCP dan AP2

Universal Commerce Protocol (UCP)

Agent Payments Protocol (AP2)

3. Menyiapkan lingkungan Anda

Penyiapan Project Google Cloud

Buat Project Google Cloud

Mulai Cloud Shell

Mengakses model Gemini

Menyiapkan Struktur Direktori

Menginstal Dependensi

4. Menentukan agen

Penjelasan Kode

Daftar Alat

5. Membangun alat agen: Penemuan & Penjelajahan

Memahami Penyiapan

Menemukan Teater

Telusuri Film

Mendapatkan Detail Film

6. Membangun alat agen: Checkout & Pembayaran

Buat Checkout

Selesaikan Pembelian

Checkpoint: Penuh tools.py

7. Jadikan dapat dijalankan

Helper protokol

Penjual tiruan

8. Menjalankan Agen dengan Web ADK

Coba

Berbicara dengan agen

9. Pembersihan

10. Selamat! 🎉

Yang telah Anda pelajari

Apa yang berbeda dalam produksi?

Langkah berikutnya

Dokumen referensi

Checkpoint: Penuh `tools.py`