Connecting and Authenticating

Choose a connection entry point

To connect flyte-sdk to a Flyte backend, provide either an endpoint or an API key. The SDK’s asynchronous initialization path forwards the authentication and transport settings to ClientSet:

async def _initialize_client(
    api_key: str | None = None,
    auth_type: AuthType = "Pkce",
    endpoint: str | None = None,
    client_config: ClientConfig | None = None,
    headless: bool = False,
    insecure: bool = False,
    insecure_skip_verify: bool = False,
    ca_cert_file_path: str | None = None,
    command: List[str] | None = None,
    proxy_command: List[str] | None = None,
    client_id: str | None = None,
    client_credentials_secret: str | None = None,
    rpc_retries: int = 3,
    http_proxy_url: str | None = None,
) -> ClientSet:
    from flyte.remote._client.controlplane import ClientSet

    if endpoint:
        return await ClientSet.for_endpoint(
            endpoint,
            insecure=insecure,
            insecure_skip_verify=insecure_skip_verify,
            auth_type=auth_type,
            headless=headless,
            ca_cert_file_path=ca_cert_file_path,
            command=command,
            proxy_command=proxy_command,
            client_id=client_id,
            client_credentials_secret=client_credentials_secret,
            client_config=client_config,
            rpc_retries=rpc_retries,
            http_proxy_url=http_proxy_url,
        )
    elif api_key:
        return await ClientSet.for_api_key(
            api_key,
            insecure=insecure,
            insecure_skip_verify=insecure_skip_verify,
            auth_type=auth_type,
            headless=headless,
            ca_cert_file_path=ca_cert_file_path,
            command=command,
            proxy_command=proxy_command,
            client_id=client_id,
            client_credentials_secret=client_credentials_secret,
            client_config=client_config,
            rpc_retries=rpc_retries,
            http_proxy_url=http_proxy_url,
        )

    raise InitializationError(
        "MissingEndpointOrApiKeyError",
        "user",
        "Either endpoint or api_key must be provided to initialize the client.",
    )

ClientSet.for_endpoint() calls create_channel(endpoint, None, ...). ClientSet.for_api_key() decodes the API key to recover the endpoint, then calls create_channel(None, api_key, ...). The API-key path therefore selects client-secret authentication inside channel creation, even if another auth_type is passed.

Once created, ClientSet exposes the asynchronous gRPC stubs through these properties:

• metadata_service and project_domain_service use the Admin service.
• task_service uses the Task service.
• run_service uses the Run service.
• dataproxy_service uses the DataProxy service.
• logs_service uses the Run Logs service.
• secrets_service uses the Secret service.

Close the underlying channel when the client is no longer needed:

await client.close()

ClientSet.for_serverless() and ClientSet.from_env() are declared connection helpers, but currently raise NotImplementedError.

Understand channel and configuration discovery

create_channel() first creates an unauthenticated gRPC channel. For a secure connection it creates a grpc.aio.secure_channel() using either default SSL credentials, a certificate loaded from ca_cert_file_path, or credentials bootstrapped with insecure_skip_verify. For insecure=True, it creates a plaintext grpc.aio.insecure_channel() instead.

The unauthenticated channel is used to construct RemoteClientConfigStore. Its get_client_config() method calls the Flyte AuthMetadataService concurrently:

async def get_client_config(self) -> ClientConfig:
    metadata_service = AuthMetadataServiceStub(self._unauthenticated_channel)
    oauth2_metadata_task = metadata_service.GetOAuth2Metadata(OAuth2MetadataRequest())
    public_client_config_task = metadata_service.GetPublicClientConfig(PublicClientAuthConfigRequest())
    oauth2_metadata, public_client_config = await asyncio.gather(
        oauth2_metadata_task, public_client_config_task
    )
    return ClientConfig(
        token_endpoint=oauth2_metadata.token_endpoint,
        authorization_endpoint=oauth2_metadata.authorization_endpoint,
        redirect_uri=public_client_config.redirect_uri,
        client_id=public_client_config.client_id,
        scopes=public_client_config.scopes,
        header_key=public_client_config.authorization_metadata_key,
        device_authorization_endpoint=oauth2_metadata.device_authorization_endpoint,
        audience=public_client_config.audience,
    )

This lets the authentication flow obtain OAuth endpoints and public-client settings before authenticated RPCs are available. create_channel() then installs default-metadata, proxy-authentication, and authentication interceptors for unary-unary, unary-stream, stream-unary, and stream-stream calls. Authentication is consequently lazy: the selected authenticator obtains credentials when an interceptor needs request metadata rather than during the initial channel construction.

Supply static configuration when needed

Use StaticClientConfigStore when the OAuth metadata is known locally. ClientConfig requires the token endpoint, authorization endpoint, redirect URI, and client ID; the device endpoint, scopes, header key, and audience are optional as shown by the model:

from flyte.remote.auth import ClientConfig
from flyte.remote._client.auth._client_config import StaticClientConfigStore

config = ClientConfig(
    token_endpoint="https://auth.example/token",
    authorization_endpoint="https://auth.example/authorize",
    redirect_uri="http://127.0.0.1:8080/callback",
    client_id="flyte-client",
    scopes=["openid", "offline_access"],
    header_key="authorization",
    audience="flyte",
)
config_store = StaticClientConfigStore(config)

The remote authenticator normally calls Authenticator._resolve_config(), which fetches the remote configuration once and then applies a supplied local client_config with ClientConfig.with_override(). Overrides use truthiness (other.value or self.value), so an empty string or empty list does not replace an existing remote value.

Select an authentication method

The public AuthType values accepted by the factory are:

auth_type	Factory class	Use it when
Pkce	PKCEAuthenticator	A user can complete browser-based login. This is the default.
DeviceFlow	DeviceCodeAuthenticator	The user login must work headlessly.
ClientSecret	ClientCredentialsAuthenticator	A service has a client ID and client secret.
ExternalCommand	AsyncCommandAuthenticator	Another process supplies the token on stdout.

The factory dispatch is explicit:

def get_async_authenticator(
    endpoint: str,
    cfg_store: ClientConfigStore,
    *,
    command: typing.Optional[typing.List[str]] = None,
    insecure_skip_verify: bool = False,
    auth_type: AuthType = "Pkce",
    ca_cert_file_path: typing.Optional[str] = None,
    **kwargs,
) -> Authenticator:
    verify = None
    if insecure_skip_verify:
        verify = False
    elif ca_cert_file_path:
        verify = True if ca_cert_file_path is not None else False

    match auth_type:
        case "Pkce":
            return PKCEAuthenticator(endpoint=endpoint, cfg_store=cfg_store, verify=verify, **kwargs)
        case "ClientSecret":
            return ClientCredentialsAuthenticator(endpoint=endpoint, cfg_store=cfg_store, verify=verify, **kwargs)
        case "ExternalCommand":
            return AsyncCommandAuthenticator(endpoint=endpoint, command=command, verify=verify, **kwargs)
        case "DeviceFlow":
            return DeviceCodeAuthenticator(endpoint=endpoint, cfg_store=cfg_store, verify=verify, **kwargs)
        case _:
            raise ValueError(
                f"Invalid auth mode [{auth_type}] specified. Please update the creds config to use a valid value"
            )

For CLI configuration, sanitize_auth_type() maps pkce to Pkce, headless, device-flow, and device_flow to DeviceFlow, client-secret, client_secret, clientsecret, app-credential, and app_credential to ClientSecret, and external-command, external_command, externalcommand, command, and custom to ExternalCommand. An unknown alias raises ValueError.

Reuse the shared credential lifecycle

All four authenticators subclass Authenticator. On construction, it uses explicitly supplied credentials, or calls KeyringStore.retrieve(endpoint) when no credentials were supplied. get_grpc_call_auth_metadata() converts the current token into a bearer metadata pair and includes the credential ID used to coordinate refreshes:

async def get_grpc_call_auth_metadata(self) -> typing.Optional[GrpcAuthMetadata]:
    creds = self.get_credentials()
    if creds:
        header_key = self._default_header_key
        if self._resolved_config is not None:
            header_key = self._resolved_config.header_key
        return GrpcAuthMetadata(
            creds_id=creds.id,
            pairs=Metadata((header_key, f"Bearer {creds.access_token}")),
        )
    return None

refresh_credentials() runs the subclass’s _do_refresh_credentials() under an asyncio.Lock, stores successful credentials in KeyringStore, and updates the credential ID. If refresh raises an exception, it deletes the endpoint’s stored access and refresh tokens and re-raises the exception. The gRPC authentication interceptors and AsyncAuthenticatedClient use the credential ID to avoid repeating a refresh after another coroutine has already refreshed the token. An HTTP 401 causes one refresh and retry; gRPC authentication failures likewise trigger a refresh and retry through the interceptors.

Credentials normalizes for_endpoint by stripping its URL scheme and computes id as the MD5 hash of access_token:

class Credentials(pydantic.BaseModel):
    access_token: str
    for_endpoint: str = "flyte-default"
    id: str = ""
    refresh_token: str | None = None
    expires_in: int | None = None

    @pydantic.field_validator("for_endpoint", mode="after")
    @classmethod
    def validate_endpoint(cls, v: str) -> str:
        return strip_scheme(v)

    @pydantic.model_validator(mode="after")
    def compute_id(self) -> "Credentials":
        if self.access_token:
            self.id = hashlib.md5(self.access_token.encode()).hexdigest()
        return self

The ID is a refresh-coordination value, not an access credential. KeyringStore uses the normalized endpoint as the keyring service and the keys access_token and refresh_token. Keyring failures are logged and treated as cache misses. Retrieval does not restore expires_in, so a retrieved Credentials object has expires_in=None.

Use browser-based PKCE

Choose Pkce when the user can authenticate in a browser. PKCEAuthenticator lazily resolves ClientConfig, generates a verifier and S256 challenge, and creates an AuthorizationClient with the configured redirect URI, client ID, scopes, authorization endpoint, and token endpoint:

async def _initialize_auth_client(self):
    if not self._auth_client:
        code_verifier = await _generate_code_verifier()
        code_challenge = await _create_code_challenge(code_verifier)

        cfg = await self._resolve_config()
        self._auth_client = AuthorizationClient(
            endpoint=self._endpoint,
            redirect_uri=cfg.redirect_uri,
            client_id=cfg.client_id,
            audience=cfg.audience,
            scopes=cfg.scopes,
            auth_endpoint=cfg.authorization_endpoint,
            token_endpoint=cfg.token_endpoint,
            verify=self._verify,
            http_session=self._http_session,
            request_auth_code_params={
                "code_challenge": code_challenge,
                "code_challenge_method": "S256",
            },
            request_access_token_params={"code_verifier": code_verifier},
            refresh_access_token_params={},
            add_request_auth_code_params_to_request_access_token_params=True,
        )

The refresh sequence is refresh-first:

async def _do_refresh_credentials(self) -> Credentials:
    await self._initialize_auth_client()
    if self._creds:
        try:
            return await self._auth_client.refresh_access_token(self._creds)
        except AccessTokenNotFoundError:
            logger.warning("Logging in...")
    return await self._auth_client.get_creds_from_remote()

If no usable refresh token exists, AuthorizationClient.get_creds_from_remote() starts a localhost callback server, opens the authorization URL with webbrowser.open_new_tab(), and waits for the callback. The callback must contain code and state; _request_access_token() rejects a state that differs from the generated state before exchanging the code. The result is cached for 60 seconds, which can prevent duplicate browser flows during closely spaced calls.

The redirect URI must provide a hostname and port because the callback server binds to them. For Auth0 deployments, the PKCE authenticator’s documentation specifically calls out scopes such as offline_access, offline, all, and openid when refresh-token caching is required; scope support still depends on the configured identity provider.

Use headless device flow

Choose DeviceFlow when the user cannot complete a localhost browser callback. DeviceCodeAuthenticator requires the resolved configuration to publish device_authorization_endpoint:

cfg = await self._resolve_config()

if cfg.device_authorization_endpoint is None:
    raise AuthenticationError(
        "Device Authentication is not available on the Flyte backend / authentication server"
    )

With a cached refresh token, it first sends a refresh-token grant. If that raises AuthenticationError or AuthenticationPending, it falls back to device login. The fallback requests a device code, prints the verification URL with the user code, and polls the token endpoint:

resp = await token_client.get_device_code(
    cfg.device_authorization_endpoint,
    cfg.client_id,
    audience=cfg.audience,
    scopes=cfg.scopes,
    http_session=self._http_session,
)
full_uri = f"{resp.verification_uri}?user_code={resp.user_code}"
click.secho(
    f"To Authenticate, navigate in a browser to the following URL: "
    f"{click.style(full_uri, fg='blue', underline=True)}"
)

token, refresh_token, expires_in = await token_client.poll_token_endpoint(
    resp,
    token_endpoint=cfg.token_endpoint,
    client_id=cfg.client_id,
    audience=cfg.audience,
    scopes=cfg.scopes,
    http_proxy_url=self._http_proxy_url,
    verify=self._verify,
    http_session=self._http_session,
)

The device flow prints a URL rather than opening a browser itself. Polling continues for pending responses according to the token client’s interval handling; an expired device code or another non-pending authentication error ends the flow.

Use client credentials

Choose ClientSecret for service or SDK use. ClientCredentialsAuthenticator rejects missing values immediately:

from flyte.remote._client.auth._authenticators.client_credentials import ClientCredentialsAuthenticator

authenticator = ClientCredentialsAuthenticator(
    client_id="service-client",
    client_credentials_secret="service-secret",
    endpoint="dns:///flyte.example:443",
    cfg_store=config_store,
)

The authenticator resolves the configuration, creates a Basic authorization header from the supplied ID and secret, and calls the token client with the configured token endpoint, scopes, audience, HTTP session, proxy, and TLS settings:

authorization_header = token_client.get_basic_authorization_header(
    self._client_id, self._client_credentials_secret
)
token, refresh_token, expires_in = await token_client.get_token(
    token_endpoint=cfg.token_endpoint,
    authorization_header=authorization_header,
    http_proxy_url=self._http_proxy_url,
    verify=self._verify,
    scopes=cfg.scopes,
    audience=cfg.audience,
    http_session=self._http_session,
)

The returned access and optional refresh tokens are placed in Credentials and then persisted by the shared Authenticator.refresh_credentials() lifecycle.

Use an external command

Choose ExternalCommand when an executable already knows how to obtain the token. Pass an argument list, not a shell command string:

from flyte.remote._client.auth._authenticators.external_command import AsyncCommandAuthenticator

authenticator = AsyncCommandAuthenticator(
    command=["my-token-command", "--audience", "flyte"],
    endpoint="dns:///flyte.example:443",
)

An empty list raises AuthenticationError. The authenticator invokes asyncio.create_subprocess_exec() with captured stdout and stderr. A zero exit code produces Credentials whose access token is stdout.decode().strip(); a nonzero exit code or execution exception becomes AuthenticationError. The command is logged for debugging, so do not put secrets in its arguments. The same AsyncCommandAuthenticator is used for proxy authentication, with the proxy path emitting proxy-authorization metadata rather than the default authorization header.

Map configuration to authentication

The initialization path forwards these PlatformConfig/admin settings into channel and authenticator setup:

Configuration	Effect
admin.endpoint / PlatformConfig.endpoint	Endpoint used by ClientSet.for_endpoint(); initialization requires this or an API key.
admin.authType / PlatformConfig.auth_mode	Selects Pkce, DeviceFlow, ClientSecret, or ExternalCommand.
admin.clientId / PlatformConfig.client_id	Client identifier used by PKCE, device flow, and client credentials.
admin.clientSecretLocation	Reads a mounted client-secret file; configuration loading selects ClientSecret when a secret is found.
admin.clientSecretEnvVar	Reads the client-credentials secret from the named environment variable and can select ClientSecret.
admin.scopes / PlatformConfig.scopes	Scopes sent to PKCE, device, and token requests.
admin.command / PlatformConfig.command	Argument list for ExternalCommand; stdout is the token.
admin.proxyCommand / PlatformConfig.proxy_command	Separate command for proxy authentication.
admin.caCertFilePath / PlatformConfig.ca_cert_file_path	CA certificate used for gRPC and HTTP TLS configuration.
admin.insecure	Uses a plaintext gRPC channel.
admin.insecureSkipVerify	Disables HTTP certificate verification and uses server bootstrap behavior for gRPC when applicable.
admin.httpProxyURL / PlatformConfig.http_proxy_url	Proxy used by OAuth and token HTTP requests.

Use insecure_skip_verify only where accepting unverified certificates is appropriate, such as local development or testing; the AuthorizationClient documentation notes that verify=False accepts certificates without normal TLS verification.

Troubleshoot common failures

• No endpoint or API key: _initialize_client() raises InitializationError with MissingEndpointOrApiKeyError.
• No OAuth configuration store: Authenticator._resolve_config() raises ValueError("ClientConfigStore is not set. Cannot resolve configuration."). OAuth authenticators need either a RemoteClientConfigStore, a StaticClientConfigStore, or an already resolved configuration path.
• Device flow unavailable: The backend returned no device_authorization_endpoint; DeviceCodeAuthenticator raises AuthenticationError before requesting a device code.
• PKCE callback does not complete: Check that the configured redirect URI has a usable hostname and port and that the callback includes both code and state.
• Client-secret login fails immediately: Both client_id and client_credentials_secret are required by ClientCredentialsAuthenticator.
• External command fails: Run the exact argument list manually and verify that it exits zero and writes only the token (plus optional whitespace) to stdout.
• Credentials appear not to persist: Keyring support is optional. KeyringStore logs backend failures and returns a cache miss; it also does not restore expires_in.
• A failed refresh removes a token: Any exception from _do_refresh_credentials() causes Authenticator.refresh_credentials() to call KeyringStore.delete() for the endpoint, removing both access- and refresh-token entries.
• Endpoint cache seems shared: Credentials.for_endpoint strips schemes such as https:// and dns:/// before keyring lookup. Host, port, and path differences remain significant, but different scheme spellings can map to the same keyring service.

The authenticator modules under remote._client.auth._authenticators are private implementation paths. Prefer the public flyte.remote/flyte.remote.auth exports and flyte-sdk’s initialization/configuration APIs for application code; use the private class imports above when directly selecting or integrating a concrete authenticator.