Shopify POS Voice AI: Build Checkout with MCP (2026)

When a sales associate has to glance at a tablet, tap through the POS, and then repeat the same steps for a return, the flow feels clunky. The real break‑point showed up in my own boutique: the clerk would lose focus while entering “add‑gift‑wrap” or “apply loyalty discount”, and the checkout queue lengthened. What if a voice‑first AI could listen, pull inventory, ring up the sale, and confirm the total—all without a single tap?

That’s exactly what we’ll build: a hands‑free retail assistant that lives on a desktop (or bundled tablet) next to the Shopify POS register, understands spoken commands, and talks to Shopify through Anthropic’s Model Context Protocol (MCP). The result is a conversational layer that feels as fast as a human clerk but stays consistent with the POS ledger.

How a custom voice AI agent integrates with Shopify POS

A custom voice AI agent integrates with Shopify POS using Anthropic’s Model Context Protocol. You write a Python MCP server exposing POS tools (e.g., “process_sale”). Claude Desktop, linked to this server, calls those tools. A speech front‑end streams audio to Claude, which interprets the command and triggers the corresponding POS action via MCP.

Understanding the MCP + Shopify POS + Voice AI Stack

**The Role of Model Context Protocol (MCP – 2026 Stable Release)** MCP is the glue that lets Claude act like a remote‑procedure‑call client. Instead of sending plain JSON, the agent describes a *tool* (name, args, description). The MCP server validates the schema, runs the function, and returns a structured result. Because the protocol lives over a persistent WebSocket, Claude can keep a conversational context while the backend performs async I/O.

**Why Voice‑Activated Shopify POS Management is a 2026 Retail Problem‑Solver** Retail staff need their hands on products, not on keyboards. Voice removes the friction of manual entry, reduces checkout time, and lowers the risk of input errors when the clerk is juggling a cart and a customer. In 2026, Shopify’s POS SDK finally exposed real‑time inventory hooks, making a voice layer feasible without polling.

**Architectural Overview**

flowchart LR
    A[Voice Input] --> B[Whisper Realtime STT]
    B --> C[Claude Desktop (Agent Mode)]
    C --> D[MCP Server (Python Async)]
    D --> E[Shopify Admin API]
    E --> F[POS Actions (Sale, Refund, Drawer)]
    F --> G[POS UI Feedback]
    C --> H[ElevenLabs TTS]
    H --> A

*Voice → STT → Claude → MCP → Shopify → POS UI; Claude also streams TTS back to the user.*

—

Prerequisites & 2026‑Specific Setup

Component	Version	Install / Configure
Python	3.12.4	`pyenv install 3.12.4 && pyenv local 3.12.4`
Anthropic SDK	`anthropic==0.23.0`	`pip install anthropic==0.23.0`
FastMCP SDK	`fastmcp==0.9.2`	`pip install fastmcp==0.9.2`
py‑shopify‑api	`2.4.0`	`pip install py-shopify-api==2.4.0`
Flask‑SocketIO	`6.2.0`	`pip install flask-socketio==6.2.0`
Whisper Realtime client	`whisper-realtime==0.5.1`	`pip install whisper-realtime==0.5.1`
ElevenLabs TTS client	`elevenlabs==0.2.3`	`pip install elevenlabs==0.2.3`
GUI (Streamlit)	`1.38.0`	`pip install streamlit==1.38.0`

1. Claude Desktop – Agent Mode

Download **Claude Desktop 3.5 Sonnet** from the Anthropic portal, open *Settings → Agent Mode*, and toggle *Enable MCP endpoint*. Point the endpoint to `ws://localhost:8765/mcp`.

2. Shopify Custom App & POS Token

1. In your Shopify admin, create a **Custom App** → *Configuration* → add scopes: `read_products`, `write_products`, `read_inventory`, `write_inventory`, `write_checkouts`, `read_checkouts`, `read_pos`, `write_pos`.
2. Install the app on your store. The installation page will give you a **POS access token** (store it as `SHOPIFY_POS_TOKEN`).

3. Python Virtual Environment

python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt   # include the libs from the table above

4. MCP SDK Installation

pip install fastmcp==0.9.2

The SDK ships `mcprun` (CLI) and a base `MCPServer` class we’ll subclass.

—

Building the Custom MCP Server for Shopify Operations

Create `mcp_server.py`:

# mcp_server.py – FastMCP 0.9.2, Python 3.12
# Handles Shopify POS tools for the Claude agent

import os
import asyncio
import logging
from typing import List, Dict, Any

import httpx
from fastmcp import MCPServer, Tool, ToolResult, ValidationError
from pydantic import BaseModel, Field, ValidationError as PydanticError

# ----------------------------------------------------------------------
# Configuration
# ----------------------------------------------------------------------
SHOPIFY_DOMAIN = os.getenv("SHOPIFY_DOMAIN")          # e.g. myshop.myshopify.com
POS_TOKEN = os.getenv("SHOPIFY_POS_TOKEN")
API_VERSION = "2026-10"
BASE_URL = f"https://{SHOPIFY_DOMAIN}/admin/api/{API_VERSION}"
HEADERS = {"X-Shopify-Access-Token": POS_TOKEN, "Content-Type": "application/json"}

# Reuse a single httpx.AsyncClient for connection pooling
http_client = httpx.AsyncClient(headers=HEADERS, timeout=10.0)

# Simple in‑memory cache for inventory (30 s TTL)
_inventory_cache: Dict[str, Any] = {}
CACHE_TTL = 30

# ----------------------------------------------------------------------
# Pydantic models for tool arguments
# ----------------------------------------------------------------------
class SaleItem(BaseModel):
    sku: str = Field(..., description="SKU of the product")
    quantity: int = Field(..., gt=0, description="Units to sell")
    price_cents: int = Field(..., gt=0, description="Price in cents per unit")

class ProcessSaleArgs(BaseModel):
    location_id: str = Field(..., description="Shopify POS location ID")
    items: List[SaleItem] = Field(..., min_items=1)
    customer_email: str | None = Field(None, description="Optional customer email")

# ----------------------------------------------------------------------
# Helper functions
# ----------------------------------------------------------------------
async def get_inventory(sku: str) -> int:
    """Return available quantity for a SKU, using cache."""
    now = asyncio.get_event_loop().time()
    cache_entry = _inventory_cache.get(sku)
    if cache_entry and now - cache_entry["ts"] < CACHE_TTL:
        return cache_entry["qty"]

    query = """
    {
      productVariants(first: 1, query: $sku) {
        edges {
          node {
            inventoryQuantity
          }
        }
      }
    }
    """
    variables = {"sku": sku}
    resp = await http_client.post(
        f"{BASE_URL}/graphql.json",
        json={"query": query, "variables": variables},
    )
    resp.raise_for_status()
    data = resp.json()
    qty = data["data"]["productVariants"]["edges"][0]["node"]["inventoryQuantity"]
    _inventory_cache[sku] = {"qty": qty, "ts": now}
    return qty

async def make_sale(args: ProcessSaleArgs) -> Dict[str, Any]:
    """Wrap Shopify's POS checkout creation and completion."""
    # 1️⃣ Build line items
    line_items = [
        {
            "variant_id": await sku_to_variant_id(it.sku),
            "quantity": it.quantity,
            "price": f"{it.price_cents/100:.2f}",
        }
        for it in args.items
    ]

    # 2️⃣ Create checkout (POST /pos/checkouts)
    checkout_payload = {
        "checkout": {
            "location_id": args.location_id,
            "line_items": line_items,
            "email": args.customer_email,
        }
    }

    try:
        resp = await http_client.post(
            f"{BASE_URL}/pos/checkouts.json", json=checkout_payload
        )
        resp.raise_for_status()
    except httpx.HTTPStatusError as exc:
        if exc.response.status_code == 429:
            # Rate limit – backoff and retry once
            await asyncio.sleep(2)
            resp = await http_client.post(
                f"{BASE_URL}/pos/checkouts.json", json=checkout_payload
            )
            resp.raise_for_status()
        else:
            raise

    checkout = resp.json()["checkout"]

    # 3️⃣ Complete checkout (POST /pos/checkouts/{id}/complete)
    complete_resp = await http_client.post(
        f"{BASE_URL}/pos/checkouts/{checkout['id']}/complete.json"
    )
    complete_resp.raise_for_status()
    return complete_resp.json()["checkout"]

async def sku_to_variant_id(sku: str) -> str:
    """Lookup variant ID from SKU (cached)."""
    query = """
    {
      productVariants(first: 1, query: $sku) {
        edges {
          node {
            id
          }
        }
      }
    }
    """
    variables = {"sku": sku}
    resp = await http_client.post(
        f"{BASE_URL}/graphql.json",
        json={"query": query, "variables": variables},
    )
    resp.raise_for_status()
    data = resp.json()
    variant_id = data["data"]["productVariants"]["edges"][0]["node"]["id"]
    return variant_id

# ----------------------------------------------------------------------
# MCP Tool definitions
# ----------------------------------------------------------------------
class GetInventoryTool(Tool):
    name = "get_inventory"
    description = "Return the current inventory count for a given SKU."
    input_schema = {"type": "object", "properties": {"sku": {"type": "string"}}, "required": ["sku"]}

    async def call(self, params: Dict) -> ToolResult:
        sku = params["sku"]
        try:
            qty = await get_inventory(sku)
            return ToolResult(output={"sku": sku, "available_quantity": qty})
        except Exception as e:
            logging.exception("Inventory fetch failed")
            return ToolResult(error=str(e))

class ProcessSaleTool(Tool):
    name = "process_sale"
    description = "Create and complete a POS sale for the given items."
    input_schema = ProcessSaleArgs.schema()

    async def call(self, params: Dict) -> ToolResult:
        try:
            args = ProcessSaleArgs(**params)
        except PydanticError as ve:
            return ToolResult(error=f"Invalid arguments: {ve}")

        try:
            result = await make_sale(args)
            return ToolResult(output={"checkout_id": result["id"], "total_price": result["total_price"]})
        except Exception as e:
            logging.exception("Sale processing failed")
            return ToolResult(error=str(e))

# ----------------------------------------------------------------------
# Server bootstrap
# ----------------------------------------------------------------------
tools = [GetInventoryTool(), ProcessSaleTool()]

class ShopifyMCPServer(MCPServer):
    def __init__(self):
        super().__init__(tools=tools, debug=True)

    async def on_error(self, exc: Exception):
        logging.error("Uncaught MCP error: %s", exc)

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    server = ShopifyMCPServer()
    asyncio.run(server.serve(host="0.0.0.0", port=8765))

**Explanation of key sections**

• **Caching** – inventory queries are cached for 30 s, as the expert quote suggests that keeping latency < 2 s requires batching and short‑term caching.
• **Rate‑limit handling** – on a 429 we back‑off 2 seconds and retry once; this pattern will be reused for other Shopify calls.
• **Typed arguments** – Pydantic validates the tool payload before hitting Shopify, preventing malformed requests that would otherwise cause a 400.

Testing the Server

# Launch the MCP server
python mcp_server.py

Open a terminal and run the built‑in inspector:

mcpr