1. Genel Bakış
Bu codelab'de, iki açık kaynaklı ticaret protokolünü kullanarak birden fazla sinema satıcısından sinema bileti ayırtan bir ADK aracısı çalıştıracaksınız:
- UCP (Evrensel Ticaret Protokolü): Ajanların satıcıları keşfetmesi, kataloglarda arama yapması ve ödeme akışlarını yönetmesi için kullanılan bir standarttır.
- AP2 (Agent Payments Protocol): Kriptografik olarak imzalanmış yetkiler kullanılarak güvenli ve doğrulanabilir ödeme yetkilendirmesi için kullanılan bir protokol.
CineAgent adlı demo uygulaması, farklı özelliklere (koltuk seçimi, özel formatlar ve ödeme yöntemleri) sahip iki sahte sinema satıcısına bağlanır ve aramadan ödemeye kadar tüm rezervasyon akışını yönetir.
Neler öğreneceksiniz?
- UCP satıcı keşfinin
/.well-known/ucpprofilleri üzerinden işleyiş şekli - ADK aracısı, kataloglarda arama yapmak ve ödeme oluşturmak için UCP'yi nasıl kullanır?
- AP2 zorunlulukları (CartMandate, PaymentMandate) işlemleri nasıl güvence altına alır?
- Uçtan uca UCP ve AP2 protokolleri, ajan tabanlı e-ticaretin güvenliğini nasıl sağlar?
İhtiyacınız olanlar
- Faturalandırmanın etkin olduğu bir Google Cloud projesi
- Chrome gibi bir web tarayıcısı
- Python 3.11 veya sonraki sürümler
Bu codelab, Python ve Google Cloud hakkında bilgi sahibi olan orta düzey geliştiriciler içindir. Bu codelab'in tamamlanması yaklaşık 15 dakika sürer.
Bu kod laboratuvarında oluşturulan kaynakların maliyeti 5 ABD dolarından az olmalıdır.
2. UCP ve AP2 protokollerini anlama
Aracı oluşturmaya başlamadan önce, bu güvenli aracı tabanlı ticareti mümkün kılan iki protokolü anlayalım.
Evrensel Ticaret Protokolü (UCP)
UCP, yapay zeka ajanlarının satıcılarla etkileşim kurma şeklini standart hâle getirir. Standartlaştırılmış bir kaynak modeli sunarak temsilcilerin her mağaza için özel API'leri öğrenmesi sorununu çözer.
İşleyiş şekli:
- Bulma: UCP'ye uygun her satıcı, standart bir konumda profil gösterir:
/.well-known/ucp. Örnek: Everlane'in UCP uç noktası.Bir temsilci bu profili okuduğunda şunları arar:- Özellikler: İşletmenin desteklediği bağımsız temel özellikler (ör. katalogda arama veya ödeme).
- Hizmetler: Veri alışverişi için kullanılan alt düzey iletişim katmanları. Örnekler: REST API, MCP (Model Context Protocol), A2A (Agent2Agent Protocol).
- Uzantılar: Bir satıcının özel davranışa ihtiyacı varsa bu profilde özel uzantılar tanımlayabilir.
- İşlemler: Ajan, keşfedildikten sonra işlemleri gerçekleştirmek için sağlanan hizmet uç noktasını kullanır. Bu codelab'de hizmet aktarımı olarak Model Context Protocol (MCP) kullanılır. Aracı, keşfedilen özellikleri (ürün arama, ödeme oluşturma ve satın alma işlemlerini tamamlama) çağırmak için bu uç noktaya JSON-RPC 2.0 çağrıları yapar.
Agent Payments Protocol (AP2)
AP2, ödemelerin kullanıcılar adına aracılar tarafından nasıl yetkilendirileceğini standartlaştırır. Hassas ödeme bilgilerini işleyen aracıların güvenlik sorununu çözer.
İşleyiş şekli:
- Sepet Yetkisi: Bir aracı, UCP Protokolü'nü kullanarak ödeme oluşturduğunda satıcı
CartMandatedöndürür. Bu, alışveriş sepeti ayrıntılarını ve satıcının kriptografik imzasını içeren bir JSON nesnesidir. Fiyat sabitleme garantisi görevi görür. Satıcı, bu yetkiyi verdikten sonra fiyatı değiştiremez. - Ödeme Talimatı: Kullanıcı (veya kullanıcı adına hareket eden temsilci), ödeme yetkisi vermek için alışveriş sepetinin içeriğini doğruladıktan sonra
PaymentMandateoluşturur. BuPaymentMandate,CartMandate'e referans verir ve kullanıcının kriptografik imzasını (veya yetkilendirme jetonunu) içerir. - Çift İmza Doğrulaması: Satıcı, her iki yetkiyi de alır.
CartMandateüzerinde kendi imzalarını,PaymentMandateüzerinde ise kullanıcının imzasını doğrularlar. Her ikisi de geçerliyse işlem devam eder.
Bu "çift kilit" sistemi, satıcıların fazla ücret almasını ve aracıların yetkisiz harcama yapmasını engeller. Üretimde, bu zorunluluklar kullanıcı gizliliğini korumak için SD-JWT (Seçici Açıklama JWT) kullanır.
3. Ortamınızı ayarlama
Google Cloud Proje Kurulumu
Google Cloud projesi oluşturma
- Google Cloud Console'daki proje seçici sayfasında bir Google Cloud projesi seçin veya oluşturun.
- Cloud projeniz için faturalandırmanın etkinleştirildiğinden emin olun. Bir projede faturalandırmanın etkin olup olmadığını kontrol etmeyi öğrenin.
Cloud Shell'i Başlatma
Cloud Shell, Google Cloud'da çalışan ve gerekli araçların önceden yüklendiği bir komut satırı ortamıdır.
- Google Cloud Console'un üst kısmından Cloud Shell'i etkinleştir'i tıklayın.
- Cloud Shell'e bağlandıktan sonra kimlik doğrulamanızı onaylayın:
gcloud auth list - Projenizin yapılandırıldığını onaylayın:
gcloud config get project - Projeniz beklendiği gibi ayarlanmamışsa şu şekilde ayarlayın:
export PROJECT_ID=<YOUR_PROJECT_ID> gcloud config set project $PROJECT_ID
Gemini modellerine erişim
Cloud Shell ortamınızda aşağıdaki komutları kopyalayıp yapıştırın. Bu işlem, Cine Agent'ın kullanacağı Gemini modellerine erişimi etkinleştirir.
export GOOGLE_CLOUD_PROJECT=$PROJECT_ID
export GOOGLE_CLOUD_LOCATION=global
export GOOGLE_GENAI_USE_VERTEXAI=True
Dizin yapısını ayarlama
Aracı için yeni bir dizin oluşturmak üzere aşağıdaki komutları kopyalayıp yapıştırın:
mkdir -p agent_payments
cd agent_payments
Bağımlılıkları yükleme
Google Cloud Shell, ortamı ve bağımlılıkları yönetmek için uv ile önceden yüklenmiş olarak gelir.
agent_paymentsklasörünüzün kök dizininde birpyproject.tomldosyası oluşturun ve aşağıdaki içeriği bu dosyaya ekleyin. Bu dosya, proje meta verilerini ve bağımlılıklarını tanımlar.
[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",
]
- Sanal ortamı oluşturmak ve tüm bağımlılıkları yüklemek için aşağıdaki komutu çalıştırın:
uv sync
uvtarafından oluşturulan sanal ortamı etkinleştirin:
source .venv/bin/activate
4. Temsilciyi tanımlama
Herhangi bir araç mantığı yazmadan önce, agent.py adlı bir dosyada aracıyı tanımlayalım. Bu ajan, film rezervasyonu akışının düzenleyicisi olarak hareket eder.
Oluşturma 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),
],
)
Kod Açıklaması
Bu aracı tanımında neler olduğunu inceleyelim:
model="gemini-3.1-pro-preview": Karmaşık akıl yürütme ve araç kullanımı için en yeni Gemini Pro önizleme modelini kullanıyoruz.instruction: Bu istem, aracının davranışını yönlendirir. Aracıya UCP ve AP2'yi kullanmasını açıkça söyler, mevcut araçları listeler ve "Asla veri uydurma" ve "Fiyatlar sent cinsindendir" gibi önemli kurallar belirler.tools: Bu, aracının kullanıcının isteklerine göre çağırmayı seçebileceği Python işlevlerinin (bir sonraki adımda oluşturacağımız) listesidir.require_confirmation: Herhangi bir aracıFunctionTool(my_function,require_confirmation=True)ile sarmalayabilirsiniz. Etkinleştirildiğinde aracı, aracı çalıştırmadan önce duraklatır ve basit bir "evet" veya "hayır" onayı bekler. Burada,complete_purchasearacını yürütmeden önce ajanın işlemi duraklatarak kullanıcı onayı alması sağlanıyor.
Araç Listesi
Temsilci tanımı, ne oluşturmamız gerektiğini belirtir. Her araç, UCP veya AP2 protokolündeki belirli bir işlemle eşlenir:
Araç | Ne işe yarar? | Protokol İşlemi |
| Satıcıları ve özelliklerini bulma | Sorgular |
| Satıcılar genelinde kataloglarda arama yapma | JSON-RPC'den MCP uç noktasına |
| Belirli bir satıcıdaki gösterim saatlerini öğrenme | JSON-RPC'den MCP uç noktasına |
| Ödeme oturumu başlatma | JSON-RPC'den MCP uç noktasına |
| Ödemeyi yetkilendirin ve siparişi tamamlayın | AP2 Yetkisini İmzalama ve MCP'ye Gönderme |
Gemini modeli, görüşmeye göre her aracın ne zaman çağrılacağına karar verir. Bir sonraki görevimiz, her aracın tools.py içinde ne yaptığını uygulamaktır.
5. Aracı oluşturma araçları: Keşif ve Göz Atma
Şimdi temsilcinin filmlere göz atmak ve filmleri keşfetmek için kullanacağı araçları uygulayalım. Her araç bir UCP işlemini kapsar.
tools.py adlı yeni bir dosya oluşturun ve aşağıdaki kodu kopyalayıp yapıştırın:
"""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()
Kurulumu Anlama
Bu codelab'in temsilci oluşturmaya odaklanmasını sağlamak için daha sonraki bir adımda inceleyeceğimiz UCPClient ve AP2Handler olmak üzere iki yardımcı sınıf kullanıyoruz.
- Bunlar nelerdir?: Bunlar, bu codelab için sahte satıcılarla etkileşimi simüle etmek amacıyla oluşturduğumuz elle yazılmış yardımcı sınıflardır. Resmi UCP ve AP2 SDK'ları henüz kullanıma sunulmadığı için bu yardımcıları kullanarak aradaki boşluğu dolduruyoruz. Üretim ortamında, kullanıma sunulduktan sonra resmi SDK'ları kullanırsınız.
- Şimdilik bunları yardımcı nesneler olarak değerlendirin:
_ucp.discover(url): Bir satıcının profilini getirir._ucp.mcp_call(url, method, params): İşletmenin MCP uç noktasına JSON-RPC 2.0 isteği gönderir.
Tiyatroları keşfetme
Bu araç, UCP akışının ilk adımıdır. Hangi satıcıların olduğunu ve hangi hizmetleri desteklediklerini bulur.
tools.py uygulamasına ekleme:
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)
Bu işlem ne yapar?
- Sunucu tarafından sağlanan satıcı URL'lerinin listesini yineler. Bu codelab'in ilerleyen bölümünde iki sahte satıcı oluşturacağız.
- Her URL için
_ucp.discover(url)işlevi çağrılır. Bu işlev,/.well-known/ucpuç noktasına ulaşır. - Adı, özellikleri ve ödeme işleyicileri özet bir listede toplar.
- Listeyi, temsilcinin okuması için JSON dizesi olarak döndürür.
Filmlerde Ara
Bu araç, keşfedilen tüm satıcıları arar ve sonuçları birleştirir. Aynı film, farklı formatlarda (IMAX, Dolby) ve farklı fiyatlarla birden fazla sinemada gösteriliyor olabileceğinden bu bilgi çok önemlidir.
tools.py uygulamasına ekleme:
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)
Bu işlem ne yapar?
- Bulunan tüm sinemaları sırayla gösterir.
- Satıcının MCP uç noktasına bir arama kataloğu JSON-RPC isteği (
_ucp.mcp_call(url, "search_catalog", {"query": query})) gönderir. - Ardından, sonuçları ayrıştırarak filmleri ve "varyantlarını" (belirli gösterim saatlerini ve formatları temsil eder) bulmak için biraz temizlik yapar. Filmleri kimliğe göre gruplandırır. Böylece kullanıcı, yinelenen film girişleri görmez.
Film Ayrıntılarını Alma
Bu araç, belirli bir sinemadaki belirli bir filme ait tüm katalog ayrıntılarını alır.
tools.py uygulamasına ekleme:
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)
Bu işlem ne yapar?
- Belirli
movie_iddeğerini ileterek satıcının MCP uç noktasındalookup_catalogyöntemini çağırır. Bu işlem, söz konusu sinemayla ilgili gösterim saatleri ve koltuk durumu gibi ayrıntılı bilgileri döndürür.
6. Aracı araçlarını oluşturma: Ödeme
Bu araçlar, ödeme ve satın alma sürecini yönetir. Güvenli işlemler için AP2 protokolü kullanılır.
Ödeme oluşturma
Bu araç, belirli bir gösterim zamanı için belirli bir sinemada ödeme oturumu başlatır.
tools.py uygulamasına ekleme:
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)
Bu işlem ne yapar?
- Satıcının MCP uç noktasında
create_checkoutyöntemini çağırır. - Kullanıcının istediği
showtime_idvequantitydeğerlerini iletir. - Satıcı, AP2 CartMandate içeren bir JSON nesnesi döndürür.
Not: Yanıt verilerini incelerseniz ap2.cart_mandate içinde merchant_authorization alanı bulunur. Bu, satıcının belirtilen fiyatı kilitleyen kriptografik imzasıdır. Bu ayar daha sonra değiştirilemez.
Satın Alma İşlemini Tamamlayın
Bu araçta üç şey olur:
- Ödeme adımından CartMandate'i alır ve doğrularız (mock).
- PaymentMandate (mock) oluşturun ve imzalayın.
- Satın alma işlemini tamamlamak için imzalı yetki belgesini satıcıya gönderin.
tools.py uygulamasına ekleme:
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)
Neden iki zorunlu kılma? CartMandate, satıcının fiyatı teklif ettikten sonra değiştirmesini engeller. PaymentMandate, aracının kullanıcının izni olmadan ödeme almasını engeller. Akış:
Merchant locks price -> User authorizes charge -> Merchant verifies both -> Order completes.
Kontrol noktası: Tam tools.py
Tam tools.py artık 5 araç işlevine ve UCP ile AP2 istemcileri için modül düzeyinde başlatmaya sahip olmalıdır. Şuna benzediğini doğrulayın:
"""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. Çalıştırılabilir hale getirme
Gerçek UCP satıcılarında, temsilciyi URL'lerine yönlendirmeniz yeterli olur. Bu codelab'de yerel olarak test etmek için iki şeye ihtiyacımız var:
- Sahte satıcılar: Test edebileceğiniz bir şey olması için UCP uç noktalarını simüle eden yerel sunucular
- Protokol yardımcıları: UCP ve AP2 için ince HTTP sarmalayıcıları (üretimde, resmi SDK'lar bunların yerini alır)
Not: Bu kodu dikkatlice okumanız gerekmez. Bu dosyalar, gerçek altyapı ve SDK'ların sağlayacağı özellikleri simüle eder. Bunları olduğu gibi kopyalayın.
Protokol yardımcıları
UCP ve AP2'nin henüz istemci SDK'ları yoktur. Bu iki dosya, HTTP bağlantısını yönetir.
Oluşturma 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()
Oluşturma 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()
Sahte satıcılar
Oluşturma 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
Bu işlem, her biri iki uç nokta içeren iki FastAPI sunucusu oluşturur:
GET /.well-known/ucp: UCP keşfi. Satıcının özelliklerini, MCP uç nokta URL'sini ve kabul edilen ödeme yöntemlerini döndürür.POST /mcp: MCP (Model Context Protocol) işlemleri. Katalog arama, ödeme, indirimler ve ödeme için JSON-RPC 2.0 çağrılarını işler.
Satıcıları yeni bir terminalde başlatın. Satıcıların çalışmaya devam etmesi gerekir:
cd agent_payments
source .venv/bin/activate
python merchants.py
Aşağıdaki bilgileri görürsünüz:
Merchants running: Meridian (:8081), StarLight (:8082)
İlk terminalinize dönerek UCP keşfini doğrulayın:
curl -s http://localhost:8081/.well-known/ucp | python -m json.tool
Satıcının özellikleri, MCP uç nokta URL'si ve ödeme işleyicileri gösterilir.
8. ADK Web ile temsilciyi çalıştırma
ADK CLI'nın yerleşik Web kullanıcı arayüzünü kullanalım. Bu, tarayıcıda bir sohbet arayüzü sağlar ve araç onay istemlerini otomatik olarak işler.
Projeniz artık aşağıdaki gibi görünecektir:
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
Dene
Mevcut terminalde, üst dizine (agent_payments'nın bir klasör üstü) gidin ve ADK Web kullanıcı arayüzünü başlatın:
cd ../
adk web --allow_origins '*'
Sunucunun çalıştığını belirten bir çıkış görmeniz gerekir. Bu çıkışta bir URL (genellikle http://localhost:8000 veya benzeri) verilir.
Temsilciyle konuşma
adk webtarafından sağlanan URL'yi tarayıcınızda açın.- Bir sohbet arayüzü görürsünüz.
- "Hangi filmler oynuyor?" diye sormayı deneyin.
- Aracı, perde arkasında sinemaları keşfeder ve kataloglarda arama yapar. Sonuçları, UCP aracılığıyla her iki satıcıdan da toplar.
- Bilet rezervasyonu yapmasını isteyin: "Oppenheimer için saat 19:00'da 2 bilet ayırt".
- Temsilci
complete_purchasenumarayı aramaya çalıştığında ADK Web kullanıcı arayüzünün nasıl bir onay iletişim kutusu veya kart gösterdiğine dikkat edin. - İşlemi yetkilendirmek için sohbette şu JSON dizesiyle yanıt verin:
{"confirmed": true}. - Temsilci, satın alma işlemini tamamlayıp bilet kodlarıyla birlikte sipariş onayınızı size geri gönderir.
Kamera arkasında yaşananlar:
create_checkout→ Satıcı, AP2 CartMandate (imzalı fiyat kilidi) döndürdü.complete_purchase→ PaymentMandate oluşturdu, imzaladı (sahte SHA-256) ve her iki yetkiyi de satıcıya gönderdi.- Satıcı her iki imzayı da doğruladı → biletler düzenlendi (mock'ta)
9. Temizleme
Yerel sunucuların çalışır durumda kalmasını önlemek için kaynakları temizleyin:
adk webkomutunun çalıştırıldığı terminalde, aracı sunucusunu durdurmak için Ctrl+C tuşlarına basın.python merchants.pykomutunun çalıştırıldığı terminalde, sahte satıcıları durdurmak için Ctrl+C tuşlarına basın.- Aşağıdaki komutu çalıştırarak sanal ortamı her iki terminalde de devre dışı bırakın:
deactivate
- (İsteğe bağlı) Bu codelab için yeni bir Google Cloud projesi oluşturduysanız ve bu projeyi silmek istiyorsanız aşağıdaki komutu çalıştırın:
gcloud projects delete $GOOGLE_CLOUD_PROJECT
10. Tebrikler! 🎉
UCP ve AP2 kullanarak satıcıları keşfeden, kataloglara göz atan ve satın alma işlemlerini tamamlayan bir ADK aracısı oluşturdunuz.
Öğrendikleriniz
Bu codelab'de, güvenli ticaret akışlarını işleyen bir ADK aracısı oluşturmuştunuz. Oluşturduğunuz öğeler ve uyguladığınız temel kavramların özeti aşağıda verilmiştir:
Oluşturduğunuz içerik:
- UCP ve AP2 işlemlerini (keşif, katalog araması, ödeme, ödeme) kapsayan 5 aracı.
- AP2 talimat imzalama: CartMandate (satıcı fiyatı kilidi) + PaymentMandate (kullanıcı yetkilendirmesi).
- Çok satıcılı arama: Tek bir temsilci birden fazla tiyatroya sorgu gönderir ve sonuçları birleştirir.
Temel Kavramlar:
Protokol | Ne işe yarar? | İşleyiş şekli |
UCP Discovery | Ajan, satıcıları ve özelliklerini bulur. |
|
UCP MCP | Ajan, kataloglara göz atar ve ödeme işlemleri oluşturur. | Satıcının MCP uç noktasına yapılan JSON-RPC 2.0 çağrıları |
AP2 CartMandate | Satıcı, belirtilen fiyatı sabitler | Satıcı tarafından imzalanır, toplam tutar ve son kullanma tarihi içerir. |
AP2 PaymentMandate | Kullanıcı ödemeyi yetkilendirir | Kullanıcı tarafından imzalanmış, CartMandate'i referans alıyor |
Üretimde farklı olan nedir?
Bu codelab'de sahte veriler kullanılır. Üretimde:
- UCP keşfi, sabit kodlanmış localhost URL'leri yerine bir kayıt defterine göre çözümlenir.
- MCP uç noktaları gerçek satıcılar tarafından barındırılır. Aynı JSON-RPC 2.0 protokolü, gerçek envanter
- AP2 zorunlulukları, SHA-256 karma değerleriyle değil, sd-jwt-vc ile imzalanır.
- Ödeme yetkilendirme, kullanıcı izni istemleri içeren bir AP2 Cüzdan SDK'sı kullanır.
- Ön uç, araç sonuçlarını zengin kullanıcı arayüzü (ürün ızgaraları, ödeme özetleri, onay kartları) olarak oluşturur.
Sonraki adımlar
- AP2 protokol spesifikasyonunu inceleyin ve ödeme özellikli kendi aracınızı oluşturun
- Ajan tabanlı ticareti etkinleştirmek için satıcınızda UCP'yi uygulayın
- Çok temsilcili ticari iş akışları için AP2'yi A2A ile bağlama
- Kripto para ödemeleri için AP2 x402 uzantısı hakkında bilgi