Universal Commerce Protocol (UCP) Explained

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
from flask import Flask, request, jsonify
from dataclasses import dataclass
from typing import List, Optional
import uuid

app = Flask(__name__)

@dataclass
class CartItem:
    product_id: str
    variant_id: str
    quantity: int
    price: float
    
@dataclass
class CheckoutSession:
    id: str
    status: str  # "open", "complete", "cancelled"
    items: List[CartItem]
    subtotal: float
    tax: float
    shipping: float
    total: float
    currency: str

# In-memory store (use a real database in production)
sessions = {}

@app.route('/ucp/v1/checkout/sessions', methods=['POST'])
def create_session():
    data = request.json
    session_id = str(uuid.uuid4())
    
    session = CheckoutSession(
        id=session_id,
        status="open",
        items=[],
        subtotal=0,
        tax=0,
        shipping=0,
        total=0,
        currency=data.get('currency', 'USD')
    )
    
    sessions[session_id] = session
    
    return jsonify({
        "session_id": session_id,
        "status": "open",
        "expires_at": "2026-01-12T12:00:00Z"
    }), 201

@app.route('/ucp/v1/checkout/sessions/<session_id>', methods=['PATCH'])
def update_session(session_id):
    if session_id not in sessions:
        return jsonify({"error": "Session not found"}), 404
    
    session = sessions[session_id]
    data = request.json
    
    # Add items to cart
    if 'add_items' in data:
        for item_data in data['add_items']:
            # Fetch product details and validate
            product = get_product(item_data['product_id'])
            
            item = CartItem(
                product_id=item_data['product_id'],
                variant_id=item_data.get('variant_id'),
                quantity=item_data['quantity'],
                price=product['price']
            )
            session.items.append(item)
    
    # Recalculate totals
    session.subtotal = sum(item.price * item.quantity for item in session.items)
    session.tax = calculate_tax(session.subtotal, data.get('shipping_address'))
    session.shipping = calculate_shipping(session.items, data.get('shipping_address'))
    session.total = session.subtotal + session.tax + session.shipping
    
    return jsonify({
        "session_id": session_id,
        "items": [vars(item) for item in session.items],
        "subtotal": session.subtotal,
        "tax": session.tax,
        "shipping": session.shipping,
        "total": session.total,
        "currency": session.currency
    })

@app.route('/ucp/v1/checkout/sessions/<session_id>/complete', methods=['POST'])
def complete_session(session_id):
    if session_id not in sessions:
        return jsonify({"error": "Session not found"}), 404
    
    session = sessions[session_id]
    data = request.json
    
    # Process payment
    payment_result = process_payment(
        amount=session.total,
        currency=session.currency,
        payment_method=data['payment_method']
    )
    
    if not payment_result['success']:
        return jsonify({"error": "Payment failed"}), 400
    
    # Create order
    order = create_order(session, payment_result)
    
    session.status = "complete"
    
    return jsonify({
        "order_id": order['id'],
        "status": "confirmed",
        "confirmation_number": order['confirmation_number'],
        "estimated_delivery": order['estimated_delivery']
    })

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
@app.route('/ucp/v1/identity/authorize', methods=['GET'])
def authorize():
    """Generate OAuth authorization URL"""
    client_id = request.args.get('client_id')
    redirect_uri = request.args.get('redirect_uri')
    scope = request.args.get('scope', 'read_orders write_checkout')
    state = request.args.get('state')
    
    # Validate client and redirect URI
    if not validate_client(client_id, redirect_uri):
        return jsonify({"error": "Invalid client"}), 400
    
    # Generate authorization URL
    auth_code = generate_auth_code(client_id, scope)
    
    return jsonify({
        "authorization_url": f"https://yourstore.com/oauth/authorize?code={auth_code}&state={state}"
    })

@app.route('/ucp/v1/identity/token', methods=['POST'])
def token():
    """Exchange authorization code for tokens"""
    data = request.json
    
    if data.get('grant_type') == 'authorization_code':
        # Validate auth code
        auth_data = validate_auth_code(data['code'])
        if not auth_data:
            return jsonify({"error": "Invalid code"}), 400
        
        # Generate tokens
        access_token = generate_access_token(auth_data)
        refresh_token = generate_refresh_token(auth_data)
        
        return jsonify({
            "access_token": access_token,
            "refresh_token": refresh_token,
            "token_type": "Bearer",
            "expires_in": 3600,
            "scope": auth_data['scope']
        })
    
    elif data.get('grant_type') == 'refresh_token':
        # Refresh the access token
        new_access_token = refresh_access_token(data['refresh_token'])
        
        return jsonify({
            "access_token": new_access_token,
            "token_type": "Bearer",
            "expires_in": 3600
        })

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
import requests
from datetime import datetime

def send_order_webhook(order, event_type):
    """Send webhook notification to registered AI agents"""
    
    webhook_payload = {
        "event": event_type,
        "timestamp": datetime.utcnow().isoformat(),
        "order": {
            "id": order.id,
            "status": order.status,
            "tracking_number": order.tracking_number,
            "tracking_url": order.tracking_url,
            "estimated_delivery": order.estimated_delivery,
            "items": [
                {
                    "product_id": item.product_id,
                    "title": item.title,
                    "quantity": item.quantity
                }
                for item in order.items
            ]
        }
    }
    
    # Get all registered webhooks for this merchant
    webhooks = get_registered_webhooks(order.merchant_id, event_type)
    
    for webhook in webhooks:
        try:
            response = requests.post(
                webhook.url,
                json=webhook_payload,
                headers={
                    "Content-Type": "application/json",
                    "X-UCP-Signature": sign_payload(webhook_payload, webhook.secret)
                },
                timeout=10
            )
            
            log_webhook_delivery(webhook.id, response.status_code)
            
        except Exception as e:
            log_webhook_failure(webhook.id, str(e))

# Webhook event types
WEBHOOK_EVENTS = [
    "order.created",
    "order.confirmed",
    "order.shipped",
    "order.out_for_delivery",
    "order.delivered",
    "order.cancelled",
    "order.refunded"
]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# Good: Short-lived tokens with refresh
ACCESS_TOKEN_EXPIRY = 3600  # 1 hour
REFRESH_TOKEN_EXPIRY = 86400 * 30  # 30 days

# Good: Scope limitations
SCOPES = {
    "read_products": "View product catalog",
    "read_orders": "View order history",
    "write_checkout": "Create and complete purchases",
    "manage_returns": "Process returns and refunds"
}

# Good: Token revocation
def revoke_token(token):
    """Allow users to disconnect AI agents anytime"""
    token_data = decode_token(token)
    add_to_revocation_list(token_data['jti'])
    return True

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
from functools import wraps
from flask import request, jsonify
import time

# Simple rate limiter
rate_limits = {}

def rate_limit(max_requests=100, window=60):
    def decorator(f):
        @wraps(f)
        def wrapped(*args, **kwargs):
            client_id = request.headers.get('X-Client-ID')
            key = f"{client_id}:{f.__name__}"
            
            now = time.time()
            window_start = now - window
            
            # Clean old entries
            rate_limits[key] = [
                t for t in rate_limits.get(key, [])
                if t > window_start
            ]
            
            if len(rate_limits.get(key, [])) >= max_requests:
                return jsonify({
                    "error": "Rate limit exceeded",
                    "retry_after": window
                }), 429
            
            rate_limits.setdefault(key, []).append(now)
            return f(*args, **kwargs)
        return wrapped
    return decorator

@app.route('/ucp/v1/checkout/sessions', methods=['POST'])
@rate_limit(max_requests=50, window=60)
def create_session():
    # ... implementation
    pass

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
import openai
from ucp_client import UCPClient

# Initialize UCP client
ucp = UCPClient(
    base_url="https://store.example.com/ucp/v1",
    access_token="user_access_token"
)

# Define tools for the AI agent
tools = [
    {
        "type": "function",
        "function": {
            "name": "search_products",
            "description": "Search for products by query",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "category": {"type": "string"},
                    "max_price": {"type": "number"}
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "add_to_cart",
            "description": "Add a product to the shopping cart",
            "parameters": {
                "type": "object",
                "properties": {
                    "product_id": {"type": "string"},
                    "quantity": {"type": "integer"}
                },
                "required": ["product_id"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "checkout",
            "description": "Complete the purchase",
            "parameters": {
                "type": "object",
                "properties": {
                    "confirm": {"type": "boolean"}
                },
                "required": ["confirm"]
            }
        }
    }
]

def search_products(query, category=None, max_price=None):
    """Search products via UCP"""
    products = ucp.products.search(
        query=query,
        category=category,
        max_price=max_price
    )
    return products

def add_to_cart(product_id, quantity=1):
    """Add item to cart via UCP"""
    if not hasattr(ucp, 'current_session'):
        ucp.current_session = ucp.checkout.create_session()
    
    ucp.checkout.add_item(
        session_id=ucp.current_session['id'],
        product_id=product_id,
        quantity=quantity
    )
    
    return ucp.checkout.get_session(ucp.current_session['id'])

def checkout(confirm):
    """Complete purchase via UCP"""
    if not confirm:
        return {"status": "cancelled", "message": "User did not confirm"}
    
    # Get user's saved payment method
    payment_method = ucp.identity.get_default_payment_method()
    
    # Complete the order
    result = ucp.checkout.complete(
        session_id=ucp.current_session['id'],
        payment_method=payment_method
    )
    
    return result

# Agent loop
def run_shopping_agent(user_request):
    messages = [
        {"role": "system", "content": """You are a shopping assistant. 
        Help users find and purchase products. Always confirm before checkout.
        Be concise and helpful."""},
        {"role": "user", "content": user_request}
    ]
    
    while True:
        response = openai.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )
        
        message = response.choices[0].message
        
        if message.tool_calls:
            for tool_call in message.tool_calls:
                func_name = tool_call.function.name
                args = json.loads(tool_call.function.arguments)
                
                # Execute the function
                if func_name == "search_products":
                    result = search_products(**args)
                elif func_name == "add_to_cart":
                    result = add_to_cart(**args)
                elif func_name == "checkout":
                    result = checkout(**args)
                
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps(result)
                })
        else:
            return message.content

# Example usage
response = run_shopping_agent("Find me running shoes under $150 and add the best one to cart")
print(response)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# MCP server exposing UCP capabilities as tools
from mcp import Server

server = Server()

@server.tool
def ucp_search_products(query: str, max_price: float = None):
    """Search products on connected stores via UCP"""
    return ucp_client.search(query, max_price=max_price)

@server.tool
def ucp_create_checkout(product_id: str, quantity: int):
    """Create a checkout session via UCP"""
    session = ucp_client.checkout.create_session()
    ucp_client.checkout.add_item(session['id'], product_id, quantity)
    return session

@server.tool
def ucp_complete_purchase(session_id: str, confirm: bool):
    """Complete a purchase via UCP (requires user confirmation)"""
    if not confirm:
        return {"error": "User must confirm purchase"}
    return ucp_client.checkout.complete(session_id)