Skip to main content

Client Architecture

One channel, several typed services​

A remote RPC in flyte-sdk starts with ClientSet in remote/_client/controlplane.py. ClientSet is an asynchronous façade over generated gRPC stubs: its constructor receives an existing grpc.aio.Channel, stores the endpoint and insecure flag, and binds six stubs to that channel. The stubs are AdminServiceStub, TaskServiceStub, RunServiceStub, DataProxyServiceStub, RunLogsServiceStub, and SecretServiceStub.

The public service attributes are properties, not factory methods. The admin stub is exposed twice because it satisfies two protocol views:

client.metadata_service       # AdminServiceStub, typed as MetadataServiceProtocol
client.project_domain_service # AdminServiceStub, typed as ProjectDomainService
client.task_service           # TaskServiceStub, typed as TaskService
client.run_service            # RunServiceStub, typed as RunService
client.dataproxy_service      # DataProxyServiceStub, typed as DataProxyService
client.logs_service           # RunLogsServiceStub, typed as RunLogsService
client.secrets_service        # SecretServiceStub, typed as SecretService

The annotations do not turn these objects into wrapper services. ClientSet.project_domain_service, for example, returns self._admin_client directly, and ClientSet.run_service returns self._run_service directly. Consequently, client.run_service() is not the access pattern; callers access client.run_service and then invoke an RPC on the generated stub.

data_proxy_channel is accepted by the constructor for compatibility, but the constructor currently creates DataProxyServiceStub(channel=channel) on the main channel. It does not install the supplied channel as a separate data-proxy transport. close() delegates to that main channel:

async def close(self, grace: float | None = None):
    return await self._channel.close(grace=grace)

Constructing the client​

The supported construction paths in ClientSet are explicit endpoint and API-key construction. for_endpoint() passes the endpoint to create_channel() and uses the resulting channel to build the façade. for_api_key() decodes the key to obtain the endpoint for ClientSet.endpoint, then passes the key to create_channel():

from flyte.remote._client.controlplane import ClientSet

client = 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,
)

The SDK's _initialize_client uses the same two branches: it calls ClientSet.for_endpoint(...) when an endpoint is configured and ClientSet.for_api_key(...) when an API key is configured, forwarding the authentication, TLS, proxy, and retry options. for_serverless() and from_env() are declared on ClientSet, but both currently raise NotImplementedError; they are not alternative working initialization paths.

An API key is not an arbitrary opaque value in this implementation. for_api_key() calls decode_api_key(), which expects a base64-decoded, colon-separated four-part value containing endpoint, client ID, client secret, and organization. create_channel() parses the key again and sets the authentication type to ClientSecret with the decoded client credentials. A malformed key therefore fails during setup rather than producing a client with an unknown endpoint.

The channel created beneath ClientSet can be insecure or TLS-based. insecure=True selects an insecure gRPC channel; TLS-related options include insecure_skip_verify and ca_cert_file_path. Other forwarded options include the authentication type, OAuth client settings, external authentication command, HTTP proxy URL, and optional proxy_command. The authentication factory supports Pkce, ClientSecret, ExternalCommand, and DeviceFlow; an invalid authentication type raises ValueError in the factory.

Protocols are structural views of generated stubs​

remote._client._protocols contains typing.Protocol definitions, not channel-owning implementations. Each protocol describes the protobuf request and response types that a generated stub must expose. This keeps higher-level remote objects coupled to capabilities such as GetRunDetails or DeployTask, rather than to a particular generated-stub class.

Projects and domains​

ProjectDomainService describes the admin API exposed through AdminServiceStub. It includes project registration and updates (RegisterProject, UpdateProject), retrieval and listing (GetProject, ListProjects), domain lookup (GetDomains), and project- or domain-attribute operations. Every method is asynchronous and takes the corresponding flyteidl.admin.project_pb2 or project_attributes_pb2 request message. For example, GetProject accepts ProjectGetRequest and returns project_pb2.Project, while UpdateProjectAttributes accepts ProjectAttributesUpdateRequest and returns ProjectAttributesUpdateResponse.

Tasks​

TaskService has a deliberately small surface: asynchronous DeployTask, GetTaskDetails, and ListTasks, using DeployTaskRequest, GetTaskDetailsRequest, and ListTasksRequest. Deployment code constructs a protobuf TaskIdentifier and sends a DeployTaskRequest to the generated stub:

task_id = task_definition_pb2.TaskIdentifier(
    org=spec.task_template.id.org,
    project=spec.task_template.id.project,
    domain=spec.task_template.id.domain,
    version=spec.task_template.id.version,
    name=spec.task_template.id.name,
)

try:
    await get_client().task_service.DeployTask(
        task_service_pb2.DeployTaskRequest(task_id=task_id, spec=spec)
    )
    logger.info(f"Deployed task {task.name} with version {task_id.version}")
except grpc.aio.AioRpcError as e:
    if e.code() == grpc.StatusCode.ALREADY_EXISTS:
        logger.info(f"Task {task.name} with image {image_uri} already exists, skipping deployment.")
        return spec
    raise

The special handling is specific to the deployment caller: ALREADY_EXISTS is logged and treated as a completed deployment, while other gRPC errors are re-raised.

Runs and actions​

RunService separates ordinary unary RPCs from watch operations. CreateRun, AbortRun, GetRunDetails, GetActionDetails, GetActionData, ListRuns, and ListActions return one protobuf response. WatchRunDetails, WatchActionDetails, WatchRuns, and WatchActions return AsyncIterator objects over their corresponding response messages.

The remote run path builds the complete request before reaching the service property. In _run.py, environment entries are converted into literals_pb2.KeyValuePair messages, then included in a run_definition_pb2.RunSpec; the resulting request is sent directly to RunServiceStub:

resp = await get_client().run_service.CreateRun(
    run_service_pb2.CreateRunRequest(
        run_id=run_id,
        project_id=project_id,
        task_spec=task_spec,
        inputs=inputs.proto_inputs,
        run_spec=run_definition_pb2.RunSpec(
            overwrite_cache=self._overwrite_cache,
            interruptible=wrappers_pb2.BoolValue(value=self._interruptible),
            annotations=annotations,
            labels=labels,
            envs=env_kv,
        ),
    )
)

This is the architectural boundary: remote domain objects assemble Flyte protobuf messages, while the client façade supplies the channel-bound generated RPC.

Secrets​

SecretService exposes asynchronous CreateSecret, UpdateSecret, GetSecret, ListSecrets, and DeleteSecret methods. The higher-level Secret façade supplies the organization, project, domain, and secret name when retrieving a secret:

resp = await get_client().secrets_service.GetSecret(
    request=payload_pb2.GetSecretRequest(
        id=definition_pb2.SecretIdentifier(
            organization=cfg.org,
            project=cfg.project,
            domain=cfg.domain,
            name=name,
        )
    )
)
return Secret(pb2=resp.secret)

Listing is paginated by the caller rather than hidden in the protocol. remote/_secret.py repeatedly sends ListSecretsRequest with the returned token and yields each response secret until resp.token is empty.

The same protocols module also defines DataProxyService for upload/download locations and data retrieval, and RunLogsService for TailLogs. RunLogsService.TailLogs is represented as a grpc.aio.UnaryStreamCall, matching the way _logs.py consumes it:

resp = get_client().logs_service.TailLogs(
    run_logs_service_pb2.TailLogsRequest(action_id=action_id, attempt=attempt)
)
async for log_set in resp:
    if log_set.logs:
        for log in log_set.logs:
            for line in log.lines:
                yield line

Authentication as a gRPC interceptor pipeline​

create_channel() builds the transport and its interceptor chain. Authentication configuration is obtained through an initially unauthenticated channel: RemoteClientConfigStore uses that channel to obtain OAuth metadata and public client configuration, and the authenticator factory uses the store to create the selected authenticator. The normal authentication factory then installs one interceptor for each gRPC call shape: unary-unary, unary-stream, stream-unary, and stream-stream. If proxy_command is configured, a parallel set of proxy-authentication interceptors can be added; without it, create_proxy_auth_interceptors() returns an empty list.

The four RPC-shape interceptors share _BaseAuthInterceptor. Its constructor takes a callable returning an Authenticator, but does not call that factory immediately. The authenticator property creates and caches the authenticator on first access. This is a per-interceptor cache: the factory is shared when the interceptors are constructed, while each interceptor stores its own _authenticator field.

call_details_with_auth_metadata() asks the authenticator for gRPC metadata. When metadata exists, it calls with_metadata() to create new call details containing the existing metadata followed by the authentication pairs, and returns the metadata's creds_id alongside the new details. When no metadata exists, it returns the original call details and an empty credential ID. The call details' method, timeout, credentials, and wait-for-ready settings are preserved by with_metadata().

Unary-unary calls retry immediately​

AuthUnaryUnaryInterceptor subclasses both _BaseAuthInterceptor and grpc.aio.UnaryUnaryClientInterceptor. It injects metadata before calling the continuation and awaits the returned call. If the RPC raises grpc.aio.AioRpcError with UNAUTHENTICATED or UNKNOWN, it refreshes credentials using the credential ID, rebuilds metadata from the original call details, and invokes the continuation once more:

updated_call_details, creds_id = await self.call_details_with_auth_metadata(client_call_details)
try:
    return await (await continuation(updated_call_details, request))
except grpc.aio.AioRpcError as e:
    if e.code() == grpc.StatusCode.UNAUTHENTICATED or e.code() == grpc.StatusCode.UNKNOWN:
        await self.authenticator.refresh_credentials(creds_id=creds_id)
        updated_call_details, _ = await self.call_details_with_auth_metadata(client_call_details)
        return await (await continuation(updated_call_details, request))
    else:
        raise e

The retry is limited to that one refresh-and-retry path. Other status codes propagate as AioRpcError.

Streaming calls authenticate lazily​

AuthUnaryStreamInterceptor and AuthStreamStreamInterceptor return the custom UnaryStreamCall wrapper instead of starting the continuation immediately. The wrapper starts the underlying call in response_iterator(), when the caller begins asynchronous iteration. It adds metadata at that point, yields responses from the underlying call, and handles UNAUTHENTICATED or UNKNOWN during iteration by refreshing credentials and creating one replacement call.

AuthStreamStreamInterceptor uses the same wrapper even though it subclasses grpc.aio.StreamStreamClientInterceptor; the wrapper's stored request can therefore be either a unary request or a request iterator. AuthStreamUnaryInterceptor is also installed by the factory and follows the immediate-call pattern for stream-unary RPCs.

UnaryStreamCall proxies the underlying call's read, initial and trailing metadata, status, details, connection waiting, cancellation, completion, remaining time, and done-callback operations. Before iteration has created an underlying call, these methods return defaults: empty Metadata, grpc.StatusCode.OK, False, None, or an empty details string. Thus, querying call state before consuming the stream does not query a live RPC.

The stream retry passes the stored request again. For a stream-stream RPC, that request is the original iterator, which may already have been consumed when the first attempt fails; replayability is therefore not guaranteed. This is an observable constraint of the wrapper's retry design.

Operational boundaries​

Several details matter when diagnosing client setup or authentication:

  • for_endpoint() requires an explicit endpoint. for_api_key() obtains its endpoint from the decoded key. At the lower create_channel() layer, missing both endpoint and API key fails with AssertionError; the higher-level initializer checks configuration earlier and raises its own initialization error.
  • insecure=True changes the transport to an insecure gRPC channel, but authentication interceptors can still be installed. TLS setup also accepts insecure_skip_verify and ca_cert_file_path.
  • auth_type is selected by the authenticator factory. The recognized values are Pkce, ClientSecret, ExternalCommand, and DeviceFlow.
  • Authentication retries occur only for grpc.StatusCode.UNAUTHENTICATED and grpc.StatusCode.UNKNOWN, and each interceptor performs at most one retry. Other AioRpcError values are propagated.
  • Authentication metadata is appended to existing metadata. If a caller has already supplied the same authentication key, duplicate metadata entries are possible.
  • Importing remote/_client/controlplane.py sets gRPC-related environment variables only when GRPC_VERBOSITY is absent: verbosity is set to ERROR, fork support is disabled, and the gLog/Abseil levels are reduced. This is a process-wide import-time side effect.
  • The repository contains no dedicated files matching the searched test or example patterns. The concrete usage paths for runs, task deployment, secrets, and logs are production SDK modules such as _run.py, _deploy.py, _secret.py, and _logs.py rather than standalone examples.