1. Обзор
В этом практическом занятии вы запустите агент ADK , который будет бронировать билеты в кино у нескольких кинотеатральных компаний, используя два протокола электронной коммерции с открытым исходным кодом:
- UCP (Universal Commerce Protocol) : Стандарт, позволяющий агентам находить продавцов, искать товары в каталогах и управлять процессами оформления заказа.
- AP2 (Agent Payments Protocol) : протокол для безопасной и проверяемой авторизации платежей с использованием криптографически подписанных мандатов.
Демонстрационное приложение CineAgent подключается к двум имитированным кинотеатрам с различными возможностями (выбор мест, специализированные форматы и способы оплаты) и управляет полным процессом бронирования от поиска до оплаты.
Что вы узнаете
- Как работает поиск продавцов UCP через профили
/.well-known/ucp - Как агент ADK использует UCP для поиска по каталогам и создания заказов.
- Как требования AP2 (CartMandate, PaymentMandate) обеспечивают безопасность транзакций
- Как работают сквозные протоколы UCP и AP2 для обеспечения безопасности агентской электронной коммерции
Что вам понадобится
- Проект Google Cloud с включенной функцией выставления счетов.
- Веб-браузер, например Chrome.
- Python 3.11+
Этот практический урок предназначен для разработчиков среднего уровня, знакомых с Python и Google Cloud. Выполнение урока занимает приблизительно 15 минут.
Стоимость ресурсов, созданных в рамках этого практического занятия, должна составлять менее 5 долларов.
2. Понимание протоколов UCP и AP2
Прежде чем приступить к созданию агента, давайте разберемся в двух протоколах, которые обеспечивают безопасную агентскую коммерцию.
Универсальный протокол коммерции ( UCP )
UCP стандартизирует взаимодействие агентов ИИ с продавцами. Он решает проблему необходимости изучения агентами пользовательских API для каждого отдельного магазина, вводя стандартизированную модель ресурсов.
Как это работает:
- Обнаружение : Каждый продавец, соответствующий стандарту UCP, предоставляет свой профиль по стандартному адресу:
/.well-known/ucp. Пример: конечная точка UCP компании Everlane . Когда агент считывает этот профиль, он ищет:- Возможности : Автономные основные функции, которые поддерживает компания, такие как поиск по каталогу или оформление заказа.
- Сервисы : Нижние уровни связи, используемые для обмена данными. Примеры: REST API, MCP (Model Context Protocol), A2A (Agent2Agent Protocol).
- Расширения : Если продавцу требуется специализированное поведение, он может определить пользовательские расширения в этом профиле.
- Операции : После обнаружения агент использует предоставленную конечную точку сервиса для выполнения операций. В этом практическом занятии мы используем протокол контекста модели (MCP) в качестве транспортного протокола сервиса. Агент выполняет вызовы JSON-RPC 2.0 к этой конечной точке для активации обнаруженных возможностей: поиска товаров, создания заказов и завершения покупок.
Протокол агентских платежей ( AP2 )
AP2 стандартизирует порядок авторизации платежей агентами от имени пользователей. Он решает проблему безопасности, связанную с обработкой агентами конфиденциальных платежных данных.
Как это работает:
- Мандат корзины : Когда агент создает заказ с использованием протокола UCP, продавец возвращает объект
CartMandate. Это JSON-объект, содержащий данные корзины и криптографическую подпись продавца. Он служит гарантией фиксации цены. Продавец не может изменить цену после выдачи этого мандата. - Платежное поручение : После проверки содержимого корзины пользователь (или его представитель) создает
PaymentMandateдля авторизации платежа. ЭтоPaymentMandateссылается на платежноеCartMandateи включает криптографическую подпись пользователя (или токен авторизации). - Двойная проверка подписи : Продавец получает оба подтверждения. Он проверяет свою подпись в подтверждении
CartMandateи подпись пользователя в подтвержденииPaymentMandate. Если обе подписи действительны, транзакция продолжается.
Эта система «двойной блокировки» гарантирует, что продавцы не смогут завышать цены, а агенты не смогут тратить деньги без разрешения. В производственной среде для защиты конфиденциальности пользователей используется SD-JWT (Selective Disclosure JWT).
3. Настройте свою среду.
Настройка проекта Google Cloud
Создайте проект в Google Cloud.
- В консоли Google Cloud на странице выбора проекта выберите или создайте проект Google Cloud .
- Убедитесь, что для вашего облачного проекта включена функция выставления счетов. Узнайте, как проверить, включена ли функция выставления счетов для проекта .
Запустить Cloud Shell
Cloud Shell — это среда командной строки, работающая в Google Cloud и поставляемая с предустановленными необходимыми инструментами.
- В верхней части консоли Google Cloud нажмите кнопку «Активировать Cloud Shell» .
- После подключения к Cloud Shell подтвердите свою аутентификацию:
gcloud auth list - Убедитесь, что ваш проект настроен:
gcloud config get project - Если параметры вашего проекта заданы не так, как ожидалось, настройте их следующим образом:
export PROJECT_ID=<YOUR_PROJECT_ID> gcloud config set project $PROJECT_ID
Доступ к моделям Gemini
В вашей среде Cloud Shell скопируйте и вставьте следующие команды. Это обеспечит доступ к моделям Gemini, которые будет использовать Cine Agent.
export GOOGLE_CLOUD_PROJECT=$PROJECT_ID
export GOOGLE_CLOUD_LOCATION=global
export GOOGLE_GENAI_USE_VERTEXAI=True
Структура каталогов установки
Скопируйте и вставьте следующие команды, чтобы создать новый каталог для агента:
mkdir -p agent_payments
cd agent_payments
Установите зависимости
В Google Cloud Shell предустановлен uv для управления средой и зависимостями.
- Создайте файл
pyproject.tomlв корневой папкеagent_paymentsи добавьте в него следующее содержимое. Этот файл определяет метаданные проекта и зависимости.
[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",
]
- Выполните следующую команду, чтобы создать виртуальное окружение и установить все зависимости:
uv sync
- Активируйте виртуальную среду, созданную с помощью
uv:
source .venv/bin/activate
4. Определите агента.
Прежде чем писать логику для работы инструмента, давайте определим самого агента в файле с именем agent.py . Этот агент будет выступать в роли координатора процесса бронирования билетов в кино.
Создайте 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),
],
)
Пояснение к коду
Давайте разберем, что происходит в этом определении агента:
-
model="gemini-3.1-pro-preview": Мы используем последнюю предварительную модель Gemini Pro для сложных рассуждений и использования инструментов. -
instruction: Это подсказка, которая направляет поведение агента. Она явно указывает агенту использовать UCP и AP2, перечисляет доступные инструменты и устанавливает важные правила, такие как «Никогда не выдумывайте данные» и «Цены указаны в центах». -
tools: Это список функций Python (которые мы создадим позже), которые агент может вызывать по запросу пользователя. -
require_confirmation: Вы можете обернуть любой инструмент функциейFunctionTool(my_function,require_confirmation=True). При срабатывании агент приостанавливает работу и ожидает простого подтверждения «да» или «нет», прежде чем выполнить инструмент. В данном случае, перед выполнением инструментаcomplete_purchase, агент приостанавливает работу для подтверждения от человека.
Список инструментов
В определении агента указано, что нам нужно построить. Каждый инструмент соответствует определенной операции в протоколе UCP или AP2:
Инструмент | Что это делает | Действие протокола |
| Найдите продавцов и узнайте об их возможностях. | Запросы |
| Поиск по каталогам различных продавцов. | JSON-RPC в конечную точку MCP |
| Узнайте расписание сеансов в конкретном заведении. | JSON-RPC в конечную точку MCP |
| Начать оформление заказа | JSON-RPC в конечную точку MCP |
| Авторизуйте платеж и завершите оформление заказа. | Подписывает мандат AP2 и направляет его в MCP. |
Модель Gemini будет определять, когда вызывать каждый инструмент, исходя из диалога. Наша следующая задача — реализовать функции каждого инструмента в tools.py .
5. Создайте инструменты агента: обнаружение и просмотр.
Теперь давайте реализуем инструменты, которые агент будет использовать для просмотра и поиска фильмов. Каждый инструмент представляет собой обертку операции UCP.
Создайте новый файл с именем tools.py и скопируйте в него следующий код:
"""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()
Понимание настройки
Чтобы в этом практическом занятии основное внимание уделялось созданию агента, мы используем два вспомогательных класса, UCPClient и AP2Handler , которые мы рассмотрим позже.
- Что это такое? Это написанные вручную вспомогательные классы, которые мы создали для этого практического занятия, чтобы имитировать взаимодействие с условными продавцами. Поскольку официальные SDK UCP и AP2 пока недоступны , мы используем эти вспомогательные классы, чтобы восполнить этот пробел. В производственной среде вы будете использовать официальные SDK, как только они станут доступны.
- Пока что рассматривайте их как вспомогательные объекты :
-
_ucp.discover(url): Получает профиль продавца. -
_ucp.mcp_call(url, method, params): Отправляет JSON-RPC 2.0 запрос на конечную точку MCP продавца.
-
Откройте для себя театры
Этот инструмент — первый шаг в процессе UCP. Он определяет, какие продавцы существуют и что они поддерживают.
Добавить в 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)
Что это делает :
- Программа перебирает список URL-адресов продавцов, предоставленный сервером. В этом практическом занятии мы создадим двух фиктивных продавцов в последующем разделе.
- Для каждого URL-адреса вызывается функция
_ucp.discover(url), которая обращается к конечной точке/.well-known/ucp. - Он собирает имя, возможности и обработчики платежей в сводный список.
- Функция возвращает список в виде строки JSON, которую затем может прочитать агент.
Поиск фильмов
Этот инструмент выполняет поиск по всем найденным продавцам и объединяет результаты. Это крайне важно, поскольку один и тот же фильм может демонстрироваться в нескольких кинотеатрах с разными форматами (IMAX, Dolby) и разными ценами.
Добавить в 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)
Что это делает :
- Он проходит через все обнаруженные театры военных действий.
- Он отправляет JSON-RPC-запрос к каталогу поиска (
_ucp.mcp_call(url, "search_catalog", {"query": query})) на конечную точку MCP продавца. - Затем выполняется небольшая обработка результатов для поиска фильмов и их «вариантов» (которые представляют собой конкретное время показа и форматы). Фильмы группируются по ID, чтобы пользователь не видел повторяющихся записей.
Получить подробную информацию о фильме
Этот инструмент позволяет получить полную информацию о конкретном фильме в конкретном кинотеатре.
Добавить в 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)
Что это делает :
- Этот метод вызывает метод
lookup_catalogв конечной точке MCP продавца, передавая конкретныйmovie_id. Он возвращает подробную информацию, такую как расписание сеансов и наличие мест в конкретном кинотеатре.
6. Создайте инструменты для агентов: оформление заказа и оплата.
Эти инструменты обрабатывают процесс оформления заказа и покупки. Именно здесь вступает в игру протокол AP2, обеспечивающий безопасность транзакций.
Создать оформление заказа
Этот инструмент запускает сеанс оформления заказа в конкретном кинотеатре на конкретное время сеанса.
Добавить в 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)
Что это делает :
- Он вызывает метод
create_checkoutна конечной точке MCP продавца. - Он передает идентификатор
showtime_idиquantityзапрошенные пользователем. - Продавец возвращает JSON-объект, содержащий AP2 CartMandate .
Примечание : Если вы изучите данные ответа, то увидите, что ap2.cart_mandate содержит поле merchant_authorization . Это криптографическая подпись продавца, фиксирующая указанную цену. Он не сможет изменить её позже!
Завершить покупку
В этом инструменте происходит три вещи:
- Мы получаем CartMandate из формы оформления заказа и проверяем его (имитация) .
- Создайте и подпишите платежный мандат (макет) .
- Отправьте подписанное поручение продавцу для завершения покупки.
Добавить в 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)
Почему два мандата? Мандат CartMandate гарантирует, что продавец не сможет изменить цену после её указания. Мандат PaymentMandate гарантирует, что агент не сможет взимать плату с пользователя без его согласия. Процесс выглядит следующим образом:
Merchant locks price -> User authorizes charge -> Merchant verifies both -> Order completes .
Контрольная точка: Полный файл tools.py
Теперь ваш файл tools.py должен содержать 5 функций инструментов и инициализацию на уровне модуля для клиентов UCP и AP2. Убедитесь, что он выглядит следующим образом:
"""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. Сделайте его работоспособным.
При работе с реальными продавцами UCP вы просто указываете агенту их URL-адреса, и всё готово. Для этого практического занятия нам нужно протестировать локально две вещи:
- Имитаторы торговых точек — локальные серверы, которые моделируют конечные точки UCP, чтобы у вас была возможность проводить тестирование.
- Вспомогательные протоколы — тонкие HTTP-обертки для UCP и AP2 (в производственной среде их заменяют официальные SDK).
Примечание : Внимательно читать этот код не нужно. Эти файлы имитируют возможности реальной инфраструктуры и SDK. Копируйте их как есть.
Вспомогательные протоколы
Для UCP и AP2 пока нет клиентских SDK — эти два файла обеспечивают работу HTTP-интерфейса.
Создайте 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()
Создайте 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()
Имитация торговцев
Создайте файл 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
Это создаст два сервера FastAPI, каждый с двумя конечными точками:
-
GET /.well-known/ucp— Обнаружение UCP. Возвращает возможности продавца, URL-адрес конечной точки MCP и принимаемые способы оплаты. -
POST /mcp— Операции MCP (Model Context Protocol). Обрабатывает вызовы JSON-RPC 2.0 для поиска в каталоге, оформления заказа, скидок и оплаты.
Запустите торговые точки в новом терминале — они должны продолжать работать:
cd agent_payments
source .venv/bin/activate
python merchants.py
Вам следует увидеть:
Merchants running: Meridian (:8081), StarLight (:8082)
Вернувшись в первый терминал , проверьте обнаружение UCP:
curl -s http://localhost:8081/.well-known/ucp | python -m json.tool
Вы должны увидеть возможности продавца, URL-адрес конечной точки MCP и обработчики платежей.
8. Запустите агент через ADK Web.
Давайте воспользуемся встроенным веб-интерфейсом ADK CLI! Он предоставляет чат в браузере и автоматически обрабатывает запросы на подтверждение использования инструмента.
Теперь ваш проект должен выглядеть так:
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
Попробуйте!
В текущем терминале перейдите в родительский каталог (на одну папку выше agent_payments ) и запустите веб-интерфейс ADK:
cd ../
adk web --allow_origins '*'
Вы должны увидеть сообщение о том, что сервер запущен, и он предоставит вам URL-адрес (обычно http://localhost:8000 или аналогичный).
Поговорите с агентом
- Откройте в браузере URL-адрес, предоставленный
adk web. - Вы увидите интерфейс чата.
- Попробуйте спросить: "Какие фильмы идут в кинотеатрах?"
- Агент будет находить кинотеатры и искать информацию в каталогах, объединяя результаты от обоих продавцов через UCP.
- Запрос на бронирование билетов: «Забронируйте 2 билета на концерт Оппенхаймера на 19:00» .
- Когда агент пытается позвонить по номеру
complete_purchase, обратите внимание, как веб-интерфейс ADK выводит диалоговое окно или карточку с подтверждением ! - Для авторизации транзакции ответьте в чате, указав следующую JSON-строку:
{"confirmed": true}. - Агент завершит покупку и вернет вам подтверждение заказа с кодами билетов!
Вот что происходило за кулисами:
-
create_checkout→ merchant вернул AP2 CartMandate (signed price lock) -
complete_purchase→ создал PaymentMandate , подписал его (имитация SHA-256), отправил оба мандата продавцу. - Продавец проверил обе подписи → выдал билеты (в имитационной игре)
9. Уборка
Чтобы избежать работы локальных серверов, очистите ресурсы:
- В терминале, где запущена
adk web, нажмите Ctrl+C , чтобы остановить сервер-агент. - В терминале, где запущена
python merchants.py, нажмите Ctrl+C, чтобы остановить запуск фиктивных продавцов. - Отключите виртуальную среду в обоих терминалах, выполнив следующую команду:
deactivate
- (Необязательно) Если вы создали новый проект Google Cloud для этого практического занятия и хотите его удалить, выполните следующую команду:
gcloud projects delete $GOOGLE_CLOUD_PROJECT
10. Поздравляем! 🎉
Вы создали агента ADK, который находит продавцов, просматривает каталоги и совершает покупки, используя UCP и AP2.
Что вы узнали
В этом практическом задании вы создали агент ADK, который обрабатывает безопасные коммерческие потоки. Вот краткое описание того, что вы создали, и ключевых концепций, которые вы применили:
Что вы построили:
- 5 инструментов для агентов, объединяющих операции UCP и AP2 — поиск, поиск по каталогу, оформление заказа, оплата.
- Подписание мандата AP2 — CartMandate (фиксация цены продавцом) + PaymentMandate (авторизация пользователя).
- Поиск по нескольким кинотеатрам — один агент отправляет запросы в несколько кинотеатров и объединяет результаты.
Ключевые понятия:
Протокол | Что это делает | Как это работает |
Открытие UCP | Агент находит продавцов и оценивает их возможности. | |
UCP MCP | Агент просматривает каталоги, оформляет заказы. | Вызовы JSON-RPC 2.0 к конечной точке MCP продавца |
AP2 CartMandate | Продавец фиксирует указанную цену. | Подпись продавца, включает общую сумму и срок действия. |
AP2 PaymentMandate | Пользователь дает разрешение на списание средств. | Подписано пользователем, содержит ссылку на CartMandate. |
В чём заключаются отличия в производстве?
В этом практическом занятии используются моки. В производственной среде:
- Обнаружение UCP осуществляется путем обращения к реестру, а не к жестко закодированным URL-адресам localhost.
- Конечные точки MCP размещаются у реальных продавцов — тот же протокол JSON-RPC 2.0, реальный ассортимент товаров.
- Мандаты AP2 подписываются с помощью sd-jwt-vc, а не хешей SHA-256.
- Авторизация платежей осуществляется с помощью SDK AP2 Wallet с запросами согласия пользователя.
- Фронтенд отображает результаты работы инструмента в виде расширенного пользовательского интерфейса (сетки товаров, сводки оформления заказа, карточки подтверждения).
Следующие шаги
- Изучите спецификацию протокола AP2 и создайте собственного агента с поддержкой платежей.
- Внедрите UCP для вашего продавца , чтобы обеспечить агентскую коммерцию.
- Подключите AP2 к A2A для многоагентных коммерческих рабочих процессов.
- Узнайте о расширении AP2 x402 для платежей в криптовалюте.