Agent Knowledge MCP

Instructions

Implementation Reference

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

  Core handler implementation for the server_upgrade tool. Handles UV availability check, config backup, PyPI version check, cache cleaning, forced reinstallation via uv tool, and intelligent configuration restoration with component reinitialization.

  @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 utility function called by server_upgrade for merging previous user configuration with the new upgraded configuration. Preserves user-customized sections (security, elasticsearch) while updating schema/version sections to latest.

  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 server_upgrade, including description and tags, registering it on the admin FastMCP app instance.

  @app.tool(
      description="Upgrade this MCP server when installed via uvx with automatic configuration backup and restoration",
      tags={"admin", "server", "upgrade", "uvx", "maintenance"}
  )

• src/main_server.py:78-80 (registration)

  Mounting of the admin_server app (containing server_upgrade) into the main FastMCP server app, exposing the tool.

  # Mount Administrative operations server with 'admin' prefix
  # This provides: admin_get_config, admin_update_config, admin_server_status, etc.
  app.mount(admin_server_app)