Advanced: Direct API Access with ClientSet

When high-level abstractions in flyte-sdk do not expose specific gRPC functionality—such as tailing logs or managing project-level attributes—you can interact directly with the control plane using the ClientSet. This provides a low-level interface to the underlying gRPC service stubs while handling channel management and authentication.

Initializing the ClientSet

Before accessing any service, you must initialize the global client state. The flyte-sdk provides the init function in _initialize.py to configure the connection to the Flyte backend.

import flyte

# Initialize with an endpoint and API key
await flyte.init(
    endpoint="flyte.example.com",
    api_key="your-api-key",
    insecure=False
)

# Access the initialized ClientSet
client = flyte.get_client()

The ClientSet class (located in remote._client.controlplane) acts as a container for various service stubs. It manages a grpc.aio.Channel and exposes specialized services through properties like run_service, task_service, and secrets_service.

Managing Executions with RunService

The RunService provides fine-grained control over workflow executions. While high-level APIs handle most execution tasks, the RunService allows you to perform operations like listing runs with complex filters or watching run details in real-time.

To create a run directly via the stub:

from flyte._protos.workflow import run_service_pb2
import flyte

client = flyte.get_client()

request = run_service_pb2.CreateRunRequest(
    project="my-project",
    domain="development",
    name="my-execution-name",
    spec=execution_spec, # run_service_pb2.ExecutionSpec
)

response = await client.run_service.CreateRun(request)
print(f"Created run: {response.run_id}")

The RunService protocol in remote._client._protocols.py also supports streaming updates through WatchRunDetails and WatchRuns, which return an AsyncIterator of responses.

Deploying and Retrieving Tasks

The TaskService is responsible for task management. You can use it to deploy task definitions or retrieve metadata for existing tasks.

from flyte._protos.workflow import task_service_pb2
import flyte

client = flyte.get_client()

# Retrieve details for a specific task
request = task_service_pb2.GetTaskDetailsRequest(
    task_id=task_service_pb2.Identifier(
        project="my-project",
        domain="development",
        name="my-task",
        version="v1"
    )
)

details = await client.task_service.GetTaskDetails(request)

Project and Domain Management

The ProjectDomainService allows you to manage the organizational structure of Flyte. This includes registering new projects and updating project-level attributes that might not be exposed through standard CLI commands.

from flyteidl.admin import project_pb2
import flyte

client = flyte.get_client()

# List all projects in the Flyte deployment
request = project_pb2.ProjectListRequest()
projects = await client.project_domain_service.ListProjects(request)

for project in projects.projects:
    print(f"Project: {project.id} - {project.name}")

Accessing Logs and Secrets

For operational tasks, the ClientSet provides access to logs and secret management.

Tailing Logs

The RunLogsService allows you to stream logs for a specific execution. Unlike most other services that use unary calls, TailLogs returns a UnaryStreamCall.

from flyte._protos.workflow import run_logs_service_pb2
import flyte

client = flyte.get_client()

request = run_logs_service_pb2.TailLogsRequest(run_id="execution-id")
log_stream = client.logs_service.TailLogs(request)

async for log_response in log_stream:
    print(log_response.log_line)

Secret Management

The SecretService (defined in remote._client._protocols.SecretService) handles the lifecycle of secrets used by tasks.

from flyte._protos.secret import payload_pb2
import flyte

client = flyte.get_client()

# Create a new secret
request = payload_pb2.CreateSecretRequest(
    id=payload_pb2.SecretIdentifier(key="my-secret", group="my-group"),
    value=b"secret-value"
)
await client.secrets_service.CreateSecret(request)

Data Proxy Operations

The DataProxyService is used to request signed URLs for uploading or downloading data. This is useful when you need to interact with Flyte's underlying storage (like S3 or GCS) without having direct credentials to the storage provider.

from flyteidl.service import dataproxy_pb2
import flyte

client = flyte.get_client()

request = dataproxy_pb2.CreateUploadLocationRequest(
    project="my-project",
    domain="development",
    filename="data.csv"
)

response = await client.dataproxy_service.CreateUploadLocation(request)
print(f"Upload to: {response.signed_url}")

Implementation Details

The ClientSet implementation in remote/_client/controlplane.py initializes specialized stubs from flyteidl and internal proto definitions:

• self._admin_client: AdminServiceStub (used for metadata_service and project_domain_service)
• self._task_service: TaskServiceStub
• self._run_service: RunServiceStub
• self._dataproxy: DataProxyServiceStub
• self._log_service: RunLogsServiceStub
• self._secrets_service: SecretServiceStub

All methods on these services are asynchronous. Ensure you call await client.close() when you are finished with the ClientSet to properly shut down the gRPC channel.