diff --git a/src/mcp_server_cli/main.py b/src/mcp_server_cli/main.py new file mode 100644 index 0000000..e2e16f3 --- /dev/null +++ b/src/mcp_server_cli/main.py @@ -0,0 +1,237 @@ +"""Command-line interface for MCP Server CLI using Click.""" + +import sys +import os +from pathlib import Path +from typing import Optional +import logging + +import click +from click.core import Context + +from mcp_server_cli.config import ConfigManager, load_config_from_path, create_config_template +from mcp_server_cli.server import run_server, create_app +from mcp_server_cli.tools import FileTools, GitTools, ShellTools + + +@click.group() +@click.version_option(version="0.1.0") +@click.option( + "--config", + "-c", + type=click.Path(exists=True), + help="Path to configuration file", +) +@click.pass_context +def main(ctx: Context, config: Optional[str]): + """MCP Server CLI - A local Model Context Protocol server.""" + ctx.ensure_object(dict) + ctx.obj["config_path"] = config + + +@main.group() +def server(): + """Server management commands.""" + pass + + +@server.command("start") +@click.option( + "--host", + "-H", + default="127.0.0.1", + help="Host to bind to", +) +@click.option( + "--port", + "-p", + default=3000, + type=int, + help="Port to listen on", +) +@click.option( + "--log-level", + "-l", + default="INFO", + type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR"]), + help="Logging level", +) +@click.pass_context +def server_start(ctx: Context, host: str, port: int, log_level: str): + """Start the MCP server.""" + config_path = ctx.obj.get("config_path") + config = None + + if config_path: + try: + config = load_config_from_path(config_path) + host = config.server.host + port = config.server.port + except Exception as e: + click.echo(f"Warning: Failed to load config: {e}", err=True) + + click.echo(f"Starting MCP server on {host}:{port}...") + run_server(host=host, port=port, log_level=log_level) + + +@server.command("status") +@click.pass_context +def server_status(ctx: Context): + """Check server status.""" + config_path = ctx.obj.get("config_path") + if config_path: + try: + config = load_config_from_path(config_path) + click.echo(f"Server configured on {config.server.host}:{config.server.port}") + return + except Exception: + pass + click.echo("Server configuration not running (check config file)") + + +@server.command("stop") +@click.pass_context +def server_stop(ctx: Context): + """Stop the server.""" + click.echo("Server stopped (not running in foreground)") + + +@main.group() +def tool(): + """Tool management commands.""" + pass + + +@tool.command("list") +@click.pass_context +def tool_list(ctx: Context): + """List available tools.""" + from mcp_server_cli.server import MCPServer + + server = MCPServer() + server.register_tool(FileTools()) + server.register_tool(GitTools()) + server.register_tool(ShellTools()) + + tools = server.list_tools() + if tools: + click.echo("Available tools:") + for tool in tools: + click.echo(f" - {tool.name}: {tool.description}") + else: + click.echo("No tools registered") + + +@tool.command("add") +@click.argument("tool_file", type=click.Path(exists=True)) +@click.pass_context +def tool_add(ctx: Context, tool_file: str): + """Add a custom tool from YAML/JSON file.""" + click.echo(f"Adding tool from {tool_file}") + + +@tool.command("remove") +@click.argument("tool_name") +@click.pass_context +def tool_remove(ctx: Context, tool_name: str): + """Remove a custom tool.""" + click.echo(f"Removing tool {tool_name}") + + +@main.group() +def config(): + """Configuration management commands.""" + pass + + +@config.command("show") +@click.pass_context +def config_show(ctx: Context): + """Show current configuration.""" + config_path = ctx.obj.get("config_path") + if config_path: + try: + config = load_config_from_path(config_path) + import json + click.echo(config.model_dump_json(indent=2)) + return + except Exception as e: + click.echo(f"Error loading config: {e}", err=True) + + config_manager = ConfigManager() + default_config = config_manager.generate_default_config() + import json + click.echo("Default configuration:") + click.echo(default_config.model_dump_json(indent=2)) + + +@config.command("init") +@click.option( + "--output", + "-o", + type=click.Path(), + default="config.yaml", + help="Output file path", +) +@click.pass_context +def config_init(ctx: Context, output: str): + """Initialize a new configuration file.""" + template = create_config_template() + path = Path(output) + + with open(path, "w") as f: + import yaml + yaml.dump(template, f, default_flow_style=False, indent=2) + + click.echo(f"Configuration written to {output}") + + +@main.command("health") +@click.pass_context +def health_check(ctx: Context): + """Check server health.""" + import httpx + + config_path = ctx.obj.get("config_path") + port = 3000 + + if config_path: + try: + config = load_config_from_path(config_path) + port = config.server.port + except Exception: + pass + + try: + response = httpx.get(f"http://127.0.0.1:{port}/health", timeout=5) + if response.status_code == 200: + data = response.json() + click.echo(f"Server status: {data.get('state', 'unknown')}") + else: + click.echo("Server not responding", err=True) + except httpx.RequestError: + click.echo("Server not running", err=True) + + +@main.command("install") +@click.pass_context +def install_completions(ctx: Context): + """Install shell completions.""" + shell = os.environ.get("SHELL", "") + if "bash" in shell: + from click import _bashcomplete + _bashcomplete.bashcomplete(main, assimilate=False, cli=ctx.command) + click.echo("Bash completions installed") + elif "zsh" in shell: + click.echo("Zsh completions: add 'eval \"$(register-python-argcomplete mcp-server)\"' to .zshrc") + else: + click.echo("Unsupported shell for auto-completion") + + +def cli_entry_point(): + """Entry point for the CLI.""" + main(obj={}) + + +if __name__ == "__main__": + cli_entry_point()