Implementing Google UCP: Technical Integration Guide (Step-by-Step)

By /

In our previous posts, we covered what Google UCP is and how to decide if it’s right for your business. Now comes the technical part: how to actually implement it.

This guide provides step-by-step implementation instructions, code examples, and best practices for integrating Google’s Universal Commerce Protocol into your ecommerce platform.

Note: This guide assumes familiarity with REST APIs, Python/FastAPI, and backend development. If you prefer turnkey implementation, our UCP implementation service handles everything for you.

Implementation Overview

UCP integration involves four main components:

  1. Discovery Endpoint (/.well-known/ucp) – Tells Google what your store supports
  2. Checkout API – Creates and manages shopping sessions
  3. Payment Integration – Processes UCP payment tokens
  4. Order Management – Handles order status and fulfillment updates

Estimated implementation time: 40-80 developer hours (5-10 business days with dedicated resources).

Step 1: Set Up Your Development Environment

Install the UCP Python SDK

Google provides an official Python SDK to accelerate development:

# Clone the UCP Python SDK
git clone https://github.com/Universal-Commerce-Protocol/python-sdk.git
cd python-sdk

# Install dependencies
pip install -r requirements.txt

# Clone sample implementations
git clone https://github.com/Universal-Commerce-Protocol/samples.git

Technology Stack

This guide uses:

  • Python 3.9+
  • FastAPI (modern Python web framework)
  • Pydantic (data validation)
  • SQLite (for demo – use PostgreSQL/MySQL in production)

You can adapt these examples to Node.js, Ruby, PHP, or any language that supports REST APIs.

Step 2: Create the Discovery Endpoint

The discovery endpoint at /.well-known/ucp tells Google AI what capabilities your store supports.

Basic Discovery Endpoint

from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()

@app.get("/.well-known/ucp")
async def ucp_discovery():
    """
    UCP Discovery endpoint - tells Google what your store supports
    """
    return JSONResponse({
        "version": "1.0",
        "merchant": {
            "id": "your_store_id",
            "name": "Your Store Name",
            "url": "https://yourstore.com",
            "logo_url": "https://yourstore.com/logo.png",
            "support_email": "support@yourstore.com",
            "support_phone": "+1-555-123-4567"
        },
        "capabilities": {
            "checkout": {
                "version": "1.0",
                "endpoint": "/api/ucp/checkout-sessions"
            },
            "order_management": {
                "version": "1.0",
                "endpoint": "/api/ucp/orders"
            }
        },
        "payment_handlers": [
            {
                "id": "stripe",
                "version": "1.0",
                "configuration": {
                    "publishable_key": "pk_live_YOUR_STRIPE_KEY"
                }
            }
        ],
        "supported_currencies": ["USD", "EUR", "GBP", "ILS"],
        "supported_countries": ["US", "GB", "IL"],
        "policies": {
            "return_policy_url": "https://yourstore.com/returns",
            "privacy_policy_url": "https://yourstore.com/privacy",
            "terms_of_service_url": "https://yourstore.com/terms"
        }
    })

Key fields explained:

  • merchant.id: Your Google Merchant Center ID
  • capabilities: What APIs you’ve implemented
  • payment_handlers: Which payment processors you support
  • supported_currencies/countries: Where you can sell

Step 3: Implement the Checkout API

The checkout API manages the shopping session lifecycle.

Data Models

First, define your data structures:

from pydantic import BaseModel, EmailStr
from typing import List, Optional
from datetime import datetime
import uuid

class LineItem(BaseModel):
    product_id: str
    variant_id: Optional[str] = None
    quantity: int
    unit_price: float

class Address(BaseModel):
    name: str
    street: str
    city: str
    state: str
    postal_code: str
    country: str
    phone: Optional[str] = None

class CreateCheckoutRequest(BaseModel):
    line_items: List[LineItem]
    currency: str = "USD"
    buyer_email: Optional[EmailStr] = None

class CheckoutSession(BaseModel):
    id: str
    status: str  # "pending", "processing", "completed", "cancelled"
    line_items: List[LineItem]
    subtotal: float
    tax: float
    shipping: float
    discount: float = 0.0
    total: float
    currency: str
    shipping_address: Optional[Address] = None
    buyer_email: Optional[EmailStr] = None
    expires_at: datetime
    created_at: datetime

Create Checkout Session

@app.post("/api/ucp/checkout-sessions")
async def create_checkout_session(request: CreateCheckoutRequest):
    """
    Create a new checkout session
    Google calls this when a customer wants to purchase
    """
    session_id = str(uuid.uuid4())

    # Validate products and get current prices
    validated_items = []
    subtotal = 0.0

    for item in request.line_items:
        # Check inventory availability
        product = get_product_from_db(item.product_id)
        if not product:
            return JSONResponse(
                status_code=404,
                content={"error": f"Product {item.product_id} not found"}
            )

        if product.stock < item.quantity:
            return JSONResponse(
                status_code=400,
                content={
                    "error": f"Insufficient stock for {product.name}",
                    "available": product.stock
                }
            )

        # Use current price (don't trust client price)
        item.unit_price = product.price
        validated_items.append(item)
        subtotal += product.price * item.quantity

    # Calculate tax (example: 8.5% sales tax)
    tax = subtotal * 0.085

    # Calculate shipping (example: flat $10)
    shipping = 10.0 if subtotal < 50 else 0.0  # Free shipping over $50

    total = subtotal + tax + shipping

    # Create session in database
    session = CheckoutSession(
        id=session_id,
        status="pending",
        line_items=validated_items,
        subtotal=subtotal,
        tax=tax,
        shipping=shipping,
        total=total,
        currency=request.currency,
        buyer_email=request.buyer_email,
        expires_at=datetime.now() + timedelta(hours=24),
        created_at=datetime.now()
    )

    save_session_to_db(session)

    return {
        "id": session.id,
        "status": session.status,
        "line_items": [item.dict() for item in session.line_items],
        "subtotal": session.subtotal,
        "tax": session.tax,
        "shipping": session.shipping,
        "total": session.total,
        "currency": session.currency,
        "expires_at": session.expires_at.isoformat()
    }

Get Checkout Session

@app.get("/api/ucp/checkout-sessions/{session_id}")
async def get_checkout_session(session_id: str):
    """
    Retrieve session details
    Google calls this to check session status
    """
    session = get_session_from_db(session_id)

    if not session:
        return JSONResponse(
            status_code=404,
            content={"error": "Session not found"}
        )

    return session.dict()

Update Checkout Session (Discounts, Shipping)

class UpdateCheckoutRequest(BaseModel):
    shipping_address: Optional[Address] = None
    discount_code: Optional[str] = None

@app.put("/api/ucp/checkout-sessions/{session_id}")
async def update_checkout_session(session_id: str, request: UpdateCheckoutRequest):
    """
    Update session with shipping address or discount code
    """
    session = get_session_from_db(session_id)

    if not session:
        return JSONResponse(
            status_code=404,
            content={"error": "Session not found"}
        )

    # Update shipping address
    if request.shipping_address:
        session.shipping_address = request.shipping_address

        # Recalculate tax based on shipping location
        tax_rate = get_tax_rate_for_location(
            request.shipping_address.state,
            request.shipping_address.country
        )
        session.tax = session.subtotal * tax_rate

    # Apply discount code
    if request.discount_code:
        discount = validate_discount_code(request.discount_code)

        if discount:
            if discount.type == "percentage":
                session.discount = session.subtotal * (discount.value / 100)
            elif discount.type == "fixed":
                session.discount = discount.value
        else:
            return JSONResponse(
                status_code=400,
                content={"error": "Invalid discount code"}
            )

    # Recalculate total
    session.total = session.subtotal + session.tax + session.shipping - session.discount

    update_session_in_db(session)

    return session.dict()

Complete Checkout

class CompleteCheckoutRequest(BaseModel):
    payment_token: str
    shipping_address: Address
    billing_address: Optional[Address] = None

@app.post("/api/ucp/checkout-sessions/{session_id}/complete")
async def complete_checkout(session_id: str, request: CompleteCheckoutRequest):
    """
    Process payment and create order
    """
    session = get_session_from_db(session_id)

    if not session:
        return JSONResponse(
            status_code=404,
            content={"error": "Session not found"}
        )

    if session.status != "pending":
        return JSONResponse(
            status_code=400,
            content={"error": f"Session already {session.status}"}
        )

    # Process payment with Stripe (or your PSP)
    try:
        payment_result = process_payment_with_stripe(
            amount=int(session.total * 100),  # Stripe uses cents
            currency=session.currency,
            payment_token=request.payment_token,
            metadata={
                "session_id": session_id,
                "source": "google_ucp"
            }
        )
    except PaymentError as e:
        return JSONResponse(
            status_code=402,
            content={
                "error": "Payment failed",
                "message": str(e)
            }
        )

    # Create order in your system
    order = create_order(
        session=session,
        shipping_address=request.shipping_address,
        billing_address=request.billing_address or request.shipping_address,
        payment_id=payment_result.id
    )

    # Update session status
    session.status = "completed"
    update_session_in_db(session)

    # Decrement inventory
    for item in session.line_items:
        decrement_inventory(item.product_id, item.quantity)

    return {
        "status": "completed",
        "order_id": order.id,
        "confirmation_number": order.confirmation_number,
        "estimated_delivery": order.estimated_delivery_date.isoformat(),
        "tracking_url": order.tracking_url
    }

Step 4: Payment Processing Integration

UCP uses tokenized payment methods. Here’s how to process them with Stripe:

import stripe

stripe.api_key = "sk_live_YOUR_SECRET_KEY"

def process_payment_with_stripe(amount: int, currency: str, payment_token: str, metadata: dict):
    """
    Process a UCP payment token with Stripe
    """
    try:
        # Create payment intent
        payment_intent = stripe.PaymentIntent.create(
            amount=amount,
            currency=currency.lower(),
            payment_method=payment_token,
            confirm=True,
            metadata=metadata,
            automatic_payment_methods={
                "enabled": True,
                "allow_redirects": "never"
            }
        )

        if payment_intent.status == "succeeded":
            return payment_intent
        else:
            raise PaymentError(f"Payment status: {payment_intent.status}")

    except stripe.error.CardError as e:
        raise PaymentError(f"Card declined: {e.user_message}")
    except stripe.error.StripeError as e:
        raise PaymentError(f"Payment processing error: {str(e)}")

For Adyen or Checkout.com, consult their UCP-specific documentation.

Step 5: Order Management Webhooks

Notify Google when order status changes:

@app.post("/api/ucp/orders/{order_id}/status")
async def update_order_status(order_id: str, status: str, tracking_number: Optional[str] = None):
    """
    Update order status and notify Google
    Call this when you ship an order or update its status
    """
    order = get_order_from_db(order_id)

    if not order:
        return JSONResponse(
            status_code=404,
            content={"error": "Order not found"}
        )

    order.status = status
    if tracking_number:
        order.tracking_number = tracking_number

    update_order_in_db(order)

    # Send webhook to Google (if order came from UCP)
    if order.source == "google_ucp":
        notify_google_order_update(order)

    return {"status": "updated"}

Step 6: Testing Your Integration

Local Testing

# Run your UCP server locally
uvicorn main:app --reload --port 8182

# Test discovery endpoint
curl http://localhost:8182/.well-known/ucp

# Test checkout session creation
curl -X POST http://localhost:8182/api/ucp/checkout-sessions   -H "Content-Type: application/json"   -d '{
    "line_items": [
      {
        "product_id": "prod_123",
        "quantity": 2,
        "unit_price": 29.99
      }
    ],
    "currency": "USD"
  }'

UCP Conformance Tests

Google provides conformance tests to validate your implementation:

git clone https://github.com/Universal-Commerce-Protocol/conformance.git
cd conformance
npm install

# Run conformance tests against your server
npm run test -- --endpoint=https://yourstore.com

# Check for errors
# All tests must pass before production deployment

Sandbox Testing with Google

After submitting your interest form to Google:

  1. You’ll receive sandbox credentials
  2. Deploy your implementation to a staging server
  3. Google will simulate AI shopping sessions
  4. Verify orders appear correctly in your system
  5. Test edge cases (out of stock, invalid discounts, payment failures)

Step 7: Production Deployment

Pre-Launch Checklist

  • ☐ All conformance tests pass
  • ☐ Discovery endpoint accessible at https://yourstore.com/.well-known/ucp
  • ☐ SSL certificate valid and up to date
  • ☐ Payment processor in live mode (not test mode)
  • ☐ Google Merchant Center feed updated and approved
  • ☐ Return policy configured
  • ☐ Error logging and monitoring enabled
  • ☐ Customer service team trained on UCP orders
  • ☐ Inventory sync working correctly
  • ☐ Tax calculation validated for all supported regions

Production Configuration

# Environment variables for production
UCP_ENVIRONMENT=production
STRIPE_SECRET_KEY=sk_live_xxx
STRIPE_PUBLISHABLE_KEY=pk_live_xxx
GOOGLE_MERCHANT_ID=your_merchant_id
SESSION_EXPIRY_HOURS=24
MAX_SESSION_ITEMS=50

Monitoring and Alerts

Set up monitoring for:

  • API uptime: Alert if discovery endpoint is down
  • Error rates: Monitor failed checkout attempts
  • Payment failures: Track declined transactions
  • Inventory sync: Ensure real-time stock updates
  • Response times: Keep API latency under 500ms

Common Issues and Troubleshooting

Issue: Discovery Endpoint Not Found

Symptom: Google cannot detect your UCP capabilities

Fix:

  • Ensure endpoint is at exact path /.well-known/ucp (no trailing slash)
  • Check that server responds with Content-Type: application/json
  • Verify HTTPS certificate is valid
  • Test with: curl -I https://yourstore.com/.well-known/ucp

Issue: Payment Token Processing Fails

Symptom: Stripe/Adyen rejects UCP payment tokens

Fix:

  • Confirm your PSP account has UCP enabled
  • Check API key permissions (need payment processing permission)
  • Validate token format matches PSP expectations
  • Review PSP-specific UCP documentation

Issue: Inventory Sync Errors

Symptom: Customers can purchase out-of-stock items

Fix:

  • Implement real-time inventory checks in checkout session creation
  • Add inventory locking during checkout (release after 24h or completion)
  • Sync Merchant Center feed hourly for stock updates

Best Practices

  1. Always validate inventory: Never trust client-provided quantities
  2. Use server-side pricing: Recalculate prices on every request
  3. Implement idempotency: Use idempotency keys to prevent duplicate orders
  4. Log everything: Comprehensive logging helps debug UCP-specific issues
  5. Set reasonable timeouts: Sessions should expire after 24 hours
  6. Handle partial fulfillment: Support backorders and split shipments
  7. Test edge cases: Out of stock, invalid addresses, payment declines
  8. Monitor performance: Keep API response times under 500ms

Next Steps

You now have a complete technical roadmap for implementing Google UCP. The key steps are:

  1. Set up your development environment with the UCP SDK
  2. Implement the discovery endpoint
  3. Build the checkout API (create, get, update, complete)
  4. Integrate payment processing
  5. Add order management webhooks
  6. Pass conformance tests
  7. Deploy to production with monitoring

Estimated timeline: 10-15 business days with dedicated developer resources.

Need Expert Implementation?

UCP integration requires careful attention to API specifications, payment processing, and edge case handling. If you’d prefer to have experts handle the implementation, our UCP implementation service provides:

  • Complete API endpoint development
  • Payment processor integration (Stripe/Adyen/Checkout.com)
  • Google Merchant Center configuration
  • Conformance testing and validation
  • Production deployment with monitoring
  • 10-day implementation timeline
  • Post-launch support

Pricing: $4,500-$7,500 depending on complexity.

Get a free consultation and quote →

This concludes our 3-part series on Google UCP. For more questions or implementation support, contact us through the UCP implementation service page.

Leave a Comment

Your email address will not be published. Required fields are marked *

Scroll to Top