1. Tổng quan
Trong lớp học lập trình này, bạn sẽ chạy một tác nhân ADK đặt vé xem phim của nhiều người bán vé xem phim bằng cách sử dụng 2 giao thức thương mại nguồn mở:
- UCP (Giao thức thương mại toàn cầu): Một tiêu chuẩn để các tác nhân khám phá người bán, tìm kiếm danh mục và quản lý quy trình thanh toán.
- AP2 (Agent Payments Protocol): Một giao thức để uỷ quyền thanh toán an toàn, có thể xác minh bằng cách sử dụng các lệnh được ký bằng mật mã.
Ứng dụng demo CineAgent kết nối với 2 người bán vé xem phim giả lập có các chức năng khác nhau (chọn ghế, định dạng chuyên biệt và phương thức thanh toán) và điều phối toàn bộ quy trình đặt vé từ tìm kiếm đến thanh toán.
Kiến thức bạn sẽ học được
- Cách hoạt động của tính năng khám phá người bán UCP thông qua hồ sơ
/.well-known/ucp - Cách một tác nhân ADK sử dụng UCP để tìm kiếm danh mục và tạo quy trình thanh toán
- Cách các chỉ thị AP2 (CartMandate, PaymentMandate) bảo mật giao dịch
- Cách giao thức UCP và AP2 mã hoá hai đầu hoạt động để bảo mật hoạt động thương mại điện tử dựa trên tác nhân
Bạn cần có
- Một dự án trên Google Cloud đã bật tính năng thanh toán
- Một trình duyệt web như Chrome
- Python 3.11 trở lên
Lớp học lập trình này dành cho các nhà phát triển có trình độ trung cấp và đã quen thuộc với Python và Google Cloud. Bạn sẽ mất khoảng 15 phút để hoàn thành lớp học lập trình này.
Các tài nguyên được tạo trong lớp học lập trình này sẽ có chi phí dưới 5 USD.
2. Tìm hiểu về giao thức UCP và AP2
Trước khi bắt đầu xây dựng tác nhân, hãy tìm hiểu hai giao thức giúp hoạt động thương mại dựa trên tác nhân an toàn này trở nên khả thi.
Giao thức thương mại toàn cầu (UCP)
UCP chuẩn hoá cách các tác nhân AI tương tác với người bán. Mô hình này giải quyết vấn đề về việc các tác nhân phải tìm hiểu API tuỳ chỉnh cho từng cửa hàng bằng cách giới thiệu một mô hình tài nguyên được chuẩn hoá.
Cách hoạt động:
- Khám phá: Mọi người bán tuân thủ UCP đều hiển thị một hồ sơ tại một vị trí tiêu chuẩn:
/.well-known/ucp. Ví dụ: Điểm cuối UCP của Everlane.Khi một tác nhân đọc hồ sơ này, tác nhân sẽ tìm:- Tính năng: Các tính năng cốt lõi độc lập mà doanh nghiệp hỗ trợ, chẳng hạn như tìm kiếm trong danh mục hoặc thanh toán.
- Dịch vụ: Các lớp giao tiếp cấp thấp được dùng để trao đổi dữ liệu. Ví dụ: API REST, MCP (Giao thức ngữ cảnh mô hình), A2A (Giao thức Agent2Agent).
- Tiện ích: Nếu cần hành vi chuyên biệt, người bán có thể xác định Tiện ích tuỳ chỉnh trong hồ sơ này.
- Thao tác: Sau khi phát hiện, tác nhân sẽ sử dụng điểm cuối dịch vụ được cung cấp để thực hiện các thao tác. Trong lớp học lập trình này, chúng ta sẽ sử dụng Giao thức ngữ cảnh mô hình (MCP) làm phương tiện truyền tải dịch vụ. Tác nhân thực hiện các lệnh gọi JSON-RPC 2.0 đến điểm cuối này để gọi các chức năng đã phát hiện: tìm kiếm sản phẩm, tạo quy trình thanh toán và hoàn tất giao dịch mua.
Giao thức thanh toán của đại lý (AP2)
AP2 chuẩn hoá cách các nhân viên thay mặt người dùng uỷ quyền thanh toán. Giải pháp này giải quyết vấn đề bảo mật khi các nhân viên xử lý thông tin xác thực thanh toán nhạy cảm.
Cách hoạt động:
- Uỷ nhiệm giỏ hàng: Khi một tác nhân tạo quy trình thanh toán bằng Giao thức UCP, người bán sẽ trả về một
CartMandate. Đây là một đối tượng JSON chứa thông tin chi tiết về giỏ hàng và một chữ ký mật mã của người bán. Chức năng này đóng vai trò như một cam kết khoá giá. Người bán không thể thay đổi giá sau khi đưa ra chỉ thị này. - Uỷ nhiệm thanh toán: Sau khi xác minh nội dung trong giỏ hàng, người dùng (hoặc tác nhân thay mặt người dùng) sẽ tạo một
PaymentMandateđể uỷ nhiệm thanh toán.PaymentMandatenày tham chiếu đếnCartMandatevà bao gồm chữ ký mã hoá (hoặc mã thông báo uỷ quyền) của người dùng. - Xác minh bằng chữ ký kép: Người bán nhận được cả hai uỷ quyền. Chúng xác minh chữ ký của chính chúng trên
CartMandatevà chữ ký của người dùng trênPaymentMandate. Nếu cả hai đều hợp lệ, giao dịch sẽ được thực hiện.
Hệ thống "khoá kép" này đảm bảo rằng người bán không thể tính phí quá mức và nhân viên không thể chi tiêu mà không được phép. Trong quá trình sản xuất, các yêu cầu này sử dụng SD-JWT (JWT công bố có chọn lọc) để bảo vệ quyền riêng tư của người dùng.
3. Thiết lập môi trường
Thiết lập dự án trên Google Cloud
Tạo một dự án trên Google Cloud
- Trong Google Cloud Console, trên trang chọn dự án, hãy chọn hoặc tạo một dự án trên Google Cloud.
- Đảm bảo bạn đã bật tính năng thanh toán cho dự án trên Cloud. Tìm hiểu cách kiểm tra xem tính năng thanh toán có được bật trên một dự án hay không.
Khởi động Cloud Shell
Cloud Shell là một môi trường dòng lệnh chạy trong Google Cloud và được tải sẵn các công cụ cần thiết.
- Nhấp vào Kích hoạt Cloud Shell ở đầu bảng điều khiển Cloud.
- Sau khi kết nối với Cloud Shell, hãy xác minh thông tin xác thực của bạn:
gcloud auth list - Xác nhận rằng dự án của bạn đã được định cấu hình:
gcloud config get project - Nếu dự án của bạn không được thiết lập như mong đợi, hãy thiết lập dự án:
export PROJECT_ID=<YOUR_PROJECT_ID> gcloud config set project $PROJECT_ID
Truy cập vào các mô hình Gemini
Trong môi trường Cloud Shell, hãy sao chép và dán các lệnh sau. Thao tác này sẽ cho phép truy cập vào các mô hình Gemini mà Cine Agent sẽ sử dụng.
export GOOGLE_CLOUD_PROJECT=$PROJECT_ID
export GOOGLE_CLOUD_LOCATION=global
export GOOGLE_GENAI_USE_VERTEXAI=True
Thiết lập cấu trúc thư mục
Sao chép và dán các lệnh sau để tạo một thư mục mới cho tác nhân:
mkdir -p agent_payments
cd agent_payments
Cài đặt các phần phụ thuộc
Google Cloud Shell được cài đặt sẵn uv để quản lý môi trường và các phần phụ thuộc.
- Tạo tệp
pyproject.tomlở thư mục gốc củaagent_paymentsrồi thêm nội dung sau vào tệp đó. Tệp này xác định siêu dữ liệu và các phần phụ thuộc của dự án.
[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",
]
- Chạy lệnh sau để tạo môi trường ảo và cài đặt tất cả các phần phụ thuộc:
uv sync
- Kích hoạt môi trường ảo do
uvtạo:
source .venv/bin/activate
4. Xác định nhân viên hỗ trợ
Trước khi viết bất kỳ logic nào cho công cụ, hãy xác định chính tác nhân trong một tệp có tên là agent.py. Tác nhân này sẽ đóng vai trò là người điều phối cho quy trình đặt vé xem phim.
Tạo 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),
],
)
Giải thích mã
Hãy phân tích những gì đang diễn ra trong định nghĩa tác nhân này:
model="gemini-3.1-pro-preview": Chúng tôi đang sử dụng mô hình Gemini Pro mới nhất (bản dùng thử) để suy luận phức tạp và sử dụng công cụ.instruction: Đây là câu lệnh hướng dẫn hành vi của trợ lý. Tệp này hướng dẫn rõ ràng cho trợ lý ảo cách sử dụng UCP và AP2, liệt kê các công cụ có sẵn và đặt các quy tắc quan trọng như "Không bao giờ bịa dữ liệu" và "Giá tính bằng xu".tools: Đây là danh sách các hàm Python (mà chúng ta sẽ tạo tiếp theo) mà tác nhân có thể chọn gọi dựa trên yêu cầu của người dùng.require_confirmation: Bạn có thể bao bọc mọi công cụ bằngFunctionTool(my_function,require_confirmation=True). Khi được kích hoạt, tác nhân sẽ tạm dừng và chờ một câu trả lời đơn giản là "có" hoặc "không" trước khi thực thi công cụ. Tại đây, trước khi thực thi công cụcomplete_purchase, nhân viên hỗ trợ sẽ tạm dừng để chờ người dùng xác nhận.
Danh sách công cụ
Định nghĩa tác nhân khai báo những gì chúng ta cần xây dựng. Mỗi công cụ liên kết với một thao tác cụ thể trong giao thức UCP hoặc AP2:
Công cụ | Ý nghĩa | Hành động theo giao thức |
| Tìm người bán và khả năng của họ | Cụm từ tìm kiếm |
| Tìm kiếm danh mục của nhiều người bán | JSON-RPC đến điểm cuối MCP |
| Nhận lịch chiếu tại một người bán cụ thể | JSON-RPC đến điểm cuối MCP |
| Bắt đầu một phiên thanh toán | JSON-RPC đến điểm cuối MCP |
| Uỷ quyền thanh toán và hoàn tất đơn đặt hàng | Ký AP2 Mandate và gửi đến MCP |
Mô hình Gemini sẽ quyết định thời điểm gọi từng công cụ dựa trên cuộc trò chuyện. Công việc tiếp theo của chúng ta là triển khai chức năng của từng công cụ trong tools.py.
5. Xây dựng các công cụ cho tác nhân: Khám phá và duyệt web
Bây giờ, hãy triển khai các công cụ mà tác nhân sẽ dùng để duyệt xem và khám phá phim. Mỗi công cụ sẽ bao bọc một thao tác UCP.
Tạo một tệp mới có tên là tools.py rồi sao chép và dán đoạn mã sau:
"""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()
Tìm hiểu về chế độ thiết lập
Để lớp học lập trình này tập trung vào việc xây dựng tác nhân, chúng ta sẽ sử dụng 2 lớp trợ giúp là UCPClient và AP2Handler. Chúng ta sẽ xem xét các lớp này ở bước sau.
- Quảng cáo khi mở ứng dụng là gì?: Đây là các lớp trợ giúp được viết tay mà chúng tôi đã tạo cho lớp học lập trình này để mô phỏng hoạt động tương tác với các người bán mô phỏng. Vì SDK UCP và AP2 chính thức chưa có sẵn, nên chúng tôi đang sử dụng những trình trợ giúp này để khắc phục vấn đề. Trong môi trường phát hành chính thức, bạn sẽ sử dụng các SDK chính thức khi chúng được cung cấp.
- Hiện tại, hãy coi chúng là các đối tượng trợ giúp:
_ucp.discover(url): Tìm nạp hồ sơ của người bán._ucp.mcp_call(url, method, params): Gửi yêu cầu JSON-RPC 2.0 đến điểm cuối MCP của người bán.
Khám phá các nhà hát
Công cụ này là bước đầu tiên trong quy trình UCP. Thư viện này tìm thấy những người bán hiện có và những người bán mà thư viện này hỗ trợ.
Thêm vào 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)
Chức năng của chế độ này:
- Thao tác này lặp lại danh sách URL của người bán do máy chủ cung cấp. Trong lớp học lập trình này, chúng ta sẽ thiết lập 2 người bán mô phỏng ở phần sau.
- Đối với mỗi URL, nó sẽ gọi
_ucp.discover(url), chạm vào điểm cuối/.well-known/ucp. - API này thu thập tên, chức năng và trình xử lý thanh toán vào một danh sách tóm tắt.
- Thao tác này trả về danh sách dưới dạng chuỗi JSON để tác nhân đọc.
Tìm kiếm Phim
Công cụ này tìm kiếm trên tất cả những người bán được phát hiện và hợp nhất kết quả. Điều này rất quan trọng vì cùng một bộ phim có thể đang chiếu ở nhiều rạp với nhiều định dạng (IMAX, Dolby) và mức giá khác nhau.
Thêm vào 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)
Chức năng của chế độ này:
- Nó lặp lại tất cả các rạp chiếu phim đã được phát hiện.
- Thao tác này sẽ gửi yêu cầu JSON-RPC danh mục tìm kiếm (
_ucp.mcp_call(url, "search_catalog", {"query": query})) đến điểm cuối MCP của người bán. - Sau đó, ứng dụng sẽ dọn dẹp một chút để phân tích kết quả nhằm tìm phim và "biến thể" của phim (biến thể này đại diện cho các suất chiếu và định dạng cụ thể). Nhóm các bộ phim theo mã nhận dạng để người dùng không thấy các mục phim trùng lặp.
Nhận thông tin chi tiết về phim
Công cụ này lấy thông tin chi tiết đầy đủ về danh mục cho một bộ phim cụ thể tại một rạp chiếu phim cụ thể.
Thêm vào 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)
Chức năng của chế độ này:
- Thao tác này gọi phương thức
lookup_catalogtrên điểm cuối MCP của người bán, truyềnmovie_idcụ thể. Thao tác này sẽ trả về thông tin chi tiết như giờ chiếu và tình trạng còn ghế của rạp chiếu phim cụ thể đó.
6. Xây dựng các công cụ cho nhân viên hỗ trợ: Thanh toán và thanh toán
Những công cụ này xử lý quy trình thanh toán và quy trình mua. Đây là nơi giao thức AP2 phát huy tác dụng để đảm bảo các giao dịch an toàn.
Tạo quy trình thanh toán
Công cụ này bắt đầu một phiên thanh toán tại một rạp chiếu phim cụ thể vào một giờ chiếu cụ thể.
Thêm vào 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)
Chức năng của chế độ này:
- Thao tác này sẽ gọi phương thức
create_checkouttrên điểm cuối MCP của người bán. - Sự kiện này chuyển
showtime_idvàquantitydo người dùng yêu cầu. - Người bán trả về một đối tượng JSON chứa AP2 CartMandate.
Lưu ý: Nếu bạn kiểm tra dữ liệu phản hồi, ap2.cart_mandate sẽ chứa trường merchant_authorization. Đây là chữ ký mã hoá của người bán khoá giá được báo giá. Họ không thể thay đổi thông tin này sau đó!
Hoàn tất giao dịch mua
Công cụ này sẽ thực hiện 3 việc:
- Chúng tôi nhận CartMandate từ quy trình thanh toán và xác minh (mô phỏng).
- Tạo và ký PaymentMandate (mock).
- Gửi Uỷ quyền đã ký cho Người bán để hoàn tất giao dịch mua.
Thêm vào 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)
Tại sao có hai nhiệm vụ? CartMandate đảm bảo rằng người bán không thể thay đổi giá sau khi báo giá. PaymentMandate đảm bảo rằng trợ lý không thể tính phí người dùng mà không có sự đồng ý của họ. Quy trình như sau:
Merchant locks price -> User authorizes charge -> Merchant verifies both -> Order completes.
Điểm kiểm tra: Đầy tools.py
tools.py hoàn chỉnh của bạn hiện phải có 5 hàm công cụ và quá trình khởi tạo cấp mô-đun cho các ứng dụng UCP và AP2. Hãy xác minh rằng ứng dụng trông như sau:
"""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. Đảm bảo có thể chạy
Với người bán UCP thực tế, bạn chỉ cần gửi cho họ URL của họ là xong. Trong lớp học lập trình này, chúng ta cần 2 thứ để kiểm thử cục bộ:
- Người bán mô phỏng – máy chủ cục bộ mô phỏng các điểm cuối UCP để bạn có thể kiểm thử
- Trợ giúp giao thức – trình bao bọc HTTP mỏng cho UCP và AP2 (trong quá trình sản xuất, các SDK chính thức sẽ thay thế những trình bao bọc này)
Lưu ý: Bạn không cần đọc kỹ mã này. Các tệp này mô phỏng những gì cơ sở hạ tầng và SDK thực tế sẽ cung cấp. Sao chép nguyên trạng.
Trợ giúp về giao thức
UCP và AP2 hiện chưa có SDK ứng dụng – hai tệp này xử lý các thành phần HTTP.
Tạo 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()
Tạo 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()
Nhà bán hàng mô phỏng
Tạo 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
Thao tác này sẽ tạo ra 2 máy chủ FastAPI, mỗi máy chủ có 2 điểm cuối:
GET /.well-known/ucp– Khám phá UCP. Trả về các chức năng của người bán, URL điểm cuối MCP và các phương thức thanh toán được chấp nhận.POST /mcp– Các thao tác MCP (Giao thức ngữ cảnh mô hình). Xử lý các lệnh gọi JSON-RPC 2.0 cho tìm kiếm danh mục, thanh toán, chiết khấu và thanh toán.
Khởi động người bán trong một thiết bị đầu cuối mới – họ cần tiếp tục hoạt động:
cd agent_payments
source .venv/bin/activate
python merchants.py
Bạn sẽ thấy:
Merchants running: Meridian (:8081), StarLight (:8082)
Trong thiết bị đầu cuối đầu tiên, hãy xác minh tính năng khám phá UCP:
curl -s http://localhost:8081/.well-known/ucp | python -m json.tool
Bạn sẽ thấy các chức năng của người bán, URL điểm cuối MCP và trình xử lý thanh toán.
8. Chạy Tác nhân bằng ADK Web
Hãy dùng Giao diện người dùng web tích hợp sẵn của ADK CLI! Điều này cung cấp một giao diện trò chuyện trong trình duyệt và tự động xử lý các lời nhắc xác nhận công cụ.
Lúc này, dự án của bạn sẽ có dạng như sau:
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
Dùng thử
Trong thiết bị đầu cuối hiện tại, hãy chuyển đến thư mục mẹ (một thư mục phía trên agent_payments) rồi khởi động Giao diện người dùng web ADK:
cd ../
adk web --allow_origins '*'
Bạn sẽ thấy kết quả cho biết máy chủ đang chạy và kết quả này sẽ cung cấp cho bạn một URL (thường là http://localhost:8000 hoặc tương tự).
Trao đổi với nhân viên hỗ trợ
- Mở URL do
adk webcung cấp trong trình duyệt. - Bạn sẽ thấy một giao diện trò chuyện.
- Thử hỏi: "Phim gì đang chiếu vậy?"
- Đằng sau hậu trường, tác nhân sẽ khám phá các rạp chiếu phim và tìm kiếm danh mục, đồng thời tổng hợp kết quả từ cả hai người bán thông qua UCP.
- Yêu cầu đặt vé: "Đặt 2 vé xem phim Oppenheimer lúc 7 giờ tối".
- Khi tác nhân cố gắng gọi
complete_purchase, hãy lưu ý cách ADK Web UI bật lên một hộp thoại xác nhận hoặc thẻ! - Để uỷ quyền cho giao dịch, hãy trả lời trong cuộc trò chuyện bằng chuỗi JSON chính xác này:
{"confirmed": true}. - Nhân viên sẽ hoàn tất giao dịch mua và gửi cho bạn thông tin xác nhận đơn đặt hàng kèm theo mã vé!
Sau đây là những gì đã xảy ra:
create_checkout→ người bán trả về một CartMandate AP2 (khoá giá đã ký)complete_purchase→ tạo một PaymentMandate, ký tên (SHA-256 mô phỏng), gửi cả hai PaymentMandate cho người bán- Người bán đã xác minh cả hai chữ ký → đã phát hành vé (trong bản mô phỏng)
9. Dọn dẹp
Để tránh để máy chủ cục bộ chạy, hãy dọn dẹp các tài nguyên:
- Trong dòng lệnh đang chạy
adk web, hãy nhấn tổ hợp phím Ctrl+C để dừng máy chủ tác nhân. - Trong thiết bị đầu cuối đang chạy
python merchants.py, hãy nhấn tổ hợp phím Ctrl+C để dừng các người bán mô phỏng. - Huỷ kích hoạt môi trường ảo trong cả hai thiết bị đầu cuối bằng cách chạy:
deactivate
- (Không bắt buộc) Nếu bạn đã tạo một dự án trên đám mây mới trên Google Cloud cho lớp học lập trình này và muốn xoá dự án đó, hãy chạy:
gcloud projects delete $GOOGLE_CLOUD_PROJECT
10. Xin chúc mừng! 🎉
Bạn đã tạo một tác nhân ADK có thể khám phá người bán, duyệt xem danh mục và hoàn tất giao dịch mua bằng UCP và AP2.
Kiến thức bạn học được
Trong lớp học lập trình này, bạn đã tạo một tác nhân ADK xử lý các quy trình thương mại an toàn. Sau đây là phần tóm tắt về những gì bạn đã tạo và các khái niệm chính mà bạn đã áp dụng:
Sản phẩm bạn đã tạo:
- 5 công cụ dành cho nhân viên hỗ trợ bao gồm các thao tác UCP và AP2 – khám phá, tìm kiếm danh mục, thanh toán.
- Ký uỷ nhiệm AP2 – CartMandate (khoá giá của người bán) + PaymentMandate (uỷ quyền của người dùng).
- Tìm kiếm nhiều người bán – một tác nhân truy vấn nhiều rạp chiếu phim, hợp nhất kết quả.
Các khái niệm chính:
Giao thức | Ý nghĩa | Cách hoạt động |
Khám phá UCP | Tác nhân tìm thấy người bán và khả năng của họ |
|
UCP MCP | Nhân viên duyệt xem danh mục, tạo quy trình thanh toán | Các lệnh gọi JSON-RPC 2.0 đến điểm cuối MCP của người bán |
AP2 CartMandate | Người bán cố định giá được báo | Do người bán ký, bao gồm tổng số tiền và ngày hết hạn |
AP2 PaymentMandate | Người dùng cho phép tính phí | Do người dùng ký, tham chiếu đến CartMandate |
Có gì khác biệt trong quá trình sản xuất?
Lớp học lập trình này sử dụng các đối tượng mô phỏng. Trong bản phát hành công khai:
- Khám phá UCP giải quyết dựa trên một sổ đăng ký, chứ không phải các URL localhost được mã hoá cứng
- Điểm cuối MCP được lưu trữ bởi những người bán thực tế – cùng một giao thức JSON-RPC 2.0, khoảng không quảng cáo thực
- Các quy định bắt buộc của AP2 được ký bằng sd-jwt-vc, chứ không phải hàm băm SHA-256
- Uỷ quyền thanh toán sử dụng AP2 Wallet SDK với lời nhắc yêu cầu sự đồng ý của người dùng
- Giao diện người dùng hiển thị kết quả của công cụ dưới dạng giao diện người dùng nhiều định dạng (lưới sản phẩm, bản tóm tắt quy trình thanh toán, thẻ xác nhận)
Các bước tiếp theo
- Khám phá quy cách giao thức AP2 và tạo tác nhân hỗ trợ thanh toán của riêng bạn
- Triển khai UCP cho người bán để bật tính năng thương mại dựa trên tác nhân
- Kết nối AP2 với A2A cho quy trình thương mại có nhiều tác nhân
- Tìm hiểu về phần mở rộng AP2 x402 cho các khoản thanh toán bằng tiền mã hoá