تأمين التجارة باستخدام "الوكيل الآمن" من خلال AP2 وبروتوكول UCP

1. نظرة عامة

في هذا الدرس التطبيقي حول الترميز، ستشغّل وكيل ADK يحجز تذاكر سينما لدى عدة تجّار باستخدام بروتوكولَي تجارة مفتوحَي المصدر:

• بروتوكول Universal Commerce Protocol (UCP): هو معيار يتيح للوكلاء العثور على التجّار والبحث في الكتالوجات وإدارة عمليات الدفع.
• بروتوكول AP2 (بروتوكول دفعات الوكيل): هو بروتوكول لتفويض الدفع الآمن الذي يمكن التحقّق منه باستخدام تفويضات موقّعة بالتشفير.

يتصل التطبيق التجريبي، 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

قبل أن نتعمّق في إنشاء الوكيل، دعونا نتعرّف على البروتوكولَين اللذين يتيحان هذه التجارة الآمنة المستندة إلى الوكلاء.

بروتوكول Universal Commerce Protocol (UCP)

يضع بروتوكول UCP معايير موحّدة لطريقة تفاعل وكلاء الذكاء الاصطناعي مع التجّار. يحلّ هذا الإطار مشكلة اضطرار الوكلاء إلى تعلُّم واجهات برمجة تطبيقات مخصّصة لكل متجر من خلال تقديم نموذج موارد موحّد.

طريقة العمل:

1. الاكتشاف: يعرض كل تاجر متوافق مع UCP ملفًا شخصيًا في موقع عادي: /.well-known/ucp. مثال: نقطة نهاية UCP في Everlane.عندما يقرأ وكيل هذا الملف الشخصي، يبحث عن:
   • الإمكانات: الميزات الأساسية المستقلة التي يوفّرها النشاط التجاري، مثل البحث في الكتالوج أو الدفع.
   • الخدمات: طبقات الاتصال ذات المستوى الأدنى المستخدَمة لتبادل البيانات أمثلة: REST API، وMCP (بروتوكول سياق النموذج)، وA2A (بروتوكول Agent2Agent).
   • الإضافات: إذا كان التاجر بحاجة إلى سلوك متخصص، يمكنه تحديد إضافات مخصّصة في هذا الملف.
2. العمليات: بعد اكتشاف نقطة نهاية الخدمة، يستخدمها الوكيل لتنفيذ العمليات. في هذا الدرس التطبيقي حول الترميز، نستخدم بروتوكول Model Context Protocol (MCP) كوسيلة نقل الخدمة. يجري الوكيل عمليات استدعاء JSON-RPC 2.0 إلى نقطة النهاية هذه لتفعيل الإمكانات التي تم اكتشافها، مثل البحث عن المنتجات وإنشاء عمليات دفع وإكمال عمليات الشراء.

بروتوكول دفعات الوكلاء (AP2)

تعمل واجهة برمجة التطبيقات AP2 على توحيد طريقة تفويض وكلاء الدفع نيابةً عن المستخدمين. يحلّ هذا الإجراء مشكلة الأمان المتعلّقة بتعامل الموظفين مع بيانات اعتماد الدفع الحساسة.

طريقة العمل:

1. تفويض عربة التسوّق: عندما ينشئ وكيل عملية دفع باستخدام بروتوكول UCP، يعرض التاجر الرمز CartMandate. هذا هو عنصر JSON يحتوي على تفاصيل سلّة التسوّق وتوقيع تشفير من التاجر. وهي تعمل كضمان لقفل السعر. لا يمكن للتاجر تغيير السعر بعد إصدار تفويض الدفع هذا.
2. تفويض الدفع: بعد التحقّق من محتويات سلّة التسوّق، ينشئ المستخدم (أو الوكيل نيابةً عن المستخدم) PaymentMandate لتفويض الدفع. يشير PaymentMandate إلى CartMandate ويتضمّن التوقيع المشفر للمستخدم (أو رمز التخويل).
3. التحقّق من التوقيع المزدوج: يتلقّى التاجر كلا التفويضَين. يتحقّق من توقيعه على CartMandate وتوقيع المستخدم على PaymentMandate. إذا كان كلاهما صالحًا، تتم المعاملة.

يضمن نظام "القفل المزدوج" هذا ألا يفرض التجّار رسومًا مفرطة وألا ينفق الوكلاء بدون إذن. في مرحلة الإنتاج، تستخدم هذه التفويضات SD-JWT (Selective Disclosure JWT) لحماية خصوصية المستخدم.

3- إعداد البيئة

إعداد مشروع Google Cloud

إنشاء مشروع على Google Cloud

1. في Google Cloud Console، في صفحة اختيار المشروع، اختَر مشروعًا على Google Cloud أو أنشِئ مشروعًا.
2. تأكَّد من تفعيل الفوترة لمشروعك على السحابة الإلكترونية. كيفية التحقّق مما إذا كانت الفوترة مفعَّلة في مشروع

بدء Cloud Shell

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

تثبيت الحِزم التابعة

يتم تثبيت uv مسبقًا في Google Cloud Shell لإدارة البيئة والملحقات.

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

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:

الأداة	وظيفتها	إجراء البروتوكول
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.

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 الخاصة بالتاجر.

اقتراحات المسارح

هذه الأداة هي الخطوة الأولى في مسار تجربة المستخدم الموحّدة. تعثر هذه الأداة على التجّار المتوفّرين والخدمات التي يقدّمونها.

الإضافة إلى 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. تتكرّر هذه العملية على قائمة بعناوين URL الخاصة بالتجّار يقدّمها الخادم. في هذا الدرس التطبيقي حول الترميز، سنعدّ متجرَين وهميَّين في القسم اللاحق.
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 المحدّد. يعرض هذا الطلب معلومات مفصّلة، مثل أوقات العرض ومدى توفّر المقاعد في قاعة السينما المحدّدة.

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)

وظيفة هذا الإعداد:

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)

لماذا يجب تقديم تفويضَين؟ يضمن 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 الخاصة بهم، وبذلك تكون قد انتهيت. في هذا الدرس التطبيقي حول الترميز، نحتاج إلى عنصرَين لاختبار التطبيق على الجهاز المحلي:

1. التجّار التجريبيون: خوادم محلية تحاكي نقاط نهاية UCP لتوفير عناصر يمكنك اختبارها
2. أدوات مساعدة البروتوكول: هي أغلفة HTTP بسيطة لبروتوكولَي UCP وAP2 (في مرحلة الإنتاج، تحلّ حِزم SDK الرسمية محلّ هذه الأدوات)

ملاحظة: لست بحاجة إلى قراءة هذا الرمز بعناية. تحاكي هذه الملفات ما توفّره البنية الأساسية وحِزم تطوير البرامج الحقيقية. يجب نسخها كما هي.

تطبيقات مساعدة في البروتوكولات

لا يتوفّر حاليًا حِزم تطوير برامج (SDK) للعملاء في UCP وAP2، إذ يتولّى هذان الملفان معالجة عمليات 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: إعلانات أثناء التصفّح في "حملات التطبيقات العامة" تعرض هذه السمة إمكانات التاجر وعنوان 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 ومعالجات الدفع.

8. تشغيل الوكيل باستخدام ADK Web

لنستخدِم واجهة مستخدم الويب المضمّنة في واجهة سطر الأوامر (CLI) الخاصة بـ ADK. يوفّر ذلك واجهة محادثة في المتصفّح ويتعامل تلقائيًا مع طلبات تأكيد الأدوات.

يُفترض أن يبدو مشروعك الآن على النحو التالي:

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. افتح عنوان URL الذي يوفّره adk web في المتصفّح.
2. ستظهر لك واجهة محادثة.
3. جرِّب طرح السؤال: "ما هي الأفلام المعروضة حاليًا؟"
4. سيكتشف الوكيل دور السينما ويبحث في الكتالوجات وراء الكواليس، ويجمع النتائج من كلا التجّار من خلال بروتوكول UCP.
5. اطلب حجز تذاكر: "أريد حجز تذكرتَين لفيلم Oppenheimer في الساعة 7 مساءً".
6. عندما يحاول الوكيل الاتصال بـ complete_purchase، لاحظ كيف تظهر نافذة تأكيد أو بطاقة في واجهة مستخدم الويب الخاصة بـ ADK.
7. لإتمام المعاملة، يُرجى الرد في المحادثة باستخدام سلسلة JSON هذه بالضبط: {"confirmed": true}.
8. سيكمل الوكيل عملية الشراء وسيرسل إليك تأكيد الطلب مع رموز التذاكر.

إليك ما حدث وراء الكواليس:

1. ‫create_checkout → ردّ التاجر على طلب AP2 CartMandate (قفل السعر الموقّع)
2. complete_purchase → أنشأ PaymentMandate ووقّعه (SHA-256 وهمي)، وأرسل كلا التفويضين إلى التاجر
3. تحقّق التاجر من كلا التوقيعَين → إصدار التذاكر (في النموذج التجريبي)

9- تَنظيم

لتجنُّب ترك الخوادم المحلية قيد التشغيل، عليك تنظيف الموارد باتّباع الخطوات التالية:

1. في الوحدة الطرفية التي يتم فيها تشغيل adk web، اضغط على Ctrl+C لإيقاف خادم الوكيل.
2. في الوحدة الطرفية التي يتم فيها تشغيل python merchants.py، اضغط على Ctrl+C لإيقاف التجار التجريبيين.
3. أوقِف البيئة الافتراضية في كلتا نافذتَي Terminal من خلال تنفيذ الأمر التالي:

deactivate

4. (اختياري) إذا أنشأت مشروعًا جديدًا على Google Cloud لهذا الدرس التطبيقي حول الترميز وأردت حذفه، نفِّذ ما يلي:

gcloud projects delete $GOOGLE_CLOUD_PROJECT

10. تهانينا! 🎉

لقد أنشأت وكيل ADK يتيح للمستخدمين العثور على التجّار وتصفّح الكتالوجات وإكمال عمليات الشراء باستخدام UCP وAP2.

ما تعلّمته

في هذا الدرس التطبيقي حول الترميز، أنشأت وكيل ADK يتعامل مع مسارات التجارة الآمنة. في ما يلي ملخّص لما أنشأته والمفاهيم الأساسية التي طبّقتها:

ما أنشأته:

• 5 أدوات وكيل لتنفيذ عمليات UCP وAP2، مثل الاستكشاف والبحث في الفهرس والدفع
• توقيع تفويض AP2: CartMandate (تثبيت سعر التاجر) + PaymentMandate (موافقة المستخدم)
• البحث لدى عدة تجار: يرسل وكيل واحد طلب بحث إلى عدة مسارح ويدمج النتائج.

المفاهيم الأساسية:

البروتوكول	وظيفتها	آلية العمل
إعلانات UCP Discovery	يعثر الوكيل على التجّار وإمكاناتهم	/.well-known/ucp → الإمكانات ونقطة نهاية MCP وطرق الدفع
UCP MCP	يتصفّح الوكيل الكتالوجات وينشئ عمليات دفع	طلبات JSON-RPC 2.0 إلى نقطة نهاية MCP الخاصة بالتاجر
AP2 CartMandate	يثبّت التاجر السعر المحدّد	موقّع من التاجر، ويتضمّن الإجمالي وتاريخ انتهاء الصلاحية
AP2 PaymentMandate	يوافق المستخدم على الرسوم	موقّع من المستخدم، يشير إلى CartMandate

ما هي الاختلافات في مرحلة الإنتاج؟

يستخدم هذا الدرس التطبيقي نماذج محاكاة. في مرحلة الإنتاج:

• يتم حلّ استكشاف UCP من خلال سجلّ، وليس من خلال عناوين URL ثابتة على المضيف المحلي.
• تستضيف الجهات التجارية الحقيقية نقاط نهاية MCP، وتستخدم بروتوكول JSON-RPC 2.0 نفسه والمستودع الفعلي.
• يتم توقيع مستندات AP2 باستخدام sd-jwt-vc، وليس تجزئات SHA-256.
• تستخدم تفويض الدفع حزمة تطوير برامج (SDK) خاصة بمحفظة AP2 مع طلبات موافقة المستخدم.
• تعرض واجهة المستخدم نتائج الأداة على شكل واجهة مستخدم غنية (شبكات المنتجات، وملخّصات الدفع، وبطاقات التأكيد).

الخطوات التالية

• استكشاف مواصفات بروتوكول AP2 وإنشاء وكيلك الخاص الذي يتيح الدفع
• تنفيذ UCP للتاجر لتفعيل التجارة المستندة إلى الوكلاء
• ربط AP2 ببروتوكول A2A لتنفيذ سير عمل التجارة المتعددة الوكلاء
• مزيد من المعلومات حول إضافة AP2 x402 لدفعات العملات المشفّرة

المستندات المرجعية

• مستندات بروتوكول AP2
• دليل المطوّر لبروتوكول UCP
• مستودع GitHub الخاص بـ AP2
• مستندات حزمة تطوير التطبيقات على Android
• مواصفات بروتوكول UCP