_client.py

# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.

from __future__ import annotations

import os
from typing import TYPE_CHECKING, Any, Dict, Type, Mapping, cast
from typing_extensions import Self, Literal, override

import httpx

from . import _exceptions
from ._qs import Querystring
from ._types import (
    Omit,
    Timeout,
    NotGiven,
    ResponseT,
    Transport,
    ProxiesTypes,
    RequestOptions,
    not_given,
)
from ._utils import (
    is_given,
    is_mapping_t,
    get_async_library,
)
from ._compat import cached_property
from ._models import FinalRequestOptions
from ._version import __version__
from ._streaming import Stream as Stream, AsyncStream as AsyncStream
from ._exceptions import KernelError, APIStatusError
from ._base_client import (
    DEFAULT_MAX_RETRIES,
    SyncAPIClient,
    AsyncAPIClient,
)
from .lib.browser_routing.routing import (
    BrowserRouteCache,
    BrowserRoutingConfig,
    strip_direct_vm_auth,
    rewrite_direct_vm_options,
    browser_routing_config_from_env,
    maybe_evict_browser_route_from_response,
    maybe_populate_browser_route_cache_from_response,
)

if TYPE_CHECKING:
    from .resources import (
        apps,
        auth,
        proxies,
        api_keys,
        browsers,
        profiles,
        projects,
        audit_logs,
        extensions,
        credentials,
        deployments,
        invocations,
        organization,
        browser_pools,
        credential_providers,
    )
    from .resources.apps import AppsResource, AsyncAppsResource
    from .resources.proxies import ProxiesResource, AsyncProxiesResource
    from .resources.api_keys import APIKeysResource, AsyncAPIKeysResource
    from .resources.profiles import ProfilesResource, AsyncProfilesResource
    from .resources.auth.auth import AuthResource, AsyncAuthResource
    from .resources.audit_logs import AuditLogsResource, AsyncAuditLogsResource
    from .resources.extensions import ExtensionsResource, AsyncExtensionsResource
    from .resources.credentials import CredentialsResource, AsyncCredentialsResource
    from .resources.deployments import DeploymentsResource, AsyncDeploymentsResource
    from .resources.invocations import InvocationsResource, AsyncInvocationsResource
    from .resources.browser_pools import BrowserPoolsResource, AsyncBrowserPoolsResource
    from .resources.browsers.browsers import BrowsersResource, AsyncBrowsersResource
    from .resources.projects.projects import ProjectsResource, AsyncProjectsResource
    from .resources.credential_providers import CredentialProvidersResource, AsyncCredentialProvidersResource
    from .resources.organization.organization import OrganizationResource, AsyncOrganizationResource

__all__ = [
    "ENVIRONMENTS",
    "Timeout",
    "Transport",
    "ProxiesTypes",
    "RequestOptions",
    "Kernel",
    "AsyncKernel",
    "Client",
    "AsyncClient",
]

ENVIRONMENTS: Dict[str, str] = {
    "production": "https://api.onkernel.com/",
    "development": "https://localhost:3001/",
}


class Kernel(SyncAPIClient):
    # client options
    api_key: str
    browser_route_cache: BrowserRouteCache

    project_id: str | None

    _environment: Literal["production", "development"] | NotGiven
    _browser_routing: BrowserRoutingConfig

    def __init__(
        self,
        *,
        api_key: str | None = None,
        project_id: str | None = None,
        environment: Literal["production", "development"] | NotGiven = not_given,
        base_url: str | httpx.URL | None | NotGiven = not_given,
        timeout: float | Timeout | None | NotGiven = not_given,
        max_retries: int = DEFAULT_MAX_RETRIES,
        default_headers: Mapping[str, str] | None = None,
        default_query: Mapping[str, object] | None = None,
        # Configure a custom httpx client.
        # We provide a `DefaultHttpxClient` class that you can pass to retain the default values we use for `limits`, `timeout` & `follow_redirects`.
        # See the [httpx documentation](https://www.python-httpx.org/api/#client) for more details.
        http_client: httpx.Client | None = None,
        # Enable or disable schema validation for data returned by the API.
        # When enabled an error APIResponseValidationError is raised
        # if the API responds with invalid data for the expected schema.
        #
        # This parameter may be removed or changed in the future.
        # If you rely on this feature, please open a GitHub issue
        # outlining your use-case to help us decide if it should be
        # part of our public interface in the future.
        _strict_response_validation: bool = False,
        _browser_route_cache: BrowserRouteCache | None = None,
    ) -> None:
        """Construct a new synchronous Kernel client instance.

        This automatically infers the `api_key` argument from the `KERNEL_API_KEY` environment variable if it is not provided.
        """
        if api_key is None:
            api_key = os.environ.get("KERNEL_API_KEY")
        if api_key is None:
            raise KernelError(
                "The api_key client option must be set either by passing api_key to the client or by setting the KERNEL_API_KEY environment variable"
            )
        self.api_key = api_key

        self.project_id = project_id

        self._environment = environment

        base_url_env = os.environ.get("KERNEL_BASE_URL")
        if is_given(base_url) and base_url is not None:
            # cast required because mypy doesn't understand the type narrowing
            base_url = cast("str | httpx.URL", base_url)  # pyright: ignore[reportUnnecessaryCast]
        elif is_given(environment):
            if base_url_env and base_url is not None:
                raise ValueError(
                    "Ambiguous URL; The `KERNEL_BASE_URL` env var and the `environment` argument are given. If you want to use the environment, you must pass base_url=None",
                )

            try:
                base_url = ENVIRONMENTS[environment]
            except KeyError as exc:
                raise ValueError(f"Unknown environment: {environment}") from exc
        elif base_url_env is not None:
            base_url = base_url_env
        else:
            self._environment = environment = "production"

            try:
                base_url = ENVIRONMENTS[environment]
            except KeyError as exc:
                raise ValueError(f"Unknown environment: {environment}") from exc

        custom_headers_env = os.environ.get("KERNEL_CUSTOM_HEADERS")
        if custom_headers_env is not None:
            parsed: dict[str, str] = {}
            for line in custom_headers_env.split("\n"):
                colon = line.find(":")
                if colon >= 0:
                    parsed[line[:colon].strip()] = line[colon + 1 :].strip()
            default_headers = {**parsed, **(default_headers if is_mapping_t(default_headers) else {})}

        super().__init__(
            version=__version__,
            base_url=base_url,
            max_retries=max_retries,
            timeout=timeout,
            http_client=http_client,
            custom_headers=default_headers,
            custom_query=default_query,
            _strict_response_validation=_strict_response_validation,
        )
        self.browser_route_cache = _browser_route_cache or BrowserRouteCache()
        self._browser_routing = browser_routing_config_from_env()

    @cached_property
    def deployments(self) -> DeploymentsResource:
        """Create and manage app deployments and stream deployment events."""
        from .resources.deployments import DeploymentsResource

        return DeploymentsResource(self)

    @cached_property
    def apps(self) -> AppsResource:
        """List applications and versions."""
        from .resources.apps import AppsResource

        return AppsResource(self)

    @cached_property
    def invocations(self) -> InvocationsResource:
        """Invoke actions and stream or query invocation status and events."""
        from .resources.invocations import InvocationsResource

        return InvocationsResource(self)

    @cached_property
    def browsers(self) -> BrowsersResource:
        """Create and manage browser sessions."""
        from .resources.browsers import BrowsersResource

        return BrowsersResource(self)

    @cached_property
    def profiles(self) -> ProfilesResource:
        """Create, list, retrieve, and delete browser profiles."""
        from .resources.profiles import ProfilesResource

        return ProfilesResource(self)

    @cached_property
    def auth(self) -> AuthResource:
        from .resources.auth import AuthResource

        return AuthResource(self)

    @cached_property
    def proxies(self) -> ProxiesResource:
        """Create and manage proxy configurations for routing browser traffic."""
        from .resources.proxies import ProxiesResource

        return ProxiesResource(self)

    @cached_property
    def extensions(self) -> ExtensionsResource:
        """Create, list, retrieve, and delete browser extensions."""
        from .resources.extensions import ExtensionsResource

        return ExtensionsResource(self)

    @cached_property
    def browser_pools(self) -> BrowserPoolsResource:
        """Create and manage browser pools for acquiring and releasing browsers."""
        from .resources.browser_pools import BrowserPoolsResource

        return BrowserPoolsResource(self)

    @cached_property
    def credentials(self) -> CredentialsResource:
        """Create and manage credentials for authentication."""
        from .resources.credentials import CredentialsResource

        return CredentialsResource(self)

    @cached_property
    def projects(self) -> ProjectsResource:
        """
        Create and manage projects for resource isolation within an organization.
        When projects are disabled for the organization, project operations return
        `404` with code `projects_disabled`.
        """
        from .resources.projects import ProjectsResource

        return ProjectsResource(self)

    @cached_property
    def organization(self) -> OrganizationResource:
        from .resources.organization import OrganizationResource

        return OrganizationResource(self)

    @cached_property
    def audit_logs(self) -> AuditLogsResource:
        """Read audit log records for the authenticated organization."""
        from .resources.audit_logs import AuditLogsResource

        return AuditLogsResource(self)

    @cached_property
    def api_keys(self) -> APIKeysResource:
        """Create and manage API keys for organization and project-scoped access."""
        from .resources.api_keys import APIKeysResource

        return APIKeysResource(self)

    @cached_property
    def credential_providers(self) -> CredentialProvidersResource:
        """Configure external credential providers like 1Password."""
        from .resources.credential_providers import CredentialProvidersResource

        return CredentialProvidersResource(self)

    @cached_property
    def with_raw_response(self) -> KernelWithRawResponse:
        return KernelWithRawResponse(self)

    @cached_property
    def with_streaming_response(self) -> KernelWithStreamedResponse:
        return KernelWithStreamedResponse(self)

    @property
    @override
    def qs(self) -> Querystring:
        return Querystring(array_format="comma")

    @property
    @override
    def auth_headers(self) -> dict[str, str]:
        api_key = self.api_key
        return {"Authorization": f"Bearer {api_key}"}

    @property
    @override
    def default_headers(self) -> dict[str, str | Omit]:
        return {
            **super().default_headers,
            "X-Stainless-Async": "false",
            "X-Kernel-Project-Id": self.project_id if self.project_id is not None else Omit(),
            **self._custom_headers,
        }

    @override
    def _prepare_options(self, options: Any) -> Any:
        options = cast(Any, super()._prepare_options(options))
        return rewrite_direct_vm_options(options, cache=self.browser_route_cache, config=self._browser_routing)

    @override
    def _prepare_request(self, request: httpx.Request) -> None:
        strip_direct_vm_auth(request, cache=self.browser_route_cache)

    @override
    def _process_response(
        self,
        *,
        cast_to: Type[ResponseT],
        options: FinalRequestOptions,
        response: httpx.Response,
        stream: bool,
        stream_cls: type[Stream[Any]] | type[AsyncStream[Any]] | None,
        retries_taken: int = 0,
    ) -> ResponseT:
        maybe_populate_browser_route_cache_from_response(response, cache=self.browser_route_cache)
        maybe_evict_browser_route_from_response(response, cache=self.browser_route_cache)
        return super()._process_response(
            cast_to=cast_to,
            options=options,
            response=response,
            stream=stream,
            stream_cls=stream_cls,
            retries_taken=retries_taken,
        )

    def copy(
        self,
        *,
        api_key: str | None = None,
        project_id: str | None = None,
        environment: Literal["production", "development"] | None = None,
        base_url: str | httpx.URL | None = None,
        timeout: float | Timeout | None | NotGiven = not_given,
        http_client: httpx.Client | None = None,
        max_retries: int | NotGiven = not_given,
        default_headers: Mapping[str, str] | None = None,
        set_default_headers: Mapping[str, str] | None = None,
        default_query: Mapping[str, object] | None = None,
        set_default_query: Mapping[str, object] | None = None,
        _browser_route_cache: BrowserRouteCache | None = None,
        _extra_kwargs: Mapping[str, Any] = {},
    ) -> Self:
        """
        Create a new client instance re-using the same options given to the current client with optional overriding.
        """
        if default_headers is not None and set_default_headers is not None:
            raise ValueError("The `default_headers` and `set_default_headers` arguments are mutually exclusive")

        if default_query is not None and set_default_query is not None:
            raise ValueError("The `default_query` and `set_default_query` arguments are mutually exclusive")

        headers = self._custom_headers
        if default_headers is not None:
            headers = {**headers, **default_headers}
        elif set_default_headers is not None:
            headers = set_default_headers

        params = self._custom_query
        if default_query is not None:
            params = {**params, **default_query}
        elif set_default_query is not None:
            params = set_default_query

        http_client = http_client or self._client
        return self.__class__(
            api_key=api_key or self.api_key,
            project_id=project_id or self.project_id,
            base_url=base_url or self.base_url,
            environment=environment or self._environment,
            timeout=self.timeout if isinstance(timeout, NotGiven) else timeout,
            http_client=http_client,
            max_retries=max_retries if is_given(max_retries) else self.max_retries,
            default_headers=headers,
            default_query=params,
            _browser_route_cache=_browser_route_cache or self.browser_route_cache,
            **_extra_kwargs,
        )

    # Alias for `copy` for nicer inline usage, e.g.
    # client.with_options(timeout=10).foo.create(...)
    with_options = copy

    @override
    def _make_status_error(
        self,
        err_msg: str,
        *,
        body: object,
        response: httpx.Response,
    ) -> APIStatusError:
        if response.status_code == 400:
            return _exceptions.BadRequestError(err_msg, response=response, body=body)

        if response.status_code == 401:
            return _exceptions.AuthenticationError(err_msg, response=response, body=body)

        if response.status_code == 403:
            return _exceptions.PermissionDeniedError(err_msg, response=response, body=body)

        if response.status_code == 404:
            return _exceptions.NotFoundError(err_msg, response=response, body=body)

        if response.status_code == 409:
            return _exceptions.ConflictError(err_msg, response=response, body=body)

        if response.status_code == 422:
            return _exceptions.UnprocessableEntityError(err_msg, response=response, body=body)

        if response.status_code == 429:
            return _exceptions.RateLimitError(err_msg, response=response, body=body)

        if response.status_code >= 500:
            return _exceptions.InternalServerError(err_msg, response=response, body=body)
        return APIStatusError(err_msg, response=response, body=body)


class AsyncKernel(AsyncAPIClient):
    # client options
    api_key: str
    browser_route_cache: BrowserRouteCache

    project_id: str | None

    _environment: Literal["production", "development"] | NotGiven
    _browser_routing: BrowserRoutingConfig

    def __init__(
        self,
        *,
        api_key: str | None = None,
        project_id: str | None = None,
        environment: Literal["production", "development"] | NotGiven = not_given,
        base_url: str | httpx.URL | None | NotGiven = not_given,
        timeout: float | Timeout | None | NotGiven = not_given,
        max_retries: int = DEFAULT_MAX_RETRIES,
        default_headers: Mapping[str, str] | None = None,
        default_query: Mapping[str, object] | None = None,
        # Configure a custom httpx client.
        # We provide a `DefaultAsyncHttpxClient` class that you can pass to retain the default values we use for `limits`, `timeout` & `follow_redirects`.
        # See the [httpx documentation](https://www.python-httpx.org/api/#asyncclient) for more details.
        http_client: httpx.AsyncClient | None = None,
        # Enable or disable schema validation for data returned by the API.
        # When enabled an error APIResponseValidationError is raised
        # if the API responds with invalid data for the expected schema.
        #
        # This parameter may be removed or changed in the future.
        # If you rely on this feature, please open a GitHub issue
        # outlining your use-case to help us decide if it should be
        # part of our public interface in the future.
        _strict_response_validation: bool = False,
        _browser_route_cache: BrowserRouteCache | None = None,
    ) -> None:
        """Construct a new async AsyncKernel client instance.

        This automatically infers the `api_key` argument from the `KERNEL_API_KEY` environment variable if it is not provided.
        """
        if api_key is None:
            api_key = os.environ.get("KERNEL_API_KEY")
        if api_key is None:
            raise KernelError(
                "The api_key client option must be set either by passing api_key to the client or by setting the KERNEL_API_KEY environment variable"
            )
        self.api_key = api_key

        self.project_id = project_id

        self._environment = environment

        base_url_env = os.environ.get("KERNEL_BASE_URL")
        if is_given(base_url) and base_url is not None:
            # cast required because mypy doesn't understand the type narrowing
            base_url = cast("str | httpx.URL", base_url)  # pyright: ignore[reportUnnecessaryCast]
        elif is_given(environment):
            if base_url_env and base_url is not None:
                raise ValueError(
                    "Ambiguous URL; The `KERNEL_BASE_URL` env var and the `environment` argument are given. If you want to use the environment, you must pass base_url=None",
                )

            try:
                base_url = ENVIRONMENTS[environment]
            except KeyError as exc:
                raise ValueError(f"Unknown environment: {environment}") from exc
        elif base_url_env is not None:
            base_url = base_url_env
        else:
            self._environment = environment = "production"

            try:
                base_url = ENVIRONMENTS[environment]
            except KeyError as exc:
                raise ValueError(f"Unknown environment: {environment}") from exc

        custom_headers_env = os.environ.get("KERNEL_CUSTOM_HEADERS")
        if custom_headers_env is not None:
            parsed: dict[str, str] = {}
            for line in custom_headers_env.split("\n"):
                colon = line.find(":")
                if colon >= 0:
                    parsed[line[:colon].strip()] = line[colon + 1 :].strip()
            default_headers = {**parsed, **(default_headers if is_mapping_t(default_headers) else {})}

        super().__init__(
            version=__version__,
            base_url=base_url,
            max_retries=max_retries,
            timeout=timeout,
            http_client=http_client,
            custom_headers=default_headers,
            custom_query=default_query,
            _strict_response_validation=_strict_response_validation,
        )
        self.browser_route_cache = _browser_route_cache or BrowserRouteCache()
        self._browser_routing = browser_routing_config_from_env()

    @cached_property
    def deployments(self) -> AsyncDeploymentsResource:
        """Create and manage app deployments and stream deployment events."""
        from .resources.deployments import AsyncDeploymentsResource

        return AsyncDeploymentsResource(self)

    @cached_property
    def apps(self) -> AsyncAppsResource:
        """List applications and versions."""
        from .resources.apps import AsyncAppsResource

        return AsyncAppsResource(self)

    @cached_property
    def invocations(self) -> AsyncInvocationsResource:
        """Invoke actions and stream or query invocation status and events."""
        from .resources.invocations import AsyncInvocationsResource

        return AsyncInvocationsResource(self)

    @cached_property
    def browsers(self) -> AsyncBrowsersResource:
        """Create and manage browser sessions."""
        from .resources.browsers import AsyncBrowsersResource

        return AsyncBrowsersResource(self)

    @cached_property
    def profiles(self) -> AsyncProfilesResource:
        """Create, list, retrieve, and delete browser profiles."""
        from .resources.profiles import AsyncProfilesResource

        return AsyncProfilesResource(self)

    @cached_property
    def auth(self) -> AsyncAuthResource:
        from .resources.auth import AsyncAuthResource

        return AsyncAuthResource(self)

    @cached_property
    def proxies(self) -> AsyncProxiesResource:
        """Create and manage proxy configurations for routing browser traffic."""
        from .resources.proxies import AsyncProxiesResource

        return AsyncProxiesResource(self)

    @cached_property
    def extensions(self) -> AsyncExtensionsResource:
        """Create, list, retrieve, and delete browser extensions."""
        from .resources.extensions import AsyncExtensionsResource

        return AsyncExtensionsResource(self)

    @cached_property
    def browser_pools(self) -> AsyncBrowserPoolsResource:
        """Create and manage browser pools for acquiring and releasing browsers."""
        from .resources.browser_pools import AsyncBrowserPoolsResource

        return AsyncBrowserPoolsResource(self)

    @cached_property
    def credentials(self) -> AsyncCredentialsResource:
        """Create and manage credentials for authentication."""
        from .resources.credentials import AsyncCredentialsResource

        return AsyncCredentialsResource(self)

    @cached_property
    def projects(self) -> AsyncProjectsResource:
        """
        Create and manage projects for resource isolation within an organization.
        When projects are disabled for the organization, project operations return
        `404` with code `projects_disabled`.
        """
        from .resources.projects import AsyncProjectsResource

        return AsyncProjectsResource(self)

    @cached_property
    def organization(self) -> AsyncOrganizationResource:
        from .resources.organization import AsyncOrganizationResource

        return AsyncOrganizationResource(self)

    @cached_property
    def audit_logs(self) -> AsyncAuditLogsResource:
        """Read audit log records for the authenticated organization."""
        from .resources.audit_logs import AsyncAuditLogsResource

        return AsyncAuditLogsResource(self)

    @cached_property
    def api_keys(self) -> AsyncAPIKeysResource:
        """Create and manage API keys for organization and project-scoped access."""
        from .resources.api_keys import AsyncAPIKeysResource

        return AsyncAPIKeysResource(self)

    @cached_property
    def credential_providers(self) -> AsyncCredentialProvidersResource:
        """Configure external credential providers like 1Password."""
        from .resources.credential_providers import AsyncCredentialProvidersResource

        return AsyncCredentialProvidersResource(self)

    @cached_property
    def with_raw_response(self) -> AsyncKernelWithRawResponse:
        return AsyncKernelWithRawResponse(self)

    @cached_property
    def with_streaming_response(self) -> AsyncKernelWithStreamedResponse:
        return AsyncKernelWithStreamedResponse(self)

    @property
    @override
    def qs(self) -> Querystring:
        return Querystring(array_format="comma")

    @property
    @override
    def auth_headers(self) -> dict[str, str]:
        api_key = self.api_key
        return {"Authorization": f"Bearer {api_key}"}

    @property
    @override
    def default_headers(self) -> dict[str, str | Omit]:
        return {
            **super().default_headers,
            "X-Stainless-Async": f"async:{get_async_library()}",
            "X-Kernel-Project-Id": self.project_id if self.project_id is not None else Omit(),
            **self._custom_headers,
        }

    @override
    async def _prepare_options(self, options: Any) -> Any:
        options = cast(Any, await super()._prepare_options(options))
        return rewrite_direct_vm_options(options, cache=self.browser_route_cache, config=self._browser_routing)

    @override
    async def _prepare_request(self, request: httpx.Request) -> None:
        strip_direct_vm_auth(request, cache=self.browser_route_cache)

    @override
    async def _process_response(
        self,
        *,
        cast_to: Type[ResponseT],
        options: FinalRequestOptions,
        response: httpx.Response,
        stream: bool,
        stream_cls: type[Stream[Any]] | type[AsyncStream[Any]] | None,
        retries_taken: int = 0,
    ) -> ResponseT:
        maybe_populate_browser_route_cache_from_response(response, cache=self.browser_route_cache)
        maybe_evict_browser_route_from_response(response, cache=self.browser_route_cache)
        return await super()._process_response(
            cast_to=cast_to,
            options=options,
            response=response,
            stream=stream,
            stream_cls=stream_cls,
            retries_taken=retries_taken,
        )

    def copy(
        self,
        *,
        api_key: str | None = None,
        project_id: str | None = None,
        environment: Literal["production", "development"] | None = None,
        base_url: str | httpx.URL | None = None,
        timeout: float | Timeout | None | NotGiven = not_given,
        http_client: httpx.AsyncClient | None = None,
        max_retries: int | NotGiven = not_given,
        default_headers: Mapping[str, str] | None = None,
        set_default_headers: Mapping[str, str] | None = None,
        default_query: Mapping[str, object] | None = None,
        set_default_query: Mapping[str, object] | None = None,
        _browser_route_cache: BrowserRouteCache | None = None,
        _extra_kwargs: Mapping[str, Any] = {},
    ) -> Self:
        """
        Create a new client instance re-using the same options given to the current client with optional overriding.
        """
        if default_headers is not None and set_default_headers is not None:
            raise ValueError("The `default_headers` and `set_default_headers` arguments are mutually exclusive")

        if default_query is not None and set_default_query is not None:
            raise ValueError("The `default_query` and `set_default_query` arguments are mutually exclusive")

        headers = self._custom_headers
        if default_headers is not None:
            headers = {**headers, **default_headers}
        elif set_default_headers is not None:
            headers = set_default_headers

        params = self._custom_query
        if default_query is not None:
            params = {**params, **default_query}
        elif set_default_query is not None:
            params = set_default_query

        http_client = http_client or self._client
        return self.__class__(
            api_key=api_key or self.api_key,
            project_id=project_id or self.project_id,
            base_url=base_url or self.base_url,
            environment=environment or self._environment,
            timeout=self.timeout if isinstance(timeout, NotGiven) else timeout,
            http_client=http_client,
            max_retries=max_retries if is_given(max_retries) else self.max_retries,
            default_headers=headers,
            default_query=params,
            _browser_route_cache=_browser_route_cache or self.browser_route_cache,
            **_extra_kwargs,
        )

    # Alias for `copy` for nicer inline usage, e.g.
    # client.with_options(timeout=10).foo.create(...)
    with_options = copy

    @override
    def _make_status_error(
        self,
        err_msg: str,
        *,
        body: object,
        response: httpx.Response,
    ) -> APIStatusError:
        if response.status_code == 400:
            return _exceptions.BadRequestError(err_msg, response=response, body=body)

        if response.status_code == 401:
            return _exceptions.AuthenticationError(err_msg, response=response, body=body)

        if response.status_code == 403:
            return _exceptions.PermissionDeniedError(err_msg, response=response, body=body)

        if response.status_code == 404:
            return _exceptions.NotFoundError(err_msg, response=response, body=body)

        if response.status_code == 409:
            return _exceptions.ConflictError(err_msg, response=response, body=body)

        if response.status_code == 422:
            return _exceptions.UnprocessableEntityError(err_msg, response=response, body=body)

        if response.status_code == 429:
            return _exceptions.RateLimitError(err_msg, response=response, body=body)

        if response.status_code >= 500:
            return _exceptions.InternalServerError(err_msg, response=response, body=body)
        return APIStatusError(err_msg, response=response, body=body)


class KernelWithRawResponse:
    _client: Kernel

    def __init__(self, client: Kernel) -> None:
        self._client = client

    @cached_property
    def deployments(self) -> deployments.DeploymentsResourceWithRawResponse:
        """Create and manage app deployments and stream deployment events."""
        from .resources.deployments import DeploymentsResourceWithRawResponse

        return DeploymentsResourceWithRawResponse(self._client.deployments)

    @cached_property
    def apps(self) -> apps.AppsResourceWithRawResponse:
        """List applications and versions."""
        from .resources.apps import AppsResourceWithRawResponse

        return AppsResourceWithRawResponse(self._client.apps)

    @cached_property
    def invocations(self) -> invocations.InvocationsResourceWithRawResponse:
        """Invoke actions and stream or query invocation status and events."""
        from .resources.invocations import InvocationsResourceWithRawResponse

        return InvocationsResourceWithRawResponse(self._client.invocations)

    @cached_property
    def browsers(self) -> browsers.BrowsersResourceWithRawResponse:
        """Create and manage browser sessions."""
        from .resources.browsers import BrowsersResourceWithRawResponse

        return BrowsersResourceWithRawResponse(self._client.browsers)

    @cached_property
    def profiles(self) -> profiles.ProfilesResourceWithRawResponse:
        """Create, list, retrieve, and delete browser profiles."""
        from .resources.profiles import ProfilesResourceWithRawResponse

        return ProfilesResourceWithRawResponse(self._client.profiles)

    @cached_property
    def auth(self) -> auth.AuthResourceWithRawResponse:
        from .resources.auth import AuthResourceWithRawResponse

        return AuthResourceWithRawResponse(self._client.auth)

    @cached_property
    def proxies(self) -> proxies.ProxiesResourceWithRawResponse:
        """Create and manage proxy configurations for routing browser traffic."""
        from .resources.proxies import ProxiesResourceWithRawResponse

        return ProxiesResourceWithRawResponse(self._client.proxies)

    @cached_property
    def extensions(self) -> extensions.ExtensionsResourceWithRawResponse:
        """Create, list, retrieve, and delete browser extensions."""
        from .resources.extensions import ExtensionsResourceWithRawResponse

        return ExtensionsResourceWithRawResponse(self._client.extensions)

    @cached_property
    def browser_pools(self) -> browser_pools.BrowserPoolsResourceWithRawResponse:
        """Create and manage browser pools for acquiring and releasing browsers."""
        from .resources.browser_pools import BrowserPoolsResourceWithRawResponse

        return BrowserPoolsResourceWithRawResponse(self._client.browser_pools)

    @cached_property
    def credentials(self) -> credentials.CredentialsResourceWithRawResponse:
        """Create and manage credentials for authentication."""
        from .resources.credentials import CredentialsResourceWithRawResponse

        return CredentialsResourceWithRawResponse(self._client.credentials)

    @cached_property
    def projects(self) -> projects.ProjectsResourceWithRawResponse:
        """
        Create and manage projects for resource isolation within an organization.
        When projects are disabled for the organization, project operations return
        `404` with code `projects_disabled`.
        """
        from .resources.projects import ProjectsResourceWithRawResponse

        return ProjectsResourceWithRawResponse(self._client.projects)

    @cached_property
    def organization(self) -> organization.OrganizationResourceWithRawResponse:
        from .resources.organization import OrganizationResourceWithRawResponse

        return OrganizationResourceWithRawResponse(self._client.organization)

    @cached_property
    def audit_logs(self) -> audit_logs.AuditLogsResourceWithRawResponse:
        """Read audit log records for the authenticated organization."""
        from .resources.audit_logs import AuditLogsResourceWithRawResponse

        return AuditLogsResourceWithRawResponse(self._client.audit_logs)

    @cached_property
    def api_keys(self) -> api_keys.APIKeysResourceWithRawResponse:
        """Create and manage API keys for organization and project-scoped access."""
        from .resources.api_keys import APIKeysResourceWithRawResponse

        return APIKeysResourceWithRawResponse(self._client.api_keys)

    @cached_property
    def credential_providers(self) -> credential_providers.CredentialProvidersResourceWithRawResponse:
        """Configure external credential providers like 1Password."""
        from .resources.credential_providers import CredentialProvidersResourceWithRawResponse

        return CredentialProvidersResourceWithRawResponse(self._client.credential_providers)


class AsyncKernelWithRawResponse:
    _client: AsyncKernel

    def __init__(self, client: AsyncKernel) -> None:
        self._client = client

    @cached_property
    def deployments(self) -> deployments.AsyncDeploymentsResourceWithRawResponse:
        """Create and manage app deployments and stream deployment events."""
        from .resources.deployments import AsyncDeploymentsResourceWithRawResponse

        return AsyncDeploymentsResourceWithRawResponse(self._client.deployments)

    @cached_property
    def apps(self) -> apps.AsyncAppsResourceWithRawResponse:
        """List applications and versions."""
        from .resources.apps import AsyncAppsResourceWithRawResponse

        return AsyncAppsResourceWithRawResponse(self._client.apps)

    @cached_property
    def invocations(self) -> invocations.AsyncInvocationsResourceWithRawResponse:
        """Invoke actions and stream or query invocation status and events."""
        from .resources.invocations import AsyncInvocationsResourceWithRawResponse

        return AsyncInvocationsResourceWithRawResponse(self._client.invocations)

    @cached_property
    def browsers(self) -> browsers.AsyncBrowsersResourceWithRawResponse:
        """Create and manage browser sessions."""
        from .resources.browsers import AsyncBrowsersResourceWithRawResponse

        return AsyncBrowsersResourceWithRawResponse(self._client.browsers)

    @cached_property
    def profiles(self) -> profiles.AsyncProfilesResourceWithRawResponse:
        """Create, list, retrieve, and delete browser profiles."""
        from .resources.profiles import AsyncProfilesResourceWithRawResponse

        return AsyncProfilesResourceWithRawResponse(self._client.profiles)

    @cached_property
    def auth(self) -> auth.AsyncAuthResourceWithRawResponse:
        from .resources.auth import AsyncAuthResourceWithRawResponse

        return AsyncAuthResourceWithRawResponse(self._client.auth)

    @cached_property
    def proxies(self) -> proxies.AsyncProxiesResourceWithRawResponse:
        """Create and manage proxy configurations for routing browser traffic."""
        from .resources.proxies import AsyncProxiesResourceWithRawResponse

        return AsyncProxiesResourceWithRawResponse(self._client.proxies)

    @cached_property
    def extensions(self) -> extensions.AsyncExtensionsResourceWithRawResponse:
        """Create, list, retrieve, and delete browser extensions."""
        from .resources.extensions import AsyncExtensionsResourceWithRawResponse

        return AsyncExtensionsResourceWithRawResponse(self._client.extensions)

    @cached_property
    def browser_pools(self) -> browser_pools.AsyncBrowserPoolsResourceWithRawResponse:
        """Create and manage browser pools for acquiring and releasing browsers."""
        from .resources.browser_pools import AsyncBrowserPoolsResourceWithRawResponse

        return AsyncBrowserPoolsResourceWithRawResponse(self._client.browser_pools)

    @cached_property
    def credentials(self) -> credentials.AsyncCredentialsResourceWithRawResponse:
        """Create and manage credentials for authentication."""
        from .resources.credentials import AsyncCredentialsResourceWithRawResponse

        return AsyncCredentialsResourceWithRawResponse(self._client.credentials)

    @cached_property
    def projects(self) -> projects.AsyncProjectsResourceWithRawResponse:
        """
        Create and manage projects for resource isolation within an organization.
        When projects are disabled for the organization, project operations return
        `404` with code `projects_disabled`.
        """
        from .resources.projects import AsyncProjectsResourceWithRawResponse

        return AsyncProjectsResourceWithRawResponse(self._client.projects)

    @cached_property
    def organization(self) -> organization.AsyncOrganizationResourceWithRawResponse:
        from .resources.organization import AsyncOrganizationResourceWithRawResponse

        return AsyncOrganizationResourceWithRawResponse(self._client.organization)

    @cached_property
    def audit_logs(self) -> audit_logs.AsyncAuditLogsResourceWithRawResponse:
        """Read audit log records for the authenticated organization."""
        from .resources.audit_logs import AsyncAuditLogsResourceWithRawResponse

        return AsyncAuditLogsResourceWithRawResponse(self._client.audit_logs)

    @cached_property
    def api_keys(self) -> api_keys.AsyncAPIKeysResourceWithRawResponse:
        """Create and manage API keys for organization and project-scoped access."""
        from .resources.api_keys import AsyncAPIKeysResourceWithRawResponse

        return AsyncAPIKeysResourceWithRawResponse(self._client.api_keys)

    @cached_property
    def credential_providers(self) -> credential_providers.AsyncCredentialProvidersResourceWithRawResponse:
        """Configure external credential providers like 1Password."""
        from .resources.credential_providers import AsyncCredentialProvidersResourceWithRawResponse

        return AsyncCredentialProvidersResourceWithRawResponse(self._client.credential_providers)


class KernelWithStreamedResponse:
    _client: Kernel

    def __init__(self, client: Kernel) -> None:
        self._client = client

    @cached_property
    def deployments(self) -> deployments.DeploymentsResourceWithStreamingResponse:
        """Create and manage app deployments and stream deployment events."""
        from .resources.deployments import DeploymentsResourceWithStreamingResponse

        return DeploymentsResourceWithStreamingResponse(self._client.deployments)

    @cached_property
    def apps(self) -> apps.AppsResourceWithStreamingResponse:
        """List applications and versions."""
        from .resources.apps import AppsResourceWithStreamingResponse

        return AppsResourceWithStreamingResponse(self._client.apps)

    @cached_property
    def invocations(self) -> invocations.InvocationsResourceWithStreamingResponse:
        """Invoke actions and stream or query invocation status and events."""
        from .resources.invocations import InvocationsResourceWithStreamingResponse

        return InvocationsResourceWithStreamingResponse(self._client.invocations)

    @cached_property
    def browsers(self) -> browsers.BrowsersResourceWithStreamingResponse:
        """Create and manage browser sessions."""
        from .resources.browsers import BrowsersResourceWithStreamingResponse

        return BrowsersResourceWithStreamingResponse(self._client.browsers)

    @cached_property
    def profiles(self) -> profiles.ProfilesResourceWithStreamingResponse:
        """Create, list, retrieve, and delete browser profiles."""
        from .resources.profiles import ProfilesResourceWithStreamingResponse

        return ProfilesResourceWithStreamingResponse(self._client.profiles)

    @cached_property
    def auth(self) -> auth.AuthResourceWithStreamingResponse:
        from .resources.auth import AuthResourceWithStreamingResponse

        return AuthResourceWithStreamingResponse(self._client.auth)

    @cached_property
    def proxies(self) -> proxies.ProxiesResourceWithStreamingResponse:
        """Create and manage proxy configurations for routing browser traffic."""
        from .resources.proxies import ProxiesResourceWithStreamingResponse

        return ProxiesResourceWithStreamingResponse(self._client.proxies)

    @cached_property
    def extensions(self) -> extensions.ExtensionsResourceWithStreamingResponse:
        """Create, list, retrieve, and delete browser extensions."""
        from .resources.extensions import ExtensionsResourceWithStreamingResponse

        return ExtensionsResourceWithStreamingResponse(self._client.extensions)

    @cached_property
    def browser_pools(self) -> browser_pools.BrowserPoolsResourceWithStreamingResponse:
        """Create and manage browser pools for acquiring and releasing browsers."""
        from .resources.browser_pools import BrowserPoolsResourceWithStreamingResponse

        return BrowserPoolsResourceWithStreamingResponse(self._client.browser_pools)

    @cached_property
    def credentials(self) -> credentials.CredentialsResourceWithStreamingResponse:
        """Create and manage credentials for authentication."""
        from .resources.credentials import CredentialsResourceWithStreamingResponse

        return CredentialsResourceWithStreamingResponse(self._client.credentials)

    @cached_property
    def projects(self) -> projects.ProjectsResourceWithStreamingResponse:
        """
        Create and manage projects for resource isolation within an organization.
        When projects are disabled for the organization, project operations return
        `404` with code `projects_disabled`.
        """
        from .resources.projects import ProjectsResourceWithStreamingResponse

        return ProjectsResourceWithStreamingResponse(self._client.projects)

    @cached_property
    def organization(self) -> organization.OrganizationResourceWithStreamingResponse:
        from .resources.organization import OrganizationResourceWithStreamingResponse

        return OrganizationResourceWithStreamingResponse(self._client.organization)

    @cached_property
    def audit_logs(self) -> audit_logs.AuditLogsResourceWithStreamingResponse:
        """Read audit log records for the authenticated organization."""
        from .resources.audit_logs import AuditLogsResourceWithStreamingResponse

        return AuditLogsResourceWithStreamingResponse(self._client.audit_logs)

    @cached_property
    def api_keys(self) -> api_keys.APIKeysResourceWithStreamingResponse:
        """Create and manage API keys for organization and project-scoped access."""
        from .resources.api_keys import APIKeysResourceWithStreamingResponse

        return APIKeysResourceWithStreamingResponse(self._client.api_keys)

    @cached_property
    def credential_providers(self) -> credential_providers.CredentialProvidersResourceWithStreamingResponse:
        """Configure external credential providers like 1Password."""
        from .resources.credential_providers import CredentialProvidersResourceWithStreamingResponse

        return CredentialProvidersResourceWithStreamingResponse(self._client.credential_providers)


class AsyncKernelWithStreamedResponse:
    _client: AsyncKernel

    def __init__(self, client: AsyncKernel) -> None:
        self._client = client

    @cached_property
    def deployments(self) -> deployments.AsyncDeploymentsResourceWithStreamingResponse:
        """Create and manage app deployments and stream deployment events."""
        from .resources.deployments import AsyncDeploymentsResourceWithStreamingResponse

        return AsyncDeploymentsResourceWithStreamingResponse(self._client.deployments)

    @cached_property
    def apps(self) -> apps.AsyncAppsResourceWithStreamingResponse:
        """List applications and versions."""
        from .resources.apps import AsyncAppsResourceWithStreamingResponse

        return AsyncAppsResourceWithStreamingResponse(self._client.apps)

    @cached_property
    def invocations(self) -> invocations.AsyncInvocationsResourceWithStreamingResponse:
        """Invoke actions and stream or query invocation status and events."""
        from .resources.invocations import AsyncInvocationsResourceWithStreamingResponse

        return AsyncInvocationsResourceWithStreamingResponse(self._client.invocations)

    @cached_property
    def browsers(self) -> browsers.AsyncBrowsersResourceWithStreamingResponse:
        """Create and manage browser sessions."""
        from .resources.browsers import AsyncBrowsersResourceWithStreamingResponse

        return AsyncBrowsersResourceWithStreamingResponse(self._client.browsers)

    @cached_property
    def profiles(self) -> profiles.AsyncProfilesResourceWithStreamingResponse:
        """Create, list, retrieve, and delete browser profiles."""
        from .resources.profiles import AsyncProfilesResourceWithStreamingResponse

        return AsyncProfilesResourceWithStreamingResponse(self._client.profiles)

    @cached_property
    def auth(self) -> auth.AsyncAuthResourceWithStreamingResponse:
        from .resources.auth import AsyncAuthResourceWithStreamingResponse

        return AsyncAuthResourceWithStreamingResponse(self._client.auth)

    @cached_property
    def proxies(self) -> proxies.AsyncProxiesResourceWithStreamingResponse:
        """Create and manage proxy configurations for routing browser traffic."""
        from .resources.proxies import AsyncProxiesResourceWithStreamingResponse

        return AsyncProxiesResourceWithStreamingResponse(self._client.proxies)

    @cached_property
    def extensions(self) -> extensions.AsyncExtensionsResourceWithStreamingResponse:
        """Create, list, retrieve, and delete browser extensions."""
        from .resources.extensions import AsyncExtensionsResourceWithStreamingResponse

        return AsyncExtensionsResourceWithStreamingResponse(self._client.extensions)

    @cached_property
    def browser_pools(self) -> browser_pools.AsyncBrowserPoolsResourceWithStreamingResponse:
        """Create and manage browser pools for acquiring and releasing browsers."""
        from .resources.browser_pools import AsyncBrowserPoolsResourceWithStreamingResponse

        return AsyncBrowserPoolsResourceWithStreamingResponse(self._client.browser_pools)

    @cached_property
    def credentials(self) -> credentials.AsyncCredentialsResourceWithStreamingResponse:
        """Create and manage credentials for authentication."""
        from .resources.credentials import AsyncCredentialsResourceWithStreamingResponse

        return AsyncCredentialsResourceWithStreamingResponse(self._client.credentials)

    @cached_property
    def projects(self) -> projects.AsyncProjectsResourceWithStreamingResponse:
        """
        Create and manage projects for resource isolation within an organization.
        When projects are disabled for the organization, project operations return
        `404` with code `projects_disabled`.
        """
        from .resources.projects import AsyncProjectsResourceWithStreamingResponse

        return AsyncProjectsResourceWithStreamingResponse(self._client.projects)

    @cached_property
    def organization(self) -> organization.AsyncOrganizationResourceWithStreamingResponse:
        from .resources.organization import AsyncOrganizationResourceWithStreamingResponse

        return AsyncOrganizationResourceWithStreamingResponse(self._client.organization)

    @cached_property
    def audit_logs(self) -> audit_logs.AsyncAuditLogsResourceWithStreamingResponse:
        """Read audit log records for the authenticated organization."""
        from .resources.audit_logs import AsyncAuditLogsResourceWithStreamingResponse

        return AsyncAuditLogsResourceWithStreamingResponse(self._client.audit_logs)

    @cached_property
    def api_keys(self) -> api_keys.AsyncAPIKeysResourceWithStreamingResponse:
        """Create and manage API keys for organization and project-scoped access."""
        from .resources.api_keys import AsyncAPIKeysResourceWithStreamingResponse

        return AsyncAPIKeysResourceWithStreamingResponse(self._client.api_keys)

    @cached_property
    def credential_providers(self) -> credential_providers.AsyncCredentialProvidersResourceWithStreamingResponse:
        """Configure external credential providers like 1Password."""
        from .resources.credential_providers import AsyncCredentialProvidersResourceWithStreamingResponse

        return AsyncCredentialProvidersResourceWithStreamingResponse(self._client.credential_providers)


Client = Kernel

AsyncClient = AsyncKernel