Python runtime API

The leoflow_runtime package runs your task callable inside the container and bridges its return value to XCom. It is installed in the DAG image (and the dev venv); your dag.py uses the Apache Airflow Task SDK (from airflow.sdk import DAG, task), and the agent invokes leoflow_runtime to execute the callable.

leoflow_runtime.runner

Run a user task callable and capture its return value.

return_value_path()

Return the path the task's return value is written to.

Overridable via LEOFLOW_RETURN_VALUE_PATH (primarily for tests).

Source code in runtime/python/leoflow_runtime/runner.py

def return_value_path() -> str:
    """Return the path the task's return value is written to.

    Overridable via ``LEOFLOW_RETURN_VALUE_PATH`` (primarily for tests).
    """
    return os.environ.get("LEOFLOW_RETURN_VALUE_PATH", DEFAULT_RETURN_VALUE_PATH)

run(entrypoint)

Import and call module:callable, writing a non-None return as JSON.

The agent reads the file and pushes it as the task's return_value XCom. A None return writes nothing, so downstream tasks see no XCom.

Emits three lifecycle lines so the UI's log panel is informative even when the user function is silent: loading <entrypoint>, resolved kwargs: {...} (when any), and returned <repr> (success) or no extra line on raise (the agent wraps the traceback). All three use [leoflow] prefix via :func:_lifecycle so they are distinguishable from user print().

Source code in runtime/python/leoflow_runtime/runner.py

def run(entrypoint: str) -> None:
    """Import and call ``module:callable``, writing a non-None return as JSON.

    The agent reads the file and pushes it as the task's ``return_value`` XCom.
    A None return writes nothing, so downstream tasks see no XCom.

    Emits three lifecycle lines so the UI's log panel is informative even when
    the user function is silent: ``loading <entrypoint>``, ``resolved kwargs:
    {...}`` (when any), and ``returned <repr>`` (success) or no extra line on
    raise (the agent wraps the traceback). All three use ``[leoflow]`` prefix
    via :func:`_lifecycle` so they are distinguishable from user ``print()``.
    """
    module_name, sep, fn_name = entrypoint.partition(":")
    if not sep or not module_name or not fn_name:
        raise ValueError(f"entrypoint must be 'module:callable', got {entrypoint!r}")

    _lifecycle(f"loading {entrypoint}")
    module = importlib.import_module(module_name)
    fn = getattr(module, fn_name)
    # Airflow TaskFlow @task decorators are not executed when called directly —
    # calling them returns an XComArg (a task reference), not the function's
    # result. Unwrap to the underlying Python function so we run the user's code
    # and capture its real return value.
    if hasattr(fn, "function"):
        fn = fn.function
    context = _operator_context()
    kwargs = _resolve_kwargs(fn, context)
    if kwargs:
        # Log keys only — the values can carry XCom-pulled secrets or any
        # user payload; per ADR 0032 they belong in the XCom tab, not in the
        # log file. The pulled lines above already report each XCom source
        # and its wire size so the operator can correlate.
        _lifecycle(f"resolved kwargs: {sorted(kwargs.keys())}")

    try:
        result = fn(**kwargs)
    except Exception as exc:  # noqa: BLE001 — surface user errors to the log
        _lifecycle(f"user function {fn_name} raised {type(exc).__name__}: {exc}")
        # Stdout is line-buffered or `-u` unbuffered; flush stderr too so the
        # ordering in the log panel matches the wall-clock order.
        sys.stdout.flush()
        sys.stderr.flush()
        # Run the @task's on_failure_callback on its terminal attempt, before the
        # re-raise, so the task's own failed outcome is unchanged (#424 inc 4b).
        _maybe_fire_on_failure_callback(context)
        # Re-raise so Python's default handler emits the traceback to stderr —
        # the agent captures it.
        raise

    _write_return(result)
    # A @task can push custom-keyed XComs via context["ti"].xcom_push — ship them the
    # same way operators do (multi-key XCom parity for native tasks, ADR 0040).
    _write_xcom_pushes(context["ti"].pushed)

run_bash(command)

Render a bash command with the run context, then exec bash -c in place so bash's stdout/stderr/exit code flow to the agent unchanged (ADR 0040). The agent only routes a command here when it contains {{ — plain bash stays a direct bash -c (no Python needed in bash-only images).

Source code in runtime/python/leoflow_runtime/runner.py

def run_bash(command: str) -> None:
    """Render a bash command with the run context, then exec ``bash -c`` in place so
    bash's stdout/stderr/exit code flow to the agent unchanged (ADR 0040). The agent
    only routes a command here when it contains ``{{`` — plain bash stays a direct
    ``bash -c`` (no Python needed in bash-only images)."""
    rendered = _render_bash(command, _operator_context())
    if rendered != command:
        _lifecycle("rendered bash command from the run context")
    # Intentionally exec bash by PATH (the agent ran the task as `bash -c` before this
    # render hop); replacing the process keeps bash's stdout/stderr/exit code intact.
    os.execvp("bash", ["bash", "-c", rendered])  # noqa: S606,S607 — deliberate bash exec

run_operator(operator_class, args)

Instantiate and execute a captured Airflow operator/sensor (ADR 0040 Phase A): import_string(class)(task_id, **args) → render_template_fields → execute. The provider is installed in the task image (declared via connectors:/ dependencies:); connections resolve from AIRFLOW_CONN_* exactly as a hook's do.

Source code in runtime/python/leoflow_runtime/runner.py

def run_operator(operator_class: str, args: dict) -> None:
    """Instantiate and execute a captured Airflow operator/sensor (ADR 0040 Phase
    A): ``import_string(class)(task_id, **args) → render_template_fields → execute``.
    The provider is installed in the task image (declared via connectors:/
    dependencies:); connections resolve from AIRFLOW_CONN_* exactly as a hook's do.
    """
    _lifecycle(f"loading operator {operator_class}")
    module_name, _, class_name = operator_class.rpartition(".")
    if not module_name:
        raise ValueError(f"operator class must be dotted, got {operator_class!r}")
    op_cls = getattr(importlib.import_module(module_name), class_name)
    task_id = os.environ.get("LEOFLOW_TASK_ID", class_name)
    op = op_cls(task_id=task_id, **_merge_operator_xcom(dict(args)))

    context = _operator_context()
    try:
        op.render_template_fields(context)
    except Exception as exc:  # noqa: BLE001 — templating is best-effort in Phase A
        _lifecycle(f"render_template_fields skipped ({type(exc).__name__}: {exc})")

    _lifecycle(f"executing {class_name}.execute()")
    try:
        result = op.execute(context)
    except Exception as exc:  # noqa: BLE001 — translate reschedule/deferral; re-raise the rest
        if _is_reschedule_exc(exc):
            _signal_reschedule(exc, class_name)  # writes the file + SystemExit, or raises
        if _is_deferral_exc(exc):
            raise RuntimeError(
                f"{class_name} asked to defer (deferrable=True), which Leoflow does not "
                f"support yet (ADR 0040 Phase C — no triggerer). Pass deferrable=False — "
                f"the operator runs synchronously in the pod (poke-style).") from exc
        # A genuine failure (not a reschedule/deferral): run the on_failure_callback
        # on the terminal attempt before re-raising, so the task's failed state is
        # unchanged (#424 inc 4b).
        _maybe_fire_on_failure_callback(context)
        raise
    _write_return(result)
    _write_extra_links(op, context["ti"].pushed)
    _write_xcom_pushes(context["ti"].pushed)

leoflow_runtime.xcom

Access XCom inputs injected into the task container by the agent.

xcom_pull(name, default=None)

Return the upstream XCom mapped to name, or default if absent.

The agent injects each declared input as LEOFLOW_XCOM_<NAME>=<json>; the name is matched case-insensitively.

Source code in runtime/python/leoflow_runtime/xcom.py

def xcom_pull(name: str, default: Any = None) -> Any:
    """Return the upstream XCom mapped to ``name``, or ``default`` if absent.

    The agent injects each declared input as ``LEOFLOW_XCOM_<NAME>=<json>``;
    the name is matched case-insensitively.
    """
    raw = os.environ.get(XCOM_ENV_PREFIX + name.upper())
    if raw is None:
        return default
    return json.loads(raw)