atlas.core.categories

atlas.core.categories

Category contracts and router for Atlas modules.

CategoryRouter

CategoryRouter(manifest: dict, registry: dict)

Query interface for finding installed modules by category or command.

Used by the MCP server and runtime to determine which verbs and tools are available given the current project's installed modules.

Source code in src/atlas/core/categories.py

def __init__(self, manifest: dict, registry: dict) -> None:
    self._manifest = manifest
    self._registry = registry

has_category_installed

has_category_installed(category: str) -> bool

Return True if at least one installed module belongs to category.

Source code in src/atlas/core/categories.py

def has_category_installed(self, category: str) -> bool:
    """Return True if at least one installed module belongs to *category*."""
    installed = self._manifest.get("installed_modules", {})
    return any(info.get("category") == category for info in installed.values())

find_all_with_command

find_all_with_command(command: str) -> list[dict]

Return every installed module that exposes command.

Each entry is {"module": <name>, "command": <cmd_string>}.

Source code in src/atlas/core/categories.py

def find_all_with_command(self, command: str) -> list[dict]:
    """Return every installed module that exposes *command*.

    Each entry is ``{"module": <name>, "command": <cmd_string>}``.
    """
    installed = self._manifest.get("installed_modules", {})
    modules = self._registry.get("modules", {})
    results = []
    for name in installed:
        reg = modules.get(name, {})
        commands = reg.get("commands", {})
        if command in commands:
            results.append({"module": name, "command": commands[command]})
    return results

find_module_for_category

find_module_for_category(category: str) -> str | None

Return the first installed module name for category, or None.

Source code in src/atlas/core/categories.py

def find_module_for_category(self, category: str) -> str | None:
    """Return the first installed module name for *category*, or None."""
    installed = self._manifest.get("installed_modules", {})
    for name, info in installed.items():
        if info.get("category") == category:
            return name
    return None

get_valid_categories

get_valid_categories() -> list[str]

Return all valid category names (installable + auto).

Source code in src/atlas/core/categories.py

def get_valid_categories() -> list[str]:
    """Return all valid category names (installable + auto)."""
    return list(ALL_CATEGORIES.keys())

is_valid_category

is_valid_category(category: str) -> bool

Return True if category is a recognised Atlas category.

Source code in src/atlas/core/categories.py

def is_valid_category(category: str) -> bool:
    """Return True if *category* is a recognised Atlas category."""
    return category in ALL_CATEGORIES

is_auto_category

is_auto_category(category: str) -> bool

Return True if category is an auto-generated (non-installable) category.

Source code in src/atlas/core/categories.py

def is_auto_category(category: str) -> bool:
    """Return True if *category* is an auto-generated (non-installable) category."""
    return category in AUTO_CATEGORIES

get_contract

get_contract(category: str) -> dict

Return the full contract dict for category, or {} if unknown.

Source code in src/atlas/core/categories.py

def get_contract(category: str) -> dict:
    """Return the full contract dict for *category*, or {} if unknown."""
    return ALL_CATEGORIES.get(category, {})

get_required_fields

get_required_fields(category: str) -> list[str]

Return the list of required module fields for category.

Source code in src/atlas/core/categories.py

def get_required_fields(category: str) -> list[str]:
    """Return the list of required module fields for *category*."""
    return ALL_CATEGORIES.get(category, {}).get("required_fields", [])

get_expected_commands

get_expected_commands(category: str) -> list[str]

Return the list of expected command keys for category.

Source code in src/atlas/core/categories.py

def get_expected_commands(category: str) -> list[str]:
    """Return the list of expected command keys for *category*."""
    return ALL_CATEGORIES.get(category, {}).get("expected_commands", [])

validate_module_against_contract

validate_module_against_contract(
    module_name: str, reg_entry: dict
) -> list[dict]

Validate reg_entry against its category's contract.

Returns a list of error dicts {"module": module_name, "error": message}. An empty list means the module is valid.

Source code in src/atlas/core/categories.py

def validate_module_against_contract(module_name: str, reg_entry: dict) -> list[dict]:
    """Validate *reg_entry* against its category's contract.

    Returns a list of error dicts ``{"module": module_name, "error": message}``.
    An empty list means the module is valid.
    """
    errors: list[dict] = []

    category = reg_entry.get("category", "")
    contract = ALL_CATEGORIES.get(category)

    if contract is None:
        errors.append(
            {
                "module": module_name,
                "error": f"unknown category: {category!r}",
            }
        )
        return errors  # can't validate further without a contract

    # Check required fields
    for field in contract.get("required_fields", []):
        if field not in reg_entry:
            errors.append(
                {
                    "module": module_name,
                    "error": f"missing required field: {field!r}",
                }
            )

    # Check expected commands
    commands_val = reg_entry.get("commands", {})
    commands = commands_val if isinstance(commands_val, dict) else {}
    for cmd in contract.get("expected_commands", []):
        if cmd not in commands:
            errors.append(
                {
                    "module": module_name,
                    "error": f"missing expected command: {cmd!r}",
                }
            )

    return errors

validate_registry_integrity

validate_registry_integrity(registry: dict) -> list[dict]

Validate every module in registry against its category contract.

registry is expected to be {"modules": {name: entry, ...}}.

Returns a flat list of error dicts across all modules. An empty list means the entire registry is valid.

Source code in src/atlas/core/categories.py

def validate_registry_integrity(registry: dict) -> list[dict]:
    """Validate every module in *registry* against its category contract.

    *registry* is expected to be ``{"modules": {name: entry, ...}}``.

    Returns a flat list of error dicts across all modules.
    An empty list means the entire registry is valid.
    """
    errors: list[dict] = []
    modules = registry.get("modules", {})
    for name, entry in modules.items():
        errors.extend(validate_module_against_contract(name, entry))
    return errors