Eight tools, one file. Wraps v3 REST under X-Auth-Token, fetches agent_commerce metafields in get_product, handles variant-tracked inventory in check_inventory, and returns a clean checkout_url from create_cart. Save as server.py, set BC_STORE_HASH and BC_API_TOKEN in .env, and run with uv run server.py.
"""
BigCommerce MCP Server — Agent Commerce Layer 3
Wraps BigCommerce v3 REST API as MCP tools.
Requires: uv add "mcp[cli]>=1.2.0" httpx python-dotenv
Run: uv run server.py
"""
import sys
import os
import json
import logging
from typing import Any, Optional
from dotenv import load_dotenv
import httpx
from mcp.server.fastmcp import FastMCP
load_dotenv()
# ─── Configuration ────────────────────────────────────────────────────────────
STORE_HASH = os.environ["BC_STORE_HASH"] # e.g. "abc123def"
API_TOKEN = os.environ["BC_API_TOKEN"] # X-Auth-Token value
BASE_URL = f"https://api.bigcommerce.com/stores/{STORE_HASH}/v3"
V2_BASE = f"https://api.bigcommerce.com/stores/{STORE_HASH}/v2"
HEADERS = {
"X-Auth-Token": API_TOKEN,
"Accept": "application/json",
"Content-Type": "application/json",
}
logging.basicConfig(stream=sys.stderr, level=logging.INFO)
logger = logging.getLogger("bigcommerce-mcp")
mcp = FastMCP("bigcommerce-store")
# ─── HTTP helpers with 429 backoff ────────────────────────────────────────────
import asyncio
async def bc_get(path: str, params: dict = None, v2: bool = False, max_retries: int = 3) -> dict:
base = V2_BASE if v2 else BASE_URL
for attempt in range(max_retries):
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.get(f"{base}{path}", headers=HEADERS, params=params)
if response.status_code == 429:
wait_ms = int(response.headers.get("X-Rate-Limit-Time-Reset-Ms", 30000))
logger.warning(f"429 backoff {wait_ms}ms (attempt {attempt+1})")
await asyncio.sleep(wait_ms / 1000)
continue
response.raise_for_status()
return response.json()
raise RuntimeError(f"Rate limit exceeded after {max_retries} retries on GET {path}")
async def bc_post(path: str, body: dict, v2: bool = False, max_retries: int = 3) -> dict:
base = V2_BASE if v2 else BASE_URL
for attempt in range(max_retries):
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(f"{base}{path}", headers=HEADERS, json=body)
if response.status_code == 429:
wait_ms = int(response.headers.get("X-Rate-Limit-Time-Reset-Ms", 30000))
await asyncio.sleep(wait_ms / 1000)
continue
response.raise_for_status()
return response.json()
raise RuntimeError(f"Rate limit exceeded after {max_retries} retries on POST {path}")
# ─── Tool 1: List Products ────────────────────────────────────────────────────
@mcp.tool()
async def list_products(
limit: int = 20,
page: int = 1,
is_featured: Optional[bool] = None,
availability: Optional[str] = None,
) -> str:
"""List products from the BigCommerce catalog.
Args:
limit: Number of products to return (max 250)
page: Page number for pagination
is_featured: Filter to featured products only
availability: 'available', 'disabled', or 'preorder'
"""
params = {
"include": "images,variants",
"include_fields": "name,sku,price,sale_price,calculated_price,inventory_level,availability,gtin,mpn,brand_id,categories,condition,custom_url",
"limit": min(limit, 250),
"page": page,
"is_visible": True,
}
if is_featured is not None: params["is_featured"] = is_featured
if availability: params["availability"] = availability
data = await bc_get("/catalog/products", params=params)
products = data.get("data", [])
result = [{
"id": p["id"],
"name": p["name"],
"sku": p.get("sku", ""),
"price": p.get("calculated_price", p.get("price", 0)),
"availability": p.get("availability", "available"),
"inventory_level": p.get("inventory_level", 0),
"gtin": p.get("gtin", ""),
"url": p.get("custom_url", {}).get("url", ""),
} for p in products]
return json.dumps({"products": result, "total": data.get("meta", {}).get("pagination", {}).get("total", len(result))}, indent=2)
# ─── Tool 2: Get Product (merges native fields + agent_commerce metafields) ───
@mcp.tool()
async def get_product(product_id: int) -> str:
"""Get full details for a BigCommerce product including variants, images, custom fields, and agent_commerce metafields.
Args:
product_id: The BigCommerce product entity ID
"""
data = await bc_get(
f"/catalog/products/{product_id}",
params={"include": "images,variants,custom_fields"}
)
product = data.get("data", {})
# Also fetch agent_commerce metafields
try:
meta_data = await bc_get(
f"/catalog/products/{product_id}/metafields",
params={"namespace": "agent_commerce"}
)
metafields = {m["key"]: m["value"] for m in meta_data.get("data", [])}
except Exception:
metafields = {}
result = {
"id": product.get("id"),
"name": product.get("name"),
"sku": product.get("sku"),
"mpn": product.get("mpn"),
"gtin": metafields.get("gtin") or product.get("gtin"),
"description": product.get("description", "")[:500],
"price": product.get("calculated_price", product.get("price")),
"sale_price": product.get("sale_price"),
"retail_price": product.get("retail_price"),
"availability": product.get("availability"),
"inventory_level": product.get("inventory_level"),
"inventory_tracking": product.get("inventory_tracking"),
"condition": product.get("condition"),
"weight": product.get("weight"),
"brand_id": product.get("brand_id"),
"categories": product.get("categories"),
"images": [img.get("url_standard") for img in product.get("images", [])[:3]],
"variants_count": len(product.get("variants", [])),
"return_policy": metafields.get("return_policy"),
"trust_score": metafields.get("trust_score"),
"custom_fields": {cf["name"]: cf["value"] for cf in product.get("custom_fields", [])},
}
return json.dumps(result, indent=2)
# ─── Tool 3: Search Products ──────────────────────────────────────────────────
@mcp.tool()
async def search_products(
keyword: str,
price_min: Optional[float] = None,
price_max: Optional[float] = None,
limit: int = 10,
) -> str:
"""Search the BigCommerce catalog by keyword, optionally filtered by price range.
Args:
keyword: Search term (matches name, description, sku, search_keywords)
price_min: Minimum price filter (USD)
price_max: Maximum price filter (USD)
limit: Number of results (max 50)
"""
params = {
"keyword": keyword,
"include": "images",
"include_fields": "name,sku,price,calculated_price,sale_price,inventory_level,availability,custom_url",
"is_visible": True,
"limit": min(limit, 50),
}
if price_min is not None: params["price_min"] = price_min
if price_max is not None: params["price_max"] = price_max
data = await bc_get("/catalog/products", params=params)
products = data.get("data", [])
result = [{
"id": p["id"],
"name": p["name"],
"sku": p.get("sku"),
"price": p.get("calculated_price", p.get("price")),
"sale_price": p.get("sale_price"),
"availability": p.get("availability"),
"in_stock": p.get("inventory_level", 1) > 0,
"image": p.get("images", [{}])[0].get("url_thumbnail", "") if p.get("images") else "",
"url": p.get("custom_url", {}).get("url", ""),
} for p in products]
return json.dumps({"query": keyword, "results": result, "count": len(result)}, indent=2)
# ─── Tool 4: Check Inventory (handles variant-tracked products) ───────────────
@mcp.tool()
async def check_inventory(product_id: int, variant_id: Optional[int] = None) -> str:
"""Check inventory level and availability for a product or specific variant.
Args:
product_id: The BigCommerce product entity ID
variant_id: Optional variant entity ID for variant-level inventory
"""
if variant_id:
data = await bc_get(
f"/catalog/products/{product_id}/variants/{variant_id}",
params={"include_fields": "inventory_level,purchasing_disabled,availability"}
)
variant = data.get("data", {})
return json.dumps({
"product_id": product_id,
"variant_id": variant_id,
"inventory_level": variant.get("inventory_level", 0),
"purchasing_disabled": variant.get("purchasing_disabled", False),
"in_stock": variant.get("inventory_level", 0) > 0,
}, indent=2)
data = await bc_get(
f"/catalog/products/{product_id}",
params={"include_fields": "inventory_level,inventory_tracking,availability,inventory_warning_level"}
)
product = data.get("data", {})
return json.dumps({
"product_id": product_id,
"inventory_level": product.get("inventory_level", 0),
"inventory_tracking": product.get("inventory_tracking", "none"),
"availability": product.get("availability", "available"),
"in_stock": product.get("availability") == "available" and product.get("inventory_level", 1) > 0,
"warning_level": product.get("inventory_warning_level", 0),
}, indent=2)
# ─── Tool 5: Create Cart (always includes channel_id and redirect_urls) ───────
@mcp.tool()
async def create_cart(
product_id: int,
quantity: int,
variant_id: Optional[int] = None,
channel_id: int = 1,
) -> str:
"""Create a cart with a single line item. Returns cart ID, total, and checkout URL.
Args:
product_id: BigCommerce product ID to add
quantity: Number of units
variant_id: Optional variant ID if product has variants
channel_id: BigCommerce channel ID (default 1; pass your registered agent channel)
"""
line_item: dict = {"product_id": product_id, "quantity": quantity}
if variant_id: line_item["variant_id"] = variant_id
body = {
"customer_id": 0,
"channel_id": channel_id,
"line_items": [line_item],
"currency": {"code": "USD"},
"locale": "en-US",
}
data = await bc_post("/carts?include=redirect_urls", body)
cart = data.get("data", {})
redirect_urls = cart.get("redirect_urls", {})
return json.dumps({
"cart_id": cart.get("id"),
"cart_amount": cart.get("cartAmount"),
"base_amount": cart.get("baseAmount"),
"discount_amount": cart.get("discountAmount"),
"currency": cart.get("currency", {}).get("code"),
"checkout_url": redirect_urls.get("checkout_url"),
"cart_url": redirect_urls.get("cart_url"),
"embedded_checkout_url": redirect_urls.get("embedded_checkout_url"),
}, indent=2)
# ─── Tool 6: Get Checkout (cart ID = checkout ID) ─────────────────────────────
@mcp.tool()
async def get_checkout(cart_id: str) -> str:
"""Retrieve the current checkout state for a cart, including shipping options.
Args:
cart_id: The cart ID returned from create_cart (same value as checkout ID)
"""
data = await bc_get(
f"/checkouts/{cart_id}",
params={"include": "consignments.available_shipping_options,promotions.banners"}
)
checkout = data.get("data", {})
return json.dumps({
"checkout_id": checkout.get("id"),
"cart_amount": checkout.get("cart", {}).get("cartAmount"),
"grand_total": checkout.get("grandTotal"),
"taxes": checkout.get("taxes"),
"coupons": checkout.get("coupons"),
"consignments": checkout.get("consignments", []),
"billing_address": checkout.get("billingAddress"),
"order_id": checkout.get("orderId"),
}, indent=2)
# ─── Tool 7: Get Order Status (v2 REST) ───────────────────────────────────────
@mcp.tool()
async def get_order_status(order_id: int) -> str:
"""Retrieve the status and summary of a BigCommerce order.
Args:
order_id: The BigCommerce order ID
"""
data = await bc_get(f"/orders/{order_id}", v2=True)
status_map = {
0: "Incomplete", 1: "Pending", 2: "Shipped",
3: "Partially Shipped", 4: "Refunded", 5: "Cancelled",
6: "Declined", 7: "Awaiting Payment", 8: "Awaiting Pickup",
9: "Awaiting Shipment", 10: "Completed", 11: "Awaiting Fulfillment",
12: "Manual Verification Required", 13: "Disputed", 14: "Partially Refunded",
}
status_id = data.get("status_id", 0)
return json.dumps({
"order_id": data.get("id"),
"status": data.get("status", status_map.get(status_id, "Unknown")),
"status_id": status_id,
"total_inc_tax": data.get("total_inc_tax"),
"payment_method": data.get("payment_method"),
"payment_status": data.get("payment_status"),
"date_created": data.get("date_created"),
"date_modified": data.get("date_modified"),
"items_total": data.get("items_total"),
"items_shipped": data.get("items_shipped"),
"shipping_cost_inc_tax": data.get("shipping_cost_inc_tax"),
"tracking_numbers": data.get("tracking_numbers", []),
}, indent=2)
# ─── Tool 8: Initiate Return (v3 refund quote → refund) ───────────────────────
@mcp.tool()
async def initiate_return(order_id: int, item_id: int, quantity: int) -> str:
"""Create a refund quote for an order item. Returns the quote for merchant review.
Args:
order_id: The BigCommerce order ID
item_id: The order line item ID
quantity: Number of units to refund
"""
body = {"items": [{"item_id": item_id, "item_type": "PRODUCT", "quantity": quantity}]}
try:
quote = await bc_post(f"/orders/{order_id}/payment_actions/refund_quotes", body)
q = quote.get("data", {})
return json.dumps({
"action": "refund_quote_created",
"order_id": order_id,
"refund_amount": q.get("total_refund_amount"),
"tax_amount": q.get("total_refund_tax_amount"),
"payment_options": q.get("payment_methods", []),
"note": "Refund quote created. Execute via POST /v3/orders/{id}/payment_actions/refunds.",
}, indent=2)
except Exception as e:
return json.dumps({"error": str(e), "order_id": order_id}, indent=2)
# ─── Entry Point ──────────────────────────────────────────────────────────────
def main():
mcp.run(transport="stdio") # switch to "streamable-http" for Fly.io/Vercel
if __name__ == "__main__":
main()
BigCommerce's Channels API lets you register any new commerce surface — including an agent channel. Use type: "storefront" with platform: "custom". The returned channel_id goes into every cart your MCP creates.
For the MCP protocol primer (JSON-RPC envelope, tool schemas, transport options) see the MCP spoke.