321 lines
11 KiB
Python
Executable file
321 lines
11 KiB
Python
Executable file
#!/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 pod_executor.containers.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 pod_executor.security.resource_limits import ResourceLimits
|
|
from mcp_forge.adapters 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, '<stdin>', '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()
|