Skip to main content

Accessing Run Inputs and Outputs

Retrieve run inputs and outputs​

Use RunDetails for a run's root action, or ActionDetails when you need a named action inside the run. The flyte CLI exposes the same two choices:

flyte get io my_run
flyte get io my_run my_action

The first command reads the root action. Supplying my_action reads that action instead. The CLI initializes its configuration, resolves the object, awaits inputs() and outputs(), and renders the result under an “Inputs & Outputs” panel.

Initialize the remote client​

The remote lookups call ensure_client() and use the configured organization, project, and domain to construct identifiers. Initialize flyte-sdk before calling either details class:

import flyte

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

flyte.init accepts either endpoint or api_key; initialization without either raises an InitializationError. Name-based lookups use the configured org, project, and domain. An organization can be supplied explicitly with org=..., or may be derived from the endpoint during initialization.

Read the root action of a run​

RunDetails.get constructs a RunIdentifier from the common configuration and requests the run through RunService.GetRunDetails. RunDetails.__post_init__ wraps the returned run's pb2.action in an ActionDetails, so the run-level accessors expose the root action's inputs and outputs directly.

import asyncio

import flyte
import flyte.remote as remote

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

run_details = remote.RunDetails.get("my_run")


async def read_run_io() -> None:
    inputs = await run_details.inputs()
    outputs = await run_details.outputs()

    print(inputs)
    print(outputs)


asyncio.run(read_run_io())

RunDetails.inputs() and RunDetails.outputs() simply delegate to run_details.action_details.inputs() and run_details.action_details.outputs(). You do not need to navigate through action_details to retrieve root-action data.

For a lower-level lookup with an already constructed protobuf identifier, use RunDetails.get_details:

from flyte._protos.workflow import identifier_pb2
from flyte.remote import RunDetails

run_details = RunDetails.get_details(
    identifier_pb2.RunIdentifier(
        org="my_org",
        project="my_project",
        domain="development",
        name="my_run",
    )
)

get_details sends a run_service_pb2.GetRunDetailsRequest to GetRunDetails and returns RunDetails for the response details. The class method is wrapped by syncify, so the call above is synchronous; async callers can use the corresponding .aio form.

Read a specific action​

When the run contains an action whose name is known, use ActionDetails.get. Without a URI, both run_name and name are required:

from flyte.remote import ActionDetails

action_details = ActionDetails.get(
    run_name="my_run",
    name="my_action",
)


async def read_action_io() -> None:
    inputs = await action_details.inputs()
    outputs = await action_details.outputs()
    print(inputs)
    print(outputs)

ActionDetails.get builds an ActionIdentifier from the configured organization, project, and domain, with run_name in its nested RunIdentifier and name as the action name. ActionDetails.get_details is the identifier-based variant and invokes GetActionDetails.

Understand the returned data​

ActionInputs and ActionOutputs retain the protobuf returned by the run service while also exposing decoded native values.

  • ActionInputs inherits UserDict. Its native values are in data, and it supports dictionary-style access.
  • ActionOutputs is a tuple subclass. Its native values are tuple elements, and the protobuf is attached as pb2.
  • Both wrappers inherit ToJSONMixin; their JSON-oriented serialization is based on pb2, not on the native container alone.

For example, the native containers can be inspected as follows:

async def inspect_io(action_details: ActionDetails) -> None:
    inputs = await action_details.inputs()
    outputs = await action_details.outputs()

    input_value = inputs["input_name"]
    first_output = outputs[0]

    print(input_value)
    print(first_output)
    print(inputs.pb2)
    print(outputs.pb2)

The key and output position must match the task interface and the actual run. The native conversion uses the task interface when the action protobuf has a task, or the trace interface when it has a trace, by calling types.guess_interface and the runtime conversion functions. A single native output is normalized to a one-element tuple; multiple outputs preserve interface order. The output protobuf retains the named literal representation, so inspect outputs.pb2 when positional access does not provide enough information.

If the action has neither a usable task nor trace interface, the loading path creates an empty native dictionary for inputs and an empty native tuple for outputs, while retaining the protobuf wrapper when corresponding response data exists. Raw protobuf data is therefore the available fallback for inspecting serialized values.

Wait for completion before reading outputs​

outputs() enforces terminal-state availability. If outputs are not already cached, it calls the action's internal _cache_data() method. That method requests data with GetActionData, converts the returned inputs and outputs, and caches them as ActionInputs and ActionOutputs. If the action is not terminal and outputs are not cached, outputs() raises:

RuntimeError: Action is not in a terminal state, outputs are not available. Please wait for the action to complete.

A retrieval flow that still displays inputs when outputs are unavailable can follow the behavior in cli/_get.py:

import asyncio
from typing import Tuple, Union

import flyte.remote as remote
from flyte.remote import ActionDetails, ActionInputs, ActionOutputs


def show_io(details: Union[remote.RunDetails, ActionDetails]) -> None:
    async def _get_io(
        details: Union[remote.RunDetails, ActionDetails],
    ) -> Tuple[ActionInputs | None, ActionOutputs | None | str]:
        inputs = await details.inputs()
        outputs: ActionOutputs | None | str = None
        try:
            outputs = await details.outputs()
        except Exception:
            outputs = "[red]not yet available[/red]"
        return inputs, outputs

    inputs, outputs = asyncio.run(_get_io(details))
    print(inputs)
    print(outputs)

The CLI catches Exception specifically so it can still render inputs when output retrieval fails. For application code, catching RuntimeError around outputs() is narrower when the caller only wants to handle the documented non-terminal case.

A run that has completed successfully can be consumed in the same way as any other terminal run. The remote image builder demonstrates the lifecycle explicitly: it waits with await run.wait.aio(quiet=True), gets details with await run.details.aio(), checks run_details.action_details.raw_phase against PHASE_SUCCEEDED, and only then calls await run_details.outputs().

Follow a run while it updates​

For action-level lifecycle handling, ActionDetails.watch streams WatchActionDetails responses until the first terminal phase. watch_updates updates the current protobuf when that terminal update arrives and can populate the input/output cache:

async def wait_for_action(action_details: ActionDetails) -> None:
    async for update in action_details.watch_updates(cache_data_on_done=True):
        print(update.phase)
        if update.done():
            break

    outputs = await action_details.outputs()
    print(outputs)

done() checks the raw phase against the action's terminal-state helper. The related properties are useful when deciding what to retrieve:

if action_details.done():
    print(action_details.phase)
    print(action_details.runtime)
    print(action_details.attempts)

if action_details.logs_available():
    print("Logs are available for the latest attempt")

logs_available(attempt=...) checks a particular one-based attempt; with no argument it checks the latest attempt recorded in status. raw_phase returns the protobuf enum, while phase returns its enum name string.

watch rejects a missing action identifier with ValueError. If the gRPC stream ends with StatusCode.CANCELLED, the cancellation is ignored; other grpc.aio.AioRpcError values are re-raised. The stream stops after yielding an update for which done() is true.

Troubleshoot retrieval failures​

  • Initialization error: Call flyte.init with an endpoint or api_key before using RunDetails or ActionDetails. All remote retrieval methods call ensure_client().
  • Wrong name-based target: Verify org, project, and domain. ActionDetails.get also requires both run_name and action name when no URI is supplied.
  • Outputs not available: Wait for a terminal action state before calling outputs(), or use watch_updates(cache_data_on_done=True) and retrieve outputs after completion.
  • Unexpected empty native values: Native decoding depends on a task or trace interface and types.guess_interface. Inspect inputs.pb2 or outputs.pb2 for the protobuf-backed representation.
  • Repeated data requests: _cache_data() returns true only when _outputs is not None. An action can have inputs without outputs, so a nonempty input response does not by itself mean that output retrieval is complete.

The CLI commands and the image-builder path are the repository's concrete user-facing references for this behavior: the CLI handles both root-run and action-specific retrieval, while the image builder waits for completion and consumes a typed ActionOutputs value.