Raising Custom Errors in Tasks

Raise a structured user error from a task

When a task needs to report a domain-specific failure instead of a generic exception, raise CustomError with the error code and message that should appear in Flyte's runtime error record:

from flyte.errors import CustomError


def validate_amount(amount: int) -> int:
    if amount <= 0:
        raise CustomError("InvalidAmount", "amount must be greater than zero")
    return amount

CustomError is imported from flyte.errors; it is not re-exported by the top-level flyte namespace. Its constructor is:

CustomError(code: str, message: str)

The code is stored on the exception as code, while message becomes the exception's string value because CustomError ultimately calls BaseRuntimeError.__init__ with that message. CustomError inherits from RuntimeUserError, so its kind is set to "user".

Inspect the error metadata

The runtime error base class stores code, kind, and worker in addition to the normal RuntimeError message. For an explicitly constructed CustomError, the values can be inspected as follows:

from flyte.errors import CustomError

error = CustomError("InvalidAmount", "amount must be greater than zero")

print(error.code)    # InvalidAmount
print(error.kind)    # user
print(str(error))    # amount must be greater than zero
print(error.worker)  # user

The last value reflects the current constructor chain: CustomError.__init__ calls RuntimeUserError as super().__init__(code, message, "user"). In RuntimeUserError, that third positional argument is the optional worker parameter; the parent supplies the "user" kind itself. Consequently, CustomError.worker is currently the literal string "user", rather than None.

The repository does not contain a user-facing task example or test that explicitly constructs CustomError; the preceding function shows the class's actual public constructor used in a task body.

Let the runtime normalize an ordinary exception

You do not have to construct CustomError yourself for an unexpected exception raised by task.execute. _internal/runtime/taskrunner.py handles that case in run_task:

except Exception as e:
    logger.exception(f"Task failed with error: {e}")
    return {}, CustomError.from_exception(e)

from_exception uses the exception's simple class name as the custom code and str(e) as the message:

from flyte.errors import CustomError

try:
    raise ValueError("bad input")
except Exception as e:
    error = CustomError.from_exception(e)

print(error.code)    # ValueError
print(str(error))    # bad input
print(error.kind)    # user
print(error.worker)  # user

This conversion does not preserve the original exception as an explicit __cause__ or __context__. run_task logs the original exception, then returns the newly created CustomError to the runtime pipeline.

Known Flyte runtime errors take a different path. run_task catches RuntimeSystemError, RuntimeUnknownError, and RuntimeUserError before its generic Exception clause and returns each one unchanged. Therefore, raising an existing RuntimeUserError subclass preserves that error's original code, message, and worker metadata; it is not converted with CustomError.from_exception.

Follow the error to the task result

convert_and_run receives the exception returned by run_task. When an error exists, it returns convert_from_native_to_error(err) instead of converting task outputs:

out, err = await run_task(tctx=tctx, controller=controller, task=task, inputs=inputs_kwargs)
if err is not None:
    return None, convert_from_native_to_error(err)

Because CustomError is a RuntimeUserError, _internal/runtime/convert.py serializes it as an execution_pb2.ExecutionError.USER record. The serialized fields are taken from the normalized exception:

elif isinstance(err, flyte.errors.RuntimeUserError):
    return Error(
        err=execution_pb2.ExecutionError(
            kind=execution_pb2.ExecutionError.USER,
            code=err.code,
            message=str(err),
            worker=err.worker,
        )
    )

The CLI runtime path, extract_download_run_upload, then uploads the serialized error rather than task outputs:

if err is not None:
    path = await upload_error(err.err, output_path)
    logger.error(f"Task {task.name} failed with error: {err}. Uploaded error to {path}")
    return

upload_error writes an ErrorDocument containing the error code, message, kind, timestamp, and worker to the error URI under the task output prefix. The runtime therefore uploads structured failure metadata; it does not raise the converted protobuf error back to the caller as the original Python exception object.

Choose the error type for the execution path

Exception type	Runtime classification	Behavior in run_task	Error conversion
CustomError(code, message)	User	Returned unchanged because it is a RuntimeUserError	ExecutionError.USER, with the supplied code and message
RuntimeUserError(code, message, worker=None)	User	Returned unchanged	ExecutionError.USER
RuntimeSystemError(code, message, worker=None)	System	Returned unchanged	ExecutionError.SYSTEM
RuntimeUnknownError(code, message, worker=None)	Unknown	Returned unchanged	ExecutionError.UNKNOWN
Any other Exception	Normalized as user	Converted with CustomError.from_exception	ExecutionError.USER, with the exception class name as code

There is also a separate normalization layer in TaskTemplate.__call__ (_task.py). It re-raises RuntimeSystemError and RuntimeUserError, but wraps other exceptions as a general RuntimeUserError:

except RuntimeSystemError:
    raise
except RuntimeUserError:
    raise
except Exception as e:
    raise RuntimeUserError(type(e).__name__, str(e)) from e

As a result, do not assume every invocation path produces CustomError: the run_task fallback explicitly calls CustomError.from_exception, while TaskTemplate.__call__ uses RuntimeUserError for its own fallback.

Troubleshoot reported codes and messages

• The code is not module-qualified. from_exception uses e.__class__.__name__, so only the simple class name is reported. Different exception classes with the same name can produce the same code.
• The message is exactly str(e). An exception with an empty message produces an empty reported message, and a custom __str__ implementation controls the text.
• Exception chaining is not added by from_exception. If you need Python-level chaining at a call site, add it separately; the conversion method itself only creates and returns a new CustomError.
• Check worker if you inspect the serialized record. The current CustomError constructor passes "user" into the inherited worker argument, so the converted record receives that value in its worker field.
• Do not expect the original Python exception remotely. After convert_and_run and extract_download_run_upload, the runtime uploads an ExecutionError/ErrorDocument representation containing the code, message, kind, and worker fields.