[DE-6014] Add support for limited access keys (#450)

* Add sdk support for limited access keys

* Update NoAPIKey error messaging

* Use cimg/python:3.7 for build_test as debian buster is EOL

* Try using debian bullseye

* Fix typecheck errors

* Fix isort import errors

* Fix api key env var names

* Fix nucleus.NucleusClient formatting in docs

* Skip test_box_pred_upload_embedding bc of issue with customObjectIndexingJobId

* Update os.environ for cli tests

* Update test cli os.environ

* Skip indexing tests

* Update NucleusClient __init__ description

* Update reasoning for skipping indexing jobs

* Update description of NucleusClient __init__

* Investigate if test_job_listing_and_retrieval doesn't work with limited access keys

* Revert "Investigate if test_job_listing_and_retrieval doesn't work with limited access keys"

This reverts commit d15e725.

* Add sleep in test_job_listing_and_retrieval

* Revert "Add sleep in test_job_listing_and_retrieval"

This reverts commit 67b8713.

Author: edwinpav

.circleci/config.yml

@@ -10,7 +10,7 @@ jobs:
     docker:
       # Important: Don't change this otherwise we will stop testing the earliest
       # python version we have to support.
-      - image: python:3.7-buster
+      - image: python:3.7-bullseye
     resource_class: medium
     parallelism: 6
     steps:

CHANGELOG.md

@@ -6,6 +6,25 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 
+## [0.17.11](https://github.com/scaleapi/nucleus-python-client/releases/tag/v0.17.11) - 2025-11-03
+
+### Added
+- Support passing a limited access key via `NucleusClient(limited_access_key=...)`. When provided, the client sends the `x-limited-access-key` header on all requests (sync and async).
+- Allow using the SDK without a standard API key when a `limited_access_key` is supplied. In this mode, Basic Auth is omitted and only the limited access header is used.
+
+Example usage:
+
+```python
+client = nucleus.NucleusClient(limited_access_key="<LIMITED_ACCESS_KEY>")
+#...
+```
+
+### Changed
+- `Connection` accepts `extra_headers` and only includes Basic Auth when `api_key` is provided. This enables header-only auth with limited access keys.
+- Header propagation applies across all request paths, including Validate endpoints and concurrent async helpers.
+- Tests updated to be tolerant of limited-access-only runs.
+- NoAPIKey error messaging updated to account for limited_access_key support.
+
 ## [0.17.10](https://github.com/scaleapi/nucleus-python-client/releases/tag/v0.17.10) - 2025-03-19
 
 ### Added

cli/client.py

@@ -12,8 +12,11 @@ def init_client():
         import nucleus
 
         api_key = os.environ.get("NUCLEUS_API_KEY", None)
-        if api_key:
-            client = nucleus.NucleusClient(api_key)
+        limited_access_key = os.environ.get("NUCLEUS_LIMITED_ACCESS_KEY", None)
+        if api_key or limited_access_key:
+            client = nucleus.NucleusClient(api_key=api_key, limited_access_key=limited_access_key)
         else:
-            raise RuntimeError("No NUCLEUS_API_KEY set")
+            raise RuntimeError(
+                "Set NUCLEUS_API_KEY or NUCLEUS_LIMITED_ACCESS_KEY"
+            )
         return client

conftest.py

@@ -11,18 +11,23 @@
 if TYPE_CHECKING:
     from nucleus import NucleusClient
 
-assert "NUCLEUS_PYTEST_API_KEY" in os.environ, (
-    "You must set the 'NUCLEUS_PYTEST_API_KEY' environment variable to a valid "
-    "Nucleus API key to run the test suite"
+assert "NUCLEUS_PYTEST_API_KEY" in os.environ or "NUCLEUS_PYTEST_LIMITED_ACCESS_KEY" in os.environ, (
+    "You must set at least one of 'NUCLEUS_PYTEST_API_KEY' or "
+    "'NUCLEUS_PYTEST_LIMITED_ACCESS_KEY' environment variables to run the test suite"
 )
 
-API_KEY = os.environ["NUCLEUS_PYTEST_API_KEY"]
+API_KEY = os.environ.get("NUCLEUS_PYTEST_API_KEY") if "NUCLEUS_PYTEST_API_KEY" in os.environ else None
+LIMITED_ACCESS_KEY = os.environ.get("NUCLEUS_PYTEST_LIMITED_ACCESS_KEY") if "NUCLEUS_PYTEST_LIMITED_ACCESS_KEY" in os.environ else None
 
 
 @pytest.fixture(scope="session")
 def CLIENT():
-    client = nucleus.NucleusClient(API_KEY)
-    return client
+    if API_KEY and LIMITED_ACCESS_KEY:
+        return nucleus.NucleusClient(api_key=API_KEY, limited_access_key=LIMITED_ACCESS_KEY)
+    if API_KEY:
+        return nucleus.NucleusClient(api_key=API_KEY)
+    # LIMITED_ACCESS_KEY only
+    return nucleus.NucleusClient(limited_access_key=LIMITED_ACCESS_KEY)
 
 
 @pytest.fixture()

nucleus/__init__.py

@@ -176,19 +176,49 @@ class NucleusClient:
 
     Parameters:
         api_key: Follow `this guide <https://scale.com/docs/account#section-api-keys>`_
-          to retrieve your API keys.
+          to retrieve your API keys. **Only** optional when ``limited_access_key`` is provided.
         use_notebook: Whether the client is being used in a notebook (toggles tqdm
           style). Default is ``False``.
         endpoint: Base URL of the API. Default is Nucleus's current production API.
+        limited_access_key: Key enabling additional, scoped access. **Only** optional when ``api_key`` is provided. Reach out to your Scale representative to obtain a limited access key.
+
+    Authentication notes:
+        Some users have Nucleus-only API keys. You can
+        instantiate the client with only ``limited_access_key`` (no ``api_key``) and the SDK
+        will authenticate requests using this key only. If both
+        ``api_key`` and ``limited_access_key`` are provided, Basic Auth (``api_key``) and the
+        additional limited access key will both be sent.
+
+        .. code-block:: python
+
+           # Using a basic auth key
+           import nucleus
+           client = nucleus.NucleusClient(api_key="YOUR_API_KEY", ...)
+
+           # Using only a limited access key (no Basic Auth)
+           import nucleus
+           client = nucleus.NucleusClient(limited_access_key="YOUR_LIMITED_KEY", ...)
+
+           # Using both keys (Basic Auth and limited access header)
+           client = nucleus.NucleusClient(
+               api_key="YOUR_API_KEY",
+               limited_access_key="YOUR_LIMITED_KEY",
+               ...
+           )
     """
 
     def __init__(
         self,
         api_key: Optional[str] = None,
         use_notebook: bool = False,
         endpoint: Optional[str] = None,
+        limited_access_key: Optional[str] = None,
     ):
-        self.api_key = self._set_api_key(api_key)
+        # Allow usage with only a limited access key
+        if api_key is None and limited_access_key:
+            self.api_key = None
+        else:
+            self.api_key = self._set_api_key(api_key)
         self.tqdm_bar = tqdm.tqdm
         if endpoint is None:
             self.endpoint = os.environ.get(
@@ -201,8 +231,11 @@ def __init__(
             import tqdm.notebook as tqdm_notebook
 
             self.tqdm_bar = tqdm_notebook.tqdm
-        self.connection = Connection(self.api_key, self.endpoint)
-        self.validate = Validate(self.api_key, self.endpoint)
+        self.extra_headers: Dict[str, str] = {}
+        if limited_access_key:
+            self.extra_headers["x-limited-access-key"] = limited_access_key
+        self.connection = Connection(self.api_key, self.endpoint, extra_headers=self.extra_headers)
+        self.validate = Validate(self.api_key, self.endpoint, extra_headers=self.extra_headers)
 
     def __repr__(self):
         return f"NucleusClient(api_key='{self.api_key}', use_notebook={self._use_notebook}, endpoint='{self.endpoint}')"

nucleus/async_utils.py

@@ -185,7 +185,14 @@ async def _post_form_data(
                 async with session.post(
                     endpoint,
                     data=form,
-                    auth=aiohttp.BasicAuth(client.api_key, ""),
+                    auth=(
+                        (lambda k: aiohttp.BasicAuth(str(k), ""))(
+                            getattr(client, "api_key", None)
+                        )
+                        if getattr(client, "api_key", None) is not None
+                        else None
+                    ),
+                    headers=getattr(client, "extra_headers", None),
                     timeout=DEFAULT_NETWORK_TIMEOUT_SEC,
                 ) as response:
                     data = await _parse_async_response(
@@ -223,7 +230,14 @@ async def _make_request(
         for sleep_time in RetryStrategy.sleep_times() + [-1]:
             async with session.get(
                 endpoint,
-                auth=aiohttp.BasicAuth(client.api_key, ""),
+                auth=(
+                    (lambda k: aiohttp.BasicAuth(str(k), ""))(
+                        getattr(client, "api_key", None)
+                    )
+                    if getattr(client, "api_key", None) is not None
+                    else None
+                ),
+                headers=getattr(client, "extra_headers", None),
                 timeout=DEFAULT_NETWORK_TIMEOUT_SEC,
             ) as response:
                 data = await _parse_async_response(

nucleus/connection.py

@@ -5,17 +5,21 @@
 import requests
 
 from .constants import DEFAULT_NETWORK_TIMEOUT_SEC
-from .errors import NucleusAPIError
+from .errors import NoAPIKey, NucleusAPIError
 from .logger import logger
 from .retry_strategy import RetryStrategy
 
 
 class Connection:
     """Wrapper of HTTP requests to the Nucleus endpoint."""
 
-    def __init__(self, api_key: str, endpoint: Optional[str] = None):
+    def __init__(self, api_key: Optional[str] = None, endpoint: Optional[str] = None, extra_headers: Optional[dict] = None):
         self.api_key = api_key
         self.endpoint = endpoint
+        self.extra_headers = extra_headers or {}
+        # Require at least one auth mechanism: Basic (api_key) or limited access header
+        if self.api_key is None and not self.extra_headers.get("x-limited-access-key"):
+            raise NoAPIKey()
 
     def __repr__(self):
         return (
@@ -63,13 +67,19 @@ def make_request(
         logger.info("Make request to %s", endpoint)
 
         for retry_wait_time in RetryStrategy.sleep_times():
+            auth_kwargs = (
+                {"auth": (self.api_key, "")} if self.api_key is not None else {}
+            )
             response = requests_command(
                 endpoint,
                 json=payload,
-                headers={"Content-Type": "application/json"},
-                auth=(self.api_key, ""),
+                headers={
+                    "Content-Type": "application/json",
+                    **(self.extra_headers or {}),
+                },
                 timeout=DEFAULT_NETWORK_TIMEOUT_SEC,
                 verify=os.environ.get("NUCLEUS_SKIP_SSL_VERIFY", None) is None,
+                **auth_kwargs,
             )
             logger.info(
                 "API request has response code %s", response.status_code

nucleus/errors.py

@@ -68,7 +68,7 @@ def __init__(
 class NoAPIKey(Exception):
     def __init__(
         self,
-        message="You need to pass an API key to the NucleusClient or set the environment variable NUCLEUS_API_KEY",
+        message="You must provide credentials to NucleusClient: pass api_key or limited_access_key, or set the environment variable NUCLEUS_API_KEY or NUCLEUS_LIMITED_ACCESS_KEY",
     ):
         self.message = message
         super().__init__(self.message)

nucleus/validate/client.py

@@ -1,4 +1,4 @@
-from typing import List
+from typing import List, Optional
 
 from nucleus.async_job import AsyncJob
 from nucleus.connection import Connection
@@ -25,8 +25,8 @@
 class Validate:
     """Model CI Python Client extension."""
 
-    def __init__(self, api_key: str, endpoint: str):
-        self.connection = Connection(api_key, endpoint)
+    def __init__(self, api_key: Optional[str], endpoint: str, extra_headers: Optional[dict] = None):
+        self.connection = Connection(api_key, endpoint, extra_headers=extra_headers)
 
     def __repr__(self):
         return f"Validate(connection='{self.connection}')"

pyproject.toml

@@ -25,7 +25,7 @@ ignore = ["E501", "E741", "E731", "F401"]  # Easy ignore for getting it running
 
 [tool.poetry]
 name = "scale-nucleus"
-version = "0.17.10"
+version = "0.17.11"
 description = "The official Python client library for Nucleus, the Data Platform for AI"
 license =  "MIT"
 authors = ["Scale AI Nucleus Team <nucleusapi@scaleapi.com>"]