Shannon MCP

#!/usr/bin/env python3 """ MCP Server Installation and Setup Script This script handles the installation and configuration of Shannon MCP server for integration testing within Claude's execution environment. """ import os import sys import json import asyncio import subprocess import tempfile from pathlib import Path from typing import Dict, Any, Optional import logging # Add parent directory to path sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent)) from shannon_mcp.managers.binary import BinaryManager from shannon_mcp.storage.database import Database logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class MCPServerInstaller: """Handles Shannon MCP server installation and setup for testing.""" def __init__(self, install_dir: Optional[Path] = None): self.install_dir = install_dir or Path.home() / ".shannon-mcp-test" self.config_dir = self.install_dir / "config" self.data_dir = self.install_dir / "data" self.logs_dir = self.install_dir / "logs" async def install(self) -> Dict[str, Any]: """ Install and configure Shannon MCP server. Returns: Installation details including paths and configuration """ logger.info("Starting Shannon MCP server installation") # Create directory structure self._create_directories() # Check if already installed if self._is_installed(): logger.info("Shannon MCP already installed, verifying installation...") return await self._verify_installation() # Install server install_result = await self._install_server() # Configure server config_result = await self._configure_server() # Verify installation verification = await self._verify_installation() return { "install_dir": str(self.install_dir), "config_dir": str(self.config_dir), "data_dir": str(self.data_dir), "logs_dir": str(self.logs_dir), "installation": install_result, "configuration": config_result, "verification": verification, "status": "success" if verification["is_valid"] else "failed" } def _create_directories(self): """Create necessary directory structure.""" for directory in [self.install_dir, self.config_dir, self.data_dir, self.logs_dir]: directory.mkdir(parents=True, exist_ok=True) logger.info(f"Created directory: {directory}") def _is_installed(self) -> bool: """Check if Shannon MCP is already installed.""" # Check for key files markers = [ self.install_dir / "shannon-mcp", self.config_dir / "config.yaml", self.data_dir / "shannon.db" ] return any(marker.exists() for marker in markers) async def _install_server(self) -> Dict[str, Any]: """Install Shannon MCP server.""" logger.info("Installing Shannon MCP server...") try: # For testing, we'll use the local development version project_root = Path(__file__).parent.parent.parent.parent # Create a virtual environment for isolated testing venv_path = self.install_dir / "venv" subprocess.run([ sys.executable, "-m", "venv", str(venv_path) ], check=True) # Install Shannon MCP in the virtual environment pip_path = venv_path / "bin" / "pip" subprocess.run([ str(pip_path), "install", "-e", str(project_root) ], check=True) # Create launcher script launcher_path = self.install_dir / "shannon-mcp" launcher_content = f"""#!/bin/bash source {venv_path}/bin/activate export SHANNON_MCP_CONFIG_DIR={self.config_dir} export SHANNON_MCP_DATA_DIR={self.data_dir} export SHANNON_MCP_LOG_DIR={self.logs_dir} python -m shannon_mcp "$@" """ launcher_path.write_text(launcher_content) launcher_path.chmod(0o755) return { "method": "local_development", "venv_path": str(venv_path), "launcher_path": str(launcher_path), "project_root": str(project_root) } except Exception as e: logger.error(f"Installation failed: {e}") return {"error": str(e), "status": "failed"} async def _configure_server(self) -> Dict[str, Any]: """Configure Shannon MCP server for testing.""" logger.info("Configuring Shannon MCP server...") # Create test configuration config = { "server": { "host": "127.0.0.1", "port": 8765, "enable_ssl": False }, "claude_code": { "discovery_timeout": 30, "update_check_interval": 86400 }, "sessions": { "defaults": { "model": "claude-3-opus-20240229", "temperature": 0.7, "max_tokens": 4000 }, "timeout": 3600, "max_concurrent": 10 }, "storage": { "base_dir": str(self.data_dir), "database": { "path": "shannon.db", "wal_mode": True }, "cas": { "path": "cas", "compression": True }, "checkpoints": { "path": "checkpoints", "max_checkpoints": 100 } }, "agents": { "max_agents": 50, "timeout": 600, "collaboration": True }, "hooks": { "enabled": True, "directory": str(self.config_dir / "hooks"), "timeout": 30, "security": { "sandboxing": True, "allowed_commands": ["echo", "cat", "ls", "test"], "blocked_paths": ["/etc/passwd", "/etc/shadow"] } }, "logging": { "level": "DEBUG", "file": str(self.logs_dir / "shannon-mcp.log"), "rotation": { "enabled": True, "max_size": "10MB", "backup_count": 5 } }, "testing": { "enabled": True, "allow_destructive_operations": True, "test_data_dir": str(self.data_dir / "test-data") } } # Write configuration config_path = self.config_dir / "config.yaml" with open(config_path, 'w') as f: import yaml yaml.dump(config, f, default_flow_style=False) # Create hooks directory hooks_dir = self.config_dir / "hooks" hooks_dir.mkdir(exist_ok=True) # Create test hook for validation test_hook = hooks_dir / "test-hook.sh" test_hook.write_text("""#!/bin/bash echo "Test hook executed at $(date)" >> /tmp/shannon-mcp-hook-test.log echo "Session: $SESSION_ID" >> /tmp/shannon-mcp-hook-test.log """) test_hook.chmod(0o755) return { "config_path": str(config_path), "hooks_dir": str(hooks_dir), "test_hook": str(test_hook), "settings": config } async def _verify_installation(self) -> Dict[str, Any]: """Verify Shannon MCP installation is working.""" logger.info("Verifying Shannon MCP installation...") verification_results = { "directories_exist": all([ self.install_dir.exists(), self.config_dir.exists(), self.data_dir.exists(), self.logs_dir.exists() ]), "config_valid": (self.config_dir / "config.yaml").exists(), "launcher_exists": (self.install_dir / "shannon-mcp").exists(), "binary_discovery": False, "database_accessible": False, "server_responsive": False } # Test binary discovery try: db_path = self.data_dir / "shannon.db" db = Database(db_path) await db.initialize() binary_manager = BinaryManager(db=db) await binary_manager.initialize() binaries = await binary_manager.discover_binaries() verification_results["binary_discovery"] = len(binaries) > 0 verification_results["discovered_binaries"] = [ {"path": str(b.path), "version": b.version} for b in binaries ] await binary_manager.cleanup() await db.close() verification_results["database_accessible"] = True except Exception as e: logger.error(f"Binary discovery failed: {e}") verification_results["binary_discovery_error"] = str(e) # Test server startup (quick check) try: launcher = self.install_dir / "shannon-mcp" if launcher.exists(): # Just check if the command runs without error result = subprocess.run( [str(launcher), "--version"], capture_output=True, text=True, timeout=5 ) verification_results["server_responsive"] = result.returncode == 0 verification_results["server_version"] = result.stdout.strip() except Exception as e: logger.error(f"Server check failed: {e}") verification_results["server_error"] = str(e) # Overall validation verification_results["is_valid"] = all([ verification_results["directories_exist"], verification_results["config_valid"], verification_results["launcher_exists"], verification_results["database_accessible"] ]) return verification_results async def uninstall(self): """Remove Shannon MCP test installation.""" logger.info(f"Removing Shannon MCP test installation from {self.install_dir}") if self.install_dir.exists() and str(self.install_dir).endswith("-test"): import shutil shutil.rmtree(self.install_dir) logger.info("Uninstallation complete") else: logger.warning("Safety check failed - not removing directory") async def main(): """Main installation entry point.""" installer = MCPServerInstaller() # Run installation result = await installer.install() # Pretty print results print("\n" + "="*60) print("Shannon MCP Server Installation Results") print("="*60) print(json.dumps(result, indent=2)) if result["status"] == "success": print("\n✅ Installation successful!") print(f"\nTo start the server:") print(f" {result['install_dir']}/shannon-mcp serve") print(f"\nConfiguration: {result['config_dir']}/config.yaml") print(f"Logs: {result['logs_dir']}/shannon-mcp.log") else: print("\n❌ Installation failed!") sys.exit(1) if __name__ == "__main__": asyncio.run(main())