מסחר עם סוכן מאובטח באמצעות AP2 ו-UCP

1. סקירה כללית

בשיעור Codelab הזה נפעיל סוכן ADK שמזמין כרטיסים לסרט אצל כמה מוכרים של כרטיסים לקולנוע באמצעות שני פרוטוקולים מסחריים בקוד פתוח:

• ‫UCP (Universal Commerce Protocol): תקן שמאפשר לסוכנים לגלות מוֹכרים, לחפש קטלוגים ולנהל תהליכי תשלום.
• ‫AP2 (Agent Payments Protocol): פרוטוקול לאישור תשלומים מאובטח וניתן לאימות באמצעות הרשאות חתומות קריפטוגרפית.

אפליקציית ההדגמה, CineAgent, מתחברת לשני מוכרים של כרטיסים לקולנוע עם יכולות שונות (בחירת מושבים, פורמטים מיוחדים ואמצעי תשלום) ומנהלת את תהליך ההזמנה המלא מחיפוש ועד תשלום.

מה תלמדו

• איך מתבצע גילוי מוכרים ב-UCP דרך /.well-known/ucp פרופילים
• איך סוכן ADK משתמש ב-UCP כדי לחפש קטלוגים וליצור דפי תשלום
• איך מנגנון AP2 (CartMandate, ‏ PaymentMandate) מאבטח עסקאות
• איך פרוטוקולי UCP ו-AP2 מקצה לקצה מאבטחים מסחר אלקטרוני באמצעות סוכן

הדרישות

• פרויקט ב-Google Cloud שהחיוב בו מופעל
• דפדפן אינטרנט כמו Chrome
• ‫Python 3.11 ואילך

שיעור ה-Codelab הזה מיועד למפתחים ברמת ביניים שיש להם היכרות מסוימת עם Python ו-Google Cloud. השלמת ה-codelab הזה תיקח בערך 15 דקות.

העלות של המשאבים שנוצרו ב-codelab הזה צריכה להיות פחות מ-5$.

2. הסבר על פרוטוקולי UCP ו-AP2

לפני שנתחיל לבנות את הסוכן, נסביר על שני הפרוטוקולים שמאפשרים מסחר מאובטח באמצעות סוכנים.

Universal Commerce Protocol‏ (UCP)

UCP יוצר סטנדרטיזציה של האינטראקציות בין סוכני AI לבין מוכרים. הוא פותר את הבעיה שסוכנים צריכים ללמוד ממשקי API מותאמים אישית לכל חנות, באמצעות הצגת מודל משאבים סטנדרטי.

איך זה עובד:

1. גילוי: כל מוכר שעומד בדרישות UCP חושף פרופיל במיקום סטנדרטי: /.well-known/ucp. דוגמה: נקודת הקצה של Everlane ב-UCP.כשסוכן קורא את הפרופיל הזה, הוא מחפש:
   • יכולות: תכונות הליבה העצמאיות שהעסק תומך בהן, כמו חיפוש קטלוג או תשלום.
   • שירותים: שכבות התקשורת ברמה הנמוכה שמשמשות להחלפת נתונים. דוגמאות: API בארכיטקטורת REST, ‏ MCP (Model Context Protocol), ‏ A2A (פרוטוקול Agent2Agent).
   • תוספים: אם מוֹכר צריך התנהגות מיוחדת, הוא יכול להגדיר תוספים מותאמים אישית בפרופיל הזה.
2. פעולות: אחרי שהסוכן מאתר את נקודת הקצה של השירות, הוא משתמש בה כדי לבצע פעולות. ב-Codelab הזה אנחנו משתמשים בModel Context Protocol‏ (MCP) כהעברת השירות. הסוכן מבצע קריאות JSON-RPC 2.0 לנקודת הקצה הזו כדי להפעיל את היכולות שזוהו: חיפוש מוצרים, יצירת דפי תשלום והשלמת רכישות.

פרוטוקול Agent Payments Protocol‏ (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, בדף לבחירת הפרויקט, בוחרים פרויקט ב-Google Cloud או יוצרים פרויקט.
2. הקפידו לוודא שהחיוב מופעל בפרויקט שלכם ב-Cloud. כך בודקים אם החיוב מופעל בפרויקט

הפעלת Cloud Shell

‫Cloud Shell היא סביבת שורת פקודה שפועלת ב-Google Cloud וכוללת מראש את הכלים הנדרשים.

1. לוחצים על Activate 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 שסוכן הקולנוע ישתמש בהם.

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

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()

הסבר על ההגדרה

כדי להתמקד בבניית הסוכן ב-Codelab הזה, אנחנו משתמשים בשני מחלקות עזר, UCPClient ו-AP2Handler, שנסביר עליהן בשלב מאוחר יותר.

• מה הם?: אלה מחלקות עזר שכתבנו ידנית במיוחד לשיעור ה-Codelab הזה, כדי לדמות אינטראקציה עם מוכרים מדומים. מאחר ש-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. הוא מבצע איטרציה על רשימה של כתובות URL של מוֹכרים שהשרת מספק. ב-codelab הזה נגדיר שני מוכרים לדוגמה בקטע הבא.
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 מהקופה ומאמתים אותו (mock).
2. יוצרים וחותמים על PaymentMandate (mock).
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 מבטיחה שהמוֹכר לא יוכל לשנות את המחיר אחרי שהוא מציין אותו. ההרשאה לחיוב מבטיחה שהסוכן לא יוכל לחייב את המשתמש ללא הסכמה. התהליך הוא:

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, פשוט מפנים את הנציג לכתובות האתרים שלהם וזהו. כדי לבדוק באופן מקומי את ה-codelab הזה, אנחנו צריכים שני דברים:

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‏ (Model Context Protocol). מטפל בקריאות JSON-RPC 2.0 לחיפוש בקטלוג, לתשלום, להנחות ולתשלום.

מפעילים את המוכרים בטרמינל חדש – הם צריכים להמשיך לפעול:

cd agent_payments
source .venv/bin/activate
python merchants.py

הפרטים שמוצגים הם:

Merchants running: Meridian (:8081), StarLight (:8082)

חוזרים אל הטרמינל הראשון ומאמתים את הגילוי של UCP:

curl -s http://localhost:8081/.well-known/ucp | python -m json.tool

יוצגו לכם היכולות של המוכר, כתובת ה-URL של נקודת הקצה של ה-MCP ומנהלי התשלומים.

8. הפעלת הסוכן באמצעות ADK Web

נשתמש בממשק המשתמש האינטרנטי המובנה ב-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. אפשר לשאול: "What movies are playing?" (אילו סרטים מוצגים עכשיו בקולנוע?)
4. הסוכן יאתר בתי קולנוע ויחפש בקטלוגים מאחורי הקלעים, ויאגד את התוצאות משני המוכרים באמצעות UCP.
5. בקשה להזמנת כרטיסים: "Book 2 tickets for Oppenheimer for 7 PM" (הזמנת 2 כרטיסים לסרט 'אופנהיימר' להקרנה בשעה 19:00).
6. כשנציג מנסה להתקשר אל complete_purchase, שימו לב איך ממשק האינטרנט של ADK מציג תיבת דו-שיח לאישור או כרטיס!
7. כדי לאשר את העסקה, צריך להשיב בצ'אט עם מחרוזת ה-JSON הבאה בדיוק: {"confirmed": true}.
8. הנציג ישלים את הרכישה וישלח לכם אישור הזמנה עם קודי הכרטיסים.

זה מה שקרה מאחורי הקלעים:

1. create_checkout ← המוכר החזיר CartMandate (נעילת מחיר חתומה) של AP2
2. ‫complete_purchase → יצר PaymentMandate, חתם עליה (גיבוב מדמה מסוג SHA-256) ושלח את שתי הרשאות התשלום למוֹכר
3. המוכר אימת את שני החתימות ← הנפיק כרטיסים (בסימולציה)

9. הסרת המשאבים

כדי למנוע הפעלה של שרתים מקומיים, צריך לנקות את המשאבים:

1. בטרמינל שבו פועל adk web, מקישים על Ctrl+C כדי לעצור את שרת הסוכן.
2. בטרמינל שבו פועל python merchants.py, מקישים על Ctrl+C כדי לעצור את המוכרים המדומים.
3. משביתים את הסביבה הווירטואלית בשני הטרמינלים על ידי הפעלת הפקודה:

deactivate

4. (אופציונלי) אם יצרתם פרויקט בענן חדש ב-Google Cloud בשביל ה-Codelab הזה ואתם רוצים למחוק אותו, מריצים את הפקודה:

gcloud projects delete $GOOGLE_CLOUD_PROJECT

10. מעולה! 🎉

יצרתם סוכן ADK שמגלה מוֹכרים, מעיין בקטלוגים ומשלים רכישות באמצעות UCP ו-AP2.

מה למדתם

ב-Codelab הזה יצרתם סוכן 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

מה שונה בסביבת הייצור?

ב-Codelab הזה נעשה שימוש ב-mocks. בסביבת הייצור:

• גילוי UCP מתבצע מול רישום, ולא מול כתובות URL של localhost שמוגדרות בהארדקוד
• נקודות הקצה של MCP מתארחות אצל מוכרים אמיתיים – אותו פרוטוקול JSON-RPC 2.0, מלאי אמיתי
• הוראות AP2 חתומות באמצעות sd-jwt-vc, ולא באמצעות גיבוב (hash) מסוג SHA-256
• אישור תשלום באמצעות AP2 Wallet SDK עם הנחיות למתן הסכמה מהמשתמש
• קצה קדמי מעבד את תוצאות הכלי כממשק משתמש עשיר (רשתות מוצרים, סיכומי תשלום, כרטיסי אישור)

השלבים הבאים

• אפשר לעיין במפרט הפרוטוקול AP2 וליצור סוכן משלכם עם אפשרות תשלום
• כדי להפעיל מסחר מבוסס-סוכן, צריך להטמיע UCP אצל המוכר
• חיבור AP2 ל-A2A כדי ליצור תהליכי עבודה מסחריים מרובי סוכנים
• מידע על התוסף AP2 x402 לתשלומים במטבעות קריפטוגרפיים

מסמכי עזר

• מסמכי תיעוד של פרוטוקול AP2
• מדריך למפתחים בנושא UCP
• מאגר GitHub של AP2
• מסמכי תיעוד של ADK
• מפרט UCP