Octogen Python SDK — async HTTP client API reference

The Octogen Python SDK gives you an async, type-safe client for the Octogen commerce API. Install it with uv, point it at your OCTO_API_KEY, and you can search products by keyword and facet, find products similar to a source product, and look up any product by URL — all backed by Pydantic models so your IDE can autocomplete every field.

Installation

Install the package

Requires Python 3.11 or later. The SDK ships from the public octogen-dev repository and is managed with uv. Clone the repository and sync the sdks/python project:

git clone https://github.com/octogen-ai/octogen-dev.git
cd octogen-dev
uv sync --project sdks/python

Run your code against the synced environment with uv run --project sdks/python python your_script.py, or activate it with source sdks/python/.venv/bin/activate.

Set your API key

The client reads your key from the OCTO_API_KEY environment variable automatically.

export OCTO_API_KEY=octo_live_...

You can also pass api_key= directly to the constructor if you prefer to manage secrets yourself (see Authentication below).

Authentication

OctogenClient resolves your API key in this order:

The api_key constructor argument, if provided.
The OCTO_API_KEY environment variable.

If neither is set, the constructor raises MissingAPIKeyError immediately — before any network request is made. Every request then sends the key as Authorization: Bearer <api-key>.

from octogen_ai_sdk import OctogenClient

# From environment variable (recommended)
client = OctogenClient()

# Explicit key
client = OctogenClient(api_key="octo_live_...")

Client constructor

OctogenClient accepts the following keyword-only arguments:

Parameter	Type	Default	Description
`api_key`	`str \| None`	`None` (reads `OCTO_API_KEY`)	Your Platform API key.
`base_url`	`str`	`https://api.octogen.ai/v1`	Override the API base URL (useful for testing).
`timeout`	`float \| httpx.Timeout`	`30.0`	Request timeout in seconds, or a full `httpx.Timeout` object.
`http_client`	`httpx.AsyncClient \| None`	`None` (SDK creates one)	Supply your own `httpx.AsyncClient` to reuse connections or configure proxies.

Using the client as an async context manager

Use async with to ensure the underlying HTTP connection pool is closed when you are done:

import asyncio
from octogen_ai_sdk import OctogenClient

async def main() -> None:
    async with OctogenClient() as client:
        results = await client.search_products(q="shirt", limit=1)
        print(results.items)

asyncio.run(main())

If you manage the client’s lifetime yourself, call await client.aclose() when finished.

Methods

`lookup_product`

async def lookup_product(
    url: str,
    *,
    resolution_mode: ProductLookupResolutionMode | str = "auto",
    on_demand_cache_policy: ProductLookupCachePolicy | str = "prefer_cache",
) -> MerchantProductUrlLookupResponse

Resolves a product page URL from Octogen’s index or with on-demand product resolution. Set resolution_mode="index_only" when your integration requires indexed identity and catalog context.

result = await client.lookup_product(
    "https://example.com/products/blue-linen-dress",
    resolution_mode="auto",
)
if result.source == "indexed":
    print(result.catalog_key, result.product.uuid)
else:
    print(result.product.title, result.product.currency)

Returns: MerchantProductUrlLookupResponse

Field	Type	Description
`source`	`Literal["indexed", "on_demand"]`	Source that produced the result.
`catalog_key`	`str \| None`	Indexed catalog, or `None` on demand.
`catalog_display_name`	`str \| None`	Indexed catalog name, or `None` on demand.
`product`	`MerchantProductView`	Full or partial-safe product view.
`resolution`	`ProductResolutionMetadata \| None`	On-demand completeness and evidence.

`search_products`

async def search_products(
    *,
    catalog: str | None = None,
    q: str | None = None,
    facets: Sequence[Facet | dict] | None = None,
    price_min: float | None = None,
    price_max: float | None = None,
    cursor: str | None = None,
    limit: int = 50,
) -> MerchantProductListPage

Searches products across all authorized catalogs by default. Pass catalog only when you want to restrict search to one catalog.

Parameter	Type	Default	Description
`catalog`	`str \| None`	`None`	Optional known active crawled catalog key. Omit to search all active crawled catalogs.
`q`	`str \| None`	`None`	Keyword search query.
`facets`	`Sequence[Facet \| dict] \| None`	`None`	List of facet filters to apply. See Faceted search.
`price_min`	`float \| None`	`None`	Minimum price filter (inclusive).
`price_max`	`float \| None`	`None`	Maximum price filter (inclusive).
`cursor`	`str \| None`	`None`	Pagination cursor from a previous response’s `next_cursor`.
`limit`	`int`	`50`	Number of results per page. Must be between 1 and 100.

Returns: MerchantProductListPage

Field	Type	Description
`items`	`list[MerchantProductListItem]`	Products on this page.
`next_cursor`	`str \| None`	Pass as `cursor` in your next call to get the next page. `None` means no more results.

Each MerchantProductListItem includes uuid, catalog_key, product_url, title, brand, current_price, original_price, image_url, images, rating, is_active, optional match scores, and updated_at.

`more_like_this_products`

async def more_like_this_products(
    *,
    source_url: str | None = None,
    source_uuid: str | None = None,
    catalog: str | None = None,
    include_facets: Sequence[Facet | dict] | None = None,
    exclude_facets: Sequence[Facet | dict] | None = None,
    price_preference: PricePreference | str = PricePreference.ANY,
    cursor: str | None = None,
    limit: int = 12,
    debug: bool = False,
) -> ProgrammaticMoreLikeThisResponse

Finds products similar to a source product URL or UUID. Provide exactly one of source_url or source_uuid. Omit catalog to search all active crawled catalogs, or pass a catalog key to keep the results within one catalog.

page = await client.more_like_this_products(
    source_url="https://warrenlotas.com/products/black-hoodie",
    catalog="warrenlotas",
    price_preference="any",
    limit=12,
)

Parameter	Type	Default	Description
`source_url`	`str \| None`	`None`	Canonical product URL to use as the source.
`source_uuid`	`str \| None`	`None`	Indexed product UUID to use as the source.
`catalog`	`str \| None`	`None`	Optional known active crawled catalog key. Omit to search all active crawled catalogs.
`include_facets`	`Sequence[Facet \| dict] \| None`	`None`	Extra include facets appended after server-generated audience facets.
`exclude_facets`	`Sequence[Facet \| dict] \| None`	`None`	Facets to exclude from similar-product results.
`price_preference`	`PricePreference \| str`	`any`	`lower`, `any`, or `higher` relative to the source product.
`cursor`	`str \| None`	`None`	Pagination cursor from a previous response’s `next_cursor`.
`limit`	`int`	`12`	Number of results per page. Must be between 1 and 100.
`debug`	`bool`	`False`	Include the public `effective_query` used for retrieval.

Returns: ProgrammaticMoreLikeThisResponse

Field	Type	Description
`source`	`ProgrammaticMoreLikeThisSourceResponse`	Source product identity.
`items`	`list[MerchantProductListItem]`	Similar products on this page.
`next_cursor`	`str \| None`	Pass as `cursor` in your next call to get the next page.
`effective_query`	`ProgrammaticMoreLikeThisEffectiveQuery \| None`	Present only when `debug` is `True`.

Complete example

The following example searches all active crawled catalogs for women’s linen dresses and prints each result with its brand and price:

import asyncio
from pprint import pformat

from octogen_ai_sdk import OctogenAPIError, OctogenClient


async def main() -> None:
    try:
        async with OctogenClient() as client:
            results = await client.search_products(
                q="women's linen summer dresses",
                limit=5,
            )

            print("Catalog scope: all active crawled catalogs")
            for product in results.items:
                brand = product.brand.name if product.brand else "Unknown brand"
                price = (
                    f"${product.current_price:.2f}"
                    if product.current_price is not None
                    else "Price unavailable"
                )
                title = product.title or "Untitled product"
                print(f"- {title} | {brand} | {price}")
                print(f"  {product.product_url}")
    except OctogenAPIError as exc:
        print(f"Octogen API error: status={exc.status_code}")
        print(pformat(exc.detail))
        raise


if __name__ == "__main__":
    asyncio.run(main())

Faceted search

Use Facet objects to filter by brand, gender, color, category, and other attributes. Pass a list of facets to the facets parameter of search_products.

from octogen_ai_sdk import Facet, FacetName, OctogenClient

async with OctogenClient() as client:
    results = await client.search_products(
        q="summer dress",
        facets=[
            Facet(name=FacetName.GENDER, values=["female"]),
            Facet(name=FacetName.COLOR_FAMILY, values=["Blue", "Green"]),
        ],
        price_max=150.0,
        limit=20,
    )
    for product in results.items:
        print(product.title, product.product_url)

You can also pass plain dicts instead of Facet instances — the SDK coerces them automatically:

facets=[
    {"name": "gender", "values": ["female"]},
    {"name": "color_family", "values": ["Blue", "Green"]},
]

FacetName is a StrEnum with constants for all built-in facet fields: BRAND_NAME, GENDER, AGE_GROUPS, COLOR, COLOR_FAMILY, IS_ACTIVEWEAR, PRODUCT_TYPE, CATEGORY_PATH_DEPTH_0 through CATEGORY_PATH_DEPTH_6, and more. You can also pass any custom attribute facet name as a plain string.

Pagination

When next_cursor is not None in a response, pass it as cursor in your next call to retrieve the following page:

cursor = None
while True:
    page = await client.search_products(
        q="linen dress",
        cursor=cursor,
        limit=50,
    )
    for product in page.items:
        print(product.title)
    if page.next_cursor is None:
        break
    cursor = page.next_cursor

Error handling

All SDK errors inherit from OctogenError. HTTP errors inherit from OctogenAPIError, which exposes status_code, detail, and response attributes.

OctogenAuthenticationError — 401

Raised when your API key is missing, malformed, or has been revoked. Check that OCTO_API_KEY is set correctly and that the key is still active in Octogen Platform.

OctogenForbiddenError — 403

Raised when a valid key attempts an action it is not authorized for — for example, accessing a catalog that has not been granted to your organization.

OctogenNotFoundError — 404

Raised when a catalog or product URL cannot be found. For lookup_product, verify the URL belongs to a catalog your key can access.

OctogenValidationError — 422

Raised when the API rejects a request due to invalid parameters. The detail attribute contains the validation error list from the API.

OctogenConnectionError

Raised when the SDK cannot reach the API — for example, due to a network timeout or DNS failure. Does not have a status_code.

Catch the base OctogenAPIError to handle all HTTP errors in one place:

from octogen_ai_sdk import (
    OctogenAPIError,
    OctogenAuthenticationError,
    OctogenConnectionError,
)

try:
    results = await client.search_products(q="dress")
except OctogenAuthenticationError:
    print("Check your OCTO_API_KEY.")
except OctogenAPIError as exc:
    print(f"API error {exc.status_code}: {exc.detail}")
except OctogenConnectionError as exc:
    print(f"Network error: {exc}")

MissingAPIKeyError is raised by the constructor, not by a network call. It will surface at client creation time if neither api_key nor OCTO_API_KEY is present.

If Octogen shares a catalog with you over BigQuery Analytics Hub, the SDK can subscribe to the listing and create a linked dataset in your own Google Cloud project, so you can query the catalog with SQL. See the Subscribe to a catalog in BigQuery guide for the full walkthrough. This helper uses your Google Cloud Application Default Credentials — not your OCTO_API_KEY — so it lives behind an optional bigquery extra. Sync it from the octogen-dev repository:

uv sync --project sdks/python --extra bigquery
gcloud auth application-default login

This installs the octogen-bq-subscribe CLI (dry-run by default; --apply to subscribe) and the octogen-bq-autosubscribe cron helper for keeping linked datasets current as new listings become available:

uv run --project sdks/python --extra bigquery octogen-bq-subscribe --listing <listing-resource> --project my-gcp-project --apply

`subscribe_to_listing`

def subscribe_to_listing(
    *,
    listing_resource: str,
    destination_project: str,
    destination_dataset_id: str | None = None,
    location: str | None = None,
    friendly_name: str | None = None,
    credentials: Any = None,
) -> BigQuerySubscriptionResult

A synchronous function (unlike the async OctogenClient). Idempotent — an existing subscription to the same listing in destination_project is returned as-is.

Parameter	Type	Default	Description
`listing_resource`	`str`	—	Listing resource name from the Platform UI BigQuery sharing page or MCP `list_bigquery_listing_resources` tool.
`destination_project`	`str`	—	Destination GCP project for the linked dataset (must be yours).
`destination_dataset_id`	`str \| None`	`octogen_<listing-id>`	Linked dataset id.
`location`	`str \| None`	The listing’s location	BigQuery dataset location.
`friendly_name`	`str \| None`	`None`	Linked dataset display name.
`credentials`	`google.auth.credentials.Credentials \| None`	`None` (uses ADC)	Explicit Google credentials, if you don’t want to use ADC.

Returns: BigQuerySubscriptionResult with listing_resource, linked_project, linked_dataset, state, already_subscribed, subscription_name, and a ready-to-run sample_query.

from octogen_ai_sdk import subscribe_to_listing

result = subscribe_to_listing(
    listing_resource="projects/octogen-prod/locations/us/dataExchanges/oneoff/listings/farfetch",
    destination_project="my-gcp-project",
)
print(result.linked_dataset, result.state)

Two BigQuery error types extend OctogenError: OctogenBigQueryError (base) and OctogenBigQueryAccessPendingError, raised when Octogen’s IAM grant on the listing hasn’t propagated yet. Because the grant is asynchronous, catch the access-pending error and retry after a few minutes.

Get Started

SDKs

MCP Integration

Guides

Octogen Python SDK — async HTTP client API reference

Installation

Authentication

Client constructor

Using the client as an async context manager

Methods

`lookup_product`

`search_products`

`more_like_this_products`

Complete example

Faceted search

Error handling

`subscribe_to_listing`

​Installation

​Authentication

​Client constructor

​Using the client as an async context manager

​Methods

​lookup_product

​search_products

​more_like_this_products

​Complete example

​Faceted search

​Pagination

​Error handling

​BigQuery subscribe (optional)

​subscribe_to_listing

Installation

Authentication

Client constructor

Using the client as an async context manager

Methods

`lookup_product`

`search_products`

`more_like_this_products`

Complete example

Faceted search

Pagination

Error handling

BigQuery subscribe (optional)

`subscribe_to_listing`