Backend quickstart

Pre-launch API. The concepts described here are settled, but the API shape is not: request and response fields, parameters and defaults can still change. Build against it, and talk to your Luigi's Box contact before you put an integration into production.

Server-side integration in five steps: token, index, verify, search, report. Examples are Python with httpx; the calls are plain HTTP and translate directly.

This is the right page for indexing and configuration, and for a storefront that renders results server-side. For a browser storefront, a distribution repository is the faster route — and the two combine: a backend that owns the catalog, a repository that owns the storefront.

What you need

	Looks like	Notes
OAuth client ID	`lbk_live_m2m_eu1_r3b6x2v8g9`
OAuth client secret	`lbe_live_eu1_…`	Secret manager only. Never in a browser or a repository.
Catalog ID	`lbc_8w3k2p`	What you index into
Channel ID	`lbn_4hj9tv`	What you search on
Region	`eu1`	It is in your credentials

/me/entitlements lists the catalogs and channels your credentials can reach.

The catalog needs content in it before step 4 returns anything. Steps 2 and 3 below put one product in and verify it; for a whole catalog use a feed and confirm the first run succeeded before you expect search to work.

1. Get a token

import time

import httpx

REGION = "eu1"
AUTH_HOST = f"https://auth.{REGION}.luigisbox.ai"
API_HOST = f"https://api.{REGION}.luigisbox.ai"

CLIENT_ID = os.environ["LBX_CLIENT_ID"]
CLIENT_SECRET = os.environ["LBX_CLIENT_SECRET"]


class TokenCache:
    """Hands out a cached access token per audience, refreshing shortly before expiry."""

    def __init__(self, http: httpx.Client) -> None:
        self._http = http
        self._tokens: dict[str, tuple[str, float]] = {}

    def get(self, audience: str) -> str:
        token, expires_at = self._tokens.get(audience, ("", 0.0))
        if time.monotonic() < expires_at:
            return token

        response = self._http.post(
            f"{AUTH_HOST}/oauth/token",
            json={
                "grant_type": "client_credentials",
                "client_id": CLIENT_ID,
                "client_secret": CLIENT_SECRET,
                "audience": audience,
            },
        )
        response.raise_for_status()
        payload = response.json()

        # Refresh 30s early so an in-flight request never races expiry.
        self._tokens[audience] = (payload["access_token"], time.monotonic() + payload["expires_in"] - 30)
        return payload["access_token"]

    def invalidate(self, audience: str) -> None:
        """Drop a cached token so the next get() mints a fresh one."""
        self._tokens.pop(audience, None)

A token is valid for one service, so a job that indexes and searches holds two:

http = httpx.Client(timeout=30.0)
tokens = TokenCache(http)

index_token = tokens.get(f"{API_HOST}/index")
discovery_token = tokens.get(f"{API_HOST}/discovery")

Verify with curl before writing any code:

curl -X POST 'https://auth.eu1.luigisbox.ai/oauth/token' \
  -H 'Content-Type: application/json' \
  -d '{
    "grant_type": "client_credentials",
    "client_id": "lbk_live_m2m_eu1_r3b6x2v8g9",
    "client_secret": "lbe_live_eu1_…",
    "audience": "https://api.eu1.luigisbox.ai/index"
  }'

2. Index a product

CATALOG_ID = "lbc_8w3k2p"

response = http.post(
    f"{API_HOST}/index/v1/{CATALOG_ID}/index/",
    headers={"Authorization": f"Bearer {tokens.get(f'{API_HOST}/index')}"},
    json={
        "objects": [
            {
                "@id": "product/sku-1001",
                "@type": "product",
                "@title": "Blue Cotton T-Shirt",
                "@category": ["category/shirts"],
                "@brand": "brand/northwear",
                "url": "https://example.com/products/blue-cotton-t-shirt",
                "image_url": "https://cdn.example.com/products/sku-1001.jpg",
                "price": 29.9,
                "list_price": 39.9,
                "availability": 1,
                "color": "blue",
                "size": "M",
                "private:margin": 8.4,
            }
        ]
    },
)
response.raise_for_status()

The path ends with a trailing slash; without it the request does not route.

{ "status": "accepted" }

202 Accepted means validated and queued. Indexing is asynchronous.

Three rules take effect with the first object you send:

Types must be consistent. 29.9 is a number; "29.90" is text. The first value a field receives fixes its type for the catalog, and a later conflicting type is rejected rather than coerced. A price sent as text can never be range-filtered.
Identities must be stable. An @id names the same product forever. Changing it loses the product’s ranking history.
Private data belongs in private:. Anything not in that namespace can be read by a browser integration.

The same call scales to a whole catalog: batch your objects and keep sending. See Batching. If you would rather hand Luigi’s Box a file to fetch on a schedule, feeds do that instead — pick one and stay with it.

3. Verify what you sent

Indexing is asynchronous, so confirm what landed. List the attributes the catalog now holds for the type:

curl -G 'https://api.eu1.luigisbox.ai/catalog/v1/lbc_8w3k2p/types/product/attributes' \
  --data-urlencode 'size=200' \
  -H 'Authorization: Bearer <catalog token>'

Every attribute comes back with its value shape. Confirm the numeric fields — price, list_price, availability — are number and not string.

Object history shows, for a single product, whether an update landed and whether it has been applied.

4. Search

CHANNEL_ID = "lbn_4hj9tv"
SURFACE_ID = "lbs_search_main"


def search(query: str, *, visitor_id: str, size: int = 24) -> dict:
    """Run one search request on behalf of a shopper."""
    response = http.post(
        f"{API_HOST}/discovery/v1/search",
        params={"channel_id": CHANNEL_ID, "surface_id": SURFACE_ID},
        json={
            "type": "product",
            "query": query,
            "size": size,
            "return_fields": ["@title", "price", "image_url", "url"],
        },
        headers={
            "Authorization": f"Bearer {tokens.get(f'{API_HOST}/discovery')}",
            "X-Lbx-Visitor-Id": visitor_id,
        },
    )
    response.raise_for_status()
    return response.json()

Two things are your responsibility when there is no browser calling directly:

Forward the shopper’s identifier. visitor_id comes from your own frontend and passes through verbatim. Generating it server-side collapses every shopper into one, which breaks personalization and analytics together.
Choose what reaches the page. A server-side token can read private: fields, so set return_fields to what you intend to render. Passing the whole hit into your template exposes every private: field it carries.

5. Report purchases server-side

If you already have an order-confirmation handler, it is a good place to report purchases from — browser-side events can be lost to ad blockers or a closed tab, and this is the event closest to revenue:

def report_transaction(order, *, visitor_id: str, consent_granted: bool) -> None:
    """Report a completed order to Luigi's Box."""
    response = http.post(
        f"{API_HOST}/events/v1/events",
        headers={
            "Authorization": f"Bearer {tokens.get(f'{API_HOST}/events')}",
            "Content-Type": "application/json",
            "X-Lbx-Visitor-Id": visitor_id,
        },
        json={
            "type": "transaction",
            "channel_id": CHANNEL_ID,
            "currency": order.currency,
            "listing_ab_test": False,
            "consent_granted": consent_granted,
            "items": [
                {
                    "reference": f"product/sku:{line.sku}",
                    "title": line.title,
                    "count": line.quantity,
                    "total_price": float(line.line_total),
                }
                for line in order.lines
            ],
        },
    )
    response.raise_for_status()

visitor_id must be the one your frontend used, or the order cannot be attributed to the searches that led to it. Store it on the order alongside everything else.

total_price is the line total, not the unit price.

Handling failures

def call_with_retry(request, audience: str, *, attempts: int = 3):
    """Retry a request once on an expired token, and with backoff on transient failures."""
    for attempt in range(attempts):
        response = request()
        if response.status_code == 401 and attempt == 0:
            tokens.invalidate(audience)   # force a fresh token, then retry
            continue
        if response.status_code in (429, 502, 503) and attempt < attempts - 1:
            time.sleep(int(response.headers.get("Retry-After", 2**attempt)))
            continue
        return response
    return response

Status	Meaning
`401`	Expired token, or the wrong audience — retry once with a fresh one
`403`	Valid token, no grant for this permission on this resource
`404`	Not found, or not visible to your credentials
`422`	Read `exception_details.validation_errors` — do not retry unchanged
`429`	Honour `Retry-After`
`502` / `503`	Transient — retry with backoff

See Errors and rate limits.

Task	Page
Load the whole catalog	Feeds · Feed management API
Keep your source field names	Field mapping
Add facets and filters	Facets · Filters
Hydrate carts and wishlists	Object lookup
Add business rules	Business rules
Add a market	Account structure