Working with Remote Tasks

When you need to interact with tasks already registered on a Flyte backend—whether to inspect their interfaces, override their resource requirements, or trigger them from a script—flyte-sdk provides the remote.Task and remote._task.TaskDetails classes. These entities allow you to treat remote tasks as local Python objects while maintaining a connection to the Flyte API.

Fetching Remote Tasks

The primary entry point for retrieving a task is Task.get(). Because fetching full task definitions (including interfaces and metadata) can be expensive, flyte-sdk returns a LazyEntity that only communicates with the backend when you explicitly request data or attempt to call the task.

Retrieving a Specific Version

If you know the exact version of the task you need, pass the name and version to Task.get(). You must then call .fetch() to resolve the LazyEntity into a TaskDetails object.

from flyte.remote import Task

# Retrieve a specific version of a task
lazy_task = Task.get(
    name="my_workflow.tasks.process_data",
    project="flytesnacks",
    domain="development",
    version="v1"
)

# Explicitly fetch the full details
task_details = lazy_task.fetch()
print(f"Fetched task: {task_details.name} version {task_details.version}")

Automatic Versioning

In scenarios like CLI tools or dynamic scripts where hardcoding a version is impractical, you can use auto_version.

• latest: Fetches the most recently created version of the task.
• current: Derives the version from the current execution context. This is only valid when called from within another running Flyte task.

# Fetch the latest version available on the backend
task = Task.get(name="my_task", auto_version="latest").fetch()

Internally, TaskDetails.get implements this by calling Task.listall with a descending sort on created_at and a limit of 1 to identify the latest version before performing a GetTaskDetails request.

Discovering Tasks

To find available tasks without knowing their names beforehand, use Task.listall(). This method returns an iterator of Task objects (which are lighter than TaskDetails).

# List the first 10 tasks in a project
tasks = Task.listall(project="flytesnacks", domain="development", limit=10)
for t in tasks:
    print(f"Found task: {t.name} ({t.version})")

Inspecting Task Interfaces

Once you have a TaskDetails object, you can inspect its signature to understand what inputs it requires. The interface property returns a NativeInterface which maps Flyte types back to Python types.

task_details = Task.get(name="math.add", version="v1").fetch()

# Inspect required vs optional arguments
print(f"Required: {task_details.required_args}")
print(f"Defaults: {task_details.default_input_args}")

# Access the full interface object
interface = task_details.interface
for name, var in interface.inputs.items():
    print(f"Input {name} is of type {var.type}")

The TaskDetails class also exposes metadata such as cache policies, secrets, and resources (requests and limits) directly from the underlying protobuf definition.

Executing Remote Tasks

You can execute a remote task by calling the LazyEntity or TaskDetails object directly.

Note: Positional arguments are not supported for remote task calls; you must use keyword arguments.

# Calling a remote task like a function
task = Task.get(name="math.multiply", auto_version="latest")
result = task(a=5, b=10)

When you call a task:

1. If it is a LazyEntity, it calls fetch() internally.
2. It validates that all required_args are provided in kwargs.
3. If the code is running inside a Flyte task context, it uses the internal controller (controller.submit_task_ref) to submit the execution to the Flyte backend.

Overriding Task Configuration

You may need to modify a task's execution parameters—such as increasing memory or setting environment variables—without changing the registered task definition. The override() method allows you to create a modified version of the task for a specific execution.

from flyte import Resources

task = Task.get(name="heavy_job", version="v1")

# Override resources and retries for this specific usage
custom_task = task.override(
    resources=Resources(cpu="2", mem="4Gi"),
    retries=3,
    env_vars={"STAGE": "production"}
)

# Execute the task with overrides
custom_task(data_path="s3://bucket/data")

The override() method modifies the internal task_template within the TaskDetails.pb2 object. It supports overriding:

• short_name: A display name for the task.
• resources: CPU, memory, and GPU requirements.
• retries: The number of times to retry on failure.
• timeout: Maximum execution time.
• env_vars: Environment variables passed to the container.
• secrets: Security context requirements.

Implementation Details

Lazy Loading with LazyEntity

The LazyEntity class in remote/_task.py acts as a proxy. It holds a getter coroutine that is only executed when fetch() is called or when the entity is invoked via __call__. This ensures that scripts listing hundreds of tasks do not trigger hundreds of heavy API calls to fetch full task details unless specifically requested.

Configuration Requirements

Interacting with remote tasks requires the flyte-sdk to be configured with a project and domain. If these are not provided in the Task.get() call, the SDK retrieves them from the global configuration via get_common_config(). If the client is not initialized, Task.listall() will trigger an ensure_client() check to verify connectivity to the Flyte service.