Skip to main content

Invoke a Task for Remote Execution

To invoke a deployed task from another Flyte task, create a lazy reference with flyte.remote.Task.get and call it with keyword arguments from the executing task:

Unverified example

This example could not be verified against this version of the codebase and may not work as shown. Validator finding: 'flyte.remote' is not present in the parsed source module

import flyte
from flyte import remote

env = flyte.TaskEnvironment(name="root")

torch_task = remote.Task.get("torch_env.torch_task", auto_version="latest")
spark_task = remote.Task.get("spark_env.spark_task", auto_version="latest")
spark_task2 = remote.Task.get("spark_env.spark_task2", auto_version="latest")


@env.task
async def root_task() -> float:
    v = await torch_task(v={"x": 12, "y": 13})
    v2 = await spark_task(x="hello")
    return await spark_task2(r=v2)

This is the cross-environment pattern used by examples/reference_tasks/root_env.py. The references are created at module scope, but Task.get returns a lazy LazyEntity; it does not immediately fetch the deployed task definition. Calling the lazy entity fetches its TaskDetails and forwards the invocation to that resolved object.

Select and resolve the deployed task​

flyte.remote.Task.get delegates to TaskDetails.get in src/flyte/remote/_task.py. Supply either an explicit version or an automatic version mode:

Unverified example

This example could not be verified against this version of the codebase and may not work as shown. Validator finding: 'flyte.remote' is not present in the parsed source module

import flyte
from flyte import remote

by_version = remote.Task.get(
    "my_env.my_task",
    project="my-project",
    domain="development",
    version="v1",
)
latest = remote.Task.get("my_env.my_task", auto_version="latest")

An explicit version is sent directly in the task-details lookup. With auto_version="latest", TaskDetails.get lists matching tasks with sort_by=("created_at", "desc") and uses the first result. If no version is returned, resolving the reference raises flyte.errors.RemoteTaskNotFoundError.

auto_version="current" is a task-context-only option. During deferred resolution, TaskDetails.get reads flyte.ctx().version; outside a task context it raises ValueError. The auto_version value must be exactly "latest" or "current", and omitting both version and auto_version raises ValueError.

The lookup uses the initialized Flyte configuration for the organization and for project and domain values that were not supplied. Initialize Flyte before fetching or invoking remote references. The example in examples/reference_tasks/root_env.py uses:

import flyte

if __name__ == "__main__":
    flyte.init_from_config()
    r = flyte.with_runcontext(labels={"x": "y"}, annotations={"x": "y"}).run(root_task)
    print(r.url)

To resolve without invoking, use TaskDetails.fetch or the lazy handle's asynchronous fetch method. fetch.aio() is important here because LazyEntity.fetch is a syncify-wrapped operation:

import flyte
import flyte.errors
from flyte.remote import Task


async def check_reference_task_exists(
    task_name: str,
    project: str | None = None,
    domain: str | None = None,
) -> bool:
    try:
        lazy_task = Task.get(
            task_name,
            project=project,
            domain=domain,
            auto_version="latest",
        )
        task_details = await lazy_task.fetch.aio()
        print(f"Version: {task_details.version}")
        print(f"Type: {task_details.task_type}")
        print(f"Required args: {task_details.required_args}")
        return True
    except flyte.errors.RemoteTaskNotFoundError:
        return False

The same resolution can be requested directly with await TaskDetails.fetch(...). A lazy handle caches the first fetched TaskDetails, so repeated await lazy_task.fetch.aio() calls return the cached object rather than resolving it again. Lookup failures therefore occur when the handle is resolved—not when Task.get constructs it.

Invoke with keyword arguments from a task​

Remote task calls must use keyword arguments. The sandbox orchestrator in examples/sandbox/remote_tasks/orchestrator.py invokes four lazy references this way:

Unverified example

This example could not be verified against this version of the codebase and may not work as shown. Validator finding: 'flyte.remote' is not present in the parsed source module

import flyte
from flyte import remote

env = flyte.TaskEnvironment(
    name="lightweight-orchestrator",
    image=flyte.Image.from_debian_base().with_pip_packages("pydantic-monty"),
)

generate_dataset = remote.Task.get("heavy-compute.generate_dataset", auto_version="latest")
fit_linear_model = remote.Task.get("heavy-compute.fit_linear_model", auto_version="latest")
compute_residuals = remote.Task.get("heavy-compute.compute_residuals", auto_version="latest")
detect_outliers = remote.Task.get("heavy-compute.detect_outliers", auto_version="latest")

@env.sandbox.orchestrator
def regression_pipeline(n_samples: int, noise: float, outlier_threshold: float) -> dict:
    dataset = generate_dataset(n_samples=n_samples, noise=noise)
    model = fit_linear_model(dataset=dataset)
    residuals = compute_residuals(dataset=dataset, model=model)
    outlier_indices = detect_outliers(residuals=residuals, threshold=outlier_threshold)
    return {
        "model": model,
        "mean_abs_error": residuals["mean_abs_error"],
        "max_abs_error": residuals["max_abs_error"],
        "n_outliers": len(outlier_indices),
        "outlier_indices": outlier_indices,
        "quality": "good" if model["r_squared"] > 0.9 else "poor",
    }

TaskDetails.__call__ rejects every non-empty positional-argument tuple with flyte.errors.RemoteTaskUsageError and instructs callers to use keyword arguments. Although its signature accepts *args, the task specification does not retain ordered inputs for positional interpretation.

Before submission, TaskDetails obtains required_args. This property returns interface input names that are not in default_input_args. The call checks that the total number of supplied arguments is at least the number of required inputs. This pre-check is count-based: it does not itself verify keyword names, unknown names, or types. Interface-dependent validation occurs in the controller's native conversion step.

How the controller submits the reference​

A resolved TaskDetails is not a local task implementation. Its __call__ method checks internal_ctx().is_task_context(), obtains the controller with get_controller(), and delegates a valid call to:

return await controller.submit_task_ref(self, *args, **kwargs)

The remote controller implementation in src/flyte/_internal/controllers/remote/_controller.py uses _task.interface to convert the keyword arguments with convert.convert_from_native_to_inputs, computes an input hash, serializes and uploads the inputs under _task.max_inline_io_bytes, and constructs a child action with Action.from_task(task_spec=_task.pb2.spec). It submits the action, waits for completion, and uses the native interface to load and convert realized outputs.

TaskDetails.max_inline_io_bytes defaults to 10 * 1024 * 1024 bytes (10 MiB). The remote controller passes this limit to the upload and output-loading paths; TaskDetails.override(max_inline_io_bytes=...) can change it.

The remote controller also requires an initialized task context and uses its current action when creating the child action. Direct TaskDetails invocation therefore requires an executing task context and an available remote controller.

Why local execution fails​

Outside a task context, TaskDetails.__call__ raises flyte.errors.RemoteTaskUsageError stating that remote tasks cannot be executed locally. The local controller has the same explicit boundary:

Unverified example

This example could not be verified against this version of the codebase and may not work as shown. Validator finding: Parsed source module 'flyte' has no member 'errors'

import flyte
from flyte.remote._task import TaskDetails


async def submit_task_ref(self, _task: TaskDetails, max_inline_io_bytes: int, *args, **kwargs) -> Any:
    raise flyte.errors.RemoteTaskUsageError(
        f"Remote tasks cannot be executed locally, only remotely. Found remote task {_task.name}"
    )

When using flyte.run instead of invoking a reference inside a task, use remote run mode. RunContext.run rejects a LazyEntity or TaskDetails when its mode is not "remote"; in remote mode, _run_remote resolves a lazy entity, converts inputs with task.interface, uses task.pb2.spec, and does not build a code bundle for the deployed task.

import flyte
from flyte.remote import Task

ref_task = Task.get("py-io.dynamic_wf", auto_version="latest")
r = flyte.run(ref_task, **data_dict)
print(r.url)

This is the reference-task pattern in examples/reftask_complex_io/pydantic_io.py. A local-mode run rejects the reference before execution.

Apply invocation-time overrides​

TaskDetails.override copies the task protobuf and returns a new frozen TaskDetails; it does not mutate the fetched object. The remote-task override test uses this configuration:

Unverified example

This example could not be verified against this version of the codebase and may not work as shown. Validator finding: 'flyte.Resources' does not accept keyword(s): cpu, memory 'flyte.RetryStrategy' accepts at most 0 positional argument(s)

import flyte
from flyte import RetryStrategy

new_td = td.override(
    short_name="new_oomer",
    resources=flyte.Resources(cpu=3, memory="100Mi"),
    retries=RetryStrategy(5),
    timeout=100,
    env_vars={"FOO": "BAR"},
    secrets=[flyte.Secret(key="openai", as_env_var="OPENAI_API_KEY")],
)

The method supports short name, resources, retries, timeout, container environment, secrets, inline-I/O limit, cache, and queue overrides. Parameters are keyword-only. Container environment overrides clear and replace the existing environment list. Extra names collected by **kwargs raise RemoteTaskUsageError.

Remote cache overrides accept cache.behavior="disable" or "override"; "auto" is rejected, and "override" requires cache.version_override. Queue is stored separately as overriden_queue (the source spelling); the remote action path currently passes queue=None to Action.from_task.

Troubleshooting remote invocations​

  • The reference appears valid, but the first call fails. Task.get is lazy. Call await reference.fetch.aio() to trigger lookup and surface RemoteTaskNotFoundError before invocation.
  • A positional call raises RemoteTaskUsageError. Rewrite every argument as a keyword. Real TaskDetails rejects positional arguments.
  • Too few values raise ValueError. Compare the supplied keyword count with task_details.required_args. Name and type checks occur during controller conversion.
  • A reference fails in ordinary Python or local flyte.run. Invoke it from a task context with a remote controller, or use remote run mode. LocalController.submit_task_ref rejects remote references.
  • auto_version="current" fails before lookup. It reads the enclosing task-context version and cannot be used outside that context.
  • A large input fails before the child action starts. Serialized inputs are uploaded under the 10 MiB default max_inline_io_bytes limit; override that value when required.