تجارت امن با نمایندگان با AP2 و UCP

۱. مرور کلی

در این آزمایشگاه کد، شما یک عامل ADK را اجرا خواهید کرد که بلیط‌های فیلم را در چندین سینما با استفاده از دو پروتکل تجاری متن‌باز رزرو می‌کند:

• UCP (پروتکل تجارت جهانی) : استانداردی برای نمایندگان جهت کشف فروشندگان، جستجوی کاتالوگ‌ها و مدیریت جریان‌های پرداخت.
• AP2 (پروتکل پرداخت‌های عامل) : پروتکلی برای مجوز پرداخت امن و قابل تأیید با استفاده از دستورات امضا شده رمزنگاری شده.

اپلیکیشن آزمایشی CineAgent به دو فروشنده‌ی سینماهای آزمایشی با قابلیت‌های مختلف (انتخاب صندلی، قالب‌های تخصصی و روش‌های پرداخت) متصل می‌شود و کل فرآیند رزرو را از جستجو تا پرداخت هماهنگ می‌کند.

آنچه یاد خواهید گرفت

• نحوه‌ی کشف تاجر UCP از طریق پروفایل‌های /.well-known/ucp چگونه است؟
• چگونه یک نماینده ADK از UCP برای جستجوی کاتالوگ‌ها و ایجاد صندوق‌ها استفاده می‌کند
• چگونه AP2 (CartMandate، PaymentMandate) تراکنش‌ها را ایمن می‌کند
• نحوه عملکرد پروتکل‌های سرتاسری UCP و AP2 برای ایمن‌سازی تجارت الکترونیک عامل‌محور

آنچه نیاز دارید

• یک پروژه گوگل کلود با قابلیت پرداخت صورتحساب
• یک مرورگر وب مانند کروم
• پایتون ۳.۱۱+

این آزمایشگاه کد برای توسعه‌دهندگان سطح متوسط ​​است که با پایتون و گوگل کلود آشنایی دارند. تکمیل این آزمایشگاه کد تقریباً ۱۵ دقیقه طول می‌کشد.

منابع ایجاد شده در این آزمایشگاه کد باید کمتر از ۵ دلار هزینه داشته باشند.

۲. درک پروتکل‌های UCP و AP2

قبل از اینکه به ساخت عامل بپردازیم، بیایید دو پروتکلی را که این تجارت عامل‌محور امن را ممکن می‌سازند، درک کنیم.

پروتکل تجارت جهانی ( UCP )

UCP نحوه تعامل عوامل هوش مصنوعی با فروشندگان را استاندارد می‌کند. این استاندارد با معرفی یک مدل منبع استاندارد، مشکل نیاز عوامل به یادگیری APIهای سفارشی برای هر فروشگاه را حل می‌کند.

چگونه کار می‌کند:

1. کشف : هر فروشنده‌ای که از UCP پیروی می‌کند، یک پروفایل را در یک مکان استاندارد قرار می‌دهد: /.well-known/ucp . مثال: نقطه پایانی UCP در Everlane . وقتی یک نماینده این پروفایل را می‌خواند، به دنبال موارد زیر می‌گردد:
   • قابلیت‌ها : ویژگی‌های اصلی مستقل که یک کسب‌وکار از آنها پشتیبانی می‌کند، مانند جستجو در کاتالوگ یا پرداخت.
   • سرویس‌ها : لایه‌های ارتباطی سطح پایین‌تر که برای تبادل داده‌ها استفاده می‌شوند. مثال‌ها: REST API، MCP (پروتکل زمینه مدل)، A2A (پروتکل عامل به عامل)
   • افزونه‌ها : اگر یک فروشنده به رفتار خاصی نیاز داشته باشد، می‌تواند افزونه‌های سفارشی را در این پروفایل تعریف کند.
2. عملیات : پس از کشف، عامل از نقطه پایانی سرویس ارائه شده برای انجام عملیات استفاده می‌کند. در این آزمایشگاه کد، ما از پروتکل زمینه مدل (MCP) به عنوان انتقال سرویس استفاده می‌کنیم. عامل فراخوانی‌های JSON-RPC 2.0 را به این نقطه پایانی انجام می‌دهد تا قابلیت‌های کشف شده را فراخوانی کند: جستجوی محصولات، ایجاد صندوق‌ها و تکمیل خریدها.

پروتکل پرداخت‌های عامل ( AP2 )

AP2 نحوه‌ی تأیید پرداخت‌ها توسط نمایندگان به نمایندگی از کاربران را استانداردسازی می‌کند. این استاندارد مشکل امنیتی مربوط به مدیریت اعتبارنامه‌های حساس پرداخت توسط نمایندگان را حل می‌کند.

چگونه کار می‌کند:

1. دستور سبد خرید : وقتی یک نماینده با استفاده از پروتکل UCP یک پرداخت ایجاد می‌کند، فروشنده یک CartMandate برمی‌گرداند. این یک شیء JSON است که شامل جزئیات سبد خرید و یک امضای رمزنگاری شده از فروشنده است. این به عنوان ضمانت قفل قیمت عمل می‌کند. فروشنده پس از صدور این دستور نمی‌تواند قیمت را تغییر دهد.
2. دستور پرداخت : پس از تأیید محتویات سبد خرید، کاربر (یا نماینده از طرف کاربر) یک PaymentMandate برای تأیید پرداخت ایجاد می‌کند. این PaymentMandate به CartMandate ارجاع می‌دهد و شامل امضای رمزنگاری کاربر (یا توکن مجوز) است.
3. تأیید امضای دوگانه : فروشنده هر دو دستور را دریافت می‌کند. آنها امضای خود را در CartMandate و امضای کاربر را در PaymentMandate تأیید می‌کنند. اگر هر دو معتبر باشند، تراکنش انجام می‌شود.

این سیستم «قفل دوگانه» تضمین می‌کند که بازرگانان نمی‌توانند بیش از حد هزینه دریافت کنند و نمایندگان نمی‌توانند بدون مجوز هزینه کنند. در عمل، این دستورالعمل‌ها از SD-JWT (Selective Disclosure JWT) برای محافظت از حریم خصوصی کاربر استفاده می‌کنند.

۳. محیط خود را آماده کنید

راه‌اندازی پروژه گوگل کلود

ایجاد یک پروژه ابری گوگل

1. در کنسول گوگل کلود ، در صفحه انتخاب پروژه، یک پروژه گوگل کلود را انتخاب یا ایجاد کنید .
2. مطمئن شوید که صورتحساب برای پروژه ابری شما فعال است. یاد بگیرید که چگونه بررسی کنید که آیا صورتحساب در یک پروژه فعال است یا خیر .

شروع پوسته ابری

Cloud Shell یک محیط خط فرمان است که در Google Cloud اجرا می‌شود و ابزارهای لازم از قبل روی آن بارگذاری شده‌اند.

1. روی فعال کردن Cloud Shell در بالای کنسول Google Cloud کلیک کنید.
2. پس از اتصال به Cloud Shell، احراز هویت خود را تأیید کنید:

   gcloud auth list
   

3. تأیید کنید که پروژه شما پیکربندی شده است:

   gcloud config get project
   

4. اگر پروژه شما مطابق انتظار تنظیم نشده است، آن را تنظیم کنید:

   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 برای مدیریت محیط و وابستگی‌ها نصب شده است.

1. یک فایل 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",
]

2. دستور زیر را برای ایجاد محیط مجازی و نصب تمام وابستگی‌ها اجرا کنید:

uv sync

3. محیط مجازی ایجاد شده توسط uv را فعال کنید:

source .venv/bin/activate

۴. عامل را تعریف کنید

قبل از نوشتن هرگونه منطق ابزار، بیایید خود عامل را در فایلی به نام 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 : این لیستی از توابع پایتون (که در ادامه خواهیم ساخت) است که عامل می‌تواند بر اساس درخواست‌های کاربر، آنها را فراخوانی کند.
• require_confirmation : می‌توانید هر ابزاری را با FunctionTool(my_function,require_confirmation=True) پوشش دهید. وقتی فعال می‌شود، عامل مکث می‌کند و قبل از اجرای ابزار، منتظر یک تأیید ساده "بله" یا "خیر" می‌ماند. در اینجا، قبل از اجرای ابزار complete_purchase ، عامل برای تأیید انسانی مکث می‌کند.

فهرست ابزارها

تعریف عامل، آنچه را که باید بسازیم، اعلام می‌کند. هر ابزار به یک عملیات خاص در پروتکل UCP یا AP2 نگاشت می‌شود:

ابزار	چه کاری انجام می‌دهد؟	اقدام پروتکلی
discover_theaters	یافتن بازرگانان و قابلیت‌های آنها	کوئری‌های /.well-known/ucp
search_movies	جستجوی کاتالوگ‌ها در بین فروشگاه‌ها	JSON-RPC به نقطه پایانی MCP
get_movie_detail	دریافت زمان‌های نمایش در یک فروشگاه خاص	JSON-RPC به نقطه پایانی MCP
create_checkout	شروع جلسه تسویه حساب	JSON-RPC به نقطه پایانی MCP
complete_purchase	پرداخت را تأیید کنید و سفارش را تکمیل کنید	امضای حکم AP2 و ارسال آن به MCP

مدل Gemini بر اساس مکالمه تصمیم می‌گیرد که چه زمانی هر ابزار را فراخوانی کند . کار بعدی ما پیاده‌سازی عملکرد هر ابزار در tools.py است.

۵. ابزارهای عامل را بسازید: کشف و مرور

حالا بیایید ابزارهایی را که عامل برای مرور و کشف فیلم‌ها استفاده خواهد کرد، پیاده‌سازی کنیم. هر ابزار یک عملیات 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)

این چه کاری انجام می‌دهد :

1. این کد، فهرستی از آدرس‌های اینترنتی فروشگاه‌های ارائه شده توسط سرور را مرور می‌کند. در این آزمایشگاه کد، در بخش بعدی دو فروشگاه آزمایشی راه‌اندازی خواهیم کرد.
2. برای هر URL، تابع _ucp.discover(url) را فراخوانی می‌کند که به نقطه پایانی /.well-known/ucp برخورد می‌کند.
3. این برنامه نام، قابلیت‌ها و گردانندگان پرداخت را در یک لیست خلاصه جمع‌آوری می‌کند.
4. این لیست را به عنوان یک رشته 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)

این چه کاری انجام می‌دهد :

1. این حلقه تمام تئاترهایی را که کشف شده‌اند، طی می‌کند.
2. این یک درخواست JSON-RPC کاتالوگ جستجو ( _ucp.mcp_call(url, "search_catalog", {"query": query}) ) را به نقطه پایانی MCP فروشنده ارسال می‌کند.
3. سپس کمی پاکسازی انجام می‌دهد تا نتایج را تجزیه کند و فیلم‌ها و «انواع» آنها (که نشان‌دهنده زمان نمایش و فرمت‌های خاص هستند) را پیدا کند. فیلم‌ها را بر اساس شناسه گروه‌بندی می‌کند تا کاربر ورودی‌های تکراری فیلم را نبیند.

جزئیات فیلم را دریافت کنید

این ابزار جزئیات کامل کاتالوگ یک فیلم خاص را در یک سینمای خاص دریافت می‌کند.

به 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 خاص را ارسال می‌کند. این متد اطلاعات دقیقی مانند زمان نمایش‌ها و در دسترس بودن صندلی برای آن سینمای خاص را برمی‌گرداند.

۶. ابزارهای نمایندگی را بسازید: پرداخت و تسویه حساب

این ابزارها جریان پرداخت و خرید را مدیریت می‌کنند. اینجاست که پروتکل 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)

این چه کاری انجام می‌دهد :

1. این متد create_checkout را در نقطه پایانی MCP فروشنده فراخوانی می‌کند.
2. این تابع showtime_id و quantity درخواستی کاربر را ارسال می‌کند.
3. فروشنده یک شیء JSON حاوی AP2 CartMandate را برمی‌گرداند.

نکته : اگر داده‌های پاسخ را بررسی کنید، ap2.cart_mandate حاوی یک فیلد merchant_authorization است. این امضای رمزنگاری‌شده‌ی فروشنده است که قیمت اعلام‌شده را قفل می‌کند. آنها نمی‌توانند بعداً آن را تغییر دهند!

خرید کامل

سه اتفاق در این ابزار رخ می‌دهد:

1. ما CartMandate را از پرداخت دریافت می‌کنیم و آن را تأیید می‌کنیم (شبیه‌سازی می‌کنیم) .
2. یک فرم PaymentMandate (آزمایشی) ایجاد و امضا کنید.
3. برای تکمیل خرید، فرم امضا شده را برای فروشنده ارسال کنید.

به 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)

چرا دو دستور؟ دستور سبد خرید تضمین می‌کند که فروشنده پس از ارائه قیمت، نمی‌تواند آن را تغییر دهد. دستور پرداخت تضمین می‌کند که نماینده نمی‌تواند بدون رضایت کاربر، هزینه‌ای از او دریافت کند. روند کار به این صورت است:

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

نقطه بازرسی: tools.py کامل.py

فایل کامل tools.py شما اکنون باید شامل ۵ تابع ابزار و مقداردهی اولیه در سطح ماژول برای کلاینت‌های 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)

۷. آن را قابل اجرا کنید

با فروشندگان واقعی UCP، شما آدرس URL آنها را به نماینده نشان می‌دهید و کار تمام است. برای این آزمایشگاه کد، ما به دو چیز برای آزمایش محلی نیاز داریم:

1. فروشندگان آزمایشی - سرورهای محلی که نقاط پایانی UCP را شبیه‌سازی می‌کنند تا شما چیزی برای آزمایش داشته باشید
2. کمک‌کننده‌های پروتکل - پوشش‌دهنده‌های 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 (پروتکل زمینه مدل). فراخوانی‌های 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 و کنترل‌کننده‌های پرداخت را ببینید.

۸. اجرای Agent با 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 یا مشابه آن).

با نماینده صحبت کنید

1. آدرس اینترنتی ارائه شده توسط adk web را در مرورگر خود باز کنید.
2. رابط چت را مشاهده خواهید کرد.
3. سعی کنید بپرسید: «چه فیلم‌هایی در حال پخش هستند؟»
4. این نماینده، سینماها را کشف کرده و کاتالوگ‌های پشت صحنه را جستجو می‌کند و نتایج هر دو فروشنده را از طریق UCP جمع‌آوری می‌کند.
5. برای رزرو بلیط بپرسید: «برای ساعت ۷ بعد از ظهر، ۲ بلیط برای اوپنهایمر رزرو کنید» .
6. وقتی عامل سعی می‌کند تابع complete_purchase را فراخوانی کند، توجه کنید که رابط کاربری وب ADK چگونه یک کادر محاوره‌ای یا کارت تأیید را نمایش می‌دهد!
7. برای تأیید تراکنش، در چت دقیقاً با این رشته JSON پاسخ دهید: {"confirmed": true} .
8. نماینده خرید را تکمیل کرده و تأیید سفارش شما را به همراه کدهای بلیط برمی‌گرداند!

در پشت صحنه چه اتفاقی افتاد:

1. create_checkout → فروشنده یک AP2 CartMandate (قفل قیمت امضا شده) را برگرداند
2. complete_purchase → یک PaymentMandate ایجاد کرد، آن را امضا کرد (SHA-256 ساختگی)، هر دو دستور را برای فروشنده ارسال کرد
3. فروشنده هر دو امضا را تأیید کرد → بلیط‌های صادر شده (در شبیه‌سازی)

۹. تمیز کردن

برای جلوگیری از اجرای سرورهای محلی، منابع را پاک کنید:

1. در ترمینالی که adk web در حال اجرا است، کلیدهای Ctrl+C را فشار دهید تا سرور عامل متوقف شود.
2. در ترمینالی که python merchants.py در حال اجرا است، کلیدهای Ctrl+C را فشار دهید تا نمایش بازرگانان ساختگی متوقف شود.
3. با اجرای دستور زیر، محیط مجازی را در هر دو ترمینال غیرفعال کنید:

deactivate

4. (اختیاری) اگر یک پروژه جدید Google Cloud برای این آزمایشگاه کد ایجاد کرده‌اید و می‌خواهید آن را حذف کنید، دستور زیر را اجرا کنید:

gcloud projects delete $GOOGLE_CLOUD_PROJECT

۱۰. تبریک می‌گویم! 🎉

شما یک عامل ADK ساختید که فروشندگان را کشف می‌کند، کاتالوگ‌ها را مرور می‌کند و با استفاده از UCP و AP2 خریدها را تکمیل می‌کند.

آنچه آموختید

در این آزمایشگاه کد، شما یک عامل ADK ساختید که جریان‌های تجاری امن را مدیریت می‌کند. در اینجا خلاصه‌ای از آنچه ساختید و مفاهیم کلیدی که به کار بردید، آمده است:

چیزی که ساختی:

• ۵ ابزار عامل که عملیات UCP و AP2 را پوشش می‌دهند - کشف، جستجوی کاتالوگ، پرداخت، پرداخت.
• امضای دستور AP2 — CartMandate (قفل قیمت فروشنده) + PaymentMandate (مجوز کاربر).
• جستجوی چند فروشنده - یک نماینده چندین سینما را جستجو می‌کند و نتایج را ادغام می‌کند.

مفاهیم کلیدی:

پروتکل	چه کاری انجام می‌دهد؟	چگونه کار می‌کند؟
کشف UCP	نماینده، بازرگانان و قابلیت‌های آنها را پیدا می‌کند	/.well-known/ucp → قابلیت‌ها، نقطه پایانی MCP، روش‌های پرداخت
UCP MCP	نماینده کاتالوگ‌ها را مرور می‌کند، صندوق‌ها را ایجاد می‌کند	فراخوانی‌های JSON-RPC 2.0 به نقطه پایانی MCP فروشنده
مجوز سبد خرید AP2	فروشنده قیمت اعلام شده را قفل می‌کند	امضا شده توسط فروشنده، شامل جمع کل + تاریخ انقضا
دستور پرداخت AP2	کاربر هزینه را تأیید می‌کند	امضا شده توسط کاربر، ارجاع به CartMandate

چه تفاوتی در تولید وجود دارد؟

این آزمایشگاه کد از mockها استفاده می‌کند. در مرحله‌ی تولید:

• کشف UCP در برابر یک رجیستری، نه URL های محلی میزبان کدگذاری شده، پاسخ می‌دهد.
• نقاط پایانی MCP توسط بازرگانان واقعی میزبانی می‌شوند - همان پروتکل JSON-RPC 2.0، موجودی واقعی
• دستورات AP2 با هش‌های sd-jwt-vc امضا می‌شوند، نه با هش‌های SHA-256
• مجوز پرداخت از یک SDK کیف پول AP2 با درخواست‌های رضایت کاربر استفاده می‌کند
• فرانت‌اند نتایج ابزار را به صورت رابط کاربری غنی (شبکه‌های محصول، خلاصه‌های پرداخت، کارت‌های تأیید) ارائه می‌دهد.

مراحل بعدی

• مشخصات پروتکل AP2 را بررسی کنید و عامل پرداخت خود را بسازید
• برای فعال کردن تجارت نمایندگی، UCP را برای تاجر خود پیاده‌سازی کنید
• اتصال AP2 به A2A برای گردش‌های کاری چندعاملی
• درباره افزونه AP2 x402 برای پرداخت‌های ارز دیجیتال بیشتر بدانید

اسناد مرجع

• مستندات پروتکل AP2
• راهنمای توسعه‌دهندگان UCP
• مخزن گیت‌هاب AP2
• مستندات ADK
• مشخصات UCP