Index

cli

Marianne CLI - modular command structure.

This package provides a modular CLI implementation for Marianne AI Compose. The CLI is built using Typer and organized into command modules for maintainability.

★ Insight ───────────────────────────────────── 1. Typer callback for global options: The @app.callback() decorator defines options that apply to ALL commands (--verbose, --quiet, --version). These callbacks run BEFORE any command, setting up shared state like output level and logging configuration.

1. Command registration via app.command(): Each command function is registered with the Typer app using app.command(). The function name becomes the CLI command name (with underscores converted to hyphens). Custom names are set via the name parameter.

2. Modular command imports: Commands are organized in separate modules (run.py, status.py, etc.) and imported here for registration. This keeps the main init.py focused on assembly while command logic lives in dedicated files. ─────────────────────────────────────────────────

Package structure

cli/ ├── init.py # This file - app assembly (~150 LOC) ├── helpers.py # Shared utilities (~600 LOC) ├── output.py # Rich formatting (~400 LOC) └── commands/ ├── init.py # Command exports ├── compose.py # compose command ├── run.py # run command ├── status.py # status, list_jobs commands ├── resume.py # resume command ├── pause.py # pause, modify commands ├── cancel.py # cancel command ├── validate.py # validate command ├── recover.py # recover command (hidden) ├── diagnose.py # logs, errors, diagnose commands ├── dashboard.py # dashboard, mcp commands └── learning.py # patterns-, learning- commands

Classes

OutputLevel

Bases: str, Enum

Output verbosity level.

Functions

create_notifiers_from_config

create_notifiers_from_config(notification_configs)

Create Notifier instances from notification configuration.

Parameters:

Name	Type	Description	Default
notification_configs	list[NotificationConfig]	List of NotificationConfig from job config.	required

Returns:

Type	Description
list[Notifier]	List of configured Notifier instances.

Source code in src/marianne/notifications/factory.py

def create_notifiers_from_config(
    notification_configs: list[NotificationConfig],
) -> list[Notifier]:
    """Create Notifier instances from notification configuration.

    Args:
        notification_configs: List of NotificationConfig from job config.

    Returns:
        List of configured Notifier instances.
    """
    from marianne.notifications.desktop import DesktopNotifier
    from marianne.notifications.slack import SlackNotifier
    from marianne.notifications.webhook import WebhookNotifier

    notifiers: list[Notifier] = []

    for config in notification_configs:
        notifier: Notifier | None = None
        # Cast Literal list to str list for from_config methods
        events: list[str] = list(config.on_events)

        if config.type == "desktop":
            notifier = DesktopNotifier.from_config(
                on_events=events,
                config=config.config,
            )
        elif config.type == "slack":
            notifier = SlackNotifier.from_config(
                on_events=events,
                config=config.config,
            )
        elif config.type == "webhook":
            notifier = WebhookNotifier.from_config(
                on_events=events,
                config=config.config,
            )
        else:
            _logger.warning("unknown_notification_type", type=config.type)
            continue

        if notifier:
            notifiers.append(notifier)

    return notifiers

version_callback

version_callback(value)

Print version and exit.

Source code in src/marianne/cli/__init__.py

def version_callback(value: bool) -> None:
    """Print version and exit."""
    if value:
        console.print(f"Marianne AI Compose v{__version__}")
        raise typer.Exit()

verbose_callback

verbose_callback(value)

Set verbose output mode.

Source code in src/marianne/cli/__init__.py

def verbose_callback(value: bool) -> None:
    """Set verbose output mode."""
    if value:
        set_output_level(OutputLevel.VERBOSE)

quiet_callback

quiet_callback(value)

Set quiet output mode.

Source code in src/marianne/cli/__init__.py

def quiet_callback(value: bool) -> None:
    """Set quiet output mode."""
    if value:
        set_output_level(OutputLevel.QUIET)

log_level_callback

log_level_callback(value)

Set log level from CLI option.

Source code in src/marianne/cli/__init__.py

def log_level_callback(value: str | None) -> str | None:
    """Set log level from CLI option."""
    if value:
        set_log_level(value)
    return value

log_file_callback

log_file_callback(value)

Set log file path from CLI option.

Source code in src/marianne/cli/__init__.py

def log_file_callback(value: Path | None) -> Path | None:
    """Set log file path from CLI option."""
    if value:
        set_log_file(value)
    return value

log_format_callback

log_format_callback(value)

Set log format from CLI option.

Source code in src/marianne/cli/__init__.py

def log_format_callback(value: str | None) -> str | None:
    """Set log format from CLI option."""
    if value:
        set_log_format(value)
    return value

conductor_clone_callback

conductor_clone_callback(value)

Set the active conductor clone name.

When --conductor-clone is passed, all daemon interactions are routed to a clone conductor instead of the production one. The clone has its own socket, PID file, state DB, and log file.

This enables safe testing without risking the production conductor.

Usage (always use = syntax): mzt --conductor-clone= status # Default clone mzt --conductor-clone=staging run x.yaml # Named clone

Source code in src/marianne/cli/__init__.py

def conductor_clone_callback(value: str | None) -> str | None:
    """Set the active conductor clone name.

    When --conductor-clone is passed, all daemon interactions are routed
    to a clone conductor instead of the production one. The clone has
    its own socket, PID file, state DB, and log file.

    This enables safe testing without risking the production conductor.

    Usage (always use = syntax):
        mzt --conductor-clone= status             # Default clone
        mzt --conductor-clone=staging run x.yaml  # Named clone
    """
    if value is not None:
        from marianne.daemon.clone import set_clone_name

        set_clone_name(value)
    return value

main

main(version=Option(False, '--version', '-V', callback=version_callback, is_eager=True, help='Show version and exit'), verbose=Option(False, '--verbose', '-v', callback=verbose_callback, is_eager=True, help='Show detailed output with additional information'), quiet=Option(False, '--quiet', '-q', callback=quiet_callback, is_eager=True, help='Show minimal output (errors only)'), conductor_clone=None, log_level=None, log_file=None, log_format=None)

Marianne AI Compose - Orchestration system for AI agent workflows.

Source code in src/marianne/cli/__init__.py

@app.callback()
def main(
    version: bool = typer.Option(
        False,
        "--version",
        "-V",
        callback=version_callback,
        is_eager=True,
        help="Show version and exit",
    ),
    verbose: bool = typer.Option(
        False,
        "--verbose",
        "-v",
        callback=verbose_callback,
        is_eager=True,
        help="Show detailed output with additional information",
    ),
    quiet: bool = typer.Option(
        False,
        "--quiet",
        "-q",
        callback=quiet_callback,
        is_eager=True,
        help="Show minimal output (errors only)",
    ),
    conductor_clone: Annotated[
        str | None,
        typer.Option(
            "--conductor-clone",
            callback=conductor_clone_callback,
            is_eager=True,
            help="Route all daemon interactions to a clone conductor. "
            "Use --conductor-clone= (with equals sign) for default clone, "
            "or --conductor-clone=NAME for a named clone. "
            "The clone has its own socket, PID file, state DB, and log.",
        ),
    ] = None,
    log_level: Annotated[
        str | None,
        typer.Option(
            "--log-level",
            "-L",
            callback=log_level_callback,
            help="Logging level (DEBUG, INFO, WARNING, ERROR)",
            envvar="MZT_LOG_LEVEL",
        ),
    ] = None,
    log_file: Annotated[
        Path | None,
        typer.Option(
            "--log-file",
            callback=log_file_callback,
            help="Path for log file output",
            envvar="MZT_LOG_FILE",
        ),
    ] = None,
    log_format: Annotated[
        str | None,
        typer.Option(
            "--log-format",
            callback=log_format_callback,
            help="Log format: json, console, or both",
            envvar="MZT_LOG_FORMAT",
        ),
    ] = None,
) -> None:
    """Marianne AI Compose - Orchestration system for AI agent workflows."""
    # Configure logging based on CLI options (called once)
    configure_global_logging(console)