Skip to main content

Remote Interaction and Control Plane

Programmatically interacting with a Flyte cluster requires the FlyteRemote client, which serves as the high-level entry point for fetching, registering, and executing entities. While flytekit is typically used to define workflows, FlyteRemote allows you to control the control plane from external scripts, CI/CD pipelines, or Jupyter notebooks.

Initializing the Remote Client​

The FlyteRemote class in flytekit.remote.remote is initialized using a Config object. The most common way to create a client is using the auto() or for_sandbox() class methods.

from flytekit.remote import FlyteRemote
from flytekit.configuration import Config

# Automatically detect configuration from environment variables or config file
remote = FlyteRemote.auto()

# Or connect to a local sandbox environment
remote_sandbox = FlyteRemote.for_sandbox()

# Or specify a specific endpoint
remote_custom = FlyteRemote.for_endpoint(
    endpoint="flyte.example.com",
    insecure=False,
    default_project="my_project",
    default_domain="development"
)

Fetching and Executing Entities​

You can fetch existing tasks, workflows, or launch plans from the Flyte Admin service using fetch_task, fetch_workflow, or fetch_launch_plan. Once fetched, these entities can be triggered using the execute method.

# Fetch a workflow by name
# If version is omitted, FlyteRemote retrieves the latest version
workflow = remote.fetch_workflow(name="my_workflow_name")

# Execute the workflow with inputs
execution = remote.execute(
    workflow,
    inputs={"input_a": 10, "input_b": "hello"},
    execution_name="manual-trigger-001"
)

print(f"Execution started: {execution.id.name}")

Executing Local Entities​

FlyteRemote.execute also supports local @task and @workflow definitions. If the entity is not yet registered on the cluster, FlyteRemote will attempt to register it before execution.

from flytekit import task, workflow

@task
def add_one(x: int) -> int:
    return x + 1

@workflow
def my_local_wf(x: int) -> int:
    return add_one(x=x)

# This will register the workflow and then execute it
execution = remote.execute(my_local_wf, inputs={"x": 5})

Monitoring and Retrieving Results​

The execute method returns a FlyteWorkflowExecution object (from flytekit.remote.executions). This object allows you to track the progress of the execution and retrieve outputs once it finishes.

Waiting for Completion​

Use the wait() method to block until the execution reaches a terminal state (Succeeded, Failed, or Aborted).

# Block until finished
execution = remote.wait(execution)

if execution.is_successful:
    # Outputs are returned as a LiteralsResolver
    results = execution.outputs
    print(f"Result: {results['o0']}")
else:
    print(f"Execution failed with error: {execution.error.message}")

Synchronizing State​

If you have an execution name and want to check its status without blocking, use fetch_execution and sync_execution.

# Fetch an existing execution by name
execution = remote.fetch_execution(name="manual-trigger-001")

# Update the local object with the latest state from the server
# Set sync_nodes=True to also fetch status for individual nodes/tasks
execution = remote.sync_execution(execution, sync_nodes=True)

print(f"Current phase: {execution.closure.phase}")

# Access node-level executions
for node_id, node_exec in execution.node_executions.items():
    print(f"Node {node_id} phase: {node_exec.closure.phase}")

Advanced Control Plane Operations​

For administrative tasks that fall outside of standard execution (like managing projects or domains), use the SynchronousFlyteClient available via the remote.client property.

from flytekit.models.project import Project

# Access the low-level SynchronousFlyteClient
client = remote.client

# List all projects
projects = client.list_projects_paginated()

# Update project metadata
my_project = Project(id="my_project", name="My Project", description="Updated description")
client.update_project(my_project)

Troubleshooting​

Accessing Outputs Too Early​

A common error is attempting to access execution.outputs before the execution is finished. The FlyteWorkflowExecution class is designed to prevent this by raising a FlyteAssertion.

# BAD: This will raise FlyteAssertion if the workflow is still RUNNING
# print(execution.outputs) 

# GOOD: Always check is_done or call wait() first
if execution.is_done:
    print(execution.outputs)

Interactive Mode Gotchas​

When interactive_mode_enabled is set to True (default in Jupyter environments), FlyteRemote uses cloudpickle to upload your local task/workflow definitions.

  • Limitation: Dynamic and Eager tasks are not supported in interactive mode.
  • Size Limit: Pickled entities are limited to 150MB. If your local context (variables, imports) exceeds this, FlyteRemote._pickle_and_upload_entity will raise a ValueError.

URI Resolution​

The remote.get() method can be used to resolve Flyte URIs (e.g., flyte://...). Depending on the URI, it may return a LiteralsResolver, a Literal, or even IPython.core.display.HTML if a Flyte Deck link is provided. Always verify the return type when using generic URI resolution.