Skip to main content

Getting Started: Fetching Remote Resources

What you will build​

You will connect flyte-sdk to a Flyte deployment, discover task versions with remote.Task.listall, fetch a deployed task definition with remote.Task.get, and inspect its TaskDetails. You will then use the same lazy task reference with flyte.with_runcontext(...).run.aio, inspect the returned remote.Run, and retrieve details for an existing run.

Prerequisites​

You need:

  • flyte-sdk installed and access to a Flyte deployment endpoint or API key;
  • a project and domain in which to list tasks and runs; and
  • credentials and transport settings accepted by flyte.init.

Remote entities call ensure_client() and read the organization, project, domain, and listing batch size from Flyte's initialized configuration. Initialize before calling any remote API method. flyte.init accepts either endpoint or api_key; initialization raises an InitializationError if neither is supplied.

import flyte

flyte.init(
    endpoint="https://flyte.example.com",
    org="my-org",
    project="my-project",
    domain="development",
)

After this call, the default project and domain are used by task and run requests unless a method supplies an override. The endpoint and authentication settings are used to create the remote client and its gRPC service stubs.

Discover deployed tasks​

List task metadata​

Import Task from the public flyte.remote package and iterate over Task.listall. The production get task command uses the same call:

from flyte.remote import Task

for task in Task.listall(limit=100):
    print(task.name, task.version)

Each item is a lightweight Task wrapping the task-list protobuf. Its name and version properties identify the deployed version, but the list result is not the complete task definition.

You can narrow the request by exact task name or by an environment prefix, and can override the initialized project or domain:

from flyte.remote import Task

for task in Task.listall(
    by_task_name="my_task",
    project="my-project",
    domain="development",
    sort_by=("created_at", "desc"),
    limit=10,
):
    print(task.name, task.version)

by_task_name creates an equality filter on name. by_task_env creates a contains filter for "<environment>." in the task name. Task listing follows server pagination tokens and stops at the requested limit. If the limit exceeds the configured batch_size (whose default is 1000), requests are split using that batch size.

The CLI uses an exact-name filter when a name is supplied without a version, and otherwise lists all tasks:

console.print(common.format("Tasks", Task.listall(by_task_name=name, limit=limit), cfg.output_format))

Here name, limit, console, common, and cfg are the command's existing CLI values; the line is from cli/_get.py.

Fetch and inspect one task​

Keep lookup lazy until you need the definition​

A single task lookup requires either an explicit version or an auto_version value. Task.get delegates to TaskDetails.get and returns a LazyEntity; it does not immediately call GetTaskDetails.

from flyte.remote import Task

reference = Task.get(name="my_task", version="20240201")
print(reference)

The visible result identifies the object as a future, using the task name (for example, Future for task with name my_task). To download the definition, call fetch:

task_details = reference.fetch()

print(task_details.name)
print(task_details.version)
print(task_details.task_type)
print(task_details.required_args)

fetch() invokes the asynchronous getter once under a lock and caches the resulting TaskDetails. The syncified method also has an asynchronous form, reference.fetch.aio(). TaskDetails then exposes the identity, task type, default input names, and required input names from the remote specification.

The fetched definition contains additional inspection data:

print(task_details.interface)
print(task_details.cache)
print(task_details.resources)
print(task_details.secrets)

interface is converted with flyte.types.guess_interface and cached as a property. cache is derived from the task metadata, resources is a (requests, limits) tuple for a container task (or () without a container), and secrets is a list of secret keys.

The CLI's versioned lookup follows this same lazy-to-materialized sequence:

v = Task.get(name=name, version=version)
t = v.fetch()
console.print(common.format(f"Task {name}", [t], "json"))

This is the path in cli/_get.py; the command currently requires both a name and version for a specific-task lookup.

Resolve a version automatically​

Use auto_version="latest" when you want TaskDetails.get to first list the named task, sort by created_at descending, and select one result:

latest = Task.get(name="my_task", auto_version="latest")
latest_details = latest.fetch()
print(latest_details.version)

If no matching task is found, fetching the lazy reference raises flyte.errors.ReferenceTaskError. The only accepted automatic values are "latest" and "current". auto_version="current" obtains the version from flyte.ctx().version and is valid only inside a Flyte task context; outside one, fetching raises ValueError.

Submit a remote task and inspect its run​

A LazyEntity can be passed directly to the run machinery. The CLI's remote-run path resolves the task and submits it through a remote run context:

async def _run():
    import flyte
    import flyte.remote

    task = flyte.remote.Task.get(self.task_name, version=self.version, auto_version="latest")

    r = await flyte.with_runcontext(
        copy_style=self.run_args.copy_style,
        mode="local" if self.run_args.local else "remote",
        name=self.run_args.name,
    ).run.aio(task, **ctx.params)

This is the production code from cli/_run.py: self.task_name, self.version, self.run_args, and ctx.params are values supplied by that Click command. In a remote invocation, the keyword arguments must match the fetched task interface. The runner fetches a lazy remote task when it needs its details.

The returned Run represents the execution and exposes its identity and state:

if isinstance(r, Run) and r.action is not None:
    print(r.name)
    print(r.phase)
    print(r.url)

Run constructs its Action from pb2.action; constructing one without an action raises RuntimeError("Run does not have an action"). url is generated from the initialized client's endpoint and the run's project, domain, and name.

Wait for completion and inspect the execution:

await r.wait.aio(quiet=True)
print(r.done())

run_details = await r.details.aio()
outputs = await run_details.outputs()

wait delegates to Action.wait and accepts wait_for="terminal" or wait_for="running". Run.details downloads RunDetails the first time and caches it on the Run; subsequent calls reuse that object. RunDetails.outputs() delegates to its ActionDetails and returns ActionOutputs.

To display logs, use the syncified show_logs method. The CLI follows a newly created run this way:

await r.show_logs.aio(max_lines=30, show_ts=True, raw=False)

Run.watch delegates to the underlying action's watch behavior, while phase, raw_phase, done, and url provide direct run inspection properties.

Retrieve an existing run​

Use Run.get when you already know the run name. The synchronous and asynchronous forms are both available because the method is syncified:

from flyte.remote import Run

run = Run.get(name="my-run")
print(run.name, run.phase)
run_details = run.details()
print(run_details.name, run_details.task_name)

The lookup resolves RunDetails.get, reconstructs a Run from the returned action data, and retains those details for the first run.details() call. In asynchronous code, the corresponding production hybrid-run path is:

from flyte._internal.runtime.task_serde import extract_code_bundle
from flyte.remote import Run

run = await Run.get.aio(name=name)
if run:
    run_details = await run.details.aio()
    spec = run_details.action_details.pb2.resolved_task_spec
    code_bundle = extract_code_bundle(spec)

That code is used by flyte-sdk's hybrid runner to recover the resolved task specification and code bundle for an existing run.

For a list of existing executions, use Run.listall:

from flyte.remote import Run

for run in Run.listall(limit=100):
    print(run.name, run.phase, run.url)

Run listing uses server pagination and enforces the requested limit; each request uses a maximum request limit of 100. The CLI's get run command uses Run.listall(limit=limit) when no run name is supplied. For a named run, that command uses RunDetails.get(name=name) directly to format detailed output.

Operational details to keep in mind​

  • Reference-task calls are keyword-only. TaskDetails.__call__ rejects positional arguments with ReferenceTaskError and checks the number of supplied arguments against required_args.
  • Calling a fetched reference task is intended for an active Flyte task context with a controller. In that context, the controller receives the task protobuf and the max_inline_io_bytes value, which defaults to 10 * 1024 * 1024. The local/non-controller path is not a completed execution path in the source.
  • LazyEntity.override fetches the definition and mutates its protobuf in place. Environment overrides clear and replace the existing container environment; short name, secrets, resources, retries, and timeout are supported by TaskDetails.override.
  • Run.sync() currently returns self; its docstring identifies it as a placeholder and it does not refresh remote state.
  • Run.abort() calls AbortRun. A NOT_FOUND gRPC response is treated as already terminated and returns silently; other RPC errors are re-raised.

You now have the complete first-contact flow: initialize the client, list lightweight Task records, materialize a TaskDetails only when needed, resolve latest versions when appropriate, submit a lazy task through a remote run context, and use Run or RunDetails to inspect executions. Next, use the task interface and required-input metadata to construct valid keyword arguments for a deployed task, then use wait, show_logs, details, and outputs as the execution progresses.