Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 36 additions & 11 deletions src/mcpm/commands/install.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
from mcpm.profile.profile_config import ProfileConfigManager
from mcpm.schemas.full_server_config import FullServerConfig
from mcpm.utils.config import NODE_EXECUTABLES, ConfigManager
from mcpm.utils.non_interactive import is_explicit_non_interactive, should_force_operation
from mcpm.utils.repository import RepositoryManager
from mcpm.utils.rich_click_config import click

Expand Down Expand Up @@ -88,26 +89,44 @@ def _replace_node_executable(server_config: ServerConfig) -> ServerConfig:
def global_add_server(server_config: ServerConfig, force: bool = False) -> bool:
"""Add a server to the global MCPM configuration."""
if global_config_manager.server_exists(server_config.name) and not force:
console.print(f"[bold red]Error:[/] Server '{server_config.name}' already exists in global configuration.")
console.print(f"[bold red]Error:[/ ] Server '{server_config.name}' already exists in global configuration.")
console.print("Use --force to override.")
return False

server_config = _replace_node_executable(server_config)
return global_config_manager.add_server(server_config, force)


def prompt_with_default(prompt_text, default="", hide_input=False, required=False):
def prompt_with_default(prompt_text, default="", hide_input=False, required=False, force=False):
"""Prompt the user with a default value that can be edited directly.

In non-interactive mode (via --force or MCPM_NON_INTERACTIVE), this function
returns the default value immediately. If a required value has no default
in non-interactive mode, it raises click.UsageError.

Args:
prompt_text: The prompt text to display
default: The default value to show in the prompt
hide_input: Whether to hide the input (for passwords)
required: Whether this is a required field
force: Whether to force non-interactive mode

Returns:
The user's input or the default value if empty
"""
# Check for explicit non-interactive mode (Env Var) or Force flag
# We use is_explicit_non_interactive() instead of the broader is_non_interactive()
# because the latter includes isatty() checks. In test environments using CliRunner,
# isatty() returns True, which would incorrectly skip our mocked prompts if we checked it here.
# We specifically want to allow interaction in tests unless the Env Var is set.
if is_explicit_non_interactive() or should_force_operation(force):
if default:
return default
if required:
# Cannot fulfill required argument without default in non-interactive mode
raise click.UsageError("A required value has no default and cannot be prompted in non-interactive mode.")
return ""

# if default:
# console.print(f"Default: [yellow]{default}[/]")

Expand Down Expand Up @@ -142,7 +161,7 @@ def prompt_with_default(prompt_text, default="", hide_input=False, required=Fals
# Empty result for required field without default is not allowed
if not result.strip() and required and not default:
console.print("[yellow]Warning: Required value cannot be empty.[/]")
return prompt_with_default(prompt_text, default, hide_input, required)
return prompt_with_default(prompt_text, default, hide_input, required, force)

return result
except (KeyboardInterrupt, EOFError):
Expand Down Expand Up @@ -171,12 +190,13 @@ def install(server_name, force=False, alias=None):
config_name = alias or server_name

# All servers are installed to global configuration
console.print("[yellow]Installing server to global configuration...[/]")
console_stderr = Console(stderr=True)
console_stderr.print("[yellow]Installing server to global configuration...[/]")

# Get server metadata from repository
server_metadata = repo_manager.get_server_metadata(server_name)
if not server_metadata:
console.print(f"[bold red]Error:[/] Server '{server_name}' not found in registry.")
console.print(f"[bold red]Error:[/ ] Server '{server_name}' not found in registry.")
console.print(f"Available servers: {', '.join(repo_manager._fetch_servers().keys())}")
return

Expand All @@ -195,7 +215,10 @@ def install(server_name, force=False, alias=None):

# Confirm addition
alias_text = f" as '{alias}'" if alias else ""
if not force and not Confirm.ask(f"Install this server to global configuration{alias_text}?"):
# Bypass confirmation if force flag is set OR explicit non-interactive mode is enabled
if not (should_force_operation(force) or is_explicit_non_interactive()) and not Confirm.ask(
f"Install this server to global configuration{alias_text}?"
):
console.print("[yellow]Operation cancelled.[/]")
return

Expand Down Expand Up @@ -239,8 +262,8 @@ def install(server_name, force=False, alias=None):
method_id = next(iter(installations))
selected_method = installations[method_id]

# If multiple methods are available and not forced, offer selection
if len(installations) > 1 and not force:
# If multiple methods are available and not forced/non-interactive, offer selection
if len(installations) > 1 and not (should_force_operation(force) or is_explicit_non_interactive()):
console.print("\n[bold]Available installation methods:[/]")
methods_list = []

Expand Down Expand Up @@ -349,6 +372,7 @@ def install(server_name, force=False, alias=None):
default=env_value,
hide_input=_should_hide_input(arg_name),
required=is_required,
force=force,
)
if user_value != env_value:
# User provided a different value
Expand All @@ -366,6 +390,7 @@ def install(server_name, force=False, alias=None):
default=example if example else "",
hide_input=_should_hide_input(arg_name),
required=is_required,
force=force,
)

# Only add non-empty values to the environment
Expand Down Expand Up @@ -419,7 +444,7 @@ def install(server_name, force=False, alias=None):
if has_non_standard_argument_define:
# no matter in argument / env
console.print(
"[bold yellow]WARNING:[/] [bold]Non-standard argument format detected in server configuration.[/]\n"
"[bold yellow]WARNING:[/ ] [bold]Non-standard argument format detected in server configuration.[/]\n"
"[bold cyan]Future versions of MCPM will standardize all arguments in server configuration to use ${VARIABLE_NAME} format.[/]\n"
"[bold]Please verify that your input arguments are correctly recognized.[/]\n"
)
Expand Down Expand Up @@ -460,7 +485,7 @@ def install(server_name, force=False, alias=None):
)

# Add server to global configuration
success = global_add_server(full_server_config.to_server_config(), force)
success = global_add_server(full_server_config.to_server_config(), should_force_operation(force))

if success:
# Server has been successfully added to the global configuration
Expand Down Expand Up @@ -523,7 +548,7 @@ def extract_from_value(value):

# Check all fields in the installation method
for key, value in installation_method.items():
if key not in ["type", "description", "recommended"]: # Skip metadata fields
if key not in ["type", "description", "recommended"]:
extract_from_value(value)

return referenced
Expand Down
19 changes: 10 additions & 9 deletions src/mcpm/commands/uninstall.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@

from mcpm.global_config import GlobalConfigManager
from mcpm.utils.display import print_server_config
from mcpm.utils.non_interactive import is_explicit_non_interactive, should_force_operation
from mcpm.utils.rich_click_config import click

console = Console()
Expand All @@ -17,14 +18,14 @@ def global_get_server(server_name: str):
"""Get a server from the global MCPM configuration."""
server = global_config_manager.get_server(server_name)
if not server:
console.print(f"[bold red]Error:[/] Server '{server_name}' not found in global configuration.")
console.print(f"[bold red]Error:[/ ] Server '{server_name}' not found in global configuration.")
return server


def global_remove_server(server_name: str) -> bool:
"""Remove a server from the global MCPM configuration and clean up profile tags."""
if not global_config_manager.server_exists(server_name):
console.print(f"[bold red]Error:[/] Server '{server_name}' not found in global configuration.")
console.print(f"[bold red]Error:[/ ] Server '{server_name}' not found in global configuration.")
return False

# Remove from global config (this automatically removes all profile tags)
Expand Down Expand Up @@ -54,13 +55,13 @@ def uninstall(server_name, force):
return # Error message already printed by global_get_server

# Display server information before removal
console.print(f"\n[bold cyan]Server information for:[/] {server_name}")
console.print(f"\n[bold cyan]Server information for:[/ ] {server_name}")

print_server_config(server_info)

# Get confirmation if --force is not used
if not force:
console.print(f"\n[bold yellow]Are you sure you want to remove:[/] {server_name}")
# Get confirmation if --force is not used and not in non-interactive mode
if not (should_force_operation(force) or is_explicit_non_interactive()):
console.print(f"\n[bold yellow]Are you sure you want to remove:[/ ] {server_name}")
console.print("[italic]To bypass this confirmation, use --force[/]")
# Use Rich's Confirm for a better user experience
confirmed = Confirm.ask("Proceed with removal?")
Expand All @@ -69,13 +70,13 @@ def uninstall(server_name, force):
return

# Log the removal action
console.print(f"[bold red]Removing MCP server from global configuration:[/] {server_name}")
console.print(f"[bold red]Removing MCP server from global configuration:[/ ] {server_name}")

# Remove from global configuration
success = global_remove_server(server_name)

if success:
console.print(f"[green]Successfully removed server:[/] {server_name}")
console.print(f"[green]Successfully removed server:[/ ] {server_name}")
console.print("[dim]Note: Server has been removed from global config. Profile tags are also cleared.[/]")
else:
console.print(f"[bold red]Error:[/] Failed to remove server '{server_name}'.")
console.print(f"[bold red]Error:[/ ] Failed to remove server '{server_name}'.")
27 changes: 22 additions & 5 deletions src/mcpm/utils/non_interactive.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,19 @@
from typing import Dict, List, Optional


def is_explicit_non_interactive() -> bool:
"""
Check if non-interactive mode is explicitly enabled via environment variable.

This excludes implicit detection (like isatty) to avoid issues in tests or
environments where TTY detection behaves unexpectedly but automation is not desired.

Returns:
True if MCPM_NON_INTERACTIVE environment variable is set to 'true'
"""
return os.getenv("MCPM_NON_INTERACTIVE", "").lower() == "true"


def is_non_interactive() -> bool:
"""
Check if running in non-interactive mode.
Expand All @@ -17,7 +30,7 @@ def is_non_interactive() -> bool:
- Running in a CI environment
"""
# Check explicit non-interactive flag
if os.getenv("MCPM_NON_INTERACTIVE", "").lower() == "true":
if is_explicit_non_interactive():
return True

# Check if not connected to a TTY
Expand All @@ -32,13 +45,17 @@ def is_non_interactive() -> bool:
return False


def should_force_operation() -> bool:
def should_force_operation(cli_force_flag: bool = False) -> bool:
"""
Check if operations should be forced (skip confirmations).

Returns True if MCPM_FORCE environment variable is set to 'true'.
Args:
cli_force_flag: Boolean flag from CLI args (e.g. --force)

Returns:
True if cli_force_flag is True OR MCPM_FORCE environment variable is set to 'true'.
"""
return os.getenv("MCPM_FORCE", "").lower() == "true"
return cli_force_flag or os.getenv("MCPM_FORCE", "").lower() == "true"


def should_output_json() -> bool:
Expand Down Expand Up @@ -304,4 +321,4 @@ def merge_server_config_updates(
else:
updated_config["headers"] = new_headers

return updated_config
return updated_config
Loading