Agent Knowledge MCP

Instructions

Implementation Reference

• src/admin/admin_server.py:808-1057 (handler)

  The primary handler for the 'server_upgrade' tool. Performs uvx-based server upgrade, creates config backup, intelligently merges user settings with new defaults, reinitializes security and Elasticsearch, and provides detailed status reports.
  @app.tool( description="Upgrade this MCP server when installed via uvx with automatic configuration backup and restoration", tags={"admin", "server", "upgrade", "uvx", "maintenance"} ) async def server_upgrade() -> str: """Upgrade the MCP server when installed via uvx with comprehensive backup and restoration.""" try: # Check if uv is available try: result = subprocess.run(["uv", "--version"], capture_output=True, check=True, timeout=10) except (subprocess.CalledProcessError, FileNotFoundError): return "❌ **UV Tool Required!**\n\n🚨 **Error:** UV is not installed or not available in PATH\n\n🛠️ **Installation Steps:**\n 1. Install UV: `curl -LsSf https://astral.sh/uv/install.sh | sh`\n 2. Restart terminal or reload shell profile\n 3. Verify installation: `uv --version`\n 4. Try server upgrade again\n\n💡 **Alternative:** Manual upgrade via pip/conda if not using uvx" except subprocess.TimeoutExpired: return "❌ **UV Command Timeout!**\n\n⏱️ **Error:** UV version check timed out\n\n💡 **Troubleshooting:**\n • Check system performance and UV installation\n • Try running `uv --version` manually in terminal\n • Restart terminal and try again\n • Consider manual upgrade if UV issues persist" # Check if this package is installed via uvx try: list_result = subprocess.run( ["uv", "tool", "list"], capture_output=True, text=True, timeout=30 ) if "agent-knowledge-mcp" not in list_result.stdout: message = "⚠️ **UV Tool Installation Required!**\n\n" message += "🚨 **Notice:** Agent Knowledge MCP server is not installed via uv tool\n\n" message += "📦 **This tool only works when the server was installed using:**\n" message += " ```bash\n uv tool install agent-knowledge-mcp\n ```\n\n" message += f"🔍 **Current UV tool packages:**\n" if list_result.stdout.strip(): message += f" {list_result.stdout.strip()}\n\n" else: message += " None installed\n\n" message += "💡 **Installation Options:**\n" message += " • Install via uvx: `uv tool install agent-knowledge-mcp`\n" message += " • Upgrade manually if using pip/conda installation\n" message += " • Contact support for installation guidance" return message except subprocess.TimeoutExpired: return "❌ **UV Tool Check Timeout!**\n\n⏱️ **Error:** UV tool list command timed out\n\n💡 **Troubleshooting:**\n • System performance issues may be affecting UV\n • Try running `uv tool list` manually in terminal\n • Check for disk space and memory availability\n • Consider system restart if problems persist" except Exception as e: return f"⚠️ **UV Installation Verification Failed!**\n\n🚨 **Error:** Cannot verify uvx installation\n🔍 **Details:** {str(e)}\n\n💡 **Resolution:**\n • Ensure agent-knowledge-mcp is installed via uvx\n • Try: `uv tool install agent-knowledge-mcp`\n • Verify UV tool is working: `uv tool list`" # Step 1: Backup current configuration config_path = Path(__file__).parent / "config.json" backup_config = None backup_status = "" if config_path.exists(): try: with open(config_path, 'r', encoding='utf-8') as f: backup_config = json.load(f) backup_status = "✅ Configuration backed up for restoration" except Exception as e: backup_status = f"⚠️ Warning: Could not backup config: {e}" else: backup_status = "ℹ️ No existing config.json to backup" # Get the latest version from PyPI first latest_version = None version_check_status = "" try: import requests response = requests.get( "https://pypi.org/pypi/agent-knowledge-mcp/json", timeout=10 ) if response.status_code == 200: data = response.json() latest_version = data["info"]["version"] version_check_status = f"📦 Latest version available: {latest_version}" else: version_check_status = f"⚠️ PyPI check failed: HTTP {response.status_code}" except ImportError: version_check_status = "⚠️ Warning: requests module not available for version check" except Exception as e: version_check_status = f"⚠️ Warning: Could not fetch latest version: {e}" # Clean UV cache first cache_status = "" try: cache_result = subprocess.run( ["uv", "cache", "clean"], capture_output=True, text=True, timeout=60 ) if cache_result.returncode == 0: cache_status = "🧹 UV cache cleaned successfully" else: cache_status = f"⚠️ UV cache clean failed: {cache_result.stderr.strip() or 'Unknown error'}" # Continue with upgrade even if cache clean fails except subprocess.TimeoutExpired: cache_status = "⚠️ UV cache clean timed out - continuing with upgrade" except Exception as e: cache_status = f"⚠️ UV cache clean error: {str(e)} - continuing anyway" # Force reinstall with specific version if available install_status = "" install_output = "" try: if latest_version: install_cmd = ["uv", "tool", "install", f"agent-knowledge-mcp=={latest_version}", "--force"] result = subprocess.run( install_cmd, capture_output=True, text=True, timeout=120 ) # If specific version fails, try without version constraint if result.returncode != 0: install_cmd = ["uv", "tool", "install", "agent-knowledge-mcp", "--force"] result = subprocess.run( install_cmd, capture_output=True, text=True, timeout=120 ) else: install_cmd = ["uv", "tool", "install", "agent-knowledge-mcp", "--force"] result = subprocess.run( install_cmd, capture_output=True, text=True, timeout=120 ) install_output = result.stdout.strip() if result.returncode == 0: # Parse installation output to check if upgrade happened upgrade_detected = False installed_version = "unknown" # Look for upgrade indicators in output if "+" in install_output and "agent-knowledge-mcp" in install_output: for line in install_output.split('\n'): if line.strip().startswith('+ agent-knowledge-mcp=='): installed_version = line.split('==')[1].strip() upgrade_detected = True break elif line.strip().startswith('- agent-knowledge-mcp==') and '+ agent-knowledge-mcp==' in install_output: upgrade_detected = True if upgrade_detected: install_status = f"🎉 Agent Knowledge MCP server upgraded successfully!" if installed_version != "unknown": install_status += f"\n📦 Installed version: {installed_version}" else: install_status = f"🔄 Agent Knowledge MCP server reinstalled successfully!" else: # Installation failed error_msg = f"❌ **Installation Failed!**\n\n" error_msg += f"🚨 **Return code:** {result.returncode}\n" if result.stderr.strip(): error_msg += f"📝 **Error output:**\n```\n{result.stderr.strip()}\n```\n" if result.stdout.strip(): error_msg += f"📄 **Standard output:**\n```\n{result.stdout.strip()}\n```\n" error_msg += f"\n🛠️ **Manual Recovery:**\n" error_msg += f" ```bash\n uv cache clean && uv tool install agent-knowledge-mcp --force\n ```\n" error_msg += f"\n💡 **Additional Help:**\n" error_msg += f" • Check UV tool is properly configured\n" error_msg += f" • Verify network connectivity to PyPI\n" error_msg += f" • Try manual installation if issues persist" return error_msg except subprocess.TimeoutExpired: return "❌ **Installation Timeout!**\n\n⏱️ **Error:** Installation process timed out (120s limit)\n\n🛠️ **Resolution:**\n • Network connectivity may be slow\n • Try manual installation: `uv tool install agent-knowledge-mcp --force`\n • Check system performance and disk space\n • Consider increasing timeout for large downloads" # Step 3: Restore configuration intelligently config_restoration_status = "" if backup_config: try: # Check if config.json exists after upgrade (it should) if config_path.exists(): # Load new config from upgrade with open(config_path, 'r', encoding='utf-8') as f: new_config = json.load(f) # Perform intelligent merge merged_config = intelligent_config_merge(new_config, backup_config) # Write merged config back with open(config_path, 'w', encoding='utf-8') as f: json.dump(merged_config, f, indent=2, ensure_ascii=False) # Reload configuration after restore config = load_config() # Reinitialize components with restored config init_security(config["security"]["allowed_base_directory"]) init_elasticsearch(config) reset_es_client() config_restoration_status = "🔧 Configuration automatically restored with intelligent merge!\n" config_restoration_status += " • Your custom settings preserved\n" config_restoration_status += " • New features from upgrade included\n" config_restoration_status += " • Deprecated settings removed" else: config_restoration_status = "⚠️ New config.json not found after upgrade" except Exception as e: config_restoration_status = f"⚠️ Warning: Could not restore configuration: {e}\n" config_restoration_status += "💡 Use 'get_config' to review and 'update_config' to customize" else: config_restoration_status = "ℹ️ No previous configuration to restore" # Build comprehensive success message message = "🎉 **Server Upgrade Completed Successfully!**\n\n" # Upgrade summary message += f"📋 **Upgrade Summary:**\n" message += f" {backup_status}\n" message += f" {version_check_status}\n" message += f" {cache_status}\n" message += f" {install_status}\n" message += f" {config_restoration_status}\n\n" # Client restart instructions message += f"🔄 **Important: Restart Your MCP Client**\n\n" message += f"To use the updated version, please restart your MCP client:\n" message += f" • **VS Code:** Reload window (Ctrl/Cmd + Shift + P → 'Reload Window')\n" message += f" • **Claude Desktop:** Restart the application\n" message += f" • **Other clients:** Restart/reload the client application\n\n" # Installation output details if install_output: message += f"📄 **Installation Output:**\n```\n{install_output}\n```\n\n" # Final success confirmation message += f"✅ **Upgrade Complete!** Restart your client to use the latest version." return message except ImportError as e: return f"❌ Module Error: Missing required dependency\n🔍 Details: {str(e)}\n💡 Some upgrade features require additional modules (requests for version checking, json for configuration management)" except subprocess.TimeoutExpired: return f"❌ Timeout Error: Upgrade process timed out\n💡 Network connectivity or system performance issues - try manual upgrade or check system resources" except Exception as e: return _format_admin_error(e, "upgrade server", "uvx package management and configuration restoration")
• src/admin/admin_server.py:716-802 (helper)

  Helper function used by server_upgrade to intelligently merge the new configuration with the user's backed-up configuration, preserving user settings while adopting new features and ignoring deprecated ones.
  def intelligent_config_merge(current_config: Dict[str, Any], backup_config: Dict[str, Any]) -> Dict[str, Any]: """ Intelligently merge configuration after server upgrade. Logic: - Some sections use LATEST config (server, schema, version info) - Some sections use INTELLIGENT merge (user settings like security, elasticsearch) - Ignore deprecated features (keys only in backup - these were removed) Args: current_config: New configuration from server upgrade backup_config: User's previous configuration (backup) Returns: Merged configuration with appropriate merge strategy per section """ # Sections that should always use the LATEST config (no merge) # These contain version info, schema definitions, server settings that must be current LATEST_CONFIG_SECTIONS = { "server", # Version info, new server settings "schema", # Schema definitions must be current "version", # Version tracking "defaults", # Default values must be current "required_fields", # Schema requirements must be current "field_types" # Schema field types must be current } # Sections that should use INTELLIGENT merge (preserve user settings) # These contain user customizations that should be preserved INTELLIGENT_MERGE_SECTIONS = { "security", # User's paths and security settings "elasticsearch", # User's ES connection settings "logging", # User's logging preferences "features", # User's feature toggles "custom" # Any custom user sections } def merge_recursive(current: Dict[str, Any], backup: Dict[str, Any], section_name: str = None) -> Dict[str, Any]: result = current.copy() # Start with current config (includes new features) for key, backup_value in backup.items(): if key in current: current_value = current[key] # Check if this is a top-level section that needs special handling if section_name is None and key in LATEST_CONFIG_SECTIONS: # Use latest config for these sections - no merge result[key] = current_value continue elif section_name is None and key in INTELLIGENT_MERGE_SECTIONS: # Use intelligent merge for these sections if isinstance(current_value, dict) and isinstance(backup_value, dict): result[key] = merge_recursive(current_value, backup_value, key) else: result[key] = backup_value # Preserve user setting continue elif section_name is None and isinstance(current_value, dict) and isinstance(backup_value, dict): # For unknown top-level sections, default to intelligent merge result[key] = merge_recursive(current_value, backup_value, key) continue # For nested values within a section, merge normally if isinstance(current_value, dict) and isinstance(backup_value, dict): # Recursively merge nested dictionaries result[key] = merge_recursive(current_value, backup_value, section_name) else: # Use backup value (user's setting) for intelligent merge sections if section_name in INTELLIGENT_MERGE_SECTIONS or section_name is None: result[key] = backup_value else: # For latest config sections, keep current value result[key] = current_value else: # Key only exists in backup # For intelligent merge sections, preserve user settings even if not in current config # BUT only if they're not clearly deprecated (e.g., "old_", "deprecated_", "legacy_") if section_name in INTELLIGENT_MERGE_SECTIONS: # Check if this looks like a deprecated setting is_deprecated = any(key.startswith(prefix) for prefix in ["old_", "deprecated_", "legacy_"]) if not is_deprecated: result[key] = backup_value # For latest config sections or deprecated keys, ignore (don't include) return result return merge_recursive(current_config, backup_config)
• src/admin/admin_server.py:808-811 (registration)

  FastMCP tool registration decorator for the server_upgrade tool, including description and tags. The admin_server_app is then mounted in main_server.py making the tool available.
  @app.tool( description="Upgrade this MCP server when installed via uvx with automatic configuration backup and restoration", tags={"admin", "server", "upgrade", "uvx", "maintenance"} )