#!/usr/bin/env python3 """ MCP-Forge CLI - Interactive Python REPL with MCP Tools A standalone command-line interface for testing and using MCP-Forge backends: - Simple executor (stateless) - Jupyter kernel (stateful) With MCP tools injected for enhanced capabilities. Usage: python mcp_forge_cli.py # Interactive mode python mcp_forge_cli.py --jupyter # Use Jupyter backend python mcp_forge_cli.py --execute "code" # Execute and exit """ import argparse import sys import os from pathlib import Path from typing import Optional # Add src to path sys.path.insert(0, str(Path(__file__).parent / "src")) from mcp_forge.config.loader import load_config from mcp_forge.podman.client import PodmanClient from mcp_forge.podman.containers import SecureContainerManager from mcp_forge.security.audit import AuditLogger from mcp_forge.security.allowlist import AllowlistValidator from mcp_forge.security.resource_limits import ResourceLimits from mcp_forge.execution.simple.backend import SimpleBackend from mcp_forge.execution.jupyter.backend import JupyterBackend class MCPForgeCLI: """Interactive CLI for MCP-Forge.""" def __init__(self, use_jupyter: bool = False, config_path: Optional[str] = None): """ Initialize CLI. Args: use_jupyter: Use Jupyter backend instead of simple executor config_path: Path to config file (default: config.yaml) """ self.use_jupyter = use_jupyter self.session_id = "cli-session" # Load configuration config_file = config_path or "config.yaml" if not Path(config_file).exists(): config_file = "config.example.yaml" print(f"Loading config from: {config_file}") self.config = load_config(config_file) # Initialize components self.audit_store = MemoryAuditStore() self.audit_logger = AuditLogger(self.audit_store) self.podman_client = PodmanClient() allowlist_manager = AllowlistManager(self.config.allowlist_file) self.validator = OperationValidator(allowlist_manager) self.container_manager = SecureContainerManager( podman_client=self.podman_client, validator=self.validator, audit_logger=self.audit_logger ) # Initialize backend if use_jupyter: print("🚀 Initializing Jupyter backend (stateful)...") self.backend = JupyterBackend( config=self.config, container_manager=self.container_manager, audit_logger=self.audit_logger ) self.backend_name = "Jupyter" else: print("🚀 Initializing Simple backend (stateless)...") self.backend = SimpleBackend( config=self.config, container_manager=self.container_manager, audit_logger=self.audit_logger ) self.backend_name = "Simple" print(f"✓ {self.backend_name} backend ready") print() def execute_code(self, code: str, show_result: bool = True) -> dict: """ Execute Python code. Args: code: Python code to execute show_result: Whether to print the result Returns: Execution result dictionary """ if self.use_jupyter: # Stateful execution with session result = self.backend.execute( code=code, session_id=self.session_id ) else: # Stateless execution result = self.backend.execute(code=code) if show_result: self._display_result(result) return result def _display_result(self, result): """Display execution result.""" if result.stdout: print(result.stdout, end='') if result.stderr: print(f"\033[31m{result.stderr}\033[0m", end='') if result.result and result.result != 'None': print(f"\033[32m{result.result}\033[0m") if not result.success and result.error: print(f"\033[31mError: {result.error}\033[0m") if result.execution_time > 0.1: print(f"\033[90m({result.execution_time:.3f}s)\033[0m") def repl(self): """Run interactive REPL.""" print("=" * 60) print(f" MCP-Forge CLI - {self.backend_name} Backend") print("=" * 60) print() print("Type Python code and press Enter to execute.") print("Special commands:") print(" .exit, .quit - Exit") print(" .clear - Clear screen") print(" .help - Show this help") if self.use_jupyter: print(" .restart - Restart kernel") print(" .vars - Show variables") print() print("Tip: Code execution is " + ("stateful (variables persist)" if self.use_jupyter else "stateless")) print() buffer = [] try: while True: try: if buffer: prompt = "... " else: prompt = ">>> " line = input(prompt) # Handle special commands if not buffer and line.startswith('.'): if line in ['.exit', '.quit']: print("\nGoodbye!") break elif line == '.clear': os.system('clear' if os.name != 'nt' else 'cls') continue elif line == '.help': self.repl() return elif line == '.restart' and self.use_jupyter: self._restart_session() continue elif line == '.vars' and self.use_jupyter: self._show_variables() continue else: print(f"Unknown command: {line}") continue # Build multi-line code buffer.append(line) # Check if we should execute code = '\n'.join(buffer) # Simple heuristic: execute if line is complete # (doesn't end with :, \, or is blank after content) if line and not line.rstrip().endswith((':', '\\')): try: compile(code, '', 'exec') # Code is complete, execute it self.execute_code(code) buffer = [] except SyntaxError as e: if 'unexpected EOF' in str(e): # Need more input continue else: # Real syntax error self.execute_code(code) buffer = [] elif not line and buffer: # Empty line after content, execute self.execute_code(code) buffer = [] except KeyboardInterrupt: print("\nKeyboardInterrupt") buffer = [] continue except EOFError: print("\nGoodbye!") break finally: self._cleanup() def _restart_session(self): """Restart Jupyter session.""" if self.use_jupyter: try: self.backend.session_manager.close_session(self.session_id) print("✓ Session restarted") except Exception as e: print(f"Error restarting session: {e}") def _show_variables(self): """Show variables in Jupyter session.""" if self.use_jupyter: try: session = self.backend.session_manager.get_session(self.session_id) kernel_id = session.kernel_id variables = self.backend.session_manager.kernel_manager.inspect_namespace(kernel_id) if variables: print("Variables:") for var in variables: print(f" - {var}") else: print("No variables defined") except Exception as e: print(f"Error showing variables: {e}") def _cleanup(self): """Cleanup resources.""" print("\nCleaning up...") if self.use_jupyter: try: self.backend.session_manager.close_session(self.session_id) except Exception: pass def main(): """Main entry point.""" parser = argparse.ArgumentParser( description="MCP-Forge CLI - Interactive Python with MCP tools", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: %(prog)s # Simple backend, interactive %(prog)s --jupyter # Jupyter backend, interactive %(prog)s --execute "print(2+2)" # Execute code and exit %(prog)s -j -e "x=5; print(x)" # Jupyter backend, execute code """ ) parser.add_argument( '-j', '--jupyter', action='store_true', help='Use Jupyter backend (stateful) instead of simple backend' ) parser.add_argument( '-e', '--execute', metavar='CODE', help='Execute code and exit (non-interactive)' ) parser.add_argument( '-c', '--config', metavar='FILE', help='Path to config file (default: config.yaml)' ) args = parser.parse_args() try: cli = MCPForgeCLI( use_jupyter=args.jupyter, config_path=args.config ) if args.execute: # Execute code and exit cli.execute_code(args.execute) else: # Interactive mode cli.repl() except KeyboardInterrupt: print("\nInterrupted") sys.exit(1) except Exception as e: print(f"\nError: {e}", file=sys.stderr) import traceback traceback.print_exc() sys.exit(1) if __name__ == '__main__': main()