Universal Commerce Protocol (UCP)

# UCP Checkout Flow

# 1. Agent discovers product via search or catalog
product = ucp.catalog.search({
    query: "running shoes size 10",
    filters: { brand: "Nike", priceMax: 150 }
})

# 2. Create cart with items
cart = ucp.checkout.createCart({
    merchantId: product.merchantId,
    items: [
        {
            productId: product.id,
            quantity: 1,
            variant: { size: "10", color: "black" }
        }
    ]
})

# 3. Cart returns dynamic pricing
# UCP handles: taxes, discounts, shipping options
print(cart.summary)
# {
#   subtotal: 129.99,
#   tax: 10.40,
#   shipping: 0.00,  # Free shipping applied
#   discount: -20.00,  # Promo code
#   total: 120.39
# }

# 4. Apply payment via Agent Payments Protocol (AP2)
payment = ucp.checkout.pay({
    cartId: cart.id,
    paymentMethod: {
        type: "agent_wallet",
        walletId: user.linkedWallet.id
    }
})

# 5. Complete checkout
order = ucp.checkout.complete({
    cartId: cart.id,
    paymentId: payment.id,
    shippingAddress: user.addresses.default
})

# 6. Order tracking via webhooks
# Agent receives real-time updates on order status

import httpx
from typing import Optional
from dataclasses import dataclass

@dataclass
class CartItem:
    product_id: str
    quantity: int
    variant: dict | None = None

class UCPClient:
    def __init__(self, api_key: str, merchant_id: str):
        self.base_url = "https://api.merchant.com/ucp/v1"
        self.client = httpx.AsyncClient(headers={
            "Authorization": f"Bearer {api_key}",
            "X-UCP-Version": "1.0"
        })
        self.merchant_id = merchant_id

    async def create_cart(
        self,
        items: list[CartItem],
        promo_code: Optional[str] = None
    ) -> dict:
        """Create a shopping cart with items."""
        response = await self.client.post(
            f"{self.base_url}/carts",
            json={
                "merchantId": self.merchant_id,
                "items": [
                    {
                        "productId": item.product_id,
                        "quantity": item.quantity,
                        "variant": item.variant
                    }
                    for item in items
                ],
                "promoCode": promo_code
            }
        )
        response.raise_for_status()
        return response.json()

    async def calculate_totals(
        self,
        cart_id: str,
        shipping_address: dict
    ) -> dict:
        """Get cart totals with tax and shipping calculated."""
        response = await self.client.post(
            f"{self.base_url}/carts/{cart_id}/calculate",
            json={"shippingAddress": shipping_address}
        )
        response.raise_for_status()
        return response.json()

    async def checkout(
        self,
        cart_id: str,
        payment_token: str,
        shipping_address: dict
    ) -> dict:
        """Complete checkout with payment."""
        response = await self.client.post(
            f"{self.base_url}/carts/{cart_id}/checkout",
            json={
                "paymentToken": payment_token,
                "shippingAddress": shipping_address
            }
        )
        response.raise_for_status()
        return response.json()

# Usage with an agent
async def purchase_product(
    ucp: UCPClient,
    product_id: str,
    user_payment_token: str,
    shipping_address: dict
):
    # Create cart
    cart = await ucp.create_cart([
        CartItem(product_id=product_id, quantity=1)
    ])

    # Calculate final price
    totals = await ucp.calculate_totals(
        cart["id"],
        shipping_address
    )
    print(f"Total: {totals['total']}")

    # Complete purchase
    order = await ucp.checkout(
        cart["id"],
        user_payment_token,
        shipping_address
    )

    return order

using System.Net.Http.Json;

public record CartItem(string ProductId, int Quantity, Dictionary<string, string>? Variant = null);

public class UcpClient : IDisposable
{
    private readonly HttpClient _client;
    private readonly string _merchantId;

    public UcpClient(string apiKey, string merchantId)
    {
        _merchantId = merchantId;
        _client = new HttpClient
        {
            BaseAddress = new Uri("https://api.merchant.com/ucp/v1/")
        };
        _client.DefaultRequestHeaders.Add("Authorization", $"Bearer {apiKey}");
        _client.DefaultRequestHeaders.Add("X-UCP-Version", "1.0");
    }

    public async Task<Cart> CreateCartAsync(
        IEnumerable<CartItem> items,
        string? promoCode = null,
        CancellationToken ct = default)
    {
        var response = await _client.PostAsJsonAsync("carts", new
        {
            merchantId = _merchantId,
            items = items.Select(i => new
            {
                productId = i.ProductId,
                quantity = i.Quantity,
                variant = i.Variant
            }),
            promoCode
        }, ct);

        response.EnsureSuccessStatusCode();
        return await response.Content.ReadFromJsonAsync<Cart>(ct)
            ?? throw new InvalidOperationException("Invalid response");
    }

    public async Task<CartTotals> CalculateTotalsAsync(
        string cartId,
        Address shippingAddress,
        CancellationToken ct = default)
    {
        var response = await _client.PostAsJsonAsync(
            $"carts/{cartId}/calculate",
            new { shippingAddress },
            ct
        );

        response.EnsureSuccessStatusCode();
        return await response.Content.ReadFromJsonAsync<CartTotals>(ct)
            ?? throw new InvalidOperationException("Invalid response");
    }

    public async Task<Order> CheckoutAsync(
        string cartId,
        string paymentToken,
        Address shippingAddress,
        CancellationToken ct = default)
    {
        var response = await _client.PostAsJsonAsync(
            $"carts/{cartId}/checkout",
            new { paymentToken, shippingAddress },
            ct
        );

        response.EnsureSuccessStatusCode();
        return await response.Content.ReadFromJsonAsync<Order>(ct)
            ?? throw new InvalidOperationException("Invalid response");
    }

    public void Dispose() => _client.Dispose();
}

// Usage
await using var ucp = new UcpClient(apiKey, merchantId);

var cart = await ucp.CreateCartAsync(new[]
{
    new CartItem("prod_123", 1, new() { ["size"] = "10" })
});

var totals = await ucp.CalculateTotalsAsync(cart.Id, userAddress);
Console.WriteLine($"Total: {totals.Total:C}");

var order = await ucp.CheckoutAsync(cart.Id, paymentToken, userAddress);
Console.WriteLine($"Order placed: {order.Id}");

from mcp.server import Server
from mcp.types import Tool, TextContent

server = Server("shopping-agent")

# MCP tool that uses UCP for checkout
@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "purchase_product":
        ucp = UCPClient(api_key=UCP_API_KEY)

        # Get user's linked identity for this merchant
        identity = await get_linked_identity(
            arguments["user_id"],
            arguments["merchant_id"]
        )

        if not identity:
            return {
                "content": [TextContent(
                    type="text",
                    text=f"Please link your {arguments['merchant_id']} account first."
                )],
                "_meta": {
                    "ui": {
                        "resourceUri": "ui://identity-linker",
                        "data": {
                            "merchantId": arguments["merchant_id"],
                            "authUrl": await ucp.get_auth_url(arguments["merchant_id"])
                        }
                    }
                }
            }

        # Create cart and calculate totals
        cart = await ucp.create_cart(
            merchant_id=arguments["merchant_id"],
            items=[{"productId": arguments["product_id"], "quantity": 1}],
            access_token=identity.access_token
        )

        totals = await ucp.calculate_totals(
            cart_id=cart["id"],
            shipping_address=arguments["shipping_address"]
        )

        # Return confirmation UI
        return {
            "content": [TextContent(
                type="text",
                text=f"Ready to purchase for {totals['total']}"
            )],
            "_meta": {
                "ui": {
                    "resourceUri": "ui://checkout-confirmation",
                    "data": {
                        "cart": cart,
                        "totals": totals,
                        "product": arguments["product_id"],
                        # Callback to complete purchase
                        "confirmAction": {
                            "tool": "confirm_purchase",
                            "args": {"cartId": cart["id"]}
                        }
                    }
                }
            }
        }

    if name == "confirm_purchase":
        # User confirmed in UI, complete the purchase
        order = await ucp.checkout(
            cart_id=arguments["cartId"],
            payment_token=arguments["paymentToken"]
        )

        return {
            "content": [TextContent(
                type="text",
                text=f"Order placed! Order ID: {order['id']}"
            )]
        }

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="purchase_product",
            description="Purchase a product from a UCP-enabled merchant",
            inputSchema={
                "type": "object",
                "properties": {
                    "merchant_id": {"type": "string"},
                    "product_id": {"type": "string"},
                    "user_id": {"type": "string"},
                    "shipping_address": {"type": "object"}
                },
                "required": ["merchant_id", "product_id", "user_id"]
            }
        ),
        Tool(
            name="confirm_purchase",
            description="Confirm a pending purchase",
            inputSchema={
                "type": "object",
                "properties": {
                    "cartId": {"type": "string"},
                    "paymentToken": {"type": "string"}
                },
                "required": ["cartId", "paymentToken"]
            }
        )
    ]