Skip to content

Jobs and Auto-Discovery

Functualize automatically discovers job functions from Python modules and registers them as CLI commands. This guide explains how the discovery mechanism works, what makes a function eligible for registration, and how to organize jobs into grouped sub-commands.

How auto-discovery works

When you create a FunctualizeApp, the job_sources parameter tells the framework where to look for job modules:

from functualize.app import FunctualizeApp, JobSources

app = FunctualizeApp(
    name="my-app",
    job_sources=JobSources(directories=["src/my_app/jobs"]),  # (1)!
)
  1. You can pass multiple directories. Each one is scanned independently.

The discovery process uses Python's pkgutil.iter_modules to scan each directory. For every Python module found (.py files), the framework:

  1. Skips sub-packages — only top-level modules in the directory are scanned, not nested packages
  2. Imports the module — loads it so functions and module-level variables can be inspected
  3. Inspects all attributes — checks each public attribute against the eligibility criteria
  4. Registers qualifying functions — adds them as Click CLI commands

Import failures are non-fatal

If a module fails to import (syntax error, missing dependency, etc.), a warning is logged and discovery continues with the remaining modules. Your application won't crash because of one broken job file.

Job names are canonical

A job's name is derived from its function name and normalized to lowercase-hyphenated form — the spelling command-line tools conventionally use, and the one Click itself defaults to:

You write The job is You run
def data_sync() data-sync func data-sync
def buildWheel() build-wheel func build-wheel
JOB_GROUP = "data_ops" + def run_etl() data-ops.run-etl func data-ops run-etl

Python identifiers cannot contain hyphens and command names conventionally do, so without a single canonical form the same job has two spellings and every consumer picks one.

Typing the Python spelling still works. func data_sync, rc.invoke("data_sync") and Deps("data_sync") all reach data-sync. That is normalization, not aliasing: there is one name, and you cannot miss it by writing it the way Python spells it. func --help always shows the canonical form.

Two places keep the underscored spelling for good reasons:

  • Environment variables — DATA_SYNC_BATCH_SIZE, because no shell can export a name containing a hyphen.
  • Config sections — [data_sync] and [data-sync] are both read, so an existing config file keeps working.

Two functions whose names normalize to the same job (build_wheel and buildWheel) are rejected at registration rather than one silently replacing the other.

Function eligibility criteria

Not every function in a job module becomes a CLI command. A function is registered only if all of the following are true:

Criterion Description
Callable The attribute must be callable
Is a function Must be an actual function (inspect.isfunction), not a class or other callable object
No underscore prefix The name must not start with _ — underscore-prefixed functions are treated as private
Defined in the module The function must be defined in the scanned module, not imported from elsewhere

The "defined in module" check prevents imported helper functions from accidentally becoming CLI commands. Only functions whose source module matches the scanned module are registered.

# jobs/data_tasks.py

from some_library import helper_function  # (1)!

JOB_GROUP = "data"


def export():  # (2)!
    """Export data to CSV."""
    print("Exporting...")


def _validate_row(row):  # (3)!
    """Internal validation helper."""
    return row is not None


class DataProcessor:  # (4)!
    """Not registered — it's a class, not a function."""
    pass
  1. helper_function is not registered — it's imported from another module.
  2. export is registered — it's a public function defined in this module.
  3. _validate_row is not registered — it starts with an underscore.
  4. DataProcessor is not registered — it's a class, not a function.

When the built-in filters cannot describe your jobs

The require_* settings describe a module by its filename, its imports, a marker, or a decorator. That covers most projects. It does not cover a host whose jobs are, say, methods on a class, or are chosen by a rule that only that host knows.

DiscoveryConfig.pre_filter takes your own predicate, asked before the module is imported:

from pathlib import Path

from functualize.app import FunctualizeApp, JobSources
from functualize.app.config import DiscoveryConfig


class HasJobSuffix:
    """Admit a module only when its name ends in `_tasks` or `_ops`."""

    def should_import(self, source_file: Path) -> bool:
        return source_file.stem.endswith(("_tasks", "_ops"))

    def fingerprint(self) -> str:
        return "has-job-suffix:v1"


app = FunctualizeApp(
    "myapp",
    job_sources=JobSources(directories=["./jobs"]),
    discovery_config=DiscoveryConfig(pre_filter=HasJobSuffix()),
)

Two things about it are worth knowing before you write one.

It composes; it does not replace

Your filter is ANDed onto the stack the require_* settings build. Setting both means a module has to satisfy both:

DiscoveryConfig(
    require_file_prefix="job_",     # filename starts with job_
    pre_filter=HasJobSuffix(),      # ...and ends with _tasks or _ops
)

It runs last, after the cheap filename checks, so an expensive predicate sees the fewest possible files.

fingerprint() is not optional

The discovery cache stores which files your filter rejected and replays those decisions on the next boot, so a scan does not re-read a file it already ruled out. That is only safe while the cache can tell that your filter still behaves the same way.

fingerprint() returns a stable string that you change when the filter's logic changes. Return the same value across processes for the same behaviour, and a new value when you edit the predicate:

    def fingerprint(self) -> str:
        return "has-job-suffix:v2"   # bumped when the suffixes changed

Identity cannot be used instead. An object's str() carries its memory address, so a digest built from the object itself would differ on every boot — invalidating the cache on every run while appearing to work. A filter with no fingerprint() is refused with a TypeError rather than silently cached wrong.

Forgetting to bump it gives you a stale cache: the same contract, and the same failure mode, as editing a require_* setting without invalidating.

lazy=False applies no discovery filter at all

JobSources(lazy=False) opts out of the cached provider and, with it, out of filtering entirely — not just pre_filter, but exclude_patterns and every require_* setting too. This is a known defect, not a design; the default (lazy=True) is the filtered path.

The protocol

ModulePreFilter is exported from functualize.plugin if you want to declare the type. There is nothing to inherit — it is satisfied structurally:

from functualize.plugin import ModulePreFilter


def check(candidate: ModulePreFilter) -> bool:
    return isinstance(candidate, ModulePreFilter)   # both methods required

JOB_GROUP and sub-command grouping

The JOB_GROUP module-level variable controls how functions are organized in the CLI hierarchy.

With JOB_GROUP (grouped)

When a module defines JOB_GROUP, all qualifying functions in that module are registered under a Click sub-command group named after the JOB_GROUP value:

# jobs/reporting.py

JOB_GROUP = "report"  # (1)!


def generate(format: str = "pdf"):
    """Generate a report."""
    print(f"Generating {format} report...")


def send(recipient: str = "team@example.com"):
    """Send the latest report."""
    print(f"Sending report to {recipient}...")
  1. All public functions in this module become sub-commands under my-app report.

This produces the following CLI structure:

my-app report generate --format pdf
my-app report send --recipient team@example.com

Without JOB_GROUP (top-level)

When a module does not define JOB_GROUP, its qualifying functions are registered as top-level commands on the main application:

# jobs/health.py
# No JOB_GROUP defined


def ping():
    """Check if the service is alive."""
    print("pong")

This registers ping directly on the app:

my-app ping

Multiple modules sharing the same JOB_GROUP

Multiple job modules can share the same JOB_GROUP value. Their functions are all registered under the same sub-command group:

# jobs/data_export.py
JOB_GROUP = "data"

def export():
    """Export data to file."""
    ...

# jobs/data_import.py
JOB_GROUP = "data"

def load():
    """Load data from file."""
    ...

Both export and load appear under the data sub-command group:

my-app data export
my-app data load

Organizing large projects

Splitting related functions across multiple files while sharing a JOB_GROUP keeps individual modules focused and manageable, while presenting a unified command group to users.

Duplicate command detection

When a command name is already registered at the same level (either within the same group or at the top level):

This can happen when:

  • Two modules with the same JOB_GROUP both define a function with the same name
  • Two modules without JOB_GROUP both define a function with the same name

Within one directory, the first module discovered wins (filesystem order from pkgutil.iter_modules) and the duplicate is not registered. Duplicate job names across providers raise a ValueError at registration. Do not rely on discovery order — rename one of the functions.

Minimal job file example

Here's a complete, minimal job file with annotations showing which elements are required for discovery:

jobs/sample_job.py
"""Sample job module demonstrating auto-discovery requirements."""

JOB_GROUP = "sample"  # (1)!


def run(target: str, dry_run: bool = False):  # (2)!
    """Execute the sample job.  # (3)!

    Args:
        target: The target resource to process.
        dry_run: If True, simulate without making changes.
    """
    print(f"Starting sample job with target: {target}")
    if dry_run:
        print("Dry run mode — skipping actual processing")
    print("Job completed successfully")
  1. JOB_GROUP — Groups this module's functions under the sample sub-command. Remove this line to register functions at the top level instead.
  2. Public function — Must be a function (not a class), must not start with _, and must be defined in this module. Parameters with type annotations become Click CLI options automatically.
  3. Docstring — Used as the command's help text in --help output.

This produces:

$ my-app sample run --help
Usage: my-app sample run [OPTIONS]

  Execute the sample job.

Options:
  --target TEXT       The target resource to process. [required]
  --dry-run / --no-dry-run
                     If True, simulate without making changes. [default: no-dry-run]
  --help             Show this message and exit.

Job Metadata

@job attaches structured metadata to job functions. It is available via JobDescriptor.declaration and useful for AI orchestrators, schema exporters, and documentation generators.

from functualize.job import job
from functualize.job import RunContext

JOB_GROUP = "deploy"


@job(
    extra_description="Deploy the application to the specified environment",
    category="deployment",
    examples=["deploy --env production", "deploy --env staging --dry-run"],
    tags=["deploy", "infrastructure", "production"],
)
def run(rc: RunContext) -> None:
    """Deploy the application."""
    ...

Identity and description parameters

Parameter Type Description
group str \| None Overrides the module-level JOB_GROUP
extra_description str \| None Description beyond the docstring summary, for AI/LLM consumption
category str \| None Grouping category for job organization
examples tuple[str, ...] \| list[str] Usage examples
tags tuple[str, ...] \| list[str] Searchable tags; also drives func mcp serve --include-tags
visibility "external" \| "internal" "internal" hides the job from MCP. Defaults to "external"
config_section str \| None Config section this job reads

@job also carries the operational contract — deps, cache, guards, exec — as self-validating value objects. See the @job API reference.

The declaration is stored as a frozen JobDeclaration on the function's __functualize_job__ attribute and incorporated into the JobDescriptor during registration. It round-trips through the discovery cache, so warm boot has it without importing the module.

Replaces @job_metadata

The @job_metadata decorator has been removed. Its ai_description argument is now extra_description; category, examples, tags and visibility keep their names.

Composable with other decorators

@job can be combined with any other decorators in any order. It does not wrap the function — it only attaches an attribute, so decorated is original always holds.

JobDescriptor Retention

After registration, every job's JobDescriptor is retained and accessible for runtime introspection:

# From application code
descriptors = app.job_registry.get_descriptors()  # All registered descriptors

descriptor = app.job_registry.get_descriptor("data-sync")  # Single descriptor by name
print(descriptor.name)           # "data-sync"
print(descriptor.group)          # "data" or None
print(descriptor.config_schema)  # Pydantic model class or None
print(descriptor.metadata)       # JobMetadataAnnotation or None

From within a job, use rc.discovery.get_job_schema(job_name) to introspect sibling jobs:

def orchestrator(rc: RunContext) -> None:
    schema = rc.discovery.get_job_schema("validate-data")
    rc.log(f"Job has {len(schema.config_schema.model_fields)} config fields")

This enables patterns like dynamic orchestration, job discovery UIs, and AI-driven job selection.

Next steps