crewAI/lib/cli/src/crewai_cli/create_crew.py

from pathlib import Path
import shutil
import sys

import click
import tomli

from crewai_cli.constants import ENV_VARS, MODELS
from crewai_cli.git import initialize_if_git_available
from crewai_cli.provider import (
    get_provider_data,
    select_model,
    select_provider,
)
from crewai_cli.utils import (
    copy_template,
    get_or_create_project_id,
    is_dmn_mode_enabled,
    load_env_vars,
    write_env_file,
)


def get_reserved_script_names() -> set[str]:
    """Get reserved script names from pyproject.toml template.

    Returns:
        Set of reserved script names that would conflict with crew folder names.
    """
    package_dir = Path(__file__).parent
    template_path = package_dir / "templates" / "crew" / "pyproject.toml"

    with open(template_path, "r") as f:
        template_content = f.read()

    template_content = template_content.replace("{{folder_name}}", "_placeholder_")
    template_content = template_content.replace("{{name}}", "placeholder")
    template_content = template_content.replace("{{crew_name}}", "Placeholder")

    template_data = tomli.loads(template_content)
    script_names = set(template_data.get("project", {}).get("scripts", {}).keys())
    script_names.discard("_placeholder_")
    return script_names


def create_folder_structure(
    name: str, parent_folder: str | None = None
) -> tuple[Path, str, str]:
    import keyword
    import re

    name = name.rstrip("/")

    if not name.strip():
        raise ValueError("Project name cannot be empty or contain only whitespace")

    folder_name = name.replace(" ", "_").replace("-", "_").lower()
    folder_name = re.sub(r"[^a-zA-Z0-9_]", "", folder_name)

    if re.match(r"^[^a-zA-Z0-9_-]+", name):
        raise ValueError(
            f"Project name '{name}' contains no valid characters for a Python module name"
        )

    if not folder_name:
        raise ValueError(
            f"Project name '{name}' contains no valid characters for a Python module name"
        )

    if folder_name[0].isdigit():
        raise ValueError(
            f"Project name '{name}' would generate folder name '{folder_name}' which cannot start with a digit (invalid Python module name)"
        )

    if keyword.iskeyword(folder_name):
        raise ValueError(
            f"Project name '{name}' would generate folder name '{folder_name}' which is a reserved Python keyword"
        )

    if not folder_name.isidentifier():
        raise ValueError(
            f"Project name '{name}' would generate invalid Python module name '{folder_name}'"
        )

    reserved_names = get_reserved_script_names()
    if folder_name in reserved_names:
        raise ValueError(
            f"Project name '{name}' would generate folder name '{folder_name}' which is reserved. "
            f"Reserved names are: {', '.join(sorted(reserved_names))}. "
            "Please choose a different name."
        )

    class_name = name.replace("_", " ").replace("-", " ").title().replace(" ", "")

    class_name = re.sub(r"[^a-zA-Z0-9_]", "", class_name)

    if not class_name:
        raise ValueError(
            f"Project name '{name}' contains no valid characters for a Python class name"
        )

    if class_name[0].isdigit():
        raise ValueError(
            f"Project name '{name}' would generate class name '{class_name}' which cannot start with a digit"
        )

    original_name_clean = re.sub(
        r"[^a-zA-Z0-9_]", "", name.replace("_", "").replace("-", "").lower()
    )
    if (
        keyword.iskeyword(original_name_clean)
        or keyword.iskeyword(class_name)
        or class_name in ("True", "False", "None")
    ):
        raise ValueError(
            f"Project name '{name}' would generate class name '{class_name}' which is a reserved Python keyword"
        )

    if not class_name.isidentifier():
        raise ValueError(
            f"Project name '{name}' would generate invalid Python class name '{class_name}'"
        )

    if parent_folder:
        folder_path = Path(parent_folder) / folder_name
    else:
        folder_path = Path(folder_name)

    if folder_path.exists():
        if is_dmn_mode_enabled():
            raise click.ClickException(f"Folder {folder_name} already exists.")
        if not click.confirm(
            f"Folder {folder_name} already exists. Do you want to override it?"
        ):
            click.secho("Operation cancelled.", fg="yellow")
            sys.exit(0)
        click.secho(f"Overriding folder {folder_name}...", fg="green", bold=True)
        shutil.rmtree(folder_path)

    click.secho(
        f"Creating {'crew' if parent_folder else 'folder'} {folder_name}...",
        fg="green",
        bold=True,
    )

    folder_path.mkdir(parents=True)
    (folder_path / "tests").mkdir(exist_ok=True)
    (folder_path / "knowledge").mkdir(exist_ok=True)
    if not parent_folder:
        (folder_path / "src" / folder_name).mkdir(parents=True)
        (folder_path / "src" / folder_name / "tools").mkdir(parents=True)
        (folder_path / "src" / folder_name / "config").mkdir(parents=True)

        package_dir = Path(__file__).parent
        agents_md_src = package_dir / "templates" / "AGENTS.md"
        if agents_md_src.exists():
            shutil.copy2(agents_md_src, folder_path / "AGENTS.md")

    return folder_path, folder_name, class_name


def copy_template_files(
    folder_path: Path, name: str, class_name: str, parent_folder: str | None
) -> None:
    package_dir = Path(__file__).parent
    templates_dir = package_dir / "templates" / "crew"

    root_template_files = (
        [
            ".gitignore",
            "pyproject.toml",
            "README.md",
            "knowledge/user_preference.txt",
        ]
        if not parent_folder
        else []
    )
    tools_template_files = ["tools/custom_tool.py", "tools/__init__.py"]
    config_template_files = ["config/agents.yaml", "config/tasks.yaml"]
    src_template_files = (
        ["__init__.py", "main.py", "crew.py"] if not parent_folder else ["crew.py"]
    )

    for file_name in root_template_files:
        src_file = templates_dir / file_name
        dst_file = folder_path / file_name
        copy_template(src_file, dst_file, name, class_name, folder_path.name)

    src_folder = (
        folder_path / "src" / folder_path.name if not parent_folder else folder_path
    )

    for file_name in src_template_files:
        src_file = templates_dir / file_name
        dst_file = src_folder / file_name
        copy_template(src_file, dst_file, name, class_name, folder_path.name)

    if not parent_folder:
        for file_name in tools_template_files + config_template_files:
            src_file = templates_dir / file_name
            dst_file = src_folder / file_name
            copy_template(src_file, dst_file, name, class_name, folder_path.name)


def create_crew(
    name: str,
    provider: str | None = None,
    skip_provider: bool = False,
    parent_folder: str | None = None,
) -> None:
    folder_path, folder_name, class_name = create_folder_structure(name, parent_folder)
    env_vars = load_env_vars(folder_path)
    if is_dmn_mode_enabled():
        skip_provider = True
    if not skip_provider:
        if not provider:
            provider_models = get_provider_data()
            if not provider_models:
                return

        existing_provider = None
        for provider, env_keys in ENV_VARS.items():
            if any(
                "key_name" in details and details["key_name"] in env_vars
                for details in env_keys
            ):
                existing_provider = provider
                break

        if existing_provider:
            if not click.confirm(
                f"Found existing environment variable configuration for {existing_provider.capitalize()}. Do you want to override it?"
            ):
                click.secho("Keeping existing provider configuration.", fg="yellow")
                return

        provider_models = get_provider_data()
        if not provider_models:
            return

        while True:
            selected_provider = select_provider(provider_models)
            if selected_provider is None:
                click.secho("Exiting...", fg="yellow")
                sys.exit(0)
            if selected_provider and isinstance(selected_provider, str):
                break
            click.secho(
                "No provider selected. Please try again or press 'q' to exit.", fg="red"
            )

        if MODELS.get(selected_provider):
            while True:
                selected_model = select_model(selected_provider, provider_models)
                if selected_model is None:
                    click.secho("Exiting...", fg="yellow")
                    sys.exit(0)
                if selected_model:
                    break
                click.secho(
                    "No model selected. Please try again or press 'q' to exit.",
                    fg="red",
                )
            env_vars["MODEL"] = selected_model

        if selected_provider in ENV_VARS:
            provider_env_vars = ENV_VARS[selected_provider]
            for details in provider_env_vars:
                if details.get("default", False):
                    for key, value in details.items():
                        if key not in ["prompt", "key_name", "default"]:
                            env_vars[key] = value
                elif "key_name" in details:
                    prompt = details["prompt"]
                    key_name = details["key_name"]
                    api_key_value = click.prompt(prompt, default="", show_default=False)

                    if api_key_value.strip():
                        env_vars[key_name] = api_key_value

        if env_vars:
            write_env_file(folder_path, env_vars)
            click.secho("API keys and model saved to .env file", fg="green")
        else:
            click.secho(
                "No API keys provided. Skipping .env file creation.", fg="yellow"
            )

        click.secho(f"Selected model: {env_vars.get('MODEL', 'N/A')}", fg="green")

    package_dir = Path(__file__).parent
    templates_dir = package_dir / "templates" / "crew"

    root_template_files = (
        [".gitignore", "pyproject.toml", "README.md", "knowledge/user_preference.txt"]
        if not parent_folder
        else []
    )
    tools_template_files = ["tools/custom_tool.py", "tools/__init__.py"]
    config_template_files = ["config/agents.yaml", "config/tasks.yaml"]
    src_template_files = (
        ["__init__.py", "main.py", "crew.py"] if not parent_folder else ["crew.py"]
    )

    for file_name in root_template_files:
        src_file = templates_dir / file_name
        dst_file = folder_path / file_name
        copy_template(src_file, dst_file, name, class_name, folder_name)

    src_folder = folder_path / "src" / folder_name if not parent_folder else folder_path

    for file_name in src_template_files:
        src_file = templates_dir / file_name
        dst_file = src_folder / file_name
        copy_template(src_file, dst_file, name, class_name, folder_name)

    if not parent_folder:
        for file_name in tools_template_files + config_template_files:
            src_file = templates_dir / file_name
            dst_file = src_folder / file_name
            copy_template(src_file, dst_file, name, class_name, folder_name)

    if not parent_folder:
        # Minted at creation so the project has a stable identity from run one.
        get_or_create_project_id(folder_path / "pyproject.toml")
        initialize_if_git_available(folder_path)

    click.secho(f"Crew {name} created successfully!", fg="green", bold=True)