src/magentic_ui/_cli.py

import asyncio
import json
import logging
import os
import signal
import sys
import types
from datetime import datetime
from pathlib import Path
from typing import Any, List, Literal, Optional, cast
from typing_extensions import Annotated

import typer
import yaml
from autogen_core import EVENT_LOGGER_NAME, CancellationToken
from .cli import Console, PrettyConsole
from .cli.pretty_console import (
    display_initial_user_task,
    display_orchestrator_welcome,
    clear_previous_lines,
    terminal_width,
)
from .task_team import get_task_team
from loguru import logger

from .agents.mcp._config import McpAgentConfig
from .magentic_ui_config import MagenticUIConfig, ModelClientConfigs, SentinelPlanConfig
from .types import RunPaths
from .utils import LLMCallFilter
from ._docker import (
    check_docker_running,
    check_browser_image,
    check_python_image,
    pull_browser_image,
    pull_python_image,
)

BOLD = "\033[1m"
RESET = "\033[0m"
MAGENTA = "\033[35m"
GREEN = "\033[32m"

# Create Typer app for CLI interface
app = typer.Typer(help="Magentic-UI CLI")


# Simple debug logging helper - no formatting, just output
# This lets the StylizedConsole handle all the formatting consistently
def log_debug(msg: str, debug: bool = False) -> None:
    """Log debug information if debug mode is enabled."""
    if debug:
        # Simple logging - StylizedConsole will handle formatting
        print(f"DEBUG: {msg}")


logging.basicConfig(level=logging.WARNING, handlers=[])
logger_llm = logging.getLogger(EVENT_LOGGER_NAME)
logger_llm.setLevel(logging.INFO)

ApprovalPolicy = Literal["always", "never", "auto-conservative", "auto-permissive"]


async def cancellable_input(
    prompt: str,
    cancellation_token: Optional[CancellationToken],
    input_type: Optional[str] = "text_input",
) -> str:
    # Copied from autogen_agentchat.agents.UserProxyAgent
    assert input_type in ["text_input", "approval"]
    log_debug(
        f"Starting cancellable_input with type: {input_type}",
        getattr(cancellation_token, "_debug", False),
    )

    # Suppress the "Enter your response:" prompt which appears at the wrong time in the UI
    if prompt.strip() == "Enter your response:":
        prompt = ""

    # Add a newline before the prompt to ensure it appears on a new line
    # This fixes the issue where the prompt appears on the same line as previous output
    if input_type == "text_input" and not prompt.startswith("\n"):
        prompt = f"\n{prompt}"

    task: asyncio.Task[str] = asyncio.create_task(asyncio.to_thread(input, prompt))
    if cancellation_token is not None:
        log_debug(
            "Linking cancellation token to input task",
            getattr(cancellation_token, "_debug", False),
        )
        cancellation_token.link_future(task)
    result = await task
    log_debug(
        f"Cancellable input completed with result length: {len(result)}",
        getattr(cancellation_token, "_debug", False),
    )
    return result


def setup_llm_logging(log_dir: str) -> None:
    """Set up logging to capture only LLM calls to a directory."""
    log_file = os.path.join(
        log_dir, f"llm_calls_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
    )
    file_handler = logging.FileHandler(log_file)
    file_handler.setLevel(logging.INFO)
    formatter = logging.Formatter(
        "%(message)s"
    )  # Only log the message since it contains the JSON
    file_handler.setFormatter(formatter)
    file_handler.addFilter(LLMCallFilter())
    for handler in logger_llm.handlers[:]:  # Remove any existing handlers
        logger_llm.removeHandler(handler)
    logger_llm.addHandler(file_handler)
    log_debug(f"LLM logging configured to: {log_file}", True)


async def get_team(
    cooperative_planning: bool,
    autonomous_execution: bool,
    work_dir: str,
    reset: bool = False,
    task: str | None = None,
    final_answer_prompt: str | None = None,
    debug: bool = False,
    state_file: str = ".magentic_ui_state.json",
    internal_workspace_root: str | None = None,
    external_workspace_root: str | None = None,
    playwright_port: int = -1,
    novnc_port: int = -1,
    inside_docker: bool = False,
    model_context_token_limit: int = 128000,
    client_config: str | None = None,
    action_policy: ApprovalPolicy = "never",
    user_proxy_type: str | None = None,
    task_metadata: str | None = None,
    hints: str | None = None,
    answer: str | None = None,
    mcp_agents: List[McpAgentConfig] | None = None,
    use_pretty_ui: bool = True,
    run_without_docker: bool = False,
    browser_headless: bool = False,
    browser_local: bool = False,
    sentinel_tasks: bool = False,
    dynamic_sentinel_sleep: bool = False,
) -> None:
    log_debug("=== Starting get_team function ===", debug)
    log_debug(
        f"Args: cooperative_planning={cooperative_planning}, autonomous_execution={autonomous_execution}, reset={reset}",
        debug,
    )
    log_debug(
        f"Args: inside_docker={inside_docker}, action_policy={action_policy}, user_proxy_type={user_proxy_type}",
        debug,
    )

    if reset:
        # delete the state file if it exists
        if os.path.exists(state_file):
            log_debug(f"Deleting existing state file at: {state_file}", debug)
            os.remove(state_file)

    if inside_docker:
        log_debug("Running inside Docker container", debug)
        # Use environment variables as fallback if paths not provided
        if internal_workspace_root is None:
            internal_workspace_root = os.environ.get("INTERNAL_WORKSPACE_ROOT")
            log_debug(
                f"Using INTERNAL_WORKSPACE_ROOT from env: {internal_workspace_root}",
                debug,
            )
        if external_workspace_root is None:
            external_workspace_root = os.environ.get("EXTERNAL_WORKSPACE_ROOT")
            log_debug(
                f"Using EXTERNAL_WORKSPACE_ROOT from env: {external_workspace_root}",
                debug,
            )

        if not internal_workspace_root or not external_workspace_root:
            error_msg = "When running inside docker, both internal and external workspace root paths must be provided either via arguments or environment variables"
            log_debug(f"ERROR: {error_msg}", debug)
            raise ValueError(error_msg)

        # In contrast to the UI -- where each session creates a new folder --
        # we need a consistent folder name so that benchmark scripts will know
        # where to place input files.
        cli_files_path = os.path.join(internal_workspace_root, "cli_files")
        log_debug(f"Creating CLI files directory at: {cli_files_path}", debug)
        os.makedirs(cli_files_path, exist_ok=True)

        paths = RunPaths(
            internal_root_dir=Path(internal_workspace_root),
            external_root_dir=Path(external_workspace_root),
            run_suffix="run",
            internal_run_dir=Path(cli_files_path),
            external_run_dir=Path(os.path.join(external_workspace_root, "cli_files")),
        )
        log_debug(
            f"RunPaths created with internal_root_dir: {paths.internal_root_dir}", debug
        )
    else:
        log_debug("Running outside Docker container", debug)
        if not work_dir:
            error_msg = "When running outside docker, work_dir path must be provided"
            log_debug(f"ERROR: {error_msg}", debug)
            raise ValueError(error_msg)

        work_dir_path = Path(work_dir)
        log_debug(f"Using work_dir: {work_dir_path}", debug)

        cli_files_path = os.path.join(work_dir, "cli_files")
        log_debug(f"Creating CLI files directory at: {cli_files_path}", debug)
        os.makedirs(cli_files_path, exist_ok=True)
        work_dir_files = work_dir_path / "cli_files"

        paths = RunPaths(
            internal_root_dir=work_dir_path,
            external_root_dir=work_dir_path,
            run_suffix="run",
            internal_run_dir=work_dir_files,
            external_run_dir=work_dir_files,
        )
        log_debug("RunPaths created with internal/external run dirs", debug)

    client_config_dict: dict[str, Any] = {}
    if client_config:
        log_debug(f"Loading client configuration from: {client_config}", debug)
        with open(client_config, "r") as f:
            client_config_dict = yaml.safe_load(f)
            log_debug(
                f"Client config loaded with {len(client_config_dict)} keys", debug
            )

    # sets the configurations for each different agent
    model_client_configs = ModelClientConfigs(
        orchestrator=client_config_dict.get("orchestrator_client", None),
        web_surfer=client_config_dict.get("web_surfer_client", None),
        coder=client_config_dict.get("coder_client", None),
        file_surfer=client_config_dict.get("file_surfer_client", None),
        action_guard=client_config_dict.get("action_guard_client", None),
    )

    mcp_agents = mcp_agents or []
    log_debug("Model client configs created for agents", debug)

    magentic_ui_config = MagenticUIConfig(
        model_client_configs=model_client_configs,
        mcp_agent_configs=mcp_agents,
        cooperative_planning=cooperative_planning,
        autonomous_execution=autonomous_execution,
        approval_policy=action_policy,
        allow_for_replans=True,
        do_bing_search=False,
        model_context_token_limit=model_context_token_limit,
        allow_follow_up_input=False,
        final_answer_prompt=final_answer_prompt,
        playwright_port=playwright_port,
        novnc_port=novnc_port,
        user_proxy_type=user_proxy_type,
        task=task_metadata,
        hints=hints,
        answer=answer,
        inside_docker=inside_docker,
        sentinel_plan=SentinelPlanConfig(
            enable_sentinel_steps=sentinel_tasks,
            dynamic_sentinel_sleep=dynamic_sentinel_sleep,
        ),
        run_without_docker=run_without_docker,
        browser_headless=browser_headless,
        browser_local=browser_local,
    )
    log_debug(
        f"MagenticUIConfig created with planning={cooperative_planning}, execution={autonomous_execution}",
        debug,
    )

    log_debug("Starting team creation", debug)

    # Creates and returns a RoundRobinGroupChat or a GroupChat with the passed configs
    log_debug("Calling get_task_team to create team object", debug)
    team = await get_task_team(
        magentic_ui_config=magentic_ui_config,
        input_func=cancellable_input,
        paths=paths,
    )
    log_debug(f"Team created with type: {type(team).__name__}", debug)

    # Team is a RoundRobinGroupChat or GroupChat instance
    log_debug("Team creation completed successfully", debug)

    if use_pretty_ui:
        display_magentic_ui_logo()
    log_debug("Logo displayed", debug)

    try:
        if state_file and os.path.exists(state_file) and not reset:
            state = None
            log_debug(f"Loading state from: {state_file}", debug)

            with open(state_file, "r") as f:
                state = json.load(f)
                log_debug("State loaded successfully", debug)

            log_debug("Calling team.load_state with loaded state", debug)
            await team.load_state(state)
            log_debug("State loading completed", debug)

        if not task:
            log_debug("No task provided, prompting user for input", debug)

            # Define flushed_input function that can be used for both UI modes
            def flushed_input(prompt: str) -> str:
                # Prompt for input, but flush the prompt to ensure it appears immediately
                print(prompt, end="", flush=True)
                user_input = input()
                log_debug(f"User input received, length: {len(user_input)}", debug)

                # Calculate lines to erase after user input
                input_lines = 1
                if user_input:
                    term_width = terminal_width()
                    input_lines += len(user_input) // term_width

                return user_input

            if use_pretty_ui:
                # Display orchestrator welcome message using pretty console formatting
                display_orchestrator_welcome()

                # The prompt is now included in the welcome message
                log_debug("Creating input task in event loop with pretty UI", debug)
                task = await asyncio.get_event_loop().run_in_executor(
                    None,
                    flushed_input,
                    "",  # prompt is already displayed with display_orchestrator_welcome
                )

                # Calculate how many lines to clear based on input length and terminal width
                term_width = terminal_width()

                # prompt line
                lines_to_clear = 1

                # Add lines for the input text based on how it would wrap in the terminal
                if task:
                    # Allow for the '> ' prefix (2 chars) in the first line
                    first_line_chars = term_width - 2
                    if len(task) > first_line_chars:
                        # First line is shorter due to the prompt
                        remaining_chars = len(task) - first_line_chars
                        # Remaining lines use full width
                        additional_lines = (
                            remaining_chars + term_width - 1
                        ) // term_width
                        lines_to_clear += additional_lines

                # Log the calculation for debugging purposes
                log_debug(
                    f"Clearing {lines_to_clear} lines for user input (term width: {term_width})",
                    debug,
                )

                # Clear just the empty line + prompt + user input lines, not the welcome message
                clear_previous_lines(lines_to_clear)

            else:
                # Use the original prompt for non-pretty UI mode
                log_debug("Creating input task in event loop", debug)
                task = await asyncio.get_event_loop().run_in_executor(
                    None,
                    flushed_input,
                    f"{MAGENTA}{BOLD}Enter your task (or press Ctrl+C to cancel): {RESET}",
                )

            log_debug("User input task completed", debug)

        log_debug(
            f"Task to execute: {task[:50] if task else ''}{'...' if task and len(task) > 50 else ''}",
            debug,
        )

        # Add an extra blank line for better spacing
        print()

        # Display the user's task in a USER box format before passing to agents
        if use_pretty_ui and task:
            # Display the initial user task with proper USER box formatting
            display_initial_user_task(task)
        elif task:
            # For non-pretty UI, we still want to display the initial task
            display_initial_user_task(task)

        log_debug("Creating team run stream with task", debug)
        stream = team.run_stream(task=task)
        log_debug(
            f"Stream created, passing to {'PrettyConsole' if use_pretty_ui else 'Console'} with debug={debug}",
            debug,
        )

        # Use PrettyConsole or the regular console based on the use_pretty_ui parameter
        if use_pretty_ui:
            await PrettyConsole(stream, debug=debug)
        # Using default Console
        else:
            await Console(stream)

        log_debug("Console processing completed", debug)

        log_debug("Saving team state", debug)
        state = await team.save_state()
        log_debug("State saved successfully", debug)

        log_debug(f"Writing state to file: {state_file}", debug)
        with open(state_file, "w") as f:
            json.dump(state, f, indent=2)
        log_debug("State file write completed", debug)

    finally:
        logger.info("Closing team...")
        log_debug("Closing team in finally block", debug)
        await team.close()
        log_debug("Team closed successfully", debug)


def display_magentic_ui_logo():
    """Display the MAGENTIC UI entry text."""

    magentic_logo = f"""{MAGENTA}{BOLD}╔═══════════════════════════════════════════════════════════════════╗
║    __  __    _    ____ _____ _   _ _____ ___ ____    _   _ ___    ║
║   |  \\/  |  / \\  / ___| ____| \\ | |_   _|_ _/ ___|  | | | |_ _|   ║
║   | |\\/| | / _ \\| |  _|  _| |  \\| | | |  | | |      | | | || |    ║
║   | |  | |/ ___ \\ |_| | |___| |\\  | | |  | | |___   | |_| || |    ║
║   |_|  |_/_/   \\_\\____|_____|_| \\_| |_| |___\\____|   \\___/|___|   ║ 
║                                                                   ║
╚═══════════════════════════════════════════════════════════════════╝\n{RESET}"""

    print(magentic_logo)


@app.callback(invoke_without_command=True)
def cli_main(
    # Required arguments
    work_dir: str = typer.Option(
        ...,
        "--work-dir",
        help="Working directory path: where the team will store its state and files (required)",
    ),
    # Basic options
    config: Optional[str] = typer.Option(
        None,
        "--config",
        help="Path to the configuration file (default: 'config.yaml')",
    ),
    run_without_docker: bool = typer.Option(
        False,
        "--run-without-docker",
        help="Run without docker. This will remove coder and filesurfer agents and disable live browser view.",
    ),
    browser_headless: Annotated[
        bool,
        typer.Option(
            "--headless",
            help="Run browser in headless mode (default: False, browser runs with GUI)",
        ),
    ] = False,
    browser_local: Annotated[
        bool,
        typer.Option(
            "--local-browser",
            help="Run browser locally on your machine instead of in Docker (default: False, uses Docker browser with noVNC)",
        ),
    ] = False,
    debug: bool = typer.Option(
        False,
        "--debug",
        help="Enable debug mode to show internal messages and detailed agent interactions (default: disabled)",
    ),
    use_state: Annotated[
        bool,
        typer.Option(
            "--use-state",
            help="Use and save the team state before and after running the task (default: always start fresh and do not use state)",
        ),
    ] = False,
    # Advanced planning options
    cooperative_planning: Annotated[
        bool,
        typer.Option(
            "--disable-planning/--enable-planning",
            help="Disable co-planning mode (default: enabled), user will not be involved in the planning process",
        ),
    ] = True,
    autonomous_execution: bool = typer.Option(
        False,
        "--autonomous-execution",
        help="Enable autonomous execution mode (default: disabled), user will not be involved in the execution",
    ),
    autonomous: bool = typer.Option(
        False,
        "--autonomous",
        help="Enable autonomous mode (default: disabled), no co-planning and no human involvment in execution",
    ),
    # Task configuration
    task: str = typer.Option(
        "",
        "--task",
        help="Specifies the initial task. If a plain string, use this input verbatim. If the string matches a filename, read the initial task from a file. Use '-' to read from stdin. (default: prompt's the user for the task)",
    ),
    final_answer_prompt: str = typer.Option(
        "",
        "--final-answer-prompt",
        help="Overrides the final answer prompt used to summarize the conversation. If a plain string, use this input verbatim. If the string matches a filename, read the prompt from a file. (default: use orchestrator's built-in prompt)",
    ),
    # Infrastructure options
    internal_workspace_root: Optional[str] = typer.Option(
        None,
        "--internal-root",
        help="Deprecated: Internal workspace root directory path (default: use INTERNAL_WORKSPACE_ROOT environment variable)",
    ),
    external_workspace_root: Optional[str] = typer.Option(
        None,
        "--external-root",
        help="Deprecated: External workspace root directory path (default: use EXTERNAL_WORKSPACE_ROOT environment variable)",
    ),
    playwright_port: int = typer.Option(
        -1,
        "--playwright-port",
        help="Port to run the Playwright browser on (default: -1 means use default port)",
    ),
    novnc_port: int = typer.Option(
        -1,
        "--novnc-port",
        help="Port to run the noVNC server on (default: -1 means use default port)",
    ),
    inside_docker: bool = typer.Option(
        False,
        "--inside-docker",
        help="Deprecated:Indicates if running inside docker container (default: False)",
    ),
    # User proxy options
    user_proxy_type: Optional[str] = typer.Option(
        None,
        "--user-proxy-type",
        help="Type of user proxy agent to use for simulations ('dummy', 'metadata', or None for default; default: None)",
    ),
    metadata_task: Optional[str] = typer.Option(
        None,
        "--metadata-task",
        help="Task description for metadata user proxy (required if user-proxy-type is 'metadata')",
    ),
    metadata_hints: Optional[str] = typer.Option(
        None,
        "--metadata-hints",
        help="Task hints for metadata user proxy (required if user-proxy-type is 'metadata')",
    ),
    metadata_answer: Optional[str] = typer.Option(
        None,
        "--metadata-answer",
        help="Task answer for metadata user proxy (required if user-proxy-type is 'metadata')",
    ),
    # Logging and monitoring
    llmlog_dir: Optional[str] = typer.Option(
        None,
        "--llmlog-dir",
        help="Directory path to save LLM call logs (if not provided, LLM logging is disabled)",
    ),
    action_policy: str = typer.Option(
        "never",
        "--action-policy",
        help="ActionGuard policy ('always', 'never', 'auto-conservative', 'auto-permissive'; default: never)",
    ),
    mcp_agents_file: Optional[str] = typer.Option(
        None,
        "--mcp-agents-file",
        help="Path to a .yaml file containing configuration compatible with MagenticUIConfig.mcp_agents",
    ),
    use_pretty_ui: Annotated[
        bool,
        typer.Option(
            "--pretty-cli/--old-cli",
            help="Use the old console without fancy formatting (default: use pretty terminal)",
        ),
    ] = True,
    sentinel_tasks: bool = typer.Option(
        False,
        "--sentinel-tasks",
        help="Use sentinel tasks to guide the agent's behavior (default: False)",
    ),
    dynamic_sentinel_sleep: bool = typer.Option(
        False,
        "--dynamic-sentinel-sleep",
        help="Enable dynamic sleep duration adaptation for sentinel tasks (only available when --sentinel-tasks is enabled, default: False)",
    ),
) -> None:
    """
    Magentic-UI CLI: A command-line interface for multi-agent task execution.

    Run tasks using a team of AI agents with web browsing, coding, and file manipulation capabilities.
    """
    log_debug(f"Command line arguments parsed: debug={debug}", debug)

    # Show summary of important arguments when debug is enabled
    if debug:
        log_debug(f"Cooperative planning: {cooperative_planning}", debug)
        log_debug(f"Autonomous execution: {autonomous_execution}", debug)
        log_debug(f"Autonomous mode: {autonomous}", debug)
        log_debug(f"Use state: {use_state}", debug)
        log_debug(f"Task specified: {bool(task)}", debug)
        log_debug(f"Action policy: {action_policy}", debug)
        log_debug(f"Inside Docker: {inside_docker}", debug)
        log_debug(f"Work directory: {work_dir}", debug)
        log_debug(f"Config file: {config}", debug)
        log_debug(f"User proxy type: {user_proxy_type}", debug)
        log_debug(f"LLM log directory: {llmlog_dir}", debug)
        log_debug(f"Sentinel tasks: {sentinel_tasks}", debug)
        log_debug(f"Dynamic sentinel sleep: {dynamic_sentinel_sleep}", debug)
        log_debug(f"Console mode: {'Pretty' if use_pretty_ui else 'Old'}", debug)
        log_debug(f"Browser headless: {browser_headless}", debug)
        log_debug(f"Browser local: {browser_local}", debug)

    # Validate user proxy type
    log_debug("Validating user proxy type", debug)
    if user_proxy_type not in [None, "dummy", "metadata"]:
        error_msg = f"Invalid user proxy type: {user_proxy_type}. Valid options are None, 'dummy', or 'metadata'."
        log_debug(f"ERROR: {error_msg}", debug)
        raise ValueError(error_msg)

    # Validate metadata user proxy parameters
    log_debug("Validating metadata user proxy parameters", debug)
    if user_proxy_type == "metadata":
        if not all([metadata_task, metadata_hints, metadata_answer]):
            error_msg = "When using metadata user proxy type, all metadata parameters (--metadata-task, --metadata-hints, --metadata-answer) must be provided."
            log_debug(f"ERROR: {error_msg}", debug)
            raise ValueError(error_msg)

    # Validate action policy
    log_debug("Validating action policy", debug)
    if action_policy not in [
        "always",
        "never",
        "auto-conservative",
        "auto-permissive",
    ]:
        error_msg = f"Invalid action policy: {action_policy}. Valid options are 'always', 'never', 'auto-conservative', 'auto-permissive'."
        log_debug(f"ERROR: {error_msg}", debug)
        raise ValueError(error_msg)

    # Validate dynamic sentinel sleep configuration
    log_debug("Validating dynamic sentinel sleep configuration", debug)
    if dynamic_sentinel_sleep and not sentinel_tasks:
        error_msg = "Dynamic sentinel sleep (--dynamic-sentinel-sleep) can only be enabled when sentinel tasks (--sentinel-tasks) are enabled."
        log_debug(f"ERROR: {error_msg}", debug)
        raise ValueError(error_msg)

    # Set up LLM logging if requested
    if llmlog_dir:
        log_debug(f"Setting up LLM logging to directory: {llmlog_dir}", debug)
        setup_llm_logging(llmlog_dir)

    # If the config file is not provided, check for the default config file
    client_config: str | None = config
    if not client_config:
        if os.path.isfile("config.yaml"):
            client_config = "config.yaml"
            log_debug(
                "Using default config.yaml file found in current directory", debug
            )
        else:
            log_debug(
                "No config file provided or found. Using default settings.", debug
            )
            logger.info("Config file not provided. Using default settings.")

    # Expand the task and final answer prompt
    log_debug("Processing task input", debug)
    processed_task: str | None = None
    if task:
        if task == "-":
            log_debug("Reading task from stdin", debug)
            processed_task = sys.stdin.buffer.read().decode("utf-8")
            log_debug(
                f"Task read from stdin, length: {len(processed_task if processed_task else '')}",
                debug,
            )
        elif os.path.isfile(task):
            log_debug(f"Reading task from file: {task}", debug)
            with open(task, "r") as f:
                processed_task = f.read()
                log_debug(
                    f"Task read from file, length: {len(processed_task if processed_task else '')}",
                    debug,
                )
        else:
            log_debug("Using task from command line argument", debug)
            processed_task = task
            log_debug(
                f"Task from argument, length: {len(processed_task if processed_task else '')}",
                debug,
            )

    if not run_without_docker:
        # Check Docker and pull images if necessary
        log_debug("Checking Docker setup...", debug)
        logger.info("Checking if Docker is running...")

        if not check_docker_running():
            logger.error("Docker is not running. Please start Docker and try again.")
            sys.exit(1)
        else:
            logger.success("Docker is running")

        # Check and pull Docker images if needed
        logger.info("Checking Docker vnc browser image...")
        if not check_browser_image():
            logger.warning("VNC browser image needs to be pulled")
            logger.info("Pulling Docker vnc image (this WILL take a few minutes)")
            pull_browser_image()
        else:
            logger.success("VNC browser image is available")

        logger.info("Checking Docker python image...")
        if not check_python_image():
            logger.warning("Python image needs to be pulled")
            logger.info("Pulling Docker python image (this WILL take a few minutes)")
            pull_python_image()
        else:
            logger.success("Python image is available")

        # Verify Docker images exist after attempted pull
        if not check_browser_image() or not check_python_image():
            logger.error(
                "Docker images not found. Please pull or build the images and try again."
            )
            sys.exit(1)

        log_debug("Docker setup completed successfully", debug)

    log_debug("Processing final answer prompt", debug)
    processed_final_answer_prompt: str | None = None
    if final_answer_prompt:
        if os.path.isfile(final_answer_prompt):
            log_debug(
                f"Reading final answer prompt from file: {final_answer_prompt}",
                debug,
            )
            with open(final_answer_prompt, "r") as f:
                processed_final_answer_prompt = f.read()
                log_debug(
                    f"Final answer prompt read from file, length: {len(processed_final_answer_prompt if processed_final_answer_prompt else '')}",
                    debug,
                )
        else:
            log_debug("Using final answer prompt from command line argument", debug)
            processed_final_answer_prompt = final_answer_prompt
            log_debug(
                f"Final answer prompt from argument, length: {len(processed_final_answer_prompt if processed_final_answer_prompt else '')}",
                debug,
            )

    # Set up autonomous execution mode if requested
    if autonomous:
        log_debug(
            "Autonomous mode enabled, setting autonomous_execution=True and cooperative_planning=False",
            debug,
        )
        autonomous_execution = True
        cooperative_planning = False

    # Try and load an MCP Agents file
    mcp_agents: List[McpAgentConfig] = []
    if mcp_agents_file:
        with open(mcp_agents_file) as fd:
            mcp_agents_data: Any = yaml.safe_load(fd)

        if not isinstance(mcp_agents_data, list):
            raise TypeError(
                f"Expected root element of mcp_agents_file to be a list but found: {type(mcp_agents_data)}"
            )

        for value in mcp_agents_data:  # type: ignore
            mcp_agent = McpAgentConfig.model_validate(value)
            mcp_agents.append(mcp_agent)
            logger.info(
                f"Loaded MCP Agent '{mcp_agent.name}' with MCP Servers [{', '.join(server.server_name for server in mcp_agent.mcp_servers)}]"
            )

    # Add a basic signal handler to log as soon as SIGINT is received
    def signal_handler(sig: int, frame: types.FrameType | None) -> Any:
        log_debug(f"Signal handler caught signal: {sig}", debug)
        logger.info("magentic-ui cli caught SIGINT...")
        log_debug("Raising KeyboardInterrupt to terminate application", debug)
        raise KeyboardInterrupt

    log_debug("Registering SIGINT signal handler", debug)
    signal.signal(signal.SIGINT, signal_handler)

    # Starts an asyncio event loop responsible for running asynchronous tasks
    log_debug("Starting asyncio event loop for get_team", debug)
    asyncio.run(
        # Passes the arguments to the get_team function
        get_team(
            cooperative_planning=cooperative_planning,
            autonomous_execution=autonomous_execution,
            reset=not use_state,  # Invert logic: if not using state, reset is True
            task=processed_task,
            final_answer_prompt=processed_final_answer_prompt,
            debug=debug,
            internal_workspace_root=internal_workspace_root,
            external_workspace_root=external_workspace_root,
            playwright_port=playwright_port,
            novnc_port=novnc_port,
            inside_docker=inside_docker,
            work_dir=work_dir,
            client_config=client_config,
            action_policy=cast(ApprovalPolicy, action_policy),
            user_proxy_type=user_proxy_type,
            task_metadata=metadata_task
            if user_proxy_type == "metadata"
            else processed_task,
            hints=metadata_hints if user_proxy_type == "metadata" else None,
            answer=metadata_answer if user_proxy_type == "metadata" else None,
            use_pretty_ui=use_pretty_ui,
            mcp_agents=mcp_agents,
            run_without_docker=run_without_docker,
            browser_headless=browser_headless,
            browser_local=browser_local,
            sentinel_tasks=sentinel_tasks,
            dynamic_sentinel_sleep=dynamic_sentinel_sleep,
        )
    )
    log_debug("Asyncio event loop and get_team function completed", debug)


def main() -> None:
    """
    Entry point for the magentic-cli command.
    Called from pyproject.toml's [project.scripts] section.
    """
    app()


if __name__ == "__main__":
    app()