diff --git a/.gitignore b/.gitignore
index 85c491c8..e19769e2 100644
--- a/.gitignore
+++ b/.gitignore
@@ -4,3 +4,8 @@
 uv.lock
 .venv
 **/node_modules
+**/*.db
+**/*.db-shm
+**/*.db-wal
+**/db_temp/
+*.log
diff --git a/rest/python/client/flower_shop/simple_happy_path_client.py b/rest/python/client/flower_shop/simple_happy_path_client.py
index f0fc55b3..3527b3b8 100644
--- a/rest/python/client/flower_shop/simple_happy_path_client.py
+++ b/rest/python/client/flower_shop/simple_happy_path_client.py
@@ -36,9 +36,9 @@
 import uuid
 import httpx
 import signing
-from ucp_sdk.models.schemas.shopping import checkout_create_request
+from ucp_sdk.models.schemas.shopping import cart_create_request
+from ucp_sdk.models.schemas.shopping import cart_update_request
 from ucp_sdk.models.schemas.shopping import checkout_update_request
-from ucp_sdk.models.schemas.shopping import payment_create_request
 from ucp_sdk.models.schemas.shopping.types import buyer_create_request
 from ucp_sdk.models.schemas.shopping.types import item_create_request
 from ucp_sdk.models.schemas.shopping.types import item_update_request
@@ -293,46 +293,30 @@ def main() -> int:
 
     # ==========================================================================
 
-    # STEP 1: Create a Checkout Session
-
+    # ==========================================================================
+    # STEP 1: Create a Cart
     # ==========================================================================
 
-    logger.info("\nSTEP 1: Creating a new Checkout Session...")
+    logger.info("\nSTEP 1: Creating a new Cart...")
 
     # We start with one item: "Red Rose"
-
     item1 = item_create_request.ItemCreateRequest(id="bouquet_roses")
-
     line_item1 = line_item_create_request.LineItemCreateRequest(
       quantity=1, item=item1
     )
 
-    # We initialize the payment section with the handlers we discovered.
-
-    # We do NOT select an instrument yet (selected_instrument_id=None).
-
-    payment_request = payment_create_request.PaymentCreateRequest(
-      instruments=[],
-      selected_instrument_id=None,
-      handlers=supported_handlers,  # Pass back what we found (or a subset)
-    )
-
-    # We include the buyer to trigger address lookup on the server
-
+    # We include the buyer
     buyer_request = buyer_create_request.BuyerCreateRequest(
       full_name="John Doe", email="john.doe@example.com"
     )
 
-    create_payload = checkout_create_request.CheckoutCreateRequest(
-      currency="USD",
+    create_payload = cart_create_request.CartCreateRequest(
       line_items=[line_item1],
-      payment=payment_request,
       buyer=buyer_request,
     )
 
     headers = get_headers()
-
-    url = "/checkout-sessions"
+    url = "/carts"
 
     json_body = create_payload.model_dump(
       mode="json", by_alias=True, exclude_none=True
@@ -344,22 +328,17 @@ def main() -> int:
       headers=headers,
     )
 
-    checkout_data = response.json()
-
-    checkout_id = checkout_data.get("id")
+    cart_data = response.json()
+    cart_id = cart_data.get("id")
 
     # Extract IDs for documentation
-
     extractions = {}
-    if checkout_id:
-      global_replacements[checkout_id] = "CHECKOUT_ID"
-      extractions["CHECKOUT_ID"] = ".id"
-
-    # We also want to capture the line item ID if possible,
-    # though it might change order. We'll grab the first one.
+    if cart_id:
+      global_replacements[cart_id] = "CART_ID"
+      extractions["CART_ID"] = ".id"
 
-    if checkout_data.get("line_items"):
-      li_id = checkout_data["line_items"][0]["id"]
+    if cart_data.get("line_items"):
+      li_id = cart_data["line_items"][0]["id"]
       global_replacements[li_id] = "LINE_ITEM_1_ID"
       extractions["LINE_ITEM_1_ID"] = ".line_items[0].id"
 
@@ -371,62 +350,47 @@ def main() -> int:
         headers,
         json_body,
         response,
-        "Step 1: Create Checkout Session",
+        "Step 1: Create Cart",
         replacements=global_replacements,
         extractions=extractions,
       )
 
     if response.status_code not in [200, 201]:
-      logger.error("Failed to create checkout: %s", response.text)
-
+      logger.error("Failed to create cart: %s", response.text)
       return 1
 
-    logger.info("Successfully created checkout session: %s", checkout_id)
-
-    logger.info(
-      "Current Total: %s cents", checkout_data["totals"][-1]["amount"]
-    )
+    logger.info("Successfully created cart: %s", cart_id)
+    logger.info("Current Total: %s cents", cart_data["totals"][-1]["amount"])
 
     # ==========================================================================
-
-    # STEP 2: Add More Items (Update Checkout)
-
+    # STEP 2: Add More Items (Update Cart)
     # ==========================================================================
 
     logger.info("\nSTEP 2: Adding a second item (Ceramic Pot)...")
 
     # Update Item 1 (Roses) - Keep quantity 1
-
     item1_update = item_update_request.ItemUpdateRequest(id="bouquet_roses")
-
     line_item1_update = line_item_update_request.LineItemUpdateRequest(
-      id=checkout_data["line_items"][0]["id"],
+      id=cart_data["line_items"][0]["id"],
       quantity=1,
       item=item1_update,
     )
 
     # Add Item 2 (Ceramic Pot) - Quantity 2
-
     item2_update = item_update_request.ItemUpdateRequest(id="pot_ceramic")
-
     line_item2_update = line_item_update_request.LineItemUpdateRequest(
-      id=str(uuid.uuid4()),
       quantity=2,
       item=item2_update,
     )
 
     # Construct the Update Payload
-
-    update_payload = checkout_update_request.CheckoutUpdateRequest(
-      id=checkout_id,
+    update_payload = cart_update_request.CartUpdateRequest(
+      id=cart_id,
       line_items=[line_item1_update, line_item2_update],
-      currency=checkout_data["currency"],
-      payment=checkout_data["payment"],
     )
 
     headers = get_headers()
-
-    url = f"/checkout-sessions/{checkout_id}"
+    url = f"/carts/{cart_id}"
 
     json_body = update_payload.model_dump(
       mode="json", by_alias=True, exclude_none=True
@@ -438,19 +402,12 @@ def main() -> int:
       headers=headers,
     )
 
-    checkout_data = response.json()
-
+    cart_data = response.json()
     extractions = {}
 
-    # Capture the new line item ID
-
-    # Assuming it's the second one since we just added it
-
-    if len(checkout_data.get("line_items", [])) > 1:
-      li_2_id = checkout_data["line_items"][1]["id"]
-
+    if len(cart_data.get("line_items", [])) > 1:
+      li_2_id = cart_data["line_items"][1]["id"]
       global_replacements[li_2_id] = "LINE_ITEM_2_ID"
-
       extractions["LINE_ITEM_2_ID"] = ".line_items[1].id"
 
     if args.export_requests_to:
@@ -461,48 +418,35 @@ def main() -> int:
         headers,
         json_body,
         response,
-        "Step 2: Add Items (Update Checkout)",
+        "Step 2: Add Items (Update Cart)",
         replacements=global_replacements,
         extractions=extractions,
       )
 
     if response.status_code != 200:
       logger.error("Failed to add items: %s", response.text)
-
       return 1
 
     logger.info("Successfully added items.")
-
-    logger.info("New Total: %s cents", checkout_data["totals"][-1]["amount"])
-
-    logger.info("Item Count: %d", len(checkout_data["line_items"]))
+    logger.info("New Total: %s cents", cart_data["totals"][-1]["amount"])
+    logger.info("Item Count: %d", len(cart_data["line_items"]))
 
     # ==========================================================================
-
-    # STEP 3: Apply Discount
-
+    # STEP 3: Apply Discount to Cart
     # ==========================================================================
 
-    logger.info("\nSTEP 3: Applying Discount (10%% OFF)...")
-
-    # Re-construct line items for update
-
-    # We need IDs from the current session
+    logger.info("\nSTEP 3: Applying Discount (10%% OFF) to Cart...")
 
     li_1 = next(
       li
-      for li in checkout_data["line_items"]
+      for li in cart_data["line_items"]
       if li["item"]["id"] == "bouquet_roses"
     )
-
     li_2 = next(
-      li
-      for li in checkout_data["line_items"]
-      if li["item"]["id"] == "pot_ceramic"
+      li for li in cart_data["line_items"] if li["item"]["id"] == "pot_ceramic"
     )
 
     item1_update = item_update_request.ItemUpdateRequest(id="bouquet_roses")
-
     line_item1_update = line_item_update_request.LineItemUpdateRequest(
       id=li_1["id"],
       quantity=1,
@@ -510,32 +454,24 @@ def main() -> int:
     )
 
     item2_update = item_update_request.ItemUpdateRequest(id="pot_ceramic")
-
     line_item2_update = line_item_update_request.LineItemUpdateRequest(
       id=li_2["id"],
       quantity=2,
       item=item2_update,
     )
 
-    # Construct the Update Payload
-
-    update_payload = checkout_update_request.CheckoutUpdateRequest(
-      id=checkout_id,
+    update_payload = cart_update_request.CartUpdateRequest(
+      id=cart_id,
       line_items=[line_item1_update, line_item2_update],
-      currency=checkout_data["currency"],
-      payment=checkout_data["payment"],
     )
 
     update_dict = update_payload.model_dump(
       mode="json", by_alias=True, exclude_none=True
     )
-
     update_dict["discounts"] = {"codes": ["10OFF"]}
 
     headers = get_headers()
-
-    url = f"/checkout-sessions/{checkout_id}"
-
+    url = f"/carts/{cart_id}"
     json_body = update_dict
 
     response = client.put(
@@ -552,45 +488,94 @@ def main() -> int:
         headers,
         json_body,
         response,
-        "Step 3: Apply Discount",
+        "Step 3: Apply Discount to Cart",
         replacements=global_replacements,
       )
 
     if response.status_code != 200:
       logger.error("Failed to apply discount: %s", response.text)
-
       return 1
 
-    checkout_data = response.json()
-
-    logger.info("Successfully applied discount.")
-
-    logger.info("New Total: %s cents", checkout_data["totals"][-1]["amount"])
-
-    discounts_applied = checkout_data.get("discounts", {}).get("applied", [])
+    cart_data = response.json()
+    logger.info("Successfully applied discount to cart.")
+    logger.info("New Total: %s cents", cart_data["totals"][-1]["amount"])
 
+    discounts_applied = cart_data.get("discounts", {}).get("applied", [])
     if discounts_applied:
       logger.info(
         "Applied Discounts: %s", [d["code"] for d in discounts_applied]
       )
-
     else:
       logger.warning("No discounts applied!")
 
+    # ==========================================================================
+    # STEP 4: Convert Cart to Checkout (Create Checkout Session)
+    # ==========================================================================
+
+    logger.info("\nSTEP 4: Creating Checkout Session from Cart...")
+
+    # We only need cart_id. The server will inherit everything else from
+    # the cart as per UCP Cart-to-Checkout conversion specification.
+    checkout_payload = {
+      "cart_id": cart_id,
+    }
+
+    headers = get_headers()
+    url = "/checkout-sessions"
+    json_body = checkout_payload
+
+    response = client.post(
+      url,
+      json=json_body,
+      headers=headers,
+    )
+
+    checkout_data = response.json()
+    checkout_id = checkout_data.get("id")
+
+    extractions = {}
+    if checkout_id:
+      global_replacements[checkout_id] = "CHECKOUT_ID"
+      extractions["CHECKOUT_ID"] = ".id"
+
+    if args.export_requests_to:
+      log_interaction(
+        args.export_requests_to,
+        "POST",
+        f"{args.server_url}{url}",
+        headers,
+        json_body,
+        response,
+        "Step 4: Create Checkout Session from Cart",
+        replacements=global_replacements,
+        extractions=extractions,
+      )
+
+    if response.status_code not in [200, 201]:
+      logger.error("Failed to create checkout from cart: %s", response.text)
+      return 1
+
+    logger.info(
+      "Successfully created checkout session from cart: %s", checkout_id
+    )
+    logger.info(
+      "Current Total: %s cents", checkout_data["totals"][-1]["amount"]
+    )
+
     # ==========================================================================
 
-    # STEP 4: Select Fulfillment Option
+    # STEP 5: Select Fulfillment Option
 
     # ==========================================================================
 
-    logger.info("\nSTEP 4: Selecting Fulfillment Option...")
+    logger.info("\nSTEP 5: Selecting Fulfillment Option...")
 
     # Ensure fulfillment options are generated
 
     if not checkout_data.get("fulfillment") or not checkout_data[
       "fulfillment"
     ].get("methods"):
-      logger.info("STEP 4: Triggering fulfillment option generation...")
+      logger.info("STEP 5: Triggering fulfillment option generation...")
 
       # Re-construct line items for update to satisfy strict validation
 
@@ -630,7 +615,7 @@ def main() -> int:
         id=checkout_id,
         line_items=[line_item1_update, line_item2_update],
         currency=checkout_data["currency"],
-        payment=checkout_data["payment"],
+        payment=checkout_data.get("payment"),
         fulfillment={
           "methods": [
             {
@@ -693,7 +678,7 @@ def main() -> int:
           headers,
           trigger_payload,
           response,
-          "Step 4: Trigger Fulfillment",
+          "Step 5: Trigger Fulfillment",
           replacements=global_replacements,
           extractions=extractions,
         )
@@ -712,7 +697,7 @@ def main() -> int:
       if method.get("destinations"):
         dest_id = method["destinations"][0]["id"]
 
-        logger.info("STEP 5: Selecting destination: %s", dest_id)
+        logger.info("STEP 6: Selecting destination: %s", dest_id)
 
         # 1. Select Destination to calculate options
 
@@ -751,7 +736,7 @@ def main() -> int:
             headers,
             payload,
             response,
-            "Step 5: Select Destination",
+            "Step 6: Select Destination",
             replacements=global_replacements,
           )
 
@@ -769,7 +754,7 @@ def main() -> int:
         if method.get("groups") and method["groups"][0].get("options"):
           option_id = method["groups"][0]["options"][0]["id"]
 
-          logger.info("STEP 6: Selecting option: %s", option_id)
+          logger.info("STEP 7: Selecting option: %s", option_id)
 
           trigger_request.fulfillment = {
             "methods": [
@@ -809,7 +794,7 @@ def main() -> int:
               headers,
               payload,
               response,
-              "Step 6: Select Option",
+              "Step 7: Select Option",
               replacements=global_replacements,
             )
 
@@ -828,11 +813,11 @@ def main() -> int:
 
     # ==========================================================================
 
-    # STEP 7: Complete Checkout (Payment)
+    # STEP 8: Complete Checkout (Payment)
 
     # ==========================================================================
 
-    logger.info("\nSTEP 7: Processing Payment...")
+    logger.info("\nSTEP 8: Processing Payment...")
 
     # We use the 'mock_payment_handler' discovered in Step 0.
 
@@ -911,7 +896,7 @@ def main() -> int:
         headers,
         final_payload,
         response,
-        "Step 7: Complete Checkout",
+        "Step 8: Complete Checkout",
         replacements=global_replacements,
         extractions=extractions,
       )
diff --git a/rest/python/server/cart_test.py b/rest/python/server/cart_test.py
new file mode 100644
index 00000000..50de25a7
--- /dev/null
+++ b/rest/python/server/cart_test.py
@@ -0,0 +1,352 @@
+#   Copyright 2026 UCP Authors
+#
+#   Licensed under the Apache License, Version 2.0 (the "License");
+#   you may not use this file except in compliance with the License.
+#   You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+#   Unless required by applicable law or agreed to in writing, software
+#   distributed under the License is distributed on an "AS IS" BASIS,
+#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#   See the License for the specific language governing permissions and
+#   limitations under the License.
+
+"""Integration tests for the UCP Cart capability."""
+
+import asyncio
+from absl.testing import absltest
+from integration_test import IntegrationTest, TestCheckout
+from models import UnifiedCart as Cart
+from sqlalchemy.sql import delete
+from ucp_sdk.models.schemas.shopping import (
+  cart_create_request as cart_create_req,
+)
+from ucp_sdk.models.schemas.shopping import (
+  cart_update_request as cart_update_req,
+)
+from ucp_sdk.models.schemas.shopping.types import (
+  item_create_request as item_create_req,
+)
+from ucp_sdk.models.schemas.shopping.types import (
+  line_item_create_request as line_item_create_req,
+)
+from ucp_sdk.models.schemas.shopping.types import (
+  line_item_update_request as line_item_update_req,
+)
+import db
+
+
+class CartIntegrationTest(IntegrationTest):
+  """Integration tests for Cart capability."""
+
+  def _create_cart_payload(
+    self,
+    items: list[tuple[str, int]],
+  ) -> cart_create_req.CartCreateRequest:
+    """Create a cart payload using SDK models."""
+    line_items = []
+    for item_id, quantity in items:
+      item = item_create_req.ItemCreateRequest(id=item_id)
+      line_item = line_item_create_req.LineItemCreateRequest(
+        quantity=quantity, item=item
+      )
+      line_items.append(line_item)
+
+    return cart_create_req.CartCreateRequest(
+      line_items=line_items,
+    )
+
+  def test_cart_lifecycle(self) -> None:
+    """Test Create, Get, Update, and Cancel Cart."""
+    with self.client:
+      # 1. Create Cart
+      payload = self._create_cart_payload([("rose", 2)])
+      response = self.client.post(
+        "/carts",
+        headers=self._get_headers(idempotency_key="cart_1", request_id="r1"),
+        json=payload.model_dump(mode="json", exclude_none=True),
+      )
+      self.assertEqual(response.status_code, 201, f"Response: {response.text}")
+      cart = Cart.model_validate(response.json())
+      self.assertTrue(cart.id.startswith("cart_"))
+      self.assertEqual(len(cart.line_items), 1)
+      self.assertEqual(cart.line_items[0].item.id, "rose")
+      self.assertEqual(cart.line_items[0].quantity, 2)
+      self.assertEqual(cart.line_items[0].item.price, 1000)
+      # Totals: subtotal = 2000, total = 2000
+      subtotal = next(t.amount for t in cart.totals if t.type == "subtotal")
+      total = next(t.amount for t in cart.totals if t.type == "total")
+      self.assertEqual(subtotal, 2000)
+      self.assertEqual(total, 2000)
+
+      cart_id = cart.id
+
+      # 2. Get Cart
+      response = self.client.get(
+        f"/carts/{cart_id}",
+        headers=self._get_headers(request_id="r2"),
+      )
+      self.assertEqual(response.status_code, 200, response.text)
+      cart = Cart.model_validate(response.json())
+      self.assertEqual(cart.id, cart_id)
+      self.assertEqual(cart.line_items[0].quantity, 2)
+
+      # 3. Update Cart (Replace items: 1 rose, 1 tulip)
+      line_items_update = [
+        line_item_update_req.LineItemUpdateRequest(
+          item={"id": "rose"},
+          quantity=1,
+        ),
+        line_item_update_req.LineItemUpdateRequest(
+          item={"id": "tulip"},
+          quantity=1,
+        ),
+      ]
+      update_payload = cart_update_req.CartUpdateRequest(
+        id=cart_id,
+        line_items=line_items_update,
+      )
+      response = self.client.put(
+        f"/carts/{cart_id}",
+        headers=self._get_headers(idempotency_key="cart_2", request_id="r3"),
+        json=update_payload.model_dump(mode="json", exclude_none=True),
+      )
+      self.assertEqual(response.status_code, 200, response.text)
+      cart = Cart.model_validate(response.json())
+      self.assertEqual(cart.id, cart_id)
+      self.assertEqual(len(cart.line_items), 2)
+      # Totals: 1000 (rose) + 800 (tulip) = 1800
+      subtotal = next(t.amount for t in cart.totals if t.type == "subtotal")
+      total = next(t.amount for t in cart.totals if t.type == "total")
+      self.assertEqual(subtotal, 1800)
+      self.assertEqual(total, 1800)
+
+      # 4. Cancel Cart
+      response = self.client.post(
+        f"/carts/{cart_id}/cancel",
+        headers=self._get_headers(idempotency_key="cart_3", request_id="r4"),
+      )
+      self.assertEqual(response.status_code, 200, response.text)
+      cart = Cart.model_validate(response.json())
+      self.assertEqual(cart.id, cart_id)
+
+      # 5. Verify Get Cart returns Not Found (HTTP 404 in our case because we
+      # raise ResourceNotFoundError)
+      response = self.client.get(
+        f"/carts/{cart_id}",
+        headers=self._get_headers(request_id="r5"),
+      )
+      self.assertEqual(response.status_code, 404, response.text)
+      data = response.json()
+      self.assertEqual(data["ucp"]["status"], "error")
+      self.assertEqual(data["messages"][0]["code"], "RESOURCE_NOT_FOUND")
+
+  def test_cart_to_checkout_conversion(self) -> None:
+    """Test converting a cart to a checkout session."""
+    with self.client:
+      # 1. Create Cart
+      payload = self._create_cart_payload([("rose", 2)])
+      response = self.client.post(
+        "/carts",
+        headers=self._get_headers(idempotency_key="c_conv_1", request_id="rc1"),
+        json=payload.model_dump(mode="json", exclude_none=True),
+      )
+      self.assertEqual(response.status_code, 201)
+      cart = Cart.model_validate(response.json())
+      cart_id = cart.id
+
+      # 2. Create Checkout using cart_id
+      checkout_payload = self._create_checkout_payload(
+        "test_checkout_from_cart", [("rose", "Red Rose", 1000, 99)]
+      ).model_dump(mode="json", exclude_none=True)
+      checkout_payload["cart_id"] = cart_id
+
+      response = self.client.post(
+        "/checkout-sessions",
+        headers=self._get_headers(idempotency_key="c_conv_2", request_id="rc2"),
+        json=checkout_payload,
+      )
+      self.assertEqual(response.status_code, 201, response.text)
+      checkout = TestCheckout.model_validate(response.json())
+      self.assertEqual(
+        self.get_resource_id(checkout.id), "test_checkout_from_cart"
+      )
+      self.assertEqual(checkout.cart_id, cart_id)
+      self.assertEqual(len(checkout.line_items), 1)
+      self.assertEqual(checkout.line_items[0].item.id, "rose")
+      self.assertEqual(checkout.line_items[0].quantity, 2)
+      subtotal = next(t.amount for t in checkout.totals if t.type == "subtotal")
+      self.assertEqual(subtotal, 2000)
+
+      # 3. Idempotent Conversion: Create another checkout with same cart_id
+      checkout_payload_2 = self._create_checkout_payload(
+        "test_checkout_from_cart_2", [("rose", "Red Rose", 1000, 1)]
+      ).model_dump(mode="json", exclude_none=True)
+      checkout_payload_2["cart_id"] = cart_id
+
+      response = self.client.post(
+        "/checkout-sessions",
+        headers=self._get_headers(idempotency_key="c_conv_3", request_id="rc3"),
+        json=checkout_payload_2,
+      )
+      self.assertEqual(response.status_code, 201)
+      checkout_2 = TestCheckout.model_validate(response.json())
+      self.assertEqual(
+        self.get_resource_id(checkout_2.id), "test_checkout_from_cart"
+      )
+      self.assertEqual(checkout_2.cart_id, cart_id)
+
+      # 4. Complete Checkout
+      payment_payload = self._create_payment_payload()
+      response = self.client.post(
+        "/checkout-sessions/test_checkout_from_cart/complete",
+        headers=self._get_headers(idempotency_key="c_conv_4", request_id="rc4"),
+        json=payment_payload,
+      )
+      self.assertEqual(response.status_code, 200, response.text)
+      checkout_comp = TestCheckout.model_validate(response.json())
+      self.assertEqual(checkout_comp.status, "completed")
+
+      # 5. Verify Cart is cleared (deleted) after completion
+      response = self.client.get(
+        f"/carts/{cart_id}",
+        headers=self._get_headers(request_id="rc5"),
+      )
+      self.assertEqual(response.status_code, 404, response.text)
+
+  def test_cart_with_discount(self) -> None:
+    """Test applying a discount code to a cart."""
+
+    async def seed_discount() -> None:
+      async with self.transactions_session_factory() as session:
+        await session.execute(delete(db.Discount))
+        session.add(
+          db.Discount(
+            code="10OFF", type="percentage", value=10, description="10% Off"
+          )
+        )
+        await session.commit()
+
+    asyncio.run(seed_discount())
+
+    with self.client:
+      # 1. Create Cart
+      payload = self._create_cart_payload([("rose", 2)])
+      response = self.client.post(
+        "/carts",
+        headers=self._get_headers(
+          idempotency_key="cart_disc_1", request_id="rd1"
+        ),
+        json=payload.model_dump(mode="json", exclude_none=True),
+      )
+      self.assertEqual(response.status_code, 201)
+      cart = Cart.model_validate(response.json())
+      cart_id = cart.id
+
+      # 2. Update Cart with discount code
+      update_payload = {
+        "id": cart_id,
+        "line_items": [
+          {"item": {"id": "rose"}, "quantity": 2},
+        ],
+        "discounts": {"codes": ["10OFF"]},
+      }
+      response = self.client.put(
+        f"/carts/{cart_id}",
+        headers=self._get_headers(
+          idempotency_key="cart_disc_2", request_id="rd2"
+        ),
+        json=update_payload,
+      )
+      self.assertEqual(response.status_code, 200, response.text)
+      cart = Cart.model_validate(response.json())
+      self.assertEqual(cart.id, cart_id)
+
+      # Verify discounts in response
+      self.assertIsNotNone(cart.discounts)
+      self.assertEqual(cart.discounts.codes, ["10OFF"])
+      self.assertEqual(len(cart.discounts.applied), 1)
+      self.assertEqual(cart.discounts.applied[0].code, "10OFF")
+      self.assertEqual(cart.discounts.applied[0].amount, 200)
+
+      # Verify totals
+      subtotal = next(t.amount for t in cart.totals if t.type == "subtotal")
+      discount = next(t.amount for t in cart.totals if t.type == "discount")
+      total = next(t.amount for t in cart.totals if t.type == "total")
+      self.assertEqual(subtotal, 2000)
+      self.assertEqual(discount, -200)
+      self.assertEqual(total, 1800)
+
+  def test_cart_to_checkout_conversion_with_discount(self) -> None:
+    """Test discount carry-forward during cart-to-checkout conversion."""
+
+    async def seed_discount() -> None:
+      async with self.transactions_session_factory() as session:
+        await session.execute(delete(db.Discount))
+        session.add(
+          db.Discount(
+            code="10OFF", type="percentage", value=10, description="10% Off"
+          )
+        )
+        await session.commit()
+
+    asyncio.run(seed_discount())
+
+    with self.client:
+      # 1. Create Cart with discount
+      create_payload = self._create_cart_payload([("rose", 2)]).model_dump(
+        mode="json", exclude_none=True
+      )
+      create_payload["discounts"] = {"codes": ["10OFF"]}
+
+      response = self.client.post(
+        "/carts",
+        headers=self._get_headers(
+          idempotency_key="cart_c_disc_1", request_id="rcd1"
+        ),
+        json=create_payload,
+      )
+      self.assertEqual(response.status_code, 201)
+      cart = Cart.model_validate(response.json())
+      cart_id = cart.id
+      self.assertEqual(len(cart.discounts.applied), 1)
+
+      # 2. Convert to Checkout
+      checkout_payload = self._create_checkout_payload(
+        "test_checkout_from_cart_disc", [("rose", "Red Rose", 1000, 2)]
+      ).model_dump(mode="json", exclude_none=True)
+      checkout_payload["cart_id"] = cart_id
+
+      response = self.client.post(
+        "/checkout-sessions",
+        headers=self._get_headers(
+          idempotency_key="cart_c_disc_2", request_id="rcd2"
+        ),
+        json=checkout_payload,
+      )
+      self.assertEqual(response.status_code, 201, response.text)
+      checkout = TestCheckout.model_validate(response.json())
+      self.assertEqual(
+        self.get_resource_id(checkout.id), "test_checkout_from_cart_disc"
+      )
+      self.assertEqual(checkout.cart_id, cart_id)
+
+      # Verify discounts carried forward
+      self.assertIsNotNone(checkout.discounts)
+      self.assertEqual(checkout.discounts.codes, ["10OFF"])
+      self.assertEqual(len(checkout.discounts.applied), 1)
+      self.assertEqual(checkout.discounts.applied[0].code, "10OFF")
+      self.assertEqual(checkout.discounts.applied[0].amount, 200)
+
+      # Verify totals
+      subtotal = next(t.amount for t in checkout.totals if t.type == "subtotal")
+      discount = next(t.amount for t in checkout.totals if t.type == "discount")
+      total = next(t.amount for t in checkout.totals if t.type == "total")
+      self.assertEqual(subtotal, 2000)
+      self.assertEqual(discount, -200)
+      self.assertEqual(total, 1800)
+
+
+if __name__ == "__main__":
+  absltest.main()
diff --git a/rest/python/server/config.py b/rest/python/server/config.py
index aec6ee9a..81f2e5d5 100644
--- a/rest/python/server/config.py
+++ b/rest/python/server/config.py
@@ -24,7 +24,7 @@
 
 FLAGS = flags.FLAGS
 
-_SERVER_VERSION_CACHE = None
+_PROFILE_CACHE = None
 
 # checkout.json annotates `currency` with `ucp_request: omit` and describes
 # it as "reflecting the merchant's market determination ... buyers provide
@@ -39,19 +39,29 @@ def get_default_currency() -> str:
   return DEFAULT_CURRENCY
 
 
-def get_server_version() -> str:
-  """Read and cache the server version from the discovery profile."""
-  global _SERVER_VERSION_CACHE
-  if _SERVER_VERSION_CACHE:
-    return _SERVER_VERSION_CACHE
+def _get_profile() -> dict:
+  global _PROFILE_CACHE
+  if _PROFILE_CACHE:
+    return _PROFILE_CACHE
 
   current_dir = Path(__file__).resolve().parent
   profile_path = current_dir / "routes" / "discovery_profile.json"
 
-  with profile_path.open() as f:
-    data = json.load(f)
-    _SERVER_VERSION_CACHE = data["ucp"]["version"]
-    return _SERVER_VERSION_CACHE
+  with profile_path.open(encoding="utf-8") as f:
+    _PROFILE_CACHE = json.load(f)
+  return _PROFILE_CACHE
+
+
+def get_server_version() -> str:
+  """Read and cache the server version from the discovery profile."""
+  profile = _get_profile()
+  return profile["ucp"]["version"]
+
+
+def get_payment_handlers() -> dict:
+  """Read and cache the payment handlers from the discovery profile."""
+  profile = _get_profile()
+  return profile["ucp"].get("payment_handlers", {})
 
 
 # Define flags only if they haven't been defined yet (to avoid duplicates
diff --git a/rest/python/server/db.py b/rest/python/server/db.py
index 16b4d969..130918f2 100644
--- a/rest/python/server/db.py
+++ b/rest/python/server/db.py
@@ -184,6 +184,15 @@ class CheckoutSession(TransactionBase):
   data = Column(JSON)
 
 
+class CartSession(TransactionBase):
+  """Cart session database model."""
+
+  __tablename__ = "carts"
+
+  id = Column(String, primary_key=True)
+  data = Column(JSON)
+
+
 class Order(TransactionBase):
   """Order database model."""
 
@@ -461,6 +470,48 @@ async def get_checkout_session(
   return None
 
 
+async def save_cart(
+  session: AsyncSession,
+  cart_id: str,
+  cart_obj: dict[str, Any],
+) -> None:
+  """Save or update a cart session."""
+  existing = await session.get(CartSession, cart_id)
+  if existing:
+    existing.data = cart_obj
+  else:
+    new_cart = CartSession(id=cart_id, data=cart_obj)
+    session.add(new_cart)
+
+
+async def get_cart_session(
+  session: AsyncSession, cart_id: str
+) -> dict[str, Any] | None:
+  """Retrieve a cart session by ID."""
+  result = await session.get(CartSession, cart_id)
+  if result:
+    return result.data
+  return None
+
+
+async def delete_cart_session(session: AsyncSession, cart_id: str) -> None:
+  """Delete a cart session by ID."""
+  existing = await session.get(CartSession, cart_id)
+  if existing:
+    await session.delete(existing)
+
+
+async def get_checkouts_by_cart_id(
+  session: AsyncSession, cart_id: str
+) -> list[dict[str, Any]]:
+  """Retrieve all checkout sessions by cart ID."""
+  stmt = select(CheckoutSession).where(
+    CheckoutSession.data["cart_id"].as_string() == cart_id
+  )
+  result = await session.execute(stmt)
+  return [r.data for r in result.scalars().all()]
+
+
 async def save_order(
   session: AsyncSession, order_id: str, order_obj: dict[str, Any]
 ) -> None:
diff --git a/rest/python/server/dependencies.py b/rest/python/server/dependencies.py
index 599e06f8..1434c7d3 100644
--- a/rest/python/server/dependencies.py
+++ b/rest/python/server/dependencies.py
@@ -36,6 +36,7 @@
 from fastapi import HTTPException
 from fastapi import Request
 from pydantic import BaseModel
+from services.cart_service import CartService
 from services.checkout_service import CheckoutService
 from services.fulfillment_service import FulfillmentService
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -256,3 +257,16 @@ def get_checkout_service(
     transactions_session,
     str(request.base_url),
   )
+
+
+def get_cart_service(
+  request: Request,
+  products_session: Annotated[AsyncSession, Depends(get_products_db)],
+  transactions_session: Annotated[AsyncSession, Depends(get_transactions_db)],
+) -> CartService:
+  """Dependency provider for CartService."""
+  return CartService(
+    products_session,
+    transactions_session,
+    str(request.base_url),
+  )
diff --git a/rest/python/server/generated_routes/ucp_routes.py b/rest/python/server/generated_routes/ucp_routes.py
index 06e0d18d..880552b2 100644
--- a/rest/python/server/generated_routes/ucp_routes.py
+++ b/rest/python/server/generated_routes/ucp_routes.py
@@ -2,6 +2,9 @@
 
 from typing import Annotated
 from fastapi import APIRouter, Body, Header
+import ucp_sdk.models.schemas.shopping.cart
+import ucp_sdk.models.schemas.shopping.cart_create_request
+import ucp_sdk.models.schemas.shopping.cart_update_request
 import ucp_sdk.models.schemas.shopping.checkout_create_request
 import ucp_sdk.models.schemas.shopping.checkout
 import ucp_sdk.models.schemas.shopping.checkout_update_request
@@ -167,3 +170,113 @@ async def order_event_webhook(
   """Order Event Webhook."""
   # TODO: Implement logic
   return {}
+
+
+@router.post(
+  "/carts",
+  response_model=ucp_sdk.models.schemas.shopping.cart.Cart,
+  response_model_exclude_none=True,
+  status_code=201,
+  operation_id="create_cart",
+  summary="Create Cart",
+)
+async def create_cart(
+  body: Annotated[
+    ucp_sdk.models.schemas.shopping.cart_create_request.CartCreateRequest,
+    Body(...),
+  ],
+  authorization: str = Header(None, alias="Authorization"),
+  x_api_key: str = Header(None, alias="X-API-Key"),
+  request_signature: str = Header(..., alias="Request-Signature"),
+  idempotency_key: str = Header(..., alias="Idempotency-Key"),
+  request_id: str = Header(..., alias="Request-Id"),
+  user_agent: str = Header(None, alias="User-Agent"),
+  content_type: str = Header(None, alias="Content-Type"),
+  accept: str = Header(None, alias="Accept"),
+  accept_language: str = Header(None, alias="Accept-Language"),
+  accept_encoding: str = Header(None, alias="Accept-Encoding"),
+):
+  """Create Cart."""
+  # TODO: Implement logic
+  return {}
+
+
+@router.get(
+  "/carts/{id}",
+  response_model=ucp_sdk.models.schemas.shopping.cart.Cart,
+  response_model_exclude_none=True,
+  status_code=200,
+  operation_id="get_cart",
+  summary="Get Cart",
+)
+async def get_cart(
+  id: str,
+  authorization: str = Header(None, alias="Authorization"),
+  x_api_key: str = Header(None, alias="X-API-Key"),
+  request_signature: str = Header(..., alias="Request-Signature"),
+  request_id: str = Header(..., alias="Request-Id"),
+  user_agent: str = Header(None, alias="User-Agent"),
+  content_type: str = Header(None, alias="Content-Type"),
+  accept: str = Header(None, alias="Accept"),
+  accept_language: str = Header(None, alias="Accept-Language"),
+  accept_encoding: str = Header(None, alias="Accept-Encoding"),
+):
+  """Get Cart."""
+  # TODO: Implement logic
+  return {}
+
+
+@router.put(
+  "/carts/{id}",
+  response_model=ucp_sdk.models.schemas.shopping.cart.Cart,
+  response_model_exclude_none=True,
+  status_code=200,
+  operation_id="update_cart",
+  summary="Update Cart",
+)
+async def update_cart(
+  id: str,
+  body: Annotated[
+    ucp_sdk.models.schemas.shopping.cart_update_request.CartUpdateRequest,
+    Body(...),
+  ],
+  authorization: str = Header(None, alias="Authorization"),
+  x_api_key: str = Header(None, alias="X-API-Key"),
+  request_signature: str = Header(..., alias="Request-Signature"),
+  idempotency_key: str = Header(..., alias="Idempotency-Key"),
+  request_id: str = Header(..., alias="Request-Id"),
+  user_agent: str = Header(None, alias="User-Agent"),
+  content_type: str = Header(None, alias="Content-Type"),
+  accept: str = Header(None, alias="Accept"),
+  accept_language: str = Header(None, alias="Accept-Language"),
+  accept_encoding: str = Header(None, alias="Accept-Encoding"),
+):
+  """Update Cart."""
+  # TODO: Implement logic
+  return {}
+
+
+@router.post(
+  "/carts/{id}/cancel",
+  response_model=ucp_sdk.models.schemas.shopping.cart.Cart,
+  response_model_exclude_none=True,
+  status_code=200,
+  operation_id="cancel_cart",
+  summary="Cancel Cart",
+)
+async def cancel_cart(
+  id: str,
+  authorization: str = Header(None, alias="Authorization"),
+  x_api_key: str = Header(None, alias="X-API-Key"),
+  request_signature: str = Header(..., alias="Request-Signature"),
+  idempotency_key: str = Header(..., alias="Idempotency-Key"),
+  request_id: str = Header(..., alias="Request-Id"),
+  user_agent: str = Header(None, alias="User-Agent"),
+  content_type: str = Header(None, alias="Content-Type"),
+  accept: str = Header(None, alias="Accept"),
+  accept_language: str = Header(None, alias="Accept-Language"),
+  accept_encoding: str = Header(None, alias="Accept-Encoding"),
+):
+  """Cancel Cart."""
+  # TODO: Implement logic
+  return {}
diff --git a/rest/python/server/integration_test.py b/rest/python/server/integration_test.py
index 9e140f72..6273b98e 100644
--- a/rest/python/server/integration_test.py
+++ b/rest/python/server/integration_test.py
@@ -93,6 +93,7 @@ class TestCheckout(
   """Checkout model supporting Fulfillment, Discount, and AP2 extensions."""
 
   platform: PlatformSchema | None = None
+  cart_id: str | None = None
 
 
 class IntegrationTest(absltest.TestCase):
diff --git a/rest/python/server/models.py b/rest/python/server/models.py
index e1197c02..c9f5de7e 100644
--- a/rest/python/server/models.py
+++ b/rest/python/server/models.py
@@ -20,6 +20,10 @@
 """
 
 from typing import Any
+from pydantic import model_validator
+from ucp_sdk.models.schemas.shopping.types.line_item_create_request import (
+  LineItemCreateRequest,
+)
 from ucp_sdk.models.schemas.shopping.ap2_mandate import Checkout as Ap2Checkout
 from ucp_sdk.models.schemas.shopping.buyer_consent import (
   Checkout as BuyerConsentCheckoutResp,
@@ -27,6 +31,7 @@
 from ucp_sdk.models.schemas.shopping.discount import (
   Checkout as DiscountCheckoutResp,
   DiscountsObject,
+  Cart as DiscountCart,
 )
 from ucp_sdk.models.schemas.shopping.fulfillment import (
   Checkout as FulfillmentCheckout,
@@ -42,12 +47,36 @@
 from ucp_sdk.models.schemas.shopping.checkout_update_request import (
   CheckoutUpdateRequest,
 )
+from ucp_sdk.models.schemas.shopping.cart_create_request import (
+  CartCreateRequest,
+)
+from ucp_sdk.models.schemas.shopping.cart_update_request import (
+  CartUpdateRequest,
+)
 
 
 class UnifiedOrder(Order):
   """Order model supporting extensions."""
 
 
+class UnifiedCart(DiscountCart):
+  """Cart model supporting Discount extension."""
+
+  pass
+
+
+class UnifiedCartCreateRequest(CartCreateRequest):
+  """Cart create request supporting Discount extension."""
+
+  discounts: DiscountsObject | None = None
+
+
+class UnifiedCartUpdateRequest(CartUpdateRequest):
+  """Cart update request supporting Discount extension."""
+
+  discounts: DiscountsObject | None = None
+
+
 class UnifiedCheckout(
   BuyerConsentCheckoutResp,
   FulfillmentCheckout,
@@ -57,14 +86,24 @@ class UnifiedCheckout(
   """Checkout model supporting various extensions."""
 
   platform: PlatformSchema | None = None
+  cart_id: str | None = None
 
 
 class UnifiedCheckoutCreateRequest(CheckoutCreateRequest):
   """Create request model combining base fields and extensions."""
 
+  line_items: list[LineItemCreateRequest] | None = None
   fulfillment: Fulfillment | None = None
   discounts: DiscountsObject | None = None
   buyer_consent: Any | None = None
+  cart_id: str | None = None
+
+  @model_validator(mode="after")
+  def validate_cart_id_or_line_items(self) -> "UnifiedCheckoutCreateRequest":
+    """Validate that either cart_id or line_items is provided."""
+    if not self.cart_id and not self.line_items:
+      raise ValueError("Either cart_id or line_items must be provided")
+    return self
 
 
 class UnifiedCheckoutUpdateRequest(CheckoutUpdateRequest):
@@ -78,4 +117,7 @@ class UnifiedCheckoutUpdateRequest(CheckoutUpdateRequest):
 UnifiedCheckout.model_rebuild()
 UnifiedCheckoutCreateRequest.model_rebuild()
 UnifiedCheckoutUpdateRequest.model_rebuild()
+UnifiedCart.model_rebuild()
+UnifiedCartCreateRequest.model_rebuild()
+UnifiedCartUpdateRequest.model_rebuild()
 UnifiedOrder.model_rebuild()
diff --git a/rest/python/server/routes/discovery_profile.json b/rest/python/server/routes/discovery_profile.json
index b359718d..5c031fea 100644
--- a/rest/python/server/routes/discovery_profile.json
+++ b/rest/python/server/routes/discovery_profile.json
@@ -20,6 +20,13 @@
           "schema": "https://ucp.dev/2026-04-08/schemas/shopping/checkout.json"
         }
       ],
+      "dev.ucp.shopping.cart": [
+        {
+          "version": "2026-04-08",
+          "spec": "https://ucp.dev/2026-04-08/specification/cart",
+          "schema": "https://ucp.dev/2026-04-08/schemas/shopping/cart.json"
+        }
+      ],
       "dev.ucp.shopping.order": [
         {
           "version": "2026-04-08",
@@ -32,7 +39,7 @@
           "version": "2026-04-08",
           "spec": "https://ucp.dev/2026-04-08/specification/discount",
           "schema": "https://ucp.dev/2026-04-08/schemas/shopping/discount.json",
-          "extends": "dev.ucp.shopping.checkout"
+          "extends": ["dev.ucp.shopping.checkout", "dev.ucp.shopping.cart"]
         }
       ],
       "dev.ucp.shopping.fulfillment": [
diff --git a/rest/python/server/routes/ucp_implementation.py b/rest/python/server/routes/ucp_implementation.py
index 1d2578f2..5bd4ef02 100644
--- a/rest/python/server/routes/ucp_implementation.py
+++ b/rest/python/server/routes/ucp_implementation.py
@@ -29,9 +29,14 @@
 from fastapi.routing import APIRoute
 import httpx
 import models
-from models import UnifiedCheckoutCreateRequest
+from models import (
+  UnifiedCartCreateRequest,
+  UnifiedCartUpdateRequest,
+  UnifiedCheckoutCreateRequest,
+)
 from pydantic import BaseModel
 from pydantic import HttpUrl
+from services.cart_service import CartService
 from services.checkout_service import CheckoutService
 from ucp_sdk.models.schemas.shopping.checkout_complete_request import (
   CheckoutCompleteRequest,
@@ -245,10 +250,71 @@ async def cancel_checkout(
   ],
 ) -> models.UnifiedCheckout:
   """Cancel Checkout Implementation."""
-  del common_headers  # Unused
   return await checkout_service.cancel_checkout(checkout_id, idempotency_key)
 
 
+async def create_cart(
+  cart_req: Annotated[
+    UnifiedCartCreateRequest,
+    Body(...),
+  ],
+  common_headers: Annotated[
+    dependencies.CommonHeaders, Depends(dependencies.common_headers)
+  ],
+  idempotency_key: Annotated[str, Depends(dependencies.idempotency_header)],
+  cart_service: Annotated[CartService, Depends(dependencies.get_cart_service)],
+) -> dict[str, Any]:
+  """Create Cart Implementation."""
+  del common_headers  # Unused
+  result = await cart_service.create_cart(cart_req, idempotency_key)
+  return result.model_dump(mode="json", by_alias=True, exclude_none=True)
+
+
+async def get_cart(
+  cart_id: Annotated[str, Path(..., alias="id")],
+  common_headers: Annotated[
+    dependencies.CommonHeaders, Depends(dependencies.common_headers)
+  ],
+  cart_service: Annotated[CartService, Depends(dependencies.get_cart_service)],
+) -> dict[str, Any]:
+  """Get Cart Implementation."""
+  del common_headers  # Unused
+  result = await cart_service.get_cart(cart_id)
+  return result.model_dump(mode="json", by_alias=True, exclude_none=True)
+
+
+async def update_cart(
+  cart_id: Annotated[str, Path(..., alias="id")],
+  cart_req: Annotated[
+    UnifiedCartUpdateRequest,
+    Body(...),
+  ],
+  common_headers: Annotated[
+    dependencies.CommonHeaders, Depends(dependencies.common_headers)
+  ],
+  idempotency_key: Annotated[str, Depends(dependencies.idempotency_header)],
+  cart_service: Annotated[CartService, Depends(dependencies.get_cart_service)],
+) -> dict[str, Any]:
+  """Update Cart Implementation."""
+  del common_headers  # Unused
+  result = await cart_service.update_cart(cart_id, cart_req, idempotency_key)
+  return result.model_dump(mode="json", by_alias=True, exclude_none=True)
+
+
+async def cancel_cart(
+  cart_id: Annotated[str, Path(..., alias="id")],
+  common_headers: Annotated[
+    dependencies.CommonHeaders, Depends(dependencies.common_headers)
+  ],
+  idempotency_key: Annotated[str, Depends(dependencies.idempotency_header)],
+  cart_service: Annotated[CartService, Depends(dependencies.get_cart_service)],
+) -> dict[str, Any]:
+  """Cancel Cart Implementation."""
+  del common_headers  # Unused
+  result = await cart_service.cancel_cart(cart_id, idempotency_key)
+  return result.model_dump(mode="json", by_alias=True, exclude_none=True)
+
+
 async def order_event_webhook(
   partner_id: str,
   payload: Annotated[Order, Body(...)],
@@ -275,6 +341,10 @@ async def order_event_webhook(
   "complete_checkout": complete_checkout,
   "cancel_checkout": cancel_checkout,
   "order_event_webhook": order_event_webhook,
+  "create_cart": create_cart,
+  "get_cart": get_cart,
+  "update_cart": update_cart,
+  "cancel_cart": cancel_cart,
 }
 
 
diff --git a/rest/python/server/services/cart_service.py b/rest/python/server/services/cart_service.py
new file mode 100644
index 00000000..87998bf8
--- /dev/null
+++ b/rest/python/server/services/cart_service.py
@@ -0,0 +1,367 @@
+#   Copyright 2026 UCP Authors
+#
+#   Licensed under the Apache License, Version 2.0 (the "License");
+#   you may not use this file except in compliance with the License.
+#   You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+#   Unless required by applicable law or agreed to in writing, software
+#   distributed under the License is distributed on an "AS IS" BASIS,
+#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#   See the License for the specific language governing permissions and
+#   limitations under the License.
+
+"""Cart service for managing the lifecycle of cart sessions."""
+
+import logging
+import uuid
+import datetime
+from typing import Any
+
+import config
+import db
+from exceptions import (
+  ResourceNotFoundError,
+  IdempotencyConflictError,
+  InvalidRequestError,
+)
+from sqlalchemy.ext.asyncio import AsyncSession
+from models import UnifiedCart as Cart
+from models import UnifiedCartCreateRequest as CartCreateRequest
+from models import UnifiedCartUpdateRequest as CartUpdateRequest
+from ucp_sdk.models.schemas.shopping.discount import (
+  DiscountsObject,
+  AppliedDiscount,
+  Allocation,
+)
+from ucp_sdk.models.schemas.shopping.types.line_item import (
+  LineItem as LineItemResponse,
+)
+from ucp_sdk.models.schemas.shopping.types.item import Item as ItemResponse
+from ucp_sdk.models.schemas.shopping.types.total import Total as TotalResponse
+from ucp_sdk.models.schemas.ucp import ResponseCartSchema
+from ucp_sdk.models.schemas.capability import ResponseSchema as Response
+from pydantic import BaseModel, AnyUrl
+import json
+import hashlib
+
+logger = logging.getLogger(__name__)
+
+
+class CartService:
+  """Service for managing cart sessions."""
+
+  def __init__(
+    self,
+    products_session: AsyncSession,
+    transactions_session: AsyncSession,
+    base_url: str,
+  ):
+    """Initialize CartService."""
+    self.products_session = products_session
+    self.transactions_session = transactions_session
+    self.base_url = base_url.rstrip("/")
+
+  def _compute_hash(self, data: Any) -> str:
+    """Compute SHA256 hash of the JSON-serialized data."""
+    if isinstance(data, BaseModel):
+      json_str = json.dumps(data.model_dump(mode="json"), sort_keys=True)
+    else:
+      json_str = json.dumps(data, sort_keys=True)
+    return hashlib.sha256(json_str.encode("utf-8")).hexdigest()
+
+  async def create_cart(
+    self,
+    cart_req: CartCreateRequest,
+    idempotency_key: str,
+  ) -> Cart:
+    """Create a new cart session."""
+    logger.info("Creating cart session")
+
+    # Idempotency Check
+    request_hash = self._compute_hash(cart_req)
+    existing_record = await db.get_idempotency_record(
+      self.transactions_session, idempotency_key
+    )
+
+    if existing_record:
+      if existing_record.request_hash != request_hash:
+        raise IdempotencyConflictError(
+          "Idempotency key reused with different parameters"
+        )
+      return Cart(**existing_record.response_body)
+
+    cart_id = f"cart_{uuid.uuid4()}"
+
+    # Map line items
+    line_items = []
+    for li_req in cart_req.line_items:
+      line_items.append(
+        LineItemResponse(
+          id=str(uuid.uuid4()),
+          item=ItemResponse(
+            id=li_req.item.id,
+            title="",
+            price=0,
+          ),
+          quantity=li_req.quantity,
+          totals=[],
+        )
+      )
+
+    cart_data = cart_req.model_dump(
+      exclude={
+        "line_items",
+      }
+    )
+
+    cart = Cart(
+      ucp=ResponseCartSchema(
+        version=config.get_server_version(),
+        capabilities={
+          "dev.ucp.shopping.cart": [
+            Response(
+              name="dev.ucp.shopping.cart",
+              version=config.get_server_version(),
+            )
+          ]
+        },
+      ),
+      id=cart_id,
+      line_items=line_items,
+      currency="USD",
+      totals=[
+        {"type": "subtotal", "amount": 0},
+        {"type": "total", "amount": 0},
+      ],
+      continue_url=AnyUrl(f"{self.base_url}/checkout?cart={cart_id}"),
+      expires_at=datetime.datetime.now(datetime.timezone.utc)
+      + datetime.timedelta(days=7),
+      **cart_data,
+    )
+
+    await self._enrich_and_recalculate(cart)
+
+    response_body = cart.model_dump(
+      mode="json", by_alias=True, exclude_none=True
+    )
+
+    # Persist cart
+    await db.save_cart(
+      self.transactions_session,
+      cart.id,
+      response_body,
+    )
+
+    # Save Idempotency Record
+    await db.save_idempotency_record(
+      self.transactions_session,
+      idempotency_key,
+      request_hash,
+      201,
+      response_body,
+    )
+
+    await self.transactions_session.commit()
+    return cart
+
+  async def get_cart(self, cart_id: str) -> Cart:
+    """Retrieve a cart session."""
+    data = await db.get_cart_session(self.transactions_session, cart_id)
+    if not data:
+      raise ResourceNotFoundError(f"Cart session {cart_id} not found")
+    return Cart(**data)
+
+  async def update_cart(
+    self,
+    cart_id: str,
+    cart_req: CartUpdateRequest,
+    idempotency_key: str,
+  ) -> Cart:
+    """Update a cart session."""
+    logger.info("Updating cart session %s", cart_id)
+
+    # Idempotency Check
+    request_hash = self._compute_hash(cart_req)
+    existing_record = await db.get_idempotency_record(
+      self.transactions_session, idempotency_key
+    )
+    if existing_record:
+      if existing_record.request_hash != request_hash:
+        raise IdempotencyConflictError(
+          "Idempotency key reused with different parameters"
+        )
+      return Cart(**existing_record.response_body)
+
+    # Verify existence
+    existing_data = await db.get_cart_session(
+      self.transactions_session, cart_id
+    )
+    if not existing_data:
+      raise ResourceNotFoundError(f"Cart session {cart_id} not found")
+    existing = Cart(**existing_data)
+
+    # Update line items
+    line_items = []
+    for li_req in cart_req.line_items:
+      line_items.append(
+        LineItemResponse(
+          id=li_req.id or str(uuid.uuid4()),
+          item=ItemResponse(
+            id=li_req.item.id,
+            title="",
+            price=0,
+          ),
+          quantity=li_req.quantity,
+          totals=[],
+          parent_id=li_req.parent_id,
+        )
+      )
+    existing.line_items = line_items
+
+    if cart_req.buyer:
+      existing.buyer = cart_req.buyer
+    if cart_req.context:
+      existing.context = cart_req.context
+    if cart_req.discounts is not None:
+      existing.discounts = cart_req.discounts
+
+    await self._enrich_and_recalculate(existing)
+
+    response_body = existing.model_dump(
+      mode="json", by_alias=True, exclude_none=True
+    )
+
+    await db.save_cart(
+      self.transactions_session,
+      cart_id,
+      response_body,
+    )
+
+    # Save Idempotency Record
+    await db.save_idempotency_record(
+      self.transactions_session,
+      idempotency_key,
+      request_hash,
+      200,
+      response_body,
+    )
+
+    await self.transactions_session.commit()
+    return existing
+
+  async def cancel_cart(
+    self,
+    cart_id: str,
+    idempotency_key: str,
+  ) -> Cart:
+    """Cancel a cart session."""
+    logger.info("Canceling cart session %s", cart_id)
+
+    # Idempotency Check
+    request_hash = self._compute_hash({})
+    existing_record = await db.get_idempotency_record(
+      self.transactions_session, idempotency_key
+    )
+    if existing_record:
+      if existing_record.request_hash != request_hash:
+        raise IdempotencyConflictError(
+          "Idempotency key reused with different parameters"
+        )
+      return Cart(**existing_record.response_body)
+
+    # Verify existence
+    existing_data = await db.get_cart_session(
+      self.transactions_session, cart_id
+    )
+    if not existing_data:
+      raise ResourceNotFoundError(f"Cart session {cart_id} not found")
+    cart = Cart(**existing_data)
+
+    # Delete cart
+    await db.delete_cart_session(self.transactions_session, cart_id)
+
+    response_body = cart.model_dump(
+      mode="json", by_alias=True, exclude_none=True
+    )
+
+    # Save Idempotency Record
+    await db.save_idempotency_record(
+      self.transactions_session,
+      idempotency_key,
+      request_hash,
+      200,
+      response_body,
+    )
+
+    await self.transactions_session.commit()
+    return cart
+
+  async def _enrich_and_recalculate(self, cart: Cart) -> None:
+    """Enrich cart items from catalog and recalculate subtotals and totals."""
+    grand_total = 0
+
+    for line in cart.line_items:
+      product_id = line.item.id
+      product = await db.get_product(self.products_session, product_id)
+      if not product:
+        raise InvalidRequestError(f"Product {product_id} not found")
+
+      line.item.price = product.price
+      line.item.title = product.title
+
+      base_amount = product.price * line.quantity
+      line.totals = [
+        TotalResponse(type="subtotal", amount=base_amount),
+        TotalResponse(type="total", amount=base_amount),
+      ]
+      grand_total += base_amount
+
+    cart.totals = [
+      TotalResponse(type="subtotal", amount=grand_total),
+    ]
+
+    # Discount Logic
+    if not cart.discounts:
+      cart.discounts = DiscountsObject()
+
+    cart.discounts.applied = None
+
+    if cart.discounts.codes:
+      discounts = await db.get_discounts_by_codes(
+        self.transactions_session, cart.discounts.codes
+      )
+      discount_map = {d.code.upper(): d for d in discounts}
+
+      for code in cart.discounts.codes:
+        discount_obj = discount_map.get(code.upper())
+        if discount_obj:
+          discount_amount = 0
+          if discount_obj.type == "percentage":
+            discount_amount = int(grand_total * (discount_obj.value / 100))
+          elif discount_obj.type == "fixed_amount":
+            discount_amount = discount_obj.value
+
+          if discount_amount > 0:
+            grand_total -= discount_amount
+            if cart.discounts.applied is None:
+              cart.discounts.applied = []
+            cart.discounts.applied.append(
+              AppliedDiscount(
+                code=discount_obj.code,
+                title=discount_obj.description,
+                amount=discount_amount,
+                allocations=[
+                  Allocation(
+                    path="$.totals[?(@.type=='subtotal')]",
+                    amount=discount_amount,
+                  )
+                ],
+              )
+            )
+            cart.totals.append(
+              TotalResponse(type="discount", amount=-discount_amount)
+            )
+
+    cart.totals.append(TotalResponse(type="total", amount=grand_total))
diff --git a/rest/python/server/services/checkout_service.py b/rest/python/server/services/checkout_service.py
index ddeecd32..31cf2804 100644
--- a/rest/python/server/services/checkout_service.py
+++ b/rest/python/server/services/checkout_service.py
@@ -158,22 +158,71 @@ async def create_checkout(
       # Return cached response
       return Checkout(**existing_record.response_body)
 
+    cart_id = getattr(checkout_req, "cart_id", None)
+
+    if cart_id:
+      # Check if incomplete checkout already exists for this cart_id
+      existing_checkouts = await db.get_checkouts_by_cart_id(
+        self.transactions_session, cart_id
+      )
+      for data in existing_checkouts:
+        if data.get("status") not in [
+          CheckoutStatus.COMPLETED,
+          CheckoutStatus.CANCELED,
+        ]:
+          logger.info(
+            "Returning existing incomplete checkout for cart %s", cart_id
+          )
+          return Checkout(**data)
+
+      # Load cart to initialize checkout
+      cart_data = await db.get_cart_session(self.transactions_session, cart_id)
+      if not cart_data:
+        raise ResourceNotFoundError(f"Cart session {cart_id} not found")
+
+      from models import UnifiedCart as CartModel
+
+      cart = CartModel(**cart_data)
+
+      # Initialize from cart
+      source_line_items = cart.line_items
+      source_buyer = cart.buyer
+      source_context = cart.context
+      source_signals = cart.signals
+      source_attribution = cart.attribution
+      source_currency = cart.currency
+      source_discounts = cart.discounts
+    else:
+      # Initialize from request
+      source_line_items = checkout_req.line_items
+      source_buyer = checkout_req.buyer
+      source_context = checkout_req.context
+      source_signals = checkout_req.signals
+      source_attribution = checkout_req.attribution
+      source_currency = config.get_default_currency()
+      source_discounts = checkout_req.discounts
+
     # Initialize full model from request
     checkout_id = getattr(checkout_req, "id", None) or str(uuid.uuid4())
 
     # Map line items
     line_items = []
-    for li_req in checkout_req.line_items:
+    for li in source_line_items:
+      item_id = li.item.id
+      quantity = li.quantity
+      parent_id = getattr(li, "parent_id", None)
+      li_id = getattr(li, "id", None) or str(uuid.uuid4())
       line_items.append(
         LineItemResponse(
-          id=str(uuid.uuid4()),
+          id=li_id,
           item=ItemResponse(
-            id=li_req.item.id,
+            id=item_id,
             title="",
             price=0,  # Will be set by recalculate_totals
           ),
-          quantity=li_req.quantity,
+          quantity=quantity,
           totals=[],
+          parent_id=parent_id,
         )
       )
 
@@ -206,6 +255,12 @@ async def create_checkout(
         "totals",
         "links",
         "fulfillment",
+        "buyer",
+        "context",
+        "signals",
+        "attribution",
+        "cart_id",
+        "discounts",
       }
     )
 
@@ -303,11 +358,11 @@ async def create_checkout(
             )
           ]
         },
-        payment_handlers={},
+        payment_handlers=config.get_payment_handlers(),
       ),
       id=checkout_id,
       status=CheckoutStatus.IN_PROGRESS,
-      currency=config.get_default_currency(),
+      currency=source_currency,
       line_items=line_items,
       totals=[
         {"type": "subtotal", "amount": 0},
@@ -323,11 +378,27 @@ async def create_checkout(
       else None,
       platform=platform_config,
       fulfillment=fulfillment_resp,
+      buyer=source_buyer.model_dump(exclude_none=True)
+      if source_buyer
+      else None,
+      context=source_context.model_dump(exclude_none=True)
+      if source_context
+      else None,
+      signals=source_signals.model_dump(exclude_none=True)
+      if source_signals
+      else None,
+      attribution=source_attribution.model_dump(exclude_none=True)
+      if source_attribution
+      else None,
+      cart_id=cart_id,
+      discounts=source_discounts.model_dump(exclude_none=True)
+      if source_discounts
+      else None,
       **checkout_data,
     )
 
     # Validate inventory and recalculate totals (Server is authority)
-    await self._recalculate_totals(checkout)
+    await self._enrich_and_recalculate(checkout)
     await self._validate_inventory(checkout)
 
     checkout.status = CheckoutStatus.READY_FOR_COMPLETE
@@ -575,7 +646,7 @@ async def update_checkout(
       existing.platform = platform_config
 
     # Validate inventory and recalculate totals (Server is authority)
-    await self._recalculate_totals(existing)
+    await self._enrich_and_recalculate(existing)
     await self._validate_inventory(existing)
 
     response_body = existing.model_dump(
@@ -805,6 +876,14 @@ async def complete_checkout(
         response_body,
       )
 
+      if checkout.cart_id:
+        logger.info(
+          "Clearing cart %s after checkout completion", checkout.cart_id
+        )
+        await db.delete_cart_session(
+          self.transactions_session, checkout.cart_id
+        )
+
       # Commit both inventory updates and checkout status update atomically
       await self.transactions_session.commit()
 
@@ -1007,11 +1086,11 @@ async def _validate_inventory(
       if qty_avail is None or qty_avail < line.quantity:
         raise OutOfStockError(f"Insufficient stock for item {product_id}")
 
-  async def _recalculate_totals(
+  async def _enrich_and_recalculate(
     self,
     checkout: Checkout,
   ) -> None:
-    """Recalculate line item subtotals and checkout totals."""
+    """Enrich items from catalog and recalculate totals."""
     grand_total = 0
 
     for line in checkout.line_items: