Using Secrets

Inject secrets into a task without persisting them in the event log.

Injects secret values into a task without ever writing them to the event log. task.with_options(secret_requests=[...]) names the secrets a task needs; Flux resolves them through the SecretManager and passes them in via the secrets argument at call time. Reach for this whenever a task needs an API key or credential that must not end up in persisted execution state.

Run it

python examples/using_secrets.py

from __future__ import annotations

from typing import Any

from flux import ExecutionContext
from flux.task import task
from flux.workflow import workflow
from flux.secret_managers import SecretManager

SECRET_NAME = "example"
SECRET_VALUE = "super secret"


@task.with_options(secret_requests=[SECRET_NAME])
async def task_with_secrets(secrets: dict[str, Any] = {}):  # Secrets are not stored in events
    # Do not print a secret value, this is just an example.
    print(f"Secret {SECRET_NAME} = {secrets[SECRET_NAME]}")
    return secrets[SECRET_NAME]


@workflow
async def using_secrets(ctx: ExecutionContext):
    return await task_with_secrets()


if __name__ == "__main__":  # pragma: no cover
    # set the secret prior to use
    secret_manager = SecretManager.current()
    secret_manager.save(SECRET_NAME, SECRET_VALUE)

    ctx = using_secrets.run()
    print(ctx.to_json())

The secrets parameter is populated by Flux from the secret_requests list, not by the workflow. Crucially, the resolved values are excluded from the recorded events — only the request names are stored — so replays re-resolve secrets rather than replaying stale or exposed values. The __main__ block seeds the secret via SecretManager.current().save(...) before the run.