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 list catalogs, search products by keyword and facet, 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:
        catalogs = await client.list_catalogs()
        print(catalogs)

asyncio.run(main())

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

Methods

`list_catalogs`

async def list_catalogs() -> list[MerchantCatalogSummary]

Returns all active catalogs granted to your API key’s organization. You can omit catalog from search_products to search all of them, or pass a catalog value to restrict search to one catalog.

catalogs = await client.list_catalogs()
for c in catalogs:
    print(c.catalog, c.display_name, c.product_count)

Returns: list[MerchantCatalogSummary] Each MerchantCatalogSummary has:

Field	Type	Description
`catalog`	`str`	Catalog identifier — pass this to `search_products` when you want to search only this catalog.
`display_name`	`str`	Human-readable catalog name.
`product_count`	`int`	Number of indexed products.
`source_base_url`	`str \| None`	Base URL of the catalog source.
`last_indexed_at`	`datetime \| None`	When the catalog was last indexed.

`lookup_product`

async def lookup_product(url: str) -> MerchantProductUrlLookupResponse

Resolves a canonical product page URL to its full product record, including pricing, images, variants, sizes, and enrichment data. The URL must belong to a catalog granted to your key.

result = await client.lookup_product("https://example.com/products/blue-linen-dress")
print(result.product.title)
print(result.product.current_price)

Returns: MerchantProductUrlLookupResponse

Field	Type	Description
`catalog_key`	`str`	Catalog the product belongs to.
`catalog_display_name`	`str`	Human-readable catalog name.
`product`	`MerchantProductView`	Full product record with all detail fields.

`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 catalog identifier from `list_catalogs()`. Omit to search all granted 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, product_url, title, brand, current_price, original_price, image_url, images, rating, and updated_at.

Complete example

The following example confirms that catalogs are available, searches all granted 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:
            catalogs = await client.list_catalogs()
            if not catalogs:
                print("No catalogs are available for this API key.")
                return

            results = await client.search_products(
                q="women's linen summer dresses",
                limit=5,
            )

            print("Catalog scope: all granted 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 the partner portal.

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.

​Installation

​Authentication

​Client constructor

​Using the client as an async context manager

​Methods

​list_catalogs

​lookup_product

​search_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

`list_catalogs`

`lookup_product`

`search_products`

Complete example

Faceted search

Pagination

Error handling

BigQuery subscribe (optional)

`subscribe_to_listing`