Secure Agent Commerce ด้วย AP2 และ UCP

1. ภาพรวม

ใน Codelab นี้ คุณจะเรียกใช้ Agent ADK ที่จองตั๋วภาพยนตร์จากผู้ขายโรงภาพยนตร์หลายรายโดยใช้โปรโตคอลการค้าแบบโอเพนซอร์ส 2 รายการ ได้แก่

  • UCP (Universal Commerce Protocol): มาตรฐานสำหรับเอเจนต์ในการค้นหาผู้ขาย ค้นหาแคตตาล็อก และจัดการขั้นตอนการชำระเงิน
  • AP2 (Agent Payments Protocol): โปรโตคอลสำหรับการให้สิทธิ์การชำระเงินที่ปลอดภัยและตรวจสอบได้โดยใช้คำสั่งที่ลงนามด้วยการเข้ารหัสลับ

แอปเดโม CineAgent เชื่อมต่อกับผู้ขายโรงภาพยนตร์จำลอง 2 รายที่มีความสามารถแตกต่างกัน (การเลือกที่นั่ง รูปแบบเฉพาะ และวิธีการชำระเงิน) และจัดระเบียบขั้นตอนการจองทั้งหมดตั้งแต่การค้นหาไปจนถึงการชำระเงิน

สิ่งที่คุณจะได้เรียนรู้

  • วิธีการค้นพบผู้ขาย UCP ผ่าน/.well-known/ucp
  • วิธีที่เอเจนต์ ADK ใช้ UCP เพื่อค้นหาแคตตาล็อกและสร้างการชำระเงิน
  • วิธีที่คำสั่ง AP2 (CartMandate, PaymentMandate) ช่วยรักษาความปลอดภัยของธุรกรรม
  • โปรโตคอล UCP และ AP2 แบบครบวงจรทำงานอย่างไรเพื่อรักษาความปลอดภัยของอีคอมเมิร์ซที่ทำงานด้วยเอเจนต์

สิ่งที่คุณต้องมี

  • โปรเจ็กต์ Google Cloud ที่เปิดใช้การเรียกเก็บเงิน
  • เว็บเบราว์เซอร์ เช่น Chrome
  • Python 3.11 ขึ้นไป

Codelab นี้เหมาะสำหรับนักพัฒนาแอปที่มีความรู้ระดับกลางซึ่งคุ้นเคยกับ Python และ Google Cloud บ้าง Codelab นี้จะใช้เวลาประมาณ 15 นาที

ทรัพยากรที่สร้างในโค้ดแล็บนี้ควรมีค่าใช้จ่ายน้อยกว่า $5

2. ทำความเข้าใจโปรโตคอล UCP และ AP2

ก่อนที่จะเจาะลึกเรื่องการสร้างเอเจนต์ เรามาทำความเข้าใจ 2 โปรโตคอลที่ทำให้การค้าแบบเอเจนต์ที่ปลอดภัยนี้เป็นไปได้

Universal Commerce Protocol (UCP)

UCP จะกำหนดมาตรฐานวิธีที่เอเจนต์ AI โต้ตอบกับผู้ขาย โดยจะช่วยแก้ปัญหาที่ตัวแทนต้องเรียนรู้ API ที่กำหนดเองสำหรับร้านค้าทุกแห่งด้วยการเปิดตัวโมเดลทรัพยากรที่ได้มาตรฐาน

วิธีการทำงาน

  1. การค้นพบ: ผู้ขายทุกรายที่ปฏิบัติตามข้อกำหนดของ UCP จะแสดงโปรไฟล์ในตำแหน่งมาตรฐาน /.well-known/ucp ตัวอย่าง: ปลายทาง UCP ของ Everlane เมื่อตัวแทนอ่านโปรไฟล์นี้ ระบบจะค้นหา
    • ความสามารถ: ฟีเจอร์หลักแบบสแตนด์อโลนที่ธุรกิจรองรับ เช่น การค้นหาแคตตาล็อกหรือการชำระเงิน
    • บริการ: เลเยอร์การสื่อสารระดับล่างที่ใช้ในการแลกเปลี่ยนข้อมูล ตัวอย่าง: REST API, MCP (Model Context Protocol), A2A (Agent2Agent Protocol)
    • ส่วนขยาย: หากผู้ขายต้องการลักษณะการทำงานเฉพาะ ก็สามารถกำหนดส่วนขยายที่กำหนดเองในโปรไฟล์นี้ได้
  2. การดำเนินการ: เมื่อค้นพบแล้ว เอเจนต์จะใช้ปลายทางบริการที่ระบุเพื่อดำเนินการ ใน Codelab นี้ เราจะใช้ Model Context Protocol (MCP) เป็นการรับส่งบริการ เอเจนต์จะเรียกใช้ JSON-RPC 2.0 ไปยังปลายทางนี้เพื่อเรียกใช้ความสามารถที่ค้นพบ ได้แก่ การค้นหาสินค้า การสร้างการชำระเงิน และการซื้อให้เสร็จสมบูรณ์

เวิร์กโฟลว์ UCP

โปรโตคอลการชำระเงินของตัวแทน (AP2)

AP2 ทำให้การให้สิทธิ์การชำระเงินโดยตัวแทนในนามของผู้ใช้เป็นมาตรฐาน ซึ่งช่วยแก้ปัญหาด้านความปลอดภัยของตัวแทนที่จัดการข้อมูลเข้าสู่ระบบการชำระเงินที่ละเอียดอ่อน

วิธีการทำงาน

  1. คำสั่งรถเข็น: เมื่อเอเจนต์สร้างการชำระเงินโดยใช้โปรโตคอล UCP ผู้ขายจะส่งคืน CartMandate ซึ่งเป็นออบเจ็กต์ JSON ที่มีรายละเอียดรถเข็นช็อปปิ้งและลายเซ็นเข้ารหัสจากผู้ขาย ซึ่งทำหน้าที่เป็นการรับประกันราคาคงที่ ผู้ขายจะเปลี่ยนราคาหลังจากออกคำสั่งนี้ไม่ได้
  2. หนังสือมอบอำนาจการชำระเงิน: หลังจากยืนยันเนื้อหาในรถเข็นแล้ว ผู้ใช้ (หรือตัวแทนในนามของผู้ใช้) จะสร้างPaymentMandate เพื่อให้สิทธิ์การชำระเงิน PaymentMandateนี้อ้างอิงถึงCartMandateและมีลายเซ็นเข้ารหัส (หรือโทเค็นการให้สิทธิ์) ของผู้ใช้
  3. การยืนยันลายเซ็น 2 ครั้ง: ผู้ขายจะได้รับทั้ง 2 คำสั่ง โดยจะยืนยันลายเซ็นของตนเองใน CartMandate และลายเซ็นของผู้ใช้ใน PaymentMandate หากทั้ง 2 อย่างถูกต้อง ธุรกรรมจะดำเนินการต่อ

ระบบ "ล็อก 2 ชั้น" นี้ช่วยให้มั่นใจได้ว่าผู้ขายจะเรียกเก็บเงินเกินไม่ได้ และตัวแทนจะใช้จ่ายโดยไม่ได้รับอนุญาตไม่ได้ ในเวอร์ชันที่ใช้งานจริง ข้อกำหนดเหล่านี้ใช้ SD-JWT (JWT การเปิดเผยข้อมูลแบบเลือก) เพื่อปกป้องความเป็นส่วนตัวของผู้ใช้

เวิร์กโฟลว์ AP2

3. ตั้งค่าสภาพแวดล้อม

การตั้งค่าโปรเจ็กต์ Google Cloud

สร้างโปรเจ็กต์ Google Cloud

  1. ในคอนโซล Google Cloud ในหน้าตัวเลือกโปรเจ็กต์ ให้เลือกหรือสร้างโปรเจ็กต์ 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

ติดตั้งการอ้างอิง

Google Cloud Shell มาพร้อมกับ uv ที่ติดตั้งไว้ล่วงหน้าเพื่อจัดการสภาพแวดล้อมและทรัพยากร Dependency

  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",
]
  1. เรียกใช้คำสั่งต่อไปนี้เพื่อสร้างสภาพแวดล้อมเสมือนและติดตั้งทรัพยากร Dependency ทั้งหมด
uv sync
  1. เปิดใช้งานสภาพแวดล้อมเสมือนที่สร้างโดย uv โดยใช้คำสั่งต่อไปนี้
source .venv/bin/activate

4. กำหนด Agent

ก่อนเขียนตรรกะของเครื่องมือใดๆ เรามากำหนดตัวแทนในไฟล์ที่ชื่อ 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),
    ],
)

คำอธิบายโค้ด

มาดูรายละเอียดสิ่งที่เกิดขึ้นในคำจำกัดความของ Agent นี้กัน

  • model="gemini-3.1-pro-preview": เราใช้โมเดลตัวอย่าง Gemini Pro ล่าสุดสำหรับการให้เหตุผลที่ซับซ้อนและการใช้เครื่องมือ
  • instruction: นี่คือพรอมต์ที่กำหนดลักษณะการทำงานของตัวแทน โดยจะบอกเอเจนต์อย่างชัดเจนให้ใช้ UCP และ AP2 แสดงรายการเครื่องมือที่มี และกำหนดกฎที่สำคัญ เช่น "ห้ามสร้างข้อมูลขึ้นมา" และ "ราคาเป็นหน่วยเซ็นต์"
  • tools: นี่คือรายการฟังก์ชัน Python (ซึ่งเราจะสร้างในขั้นตอนถัดไป) ที่เอเจนต์สามารถเลือกเรียกใช้ตามคำขอของผู้ใช้
  • require_confirmation: คุณสามารถรวมเครื่องมือใดก็ได้กับ FunctionTool(my_function,require_confirmation=True) เมื่อทริกเกอร์แล้ว เอเจนต์จะหยุดชั่วคราวและรอการอนุมัติแบบง่ายๆ ว่า "ใช่" หรือ "ไม่" ก่อนที่จะเรียกใช้เครื่องมือ ในที่นี้ ก่อนที่จะเรียกใช้complete_purchase เครื่องมือ Agent จะหยุดชั่วคราวเพื่อรอการยืนยันจากมนุษย์

รายการเครื่องมือ

คำจำกัดความของ Agent จะประกาศสิ่งที่เราต้องสร้าง เครื่องมือแต่ละอย่างจะเชื่อมโยงกับการดำเนินการที่เฉพาะเจาะจงในโปรโตคอล 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

ให้สิทธิ์การชำระเงินและทำคำสั่งซื้อให้เสร็จสมบูรณ์

Signs AP2 Mandate & sends to 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()

ทำความเข้าใจการตั้งค่า

เราใช้คลาสตัวช่วย 2 คลาส ได้แก่ UCPClient และ AP2Handler เพื่อให้ Codelab นี้มุ่งเน้นที่การสร้างเอเจนต์ ซึ่งเราจะดูในขั้นตอนถัดไป

  • การดูผ่านคืออะไร ซึ่งเป็นคลาส Helper ที่เขียนด้วยมือที่เราสร้างขึ้นสำหรับ 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 รายในส่วนถัดไป
  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 นี่คือลายเซ็นเข้ารหัสของผู้ขายที่ล็อกราคาที่เสนอ โดยผู้ใช้จะเปลี่ยนชื่อในภายหลังไม่ได้

ดำเนินการซื้อให้เสร็จสมบูรณ์

เครื่องมือนี้จะดำเนินการ 3 อย่าง ได้แก่

  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)

เหตุใดจึงต้องมีคำสั่ง 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 ให้ตัวแทนดูและจบกระบวนการ สำหรับ Codelab นี้ เราต้องมี 2 สิ่งเพื่อทดสอบในเครื่อง ได้แก่

  1. ผู้ขายจำลอง - เซิร์ฟเวอร์ในพื้นที่ที่จำลองปลายทาง UCP เพื่อให้คุณมีสิ่งที่ใช้ทดสอบ
  2. ตัวช่วยโปรโตคอล - Wrapper HTTP แบบบางสำหรับ UCP และ AP2 (ในเวอร์ชันที่ใช้งานจริง SDK อย่างเป็นทางการจะแทนที่ตัวช่วยเหล่านี้)

หมายเหตุ: คุณไม่จำเป็นต้องอ่านโค้ดนี้อย่างละเอียด ไฟล์เหล่านี้จำลองสิ่งที่โครงสร้างพื้นฐานและ SDK จริงจะให้ คัดลอกตามที่เห็น

ผู้ช่วยด้านโปรโตคอล

UCP และ AP2 ยังไม่มี SDK ของไคลเอ็นต์ โดยไฟล์ทั้ง 2 นี้จะจัดการการเชื่อมต่อ 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 2 เครื่อง โดยแต่ละเครื่องจะมีปลายทาง 2 รายการ

  • 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. เรียกใช้ Agent ด้วย ADK Web

มาใช้ Web UI ในตัวของ 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 Web UI โดยทำดังนี้

cd ../
adk web --allow_origins '*'

คุณควรเห็นเอาต์พุตที่ระบุว่าเซิร์ฟเวอร์กำลังทำงาน และจะแสดง URL (โดยปกติคือ http://localhost:8000 หรือคล้ายกัน)

พูดคุยกับตัวแทน

  1. เปิด URL ที่ adk web ให้ไว้ในเบราว์เซอร์
  2. คุณจะเห็นอินเทอร์เฟซแชท
  3. ลองถามว่า "มีหนังเรื่องอะไรบ้าง"
  4. เอเจนต์จะค้นพบโรงภาพยนตร์และค้นหาแคตตาล็อกเบื้องหลัง โดยรวบรวมผลลัพธ์จากผู้ขายทั้ง 2 รายผ่าน UCP
  5. ขอจองตั๋ว: "จองตั๋ว Oppenheimer 2 ใบสำหรับรอบ 19:00 น."
  6. เมื่อตัวแทนพยายามโทรหา complete_purchase ให้สังเกตว่า ADK Web UI จะแสดงกล่องโต้ตอบการยืนยันหรือการ์ดอย่างไร
  7. หากต้องการให้สิทธิ์ทำธุรกรรม ให้ตอบกลับในแชทด้วยสตริง JSON นี้ {"confirmed": true}
  8. ตัวแทนจะดำเนินการซื้อให้เสร็จสมบูรณ์และส่งการยืนยันคำสั่งซื้อพร้อมรหัสตั๋วให้คุณ

สิ่งที่เกิดขึ้นเบื้องหลังมีดังนี้

  1. create_checkout → ผู้ขายส่งคืน CartMandate (การล็อกราคาที่ลงนาม) ของ AP2
  2. complete_purchase → สร้าง PaymentMandate, ลงนาม (SHA-256 แบบจำลอง), ส่งทั้ง 2 คำสั่งไปยังผู้ขาย
  3. ผู้ขายยืนยันลายเซ็นทั้ง 2 รายการ → ออกตั๋ว (ในการจำลอง)

9. ล้างข้อมูล

หากไม่ต้องการให้เซิร์ฟเวอร์ภายในทำงานต่อไป ให้ล้างข้อมูลทรัพยากรโดยทำดังนี้

  1. ในเทอร์มินัลที่เรียกใช้ adk web ให้กด Ctrl+C เพื่อหยุดเซิร์ฟเวอร์ตัวแทน
  2. ในเทอร์มินัลที่เรียกใช้ python merchants.py ให้กด Ctrl+C เพื่อหยุดผู้ขายจำลอง
  3. ปิดใช้งานสภาพแวดล้อมเสมือนในทั้ง 2 เทอร์มินัลโดยเรียกใช้คำสั่งต่อไปนี้
deactivate
  1. (ไม่บังคับ) หากคุณสร้างโปรเจ็กต์ที่อยู่ในระบบคลาวด์ของ Google ใหม่สำหรับ Codelab นี้และต้องการลบ ให้เรียกใช้คำสั่งต่อไปนี้
gcloud projects delete $GOOGLE_CLOUD_PROJECT

10. ยินดีด้วย 🎉

คุณสร้าง ADK Agent ที่ค้นพบผู้ขาย เรียกดูแคตตาล็อก และทำการซื้อให้เสร็จสมบูรณ์โดยใช้ UCP และ AP2

สิ่งที่คุณได้เรียนรู้

ใน Codelab นี้ คุณได้สร้างเอเจนต์ ADK ที่จัดการโฟลว์การซื้อขายที่ปลอดภัย ข้อมูลสรุปเกี่ยวกับสิ่งที่คุณสร้างและแนวคิดหลักที่คุณนำไปใช้มีดังนี้

สิ่งที่คุณสร้าง

  • เครื่องมือสำหรับตัวแทน 5 รายการที่ครอบคลุมการดำเนินการ UCP และ AP2 ได้แก่ การค้นพบ การค้นหาแคตตาล็อก การชำระเงิน และการชำระเงิน
  • การลงนามในหนังสือมอบอำนาจ AP2 - CartMandate (การล็อกราคาของผู้ขาย) + PaymentMandate (การให้สิทธิ์ของผู้ใช้)
  • การค้นหาหลายผู้ขาย - ตัวแทน 1 รายที่ค้นหาโรงภาพยนตร์หลายแห่งและผสานผลลัพธ์

แนวคิดหลัก:

โปรโตคอล

การทำงาน

วิธีการทำงาน

การค้นพบ UCP

เอเจนต์ค้นหาผู้ขายและความสามารถของผู้ขาย

/.well-known/ucp → ความสามารถ, อุปกรณ์ปลายทาง MCP, วิธีการชำระเงิน

UCP MCP

Agent เรียกดูแคตตาล็อกและสร้างการชำระเงิน

การเรียก JSON-RPC 2.0 ไปยังปลายทาง MCP ของผู้ขาย

AP2 CartMandate

ผู้ขายล็อกราคาที่เสนอ

ผู้ขายลงนาม รวมถึงจำนวนเงินทั้งหมด + วันหมดอายุ

AP2 PaymentMandate

ผู้ใช้ให้สิทธิ์เรียกเก็บเงิน

ลงนามโดยผู้ใช้ อ้างอิงถึง CartMandate

การผลิตแตกต่างกันอย่างไร

Codelab นี้ใช้การจำลอง ใช้งานจริง

  • การค้นพบ UCP จะได้รับการแก้ไขเทียบกับรีจิสทรี ไม่ใช่ URL localhost ที่ฮาร์ดโค้ด
  • ปลายทาง MCP โฮสต์โดยผู้ขายจริง ซึ่งใช้โปรโตคอล JSON-RPC 2.0 เดียวกันและมีสินค้าคงคลังจริง
  • ข้อบังคับของ AP2 ลงนามด้วย sd-jwt-vc ไม่ใช่แฮช SHA-256
  • การให้สิทธิ์การชำระเงินใช้ AP2 Wallet SDK พร้อมข้อความแจ้งความยินยอมของผู้ใช้
  • ฟรอนท์เอนด์แสดงผลลัพธ์ของเครื่องมือเป็น UI แบบริชมีเดีย (ตารางผลิตภัณฑ์ ข้อมูลสรุปการชำระเงิน การ์ดยืนยัน)

ขั้นตอนถัดไป

  • สำรวจข้อกำหนดโปรโตคอล AP2 และสร้างเอเจนต์ที่เปิดใช้การชำระเงินของคุณเอง
  • ใช้ UCP สำหรับผู้ขายเพื่อเปิดใช้การค้าแบบเอเจนต์
  • เชื่อมต่อ AP2 กับ A2A สำหรับเวิร์กโฟลว์การค้าแบบหลายเอเจนต์
  • ดูข้อมูลเกี่ยวกับส่วนขยาย AP2 x402 สำหรับการชำระเงินด้วยสกุลเงินดิจิทัล

เอกสารอ้างอิง