> ## Documentation Index > Fetch the complete documentation index at: https://www.octogen.ai/docs/llms.txt > Use this file to discover all available pages before exploring further. # Octogen Python SDK — async HTTP client API reference > Install the async Python SDK for Octogen, set your API key, and search or look up products from your granted catalogs with typed Pydantic models. 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, 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 Requires Python 3.11 or later. The SDK ships from the public [`octogen-dev`](https://github.com/octogen-ai/octogen-dev) repository and is managed with [uv](https://docs.astral.sh/uv/). Clone the repository and sync the `sdks/python` project: ```bash theme={null} 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`. The client reads your key from the `OCTO_API_KEY` environment variable automatically. ```bash theme={null} 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](#authentication) below). ## Authentication `OctogenClient` resolves your API key in this order: 1. The `api_key` constructor argument, if provided. 2. 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 `. ```python theme={null} 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: ```python theme={null} 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` ```python theme={null} 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. ```python theme={null} 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` ```python theme={null} 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. ```python theme={null} 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` ```python theme={null} 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](#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` ```python theme={null} 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 granted catalogs, or pass a catalog key to keep the results within one catalog. ```python theme={null} 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 catalog identifier from `list_catalogs()`. Omit to search all granted 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 confirms that catalogs are available, searches all granted catalogs for women's linen dresses, and prints each result with its brand and price: ```python theme={null} 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`. ```python theme={null} 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: ```python theme={null} 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: ```python theme={null} 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. 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. 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. Raised when a catalog or product URL cannot be found. For `lookup_product`, verify the URL belongs to a catalog your key can access. Raised when the API rejects a request due to invalid parameters. The `detail` attribute contains the validation error list from the API. 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: ```python theme={null} 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. ## BigQuery subscribe (optional) If Octogen shares a catalog with you over [BigQuery Analytics Hub](https://cloud.google.com/bigquery/docs/analytics-hub-introduction), 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](/guides/bigquery-subscribe) 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`](https://github.com/octogen-ai/octogen-dev) repository: ```bash theme={null} 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: ```bash theme={null} uv run --project sdks/python --extra bigquery octogen-bq-subscribe --listing --project my-gcp-project --apply ``` ### `subscribe_to_listing` ```python theme={null} 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_` | 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`. ```python theme={null} 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.