Connect to Flyte and Authenticate Remote Requests

Choose a connection path

To access Flyte tasks, runs, and other remote resources, initialize an authenticated client with an endpoint, an encoded API key, or request-scoped metadata. The low-level entry point is the asynchronous ClientSet factory; normal SDK initialization in flyte._initialize._initialize_client selects the same factory for you.

Connect to an endpoint

from flyte.remote._client.controlplane import ClientSet

client = await ClientSet.for_endpoint(
    "dns:///flyte.example.com",
    insecure=False,
    auth_type="Pkce",
    disable_keyring=False,
)

runs = client.run_service
console_url = client.console.task_url(
    project="myproj",
    domain="dev",
    task_name="mytask",
)

ClientSet.for_endpoint is an async classmethod. It calls create_session_config, then passes the resulting SessionConfig to ClientSet. Consequently, callers must await it (or use one of Flyte's higher-level initialization APIs that handles sync/async initialization).

The constructed ClientSet creates ConnectRPC clients for the project/domain, task, app, run, logs, secrets, identity, trigger, and cluster services. Its dataproxy_service is a cluster-aware data proxy that selects a cluster through ClusterService.SelectCluster and caches per-cluster clients. Useful accessors include:

• project_domain_service
• task_service, app_service, and run_service
• logs_service, secrets_service, and identity_service
• trigger_service, cluster_service, and dataproxy_service
• endpoint, exposing the normalized session endpoint
• session_config, for creating additional ConnectRPC clients that share the transport and authentication interceptors
• console, a Console URL builder

ClientSet.for_serverless() and ClientSet.from_env() currently raise NotImplementedError; use for_endpoint, for_api_key, or the higher-level initialization path instead.

Initialize through Flyte's normal SDK path

The production initialization function forwards authentication and transport controls to ClientSet:

from flyte._initialize import _initialize_client

client = await _initialize_client(
    endpoint="dns:///flyte.example.com",
    auth_type="Pkce",
    insecure=False,
    insecure_skip_verify=False,
    ca_cert_file_path=None,
    http_proxy_url=None,
    disable_keyring=False,
)

For public application code, use the corresponding flyte.init/init.aio initialization API rather than relying on the private _initialize_client helper. The helper shows the actual forwarding contract: auth_type, headless, command, proxy_command, client_id, client_credentials_secret, client_config, rpc_retries, http_proxy_url, and disable_keyring are passed to the endpoint or API-key factory.

Connect with an API key

from flyte.remote._client.controlplane import ClientSet

client = await ClientSet.for_api_key(
    encoded_api_key,
    insecure=False,
    disable_keyring=True,
)

for_api_key passes the key to create_session_config. That function decodes the key into an endpoint, client ID, client secret, and organization, then overwrites the authentication settings with auth_type="ClientSecret", the decoded client ID, and the decoded secret. An API-key connection therefore uses the client-credentials path; it is not a PKCE login.

The higher-level init_from_api_key path uses the FLYTE_API_KEY environment variable when no api_key argument is supplied. The encoded key contains the endpoint, client ID, client secret, and organization. create_session_config requires either endpoint or api_key and raises an assertion if neither is available.

How the session authenticates RPCs

create_session_config normalizes the endpoint before constructing the session. DNS-style endpoints such as dns:///flyte.example.com and bare host/port forms become HTTP(S) endpoints; unsupported schemes such as ftp:// are rejected by endpoint normalization. The session then builds:

1. A pyqwest client for ConnectRPC transport, with the resolved CA certificate.
2. A separate httpx.AsyncClient session for OAuth and other authentication HTTP requests.
3. A DefaultMetadataInterceptor.
4. Four lazy authentication interceptors (unary, client-stream, server-stream, and bidirectional stream) when the session is not insecure.
5. Optional RPC retry interceptors when rpc_retries is greater than zero.

insecure=True selects plaintext transport and causes create_session_config to skip authentication interceptors. The session source explicitly distinguishes this from a general “disable authentication” switch: insecure means no TLS, and the implementation assumes plaintext/local-development endpoints do not have an authentication server. For TLS endpoints, use ca_cert_file_path for a CA chain or insecure_skip_verify for the certificate bootstrap behavior. http_proxy_url applies to authentication HTTP requests; proxy_command additionally creates proxy-authentication support and proxy interceptors.

The authentication factory maps the auth_type value to an authenticator:

auth_type	Authenticator	Use in flyte-sdk
Pkce	PKCEAuthenticator	Browser-based authorization-code login; the default value in _initialize_client
DeviceFlow	DeviceCodeAuthenticator	Headless device authorization
ExternalCommand	AsyncCommandAuthenticator	Token supplied by an executable
Passthrough	PassthroughAuthenticator	Headers supplied by remote.auth_metadata
ClientSecret	ClientCredentialsAuthenticator	API-key/client-credentials authentication

The factory initializes authenticators lazily, when an RPC interceptor first needs one. Before a request, the interceptor calls get_auth_headers and adds the returned headers to the ConnectRPC request context. If the call fails with UNAUTHENTICATED, UNKNOWN, or an error message containing unauthorized or unauthenticated, it refreshes credentials and retries once. Before reinjecting headers, it removes the headers from the previous attempt, so a changed configured header key does not leave a stale key in the request.

For client-stream calls, the retry reuses the original async request iterator. The interceptor source notes that this is safe when authentication fails before the iterator is consumed—the usual immediate-header-rejection case—but a partially consumed stream cannot be replayed completely.

Select an authentication flow

Browser login with PKCE

Use auth_type="Pkce" for the normal interactive flow:

client = await ClientSet.for_endpoint(
    "https://flyte.example.com",
    auth_type="Pkce",
    client_config=client_config_store_config,
)

PKCEAuthenticator creates its AuthorizationClient only when it first needs to refresh credentials. It generates a verifier and an S256 challenge, resolves the server's client configuration, and supplies the configured redirect URI, client ID, scopes, audience, authorization endpoint, and token endpoint. The authorization request includes code_challenge and code_challenge_method="S256"; the token request includes the verifier.

When credentials are already available, _do_refresh_credentials first calls AuthorizationClient.refresh_access_token. If there is no usable cached credential or the refresh raises AccessTokenNotFoundError, it falls back to the full browser flow. The authorization client opens the login URL, waits for its local callback server, validates OAuth state, exchanges the callback code, and returns credentials. Its authorization result is cached for 60 seconds to coordinate concurrent callers. A usable redirect URI and local callback port are therefore required.

The PKCE class docstring specifically calls out Auth0 scope configuration: refresh-token caching may require scopes such as offline_access, offline, all, and openid, subject to the deployment's server configuration.

Headless device-code login

Use DeviceCodeAuthenticator when the process cannot complete a browser callback locally:

from flyte.remote._client.auth._authenticators.device_code import DeviceCodeAuthenticator

authenticator = DeviceCodeAuthenticator(
    endpoint="https://flyte.example.com",
    cfg_store=config_store,
    disable_keyring=True,
)
await authenticator.refresh_credentials()

The cfg_store must provide a ClientConfig. The device flow first resolves that configuration and requires device_authorization_endpoint. If the server does not advertise that endpoint, the authenticator raises AuthenticationError.

If a cached refresh token exists, it attempts a refresh-token grant first. Only AuthenticationError and AuthenticationPending cause it to fall back to device authorization. Otherwise it requests a device code, prints a verification URL with the user code appended as ?user_code=..., and polls the token endpoint. The returned Credentials contains the access token, refresh token, expiry value, and configured endpoint.

Obtain a token from an external command

Pass an executable and its arguments as a list, not as a shell command string:

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

authenticator = AsyncCommandAuthenticator(
    ["flyte-token-helper", "--format", "token"],
    endpoint="https://flyte.example.com",
    disable_keyring=True,
)
await authenticator.refresh_credentials()

AsyncCommandAuthenticator rejects an empty or missing list during construction. It invokes the list directly with asyncio.create_subprocess_exec, captures stdout and stderr, strips stdout, and uses the result as the access token. Shell syntax is not interpreted. A non-zero exit status, subprocess failure, or other exception is logged and re-raised as AuthenticationError with an instruction to execute the command manually. The implementation treats even empty stdout as the resulting token, so the command itself must guarantee that it prints the expected token and nothing else.

The same list-oriented contract applies to proxy_command, although that command is used for proxy authorization rather than the request's Flyte access token. The factory sets the proxy authenticator's header key to proxy-authorization.

Pass through delegated request metadata

For webhook, proxy, or delegated-authentication code, initialize passthrough mode and provide metadata only around the calls that should receive it:

import flyte
import flyte.remote as remote

flyte.init_passthrough("endpoint")

with remote.auth_metadata(("key1", "value1")):
    print(remote.Run.listall(limit=2))

PassthroughAuthenticator deliberately does not call Authenticator.__init__. It creates a dummy Credentials object with access token "passthrough" only so the interceptor knows to ask for headers. get_auth_headers reads tuples from flyte.remote._auth_metadata.get_auth_metadata(), converts them to a dictionary, and returns them with the fixed credential ID "passthrough". Outside the context it returns None, and refresh_credentials is a no-op.

Because metadata is converted to a dictionary, duplicate keys follow normal dictionary behavior and the last value wins. auth_metadata must receive tuples; calling it without tuples raises ValueError. The context is request-scoped and is covered by tests for concurrent coroutine and sync/thread isolation.

Supply client configuration

OAuth authenticators resolve their settings through ClientConfigStore. RemoteClientConfigStore obtains OAuth metadata and public-client configuration from Flyte Server's AuthMetadataService:

from flyte.remote._client.auth._client_config import RemoteClientConfigStore

config_store = RemoteClientConfigStore("https://flyte.example.com")

Its get_client_config method concurrently requests OAuth2 metadata and public-client configuration using asyncio.gather. It maps the responses to ClientConfig fields: token endpoint, authorization endpoint, redirect URI, client ID, scopes, authorization metadata header key, optional device endpoint, and audience.

For an explicitly supplied configuration, use StaticClientConfigStore:

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

config = ClientConfig(
    token_endpoint="https://identity.example.com/oauth/token",
    authorization_endpoint="https://identity.example.com/oauth/authorize",
    redirect_uri="http://127.0.0.1:8080/callback",
    client_id="flyte-client",
    scopes=["offline", "all"],
    header_key="authorization",
)
config_store = StaticClientConfigStore(config)

ClientConfig requires token_endpoint, authorization_endpoint, redirect_uri, and client_id. device_authorization_endpoint, scopes, and audience are optional; header_key defaults to "authorization". StaticClientConfigStore.get_client_config returns the same object it was given.

Authenticator._resolve_config fetches remote configuration only once and caches the result. When both a store and local client_config are supplied, ClientConfig.with_override returns a new configuration whose truthy local values replace remote values. A local configuration by itself does not bypass the store requirement: if _cfg_store is absent, _resolve_config raises ValueError. Also, overrides use other.value or self.value, so an empty string or empty scopes list cannot replace a non-empty remote value.

Refresh and store credentials

All token authenticators inherit the asynchronous lifecycle from Authenticator:

headers = await authenticator.get_auth_headers()
if headers is None:
    await authenticator.refresh_credentials()
    headers = await authenticator.get_auth_headers()

On construction, Authenticator uses explicitly supplied credentials, or calls KeyringStore.retrieve(endpoint) unless disable_keyring=True. get_auth_headers returns an AuthHeaders value containing the credential ID and a bearer header. The header key is the configured ClientConfig.header_key after configuration has been resolved; before that it uses default_header_key (normally "authorization"). With no credentials, it returns None.

refresh_credentials accepts the credential ID observed by the caller. It uses that ID and an asyncio.Lock to avoid duplicate refresh work:

• If another caller has already replaced the credential ID, the method returns without refreshing again.
• Otherwise it calls the subclass's _do_refresh_credentials implementation.
• Successful credentials are stored through KeyringStore.store and the current ID is updated.
• Any refresh exception causes KeyringStore.delete for the endpoint before the exception is re-raised.

Credential IDs are generated by Credentials.compute_id as the MD5 hex digest of the access token. They coordinate refresh de-duplication; Authenticator does not proactively refresh based on expires_in. A refresh is normally initiated by an authentication failure and retry, or by an explicit refresh call.

Control keyring persistence

Credentials normalizes for_endpoint by stripping URL schemes. KeyringStore applies the same normalization before lookup, so https://flyte.example.com and dns:///flyte.example.com use flyte.example.com as the service name. Paths are retained, and other endpoint spelling must remain consistent.

KeyringStore.store writes the access token under access_token and writes refresh_token when one exists. retrieve returns None when neither token is present. If only a refresh token is stored, it returns a Credentials object with an empty access token so an authenticator can attempt refresh. delete removes both token entries.

The keyring package is imported lazily. Backend failures, including an unavailable keyring, are logged and swallowed, so a process can continue without persistence. disable_keyring=True short-circuits store, retrieve, and delete before importing keyring; use it for ephemeral runtimes or when credentials must not persist between processes. Note that any failed authenticator refresh deletes the endpoint's stored access and refresh tokens, including when the failure is transient.

Configuration checklist and troubleshooting

The higher-level admin configuration and programmatic initialization expose these controls:

• admin.endpoint, admin.insecure, admin.insecureSkipVerify, admin.caCertFilePath, and admin.httpProxyURL select the endpoint and transport behavior.
• admin.authType accepts Pkce, ClientSecret, ExternalCommand, DeviceFlow, or Passthrough.
• admin.clientId supplies the OAuth/client-credentials client ID. admin.clientSecretLocation reads a mounted secret file (trimming a trailing newline), and admin.clientSecretEnvVar reads a secret from an environment variable; finding a client secret selects ClientSecret mode.
• admin.command supplies the argv list for AsyncCommandAuthenticator.
• admin.proxyCommand supplies proxy authorization argv.
• admin.scopes supplies OAuth scopes, including scopes needed for refresh-token behavior on Auth0 deployments.
• admin.disableKeyring disables credential retrieval, persistence, and cleanup.
• FLYTE_API_KEY is the environment fallback for API-key initialization.

When diagnosing a connection, check the following concrete failure modes:

• A direct ClientSet factory call was not awaited.
• Neither an endpoint nor an API key was supplied.
• insecure=True was expected to mean “keep authentication,” but the session intentionally omits auth interceptors in that mode.
• A token authenticator was constructed without a cfg_store; _resolve_config cannot resolve OAuth settings.
• Device flow was selected but the server's ClientConfig has no device_authorization_endpoint.
• PKCE has no usable redirect URI or cannot bind its local callback port. Its callback handler also indexes code and state directly, so malformed callback data can raise KeyError.
• An external command was passed as one shell-style string, or printed diagnostics along with its token. It must be an argv list and stdout is used verbatim after stripping.
• A cached keyring credential has an empty access token because only a refresh token was available; this is a refreshable state, not a guaranteed usable bearer token.
• A streamed RPC was partially consumed before an authentication failure. The auth retry reuses the iterator and cannot reconstruct consumed items.

Targeted tests cover endpoint/session behavior, API-key forwarding, interceptor retry and header replacement, passthrough context isolation, and keyring storage semantics. The repository does not have direct complete-flow test suites for DeviceCodeAuthenticator, PKCEAuthenticator, AsyncCommandAuthenticator, RemoteClientConfigStore, StaticClientConfigStore, or the complete ClientSet factory flow, so validate those integrations against the server's advertised configuration and the operational requirements above.