# 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,

prepare_direct_vm_request,

rewrite_direct_vm_options,

browser_routing_config_from_env,

install_direct_vm_auth_stripping,

is_stale_direct_vm_auth_response,

should_retry_stale_direct_vm_auth,

install_stale_direct_vm_auth_eviction,

install_async_direct_vm_auth_stripping,

maybe_evict_browser_route_from_response,

should_retry_direct_vm_connection_error,

install_async_stale_direct_vm_auth_eviction,

direct_vm_request_body_is_known_unreplayable,

maybe_populate_browser_route_cache_from_response,

)

if TYPE_CHECKING:

from .resources import (

apps,

auth,

search,

vaults,

proxies,

api_keys,

browsers,

profiles,

projects,

telemetry,

audit_logs,

extensions,

credentials,

deployments,

invocations,

organization,

browser_pools,

config_registry,

credential_providers,

vault_provider_configs,

)

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.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.search.search import SearchResource, AsyncSearchResource

from .resources.vaults.vaults import VaultsResource, AsyncVaultsResource

from .resources.browsers.browsers import BrowsersResource, AsyncBrowsersResource

from .resources.projects.projects import ProjectsResource, AsyncProjectsResource

from .resources.telemetry.telemetry import TelemetryResource, AsyncTelemetryResource

from .resources.credential_providers import CredentialProvidersResource, AsyncCredentialProvidersResource

from .resources.audit_logs.audit_logs import AuditLogsResource, AsyncAuditLogsResource

from .resources.vault_provider_configs import VaultProviderConfigsResource, AsyncVaultProviderConfigsResource

from .resources.organization.organization import OrganizationResource, AsyncOrganizationResource

from .resources.config_registry.config_registry import ConfigRegistryResource, AsyncConfigRegistryResource

__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

project: str | None

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

_browser_routing: BrowserRoutingConfig

def __init__(

self,

*,

api_key: str | None = None,

project_id: str | None = None,

project: 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.project = project

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()

install_direct_vm_auth_stripping(self._client)

install_stale_direct_vm_auth_eviction(self._client)

@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 config_registry(self) -> ConfigRegistryResource:

"""Resolve browser and proxy recommendations for bot-protected sites."""

from .resources.config_registry import ConfigRegistryResource

return ConfigRegistryResource(self)

@cached_property

def browsers(self) -> BrowsersResource:

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 telemetry(self) -> TelemetryResource:

from .resources.telemetry import TelemetryResource

return TelemetryResource(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 vault_provider_configs(self) -> VaultProviderConfigsResource:

from .resources.vault_provider_configs import VaultProviderConfigsResource

return VaultProviderConfigsResource(self)

@cached_property

def vaults(self) -> VaultsResource:

from .resources.vaults import VaultsResource

return VaultsResource(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 search(self) -> SearchResource:

"""Search the web and retrieve content for selected results."""

from .resources.search import SearchResource

return SearchResource(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(),

"X-Kernel-Project": self.project if self.project 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:

prepare_direct_vm_request(request, cache=self.browser_route_cache)

@override

def _should_retry_on_connection_error(self, request: httpx.Request) -> bool:

return should_retry_direct_vm_connection_error(request)

@override

def _should_retry(self, response: httpx.Response) -> bool:

if direct_vm_request_body_is_known_unreplayable(response.request):

return False

if is_stale_direct_vm_auth_response(response):

# The route was already evicted by the response hook; retry only when

# the body can be rebuilt, otherwise the caller sees the original auth

# failure and a later call goes to the control plane.

return should_retry_stale_direct_vm_auth(response)

return super()._should_retry(response)

@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,

project: 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,

project=project or self.project,

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

project: str | None

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

_browser_routing: BrowserRoutingConfig

def __init__(

self,

*,

api_key: str | None = None,

project_id: str | None = None,

project: 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.project = project

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()

install_async_direct_vm_auth_stripping(self._client)

install_async_stale_direct_vm_auth_eviction(self._client)

@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 config_registry(self) -> AsyncConfigRegistryResource:

"""Resolve browser and proxy recommendations for bot-protected sites."""

from .resources.config_registry import AsyncConfigRegistryResource

return AsyncConfigRegistryResource(self)

@cached_property

def browsers(self) -> AsyncBrowsersResource:

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 telemetry(self) -> AsyncTelemetryResource:

from .resources.telemetry import AsyncTelemetryResource

return AsyncTelemetryResource(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 vault_provider_configs(self) -> AsyncVaultProviderConfigsResource:

from .resources.vault_provider_configs import AsyncVaultProviderConfigsResource

return AsyncVaultProviderConfigsResource(self)

@cached_property

def vaults(self) -> AsyncVaultsResource:

from .resources.vaults import AsyncVaultsResource

return AsyncVaultsResource(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 search(self) -> AsyncSearchResource:

"""Search the web and retrieve content for selected results."""

from .resources.search import AsyncSearchResource

return AsyncSearchResource(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(),

"X-Kernel-Project": self.project if self.project 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:

prepare_direct_vm_request(request, cache=self.browser_route_cache)

@override

def _should_retry_on_connection_error(self, request: httpx.Request) -> bool:

return should_retry_direct_vm_connection_error(request)

@override

def _should_retry(self, response: httpx.Response) -> bool:

if direct_vm_request_body_is_known_unreplayable(response.request):

return False

if is_stale_direct_vm_auth_response(response):

# The route was already evicted by the response hook; retry only when

# the body can be rebuilt, otherwise the caller sees the original auth

# failure and a later call goes to the control plane.

return should_retry_stale_direct_vm_auth(response)

return super()._should_retry(response)

@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,

project: 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,

project=project or self.project,

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 config_registry(self) -> config_registry.ConfigRegistryResourceWithRawResponse:

"""Resolve browser and proxy recommendations for bot-protected sites."""

from .resources.config_registry import ConfigRegistryResourceWithRawResponse

return ConfigRegistryResourceWithRawResponse(self._client.config_registry)

@cached_property

def browsers(self) -> browsers.BrowsersResourceWithRawResponse:

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 telemetry(self) -> telemetry.TelemetryResourceWithRawResponse:

from .resources.telemetry import TelemetryResourceWithRawResponse

return TelemetryResourceWithRawResponse(self._client.telemetry)

@cached_property

def proxies(self) -> proxies.ProxiesResourceWithRawResponse: