diff --git a/examples/05_skills_and_plugins/02_loading_plugins/main.py b/examples/05_skills_and_plugins/02_loading_plugins/main.py index fce9c46135..54581efac0 100644 --- a/examples/05_skills_and_plugins/02_loading_plugins/main.py +++ b/examples/05_skills_and_plugins/02_loading_plugins/main.py @@ -1,20 +1,38 @@ -"""Example: Loading Plugins via Conversation +"""Example: Loading and Managing Plugins -Demonstrates the recommended way to load plugins using the `plugins` parameter -on Conversation. Plugins bundle skills, hooks, and MCP config together. +This example demonstrates plugin loading and management in the SDK: + +1. Loading plugins from GitHub via Conversation (PluginSource) +2. Installing plugins to persistent storage (local and GitHub) +3. Listing, loading, and uninstalling plugins + +Plugins bundle skills, hooks, and MCP config together. + +Supported plugin sources: +- Local path: /path/to/plugin +- GitHub shorthand: github:owner/repo +- Git URL: https://github.com/owner/repo.git +- With ref: branch, tag, or commit SHA +- With repo_path: subdirectory for monorepos For full documentation, see: https://docs.all-hands.dev/sdk/guides/plugins """ import os -import sys import tempfile from pathlib import Path from pydantic import SecretStr from openhands.sdk import LLM, Agent, Conversation -from openhands.sdk.plugin import PluginSource +from openhands.sdk.plugin import ( + PluginFetchError, + PluginSource, + install_plugin, + list_installed_plugins, + load_installed_plugins, + uninstall_plugin, +) from openhands.sdk.tool import Tool from openhands.tools.file_editor import FileEditorTool from openhands.tools.terminal import TerminalTool @@ -22,57 +40,188 @@ # Locate example plugin directory script_dir = Path(__file__).parent -plugin_path = script_dir / "example_plugins" / "code-quality" - -# Define plugins to load -# Supported sources: local path, "github:owner/repo", or git URL -# Optional: ref (branch/tag/commit), repo_path (for monorepos) -plugins = [ - PluginSource(source=str(plugin_path)), - # PluginSource(source="github:org/security-plugin", ref="v2.0.0"), - # PluginSource(source="github:org/monorepo", repo_path="plugins/logging"), -] - -# Check for API key -api_key = os.getenv("LLM_API_KEY") -if not api_key: - print("Set LLM_API_KEY to run this example") - print("EXAMPLE_COST: 0") - sys.exit(0) - -# Configure LLM and Agent -model = os.getenv("LLM_MODEL", "anthropic/claude-sonnet-4-5-20250929") -llm = LLM( - usage_id="plugin-demo", - model=model, - api_key=SecretStr(api_key), - base_url=os.getenv("LLM_BASE_URL"), -) -agent = Agent( - llm=llm, tools=[Tool(name=TerminalTool.name), Tool(name=FileEditorTool.name)] -) +local_plugin_path = script_dir / "example_plugins" / "code-quality" -# Create conversation with plugins - skills, MCP config, and hooks are merged -# Note: Plugins are loaded lazily on first send_message() or run() call -with tempfile.TemporaryDirectory() as tmpdir: - conversation = Conversation( - agent=agent, - workspace=tmpdir, - plugins=plugins, - ) - # Test: The "lint" keyword triggers the python-linting skill - # This first send_message() call triggers lazy plugin loading - conversation.send_message("How do I lint Python code? Brief answer please.") +def demo_conversation_with_github_plugin(llm: LLM) -> None: + """Demo 1: Load plugin from GitHub via Conversation. + + This demonstrates loading a plugin directly from GitHub using PluginSource. + The plugin is fetched and loaded lazily when the conversation starts. + + We load the anthropics/skills repository which contains the "document-skills" + plugin with skills for pptx, xlsx, docx, and pdf document processing. + """ + print("\n" + "=" * 60) + print("DEMO 1: Loading plugin from GitHub via Conversation") + print("=" * 60) + + # Load the anthropics/skills repository which contains the document-skills plugin + # This plugin bundles multiple document processing skills including pptx + plugins = [ + PluginSource( + source="github:anthropics/skills", + ref="main", + ), + ] - # Verify skills were loaded from the plugin (after lazy loading) - skills = ( - conversation.agent.agent_context.skills - if conversation.agent.agent_context - else [] + agent = Agent( + llm=llm, + tools=[Tool(name=TerminalTool.name), Tool(name=FileEditorTool.name)], ) - print(f"Loaded {len(skills)} skill(s) from plugins") - conversation.run() + with tempfile.TemporaryDirectory() as tmpdir: + try: + conversation = Conversation( + agent=agent, + workspace=tmpdir, + plugins=plugins, + ) + + # Verify skills were loaded + skills = ( + conversation.agent.agent_context.skills + if conversation.agent.agent_context + else [] + ) + print(f"✓ Loaded {len(skills)} skill(s) from GitHub plugin") + for skill in skills: + print(f" - {skill.name}") + + # Ask a question that uses the pptx skill + conversation.send_message( + "What's the best way to create a PowerPoint presentation " + "programmatically? Check the skill before you answer." + ) + + conversation.run() + + except PluginFetchError as e: + print(f"⚠ Could not fetch from GitHub: {e}") + print(" Skipping this demo (network or rate limiting issue)") + + +def demo_install_local_plugin(installed_dir: Path) -> None: + """Demo 2: Install a plugin from a local path. + + Useful for development or local-only plugins. + """ + print("\n" + "=" * 60) + print("DEMO 2: Installing plugin from local path") + print("=" * 60) + + info = install_plugin(source=str(local_plugin_path), installed_dir=installed_dir) + print(f"✓ Installed: {info.name} v{info.version}") + print(f" Source: {info.source}") + print(f" Path: {info.install_path}") + + +def demo_install_github_plugin(installed_dir: Path) -> None: + """Demo 3: Install a plugin from GitHub to persistent storage. + + Demonstrates loading the anthropics/skills repository which contains + multiple document processing skills (pptx, xlsx, docx, pdf). + """ + print("\n" + "=" * 60) + print("DEMO 3: Installing plugin from GitHub") + print("=" * 60) + + try: + # Install the anthropics/skills repository (contains document-skills plugin) + info = install_plugin( + source="github:anthropics/skills", + ref="main", + installed_dir=installed_dir, + ) + print(f"✓ Installed: {info.name} v{info.version}") + print(f" Source: {info.source}") + print(f" Resolved ref: {info.resolved_ref}") + + # Show the skills loaded from the plugin + plugins = load_installed_plugins(installed_dir=installed_dir) + for plugin in plugins: + if plugin.name == info.name: + skills = plugin.get_all_skills() + print(f" Skills: {len(skills)}") + for skill in skills[:5]: # Show first 5 skills + desc = skill.description or "(no description)" + print(f" - {skill.name}: {desc[:50]}...") + if len(skills) > 5: + print(f" ... and {len(skills) - 5} more skills") + + except PluginFetchError as e: + print(f"⚠ Could not fetch from GitHub: {e}") + print(" (Network or rate limiting issue)") + + +def demo_list_and_load_plugins(installed_dir: Path) -> None: + """Demo 4: List and load installed plugins.""" + print("\n" + "=" * 60) + print("DEMO 4: List and load installed plugins") + print("=" * 60) + + # List installed plugins + print("Installed plugins:") + for info in list_installed_plugins(installed_dir=installed_dir): + print(f" - {info.name} v{info.version} ({info.source})") + + # Load plugins as Plugin objects + plugins = load_installed_plugins(installed_dir=installed_dir) + print(f"\nLoaded {len(plugins)} plugin(s):") + for plugin in plugins: + skills = plugin.get_all_skills() + print(f" - {plugin.name}: {len(skills)} skill(s)") + + +def demo_uninstall_plugins(installed_dir: Path) -> None: + """Demo 5: Uninstall plugins.""" + print("\n" + "=" * 60) + print("DEMO 5: Uninstalling plugins") + print("=" * 60) + + for info in list_installed_plugins(installed_dir=installed_dir): + uninstall_plugin(info.name, installed_dir=installed_dir) + print(f"✓ Uninstalled: {info.name}") + + remaining = list_installed_plugins(installed_dir=installed_dir) + print(f"\nRemaining plugins: {len(remaining)}") + + +# Main execution +if __name__ == "__main__": + api_key = os.getenv("LLM_API_KEY") + if not api_key: + print("Set LLM_API_KEY to run the full example") + print("Running install/uninstall demos only...") + llm = None + else: + model = os.getenv("LLM_MODEL", "anthropic/claude-sonnet-4-5-20250929") + llm = LLM( + usage_id="plugin-demo", + model=model, + api_key=SecretStr(api_key), + base_url=os.getenv("LLM_BASE_URL"), + ) + + with tempfile.TemporaryDirectory() as tmpdir: + installed_dir = Path(tmpdir) / "installed" + installed_dir.mkdir() + + # Demo 1: Conversation with GitHub plugin (requires LLM) + if llm: + demo_conversation_with_github_plugin(llm) + + # Demo 2-5: Plugin management (no LLM required) + demo_install_local_plugin(installed_dir) + demo_install_github_plugin(installed_dir) + demo_list_and_load_plugins(installed_dir) + demo_uninstall_plugins(installed_dir) + + print("\n" + "=" * 60) + print("EXAMPLE COMPLETED SUCCESSFULLY") + print("=" * 60) - print(f"EXAMPLE_COST: {llm.metrics.accumulated_cost:.4f}") + if llm: + print(f"EXAMPLE_COST: {llm.metrics.accumulated_cost:.4f}") + else: + print("EXAMPLE_COST: 0") diff --git a/openhands-sdk/openhands/sdk/plugin/__init__.py b/openhands-sdk/openhands/sdk/plugin/__init__.py index b3dce51e7e..1a8c6656d9 100644 --- a/openhands-sdk/openhands/sdk/plugin/__init__.py +++ b/openhands-sdk/openhands/sdk/plugin/__init__.py @@ -5,12 +5,26 @@ It also provides support for plugin marketplaces - directories that list available plugins with their metadata and source locations. + +Additionally, it provides utilities for managing installed plugins in the +user's home directory (~/.openhands/plugins/installed/). """ from openhands.sdk.plugin.fetch import ( PluginFetchError, fetch_plugin_with_resolution, ) +from openhands.sdk.plugin.installed import ( + InstalledPluginInfo, + InstalledPluginsMetadata, + get_installed_plugin, + get_installed_plugins_dir, + install_plugin, + list_installed_plugins, + load_installed_plugins, + uninstall_plugin, + update_plugin, +) from openhands.sdk.plugin.loader import load_plugins from openhands.sdk.plugin.plugin import Plugin from openhands.sdk.plugin.types import ( @@ -45,4 +59,14 @@ "MarketplacePluginEntry", "MarketplacePluginSource", "MarketplaceMetadata", + # Installed plugins management + "InstalledPluginInfo", + "InstalledPluginsMetadata", + "install_plugin", + "uninstall_plugin", + "list_installed_plugins", + "load_installed_plugins", + "get_installed_plugins_dir", + "get_installed_plugin", + "update_plugin", ] diff --git a/openhands-sdk/openhands/sdk/plugin/installed.py b/openhands-sdk/openhands/sdk/plugin/installed.py new file mode 100644 index 0000000000..ca47783cd3 --- /dev/null +++ b/openhands-sdk/openhands/sdk/plugin/installed.py @@ -0,0 +1,518 @@ +"""Installed plugins management for OpenHands SDK. + +This module provides utilities for managing plugins installed in the user's +home directory (~/.openhands/plugins/installed/). + +The installed plugins directory structure follows the Claude Code pattern:: + + ~/.openhands/plugins/installed/ + ├── plugin-name-1/ + │ ├── .plugin/ + │ │ └── plugin.json + │ ├── skills/ + │ └── ... + ├── plugin-name-2/ + │ └── ... + └── .installed.json # Metadata about installed plugins +""" + +from __future__ import annotations + +import json +import re +import shutil +from datetime import UTC, datetime +from pathlib import Path + +from pydantic import BaseModel, Field + +from openhands.sdk.logger import get_logger +from openhands.sdk.plugin.fetch import ( + fetch_plugin_with_resolution, +) +from openhands.sdk.plugin.plugin import Plugin + + +logger = get_logger(__name__) + +# Default directory for installed plugins +DEFAULT_INSTALLED_PLUGINS_DIR = Path.home() / ".openhands" / "plugins" / "installed" + +# Metadata file for tracking installed plugins +_METADATA_FILENAME = ".installed.json" + +_PLUGIN_NAME_PATTERN = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$") + + +def _resolve_installed_dir(installed_dir: Path | None) -> Path: + """Return installed_dir or the default if None.""" + return installed_dir if installed_dir is not None else DEFAULT_INSTALLED_PLUGINS_DIR + + +def get_installed_plugins_dir() -> Path: + """Get the default directory for installed plugins. + + Returns: + Path to ~/.openhands/plugins/installed/ + """ + return DEFAULT_INSTALLED_PLUGINS_DIR + + +def _validate_plugin_name(name: str) -> None: + """Validate plugin name is Claude-like kebab-case. + + This protects filesystem operations (install/uninstall) from path traversal. + """ + if not _PLUGIN_NAME_PATTERN.fullmatch(name): + raise ValueError( + f"Invalid plugin name. Expected kebab-case like 'my-plugin' (got {name!r})." + ) + + +class InstalledPluginInfo(BaseModel): + """Information about an installed plugin. + + This model tracks metadata about a plugin installation, including + where it was installed from and when. + """ + + name: str = Field(description="Plugin name (from manifest)") + version: str = Field(default="1.0.0", description="Plugin version") + description: str = Field(default="", description="Plugin description") + source: str = Field(description="Original source (e.g., 'github:owner/repo')") + resolved_ref: str | None = Field( + default=None, + description="Resolved git commit SHA (for version pinning)", + ) + repo_path: str | None = Field( + default=None, + description="Subdirectory path within the repository (for monorepos)", + ) + installed_at: str = Field( + description="ISO 8601 timestamp of installation", + ) + install_path: str = Field( + description="Path where the plugin is installed", + ) + + @classmethod + def from_plugin( + cls, + plugin: Plugin, + source: str, + resolved_ref: str | None, + repo_path: str | None, + install_path: Path, + ) -> InstalledPluginInfo: + """Create InstalledPluginInfo from a loaded Plugin.""" + return cls( + name=plugin.name, + version=plugin.version, + description=plugin.description, + source=source, + resolved_ref=resolved_ref, + repo_path=repo_path, + installed_at=datetime.now(UTC).isoformat(), + install_path=str(install_path), + ) + + +class InstalledPluginsMetadata(BaseModel): + """Metadata file for tracking all installed plugins.""" + + plugins: dict[str, InstalledPluginInfo] = Field( + default_factory=dict, + description="Map of plugin name to installation info", + ) + + @classmethod + def get_path(cls, installed_dir: Path) -> Path: + """Get the metadata file path for the given installed plugins directory.""" + return installed_dir / _METADATA_FILENAME + + @classmethod + def load_from_dir(cls, installed_dir: Path) -> InstalledPluginsMetadata: + """Load metadata from the installed plugins directory.""" + metadata_path = cls.get_path(installed_dir) + if not metadata_path.exists(): + return cls() + try: + with open(metadata_path) as f: + data = json.load(f) + return cls.model_validate(data) + except Exception as e: + logger.warning(f"Failed to load installed plugins metadata: {e}") + return cls() + + def save_to_dir(self, installed_dir: Path) -> None: + """Save metadata to the installed plugins directory.""" + metadata_path = self.get_path(installed_dir) + metadata_path.parent.mkdir(parents=True, exist_ok=True) + with open(metadata_path, "w") as f: + json.dump(self.model_dump(), f, indent=2) + + +def install_plugin( + source: str, + ref: str | None = None, + repo_path: str | None = None, + installed_dir: Path | None = None, + force: bool = False, +) -> InstalledPluginInfo: + """Install a plugin from a source. + + Fetches the plugin from the source, copies it to the installed plugins + directory, and records the installation metadata. + + Args: + source: Plugin source - can be: + - "github:owner/repo" - GitHub shorthand + - Any git URL (GitHub, GitLab, Bitbucket, etc.) + - Local path (for development/testing) + ref: Optional branch, tag, or commit to install. + repo_path: Subdirectory path within the repository (for monorepos). + installed_dir: Directory for installed plugins. + Defaults to ~/.openhands/plugins/installed/ + force: If True, overwrite existing installation. If False, raise error + if plugin is already installed. + + Returns: + InstalledPluginInfo with details about the installation. + + Raises: + PluginFetchError: If fetching the plugin fails. + FileExistsError: If plugin is already installed and force=False. + ValueError: If the plugin manifest is invalid. + + Example: + >>> info = install_plugin("github:owner/my-plugin", ref="v1.0.0") + >>> print(f"Installed {info.name} from {info.source}") + """ + installed_dir = _resolve_installed_dir(installed_dir) + + # Fetch the plugin (downloads to cache if remote) + logger.info(f"Fetching plugin from {source}") + fetched_path, resolved_ref = fetch_plugin_with_resolution( + source=source, + ref=ref, + repo_path=repo_path, + update=True, + ) + + # Load the plugin to get its metadata + plugin = Plugin.load(fetched_path) + plugin_name = plugin.name + _validate_plugin_name(plugin_name) + + # Check if already installed + install_path = installed_dir / plugin_name + if install_path.exists() and not force: + raise FileExistsError( + f"Plugin '{plugin_name}' is already installed at {install_path}. " + f"Use force=True to overwrite." + ) + + # Remove existing installation if force=True + if install_path.exists(): + logger.info(f"Removing existing installation of '{plugin_name}'") + shutil.rmtree(install_path) + + # Copy plugin to installed directory + logger.info(f"Installing plugin '{plugin_name}' to {install_path}") + installed_dir.mkdir(parents=True, exist_ok=True) + shutil.copytree(fetched_path, install_path) + + # Create installation info + info = InstalledPluginInfo.from_plugin( + plugin=plugin, + source=source, + resolved_ref=resolved_ref, + repo_path=repo_path, + install_path=install_path, + ) + + # Update metadata + metadata = InstalledPluginsMetadata.load_from_dir(installed_dir) + metadata.plugins[plugin_name] = info + metadata.save_to_dir(installed_dir) + + logger.info(f"Successfully installed plugin '{plugin_name}' v{plugin.version}") + return info + + +def uninstall_plugin( + name: str, + installed_dir: Path | None = None, +) -> bool: + """Uninstall a plugin by name. + + Only plugins tracked in the installed plugins metadata file can be uninstalled. + This avoids deleting arbitrary directories in the installed plugins directory. + + Args: + name: Name of the plugin to uninstall. + installed_dir: Directory for installed plugins. + Defaults to ~/.openhands/plugins/installed/ + + Returns: + True if the plugin was uninstalled, False if it wasn't installed. + + Example: + >>> if uninstall_plugin("my-plugin"): + ... print("Plugin uninstalled") + ... else: + ... print("Plugin was not installed") + """ + _validate_plugin_name(name) + installed_dir = _resolve_installed_dir(installed_dir) + + metadata = InstalledPluginsMetadata.load_from_dir(installed_dir) + if name not in metadata.plugins: + logger.warning(f"Plugin '{name}' is not installed") + return False + + plugin_path = installed_dir / name + if plugin_path.exists(): + logger.info(f"Uninstalling plugin '{name}' from {plugin_path}") + shutil.rmtree(plugin_path) + else: + logger.warning( + f"Plugin '{name}' was tracked but its directory is missing: {plugin_path}" + ) + + del metadata.plugins[name] + metadata.save_to_dir(installed_dir) + + logger.info(f"Successfully uninstalled plugin '{name}'") + return True + + +def _validate_tracked_plugins( + metadata: InstalledPluginsMetadata, installed_dir: Path +) -> tuple[list[InstalledPluginInfo], bool]: + """Validate tracked plugins exist on disk. + + Returns: + Tuple of (valid plugins list, whether metadata was modified). + """ + valid_plugins: list[InstalledPluginInfo] = [] + changed = False + + for name, info in list(metadata.plugins.items()): + try: + _validate_plugin_name(name) + except ValueError as e: + logger.warning(f"Invalid tracked plugin name {name!r}, removing: {e}") + del metadata.plugins[name] + changed = True + continue + + plugin_path = installed_dir / name + if plugin_path.exists(): + valid_plugins.append(info) + else: + logger.warning(f"Plugin '{name}' directory missing, removing from metadata") + del metadata.plugins[name] + changed = True + + return valid_plugins, changed + + +def _discover_untracked_plugins( + metadata: InstalledPluginsMetadata, installed_dir: Path +) -> tuple[list[InstalledPluginInfo], bool]: + """Discover plugin directories not tracked in metadata. + + Returns: + Tuple of (discovered plugins list, whether metadata was modified). + """ + discovered: list[InstalledPluginInfo] = [] + changed = False + + for item in installed_dir.iterdir(): + if not item.is_dir() or item.name.startswith("."): + continue + if item.name in metadata.plugins: + continue + + try: + _validate_plugin_name(item.name) + except ValueError: + logger.debug(f"Skipping directory with invalid plugin name: {item}") + continue + + try: + plugin = Plugin.load(item) + except Exception as e: + logger.debug(f"Skipping directory {item}: {e}") + continue + + if plugin.name != item.name: + logger.warning( + "Skipping plugin directory because manifest name doesn't match " + f"directory name: dir={item.name!r}, manifest={plugin.name!r}" + ) + continue + + info = InstalledPluginInfo( + name=plugin.name, + version=plugin.version, + description=plugin.description, + source="local", + installed_at=datetime.now(UTC).isoformat(), + install_path=str(item), + ) + discovered.append(info) + metadata.plugins[item.name] = info + changed = True + logger.info(f"Discovered untracked plugin: {plugin.name}") + + return discovered, changed + + +def list_installed_plugins( + installed_dir: Path | None = None, +) -> list[InstalledPluginInfo]: + """List all installed plugins. + + This function is self-healing: it may update the installed plugins metadata + file to remove entries whose directories were deleted, and to add entries for + plugin directories that were manually copied into the installed dir. + + Args: + installed_dir: Directory for installed plugins. + Defaults to ~/.openhands/plugins/installed/ + + Returns: + List of InstalledPluginInfo for each installed plugin. + + Example: + >>> for info in list_installed_plugins(): + ... print(f"{info.name} v{info.version} - {info.description}") + """ + installed_dir = _resolve_installed_dir(installed_dir) + + if not installed_dir.exists(): + return [] + + metadata = InstalledPluginsMetadata.load_from_dir(installed_dir) + + # Validate tracked plugins and discover untracked ones + valid_plugins, tracked_changed = _validate_tracked_plugins(metadata, installed_dir) + discovered, discovered_changed = _discover_untracked_plugins( + metadata, installed_dir + ) + + if tracked_changed or discovered_changed: + metadata.save_to_dir(installed_dir) + + return valid_plugins + discovered + + +def load_installed_plugins( + installed_dir: Path | None = None, +) -> list[Plugin]: + """Load all installed plugins. + + Loads Plugin objects for all plugins in the installed plugins directory. + This is useful for integrating installed plugins into an agent. + + Args: + installed_dir: Directory for installed plugins. + Defaults to ~/.openhands/plugins/installed/ + + Returns: + List of loaded Plugin objects. + + Example: + >>> plugins = load_installed_plugins() + >>> for plugin in plugins: + ... print(f"Loaded {plugin.name} with {len(plugin.skills)} skills") + """ + installed_dir = _resolve_installed_dir(installed_dir) + + if not installed_dir.exists(): + return [] + + return Plugin.load_all(installed_dir) + + +def get_installed_plugin( + name: str, + installed_dir: Path | None = None, +) -> InstalledPluginInfo | None: + """Get information about a specific installed plugin. + + Args: + name: Name of the plugin to look up. + installed_dir: Directory for installed plugins. + Defaults to ~/.openhands/plugins/installed/ + + Returns: + InstalledPluginInfo if the plugin is installed, None otherwise. + + Example: + >>> info = get_installed_plugin("my-plugin") + >>> if info: + ... print(f"Installed from {info.source} at {info.installed_at}") + """ + _validate_plugin_name(name) + installed_dir = _resolve_installed_dir(installed_dir) + + metadata = InstalledPluginsMetadata.load_from_dir(installed_dir) + info = metadata.plugins.get(name) + + # Verify the plugin directory still exists + if info is not None: + plugin_path = installed_dir / name + if not plugin_path.exists(): + return None + + return info + + +def update_plugin( + name: str, + installed_dir: Path | None = None, +) -> InstalledPluginInfo | None: + """Update an installed plugin to the latest version. + + Re-fetches the plugin from its original source and reinstalls it. + + This always updates to the latest version available from the original source + (i.e., it does not preserve a pinned ref). + + Args: + name: Name of the plugin to update. + installed_dir: Directory for installed plugins. + Defaults to ~/.openhands/plugins/installed/ + + Returns: + Updated InstalledPluginInfo if successful, None if plugin not installed. + + Raises: + PluginFetchError: If fetching the updated plugin fails. + + Example: + >>> info = update_plugin("my-plugin") + >>> if info: + ... print(f"Updated to v{info.version}") + """ + _validate_plugin_name(name) + installed_dir = _resolve_installed_dir(installed_dir) + + # Get current installation info + current_info = get_installed_plugin(name, installed_dir) + if current_info is None: + logger.warning(f"Plugin '{name}' is not installed") + return None + + # Re-install from the original source + logger.info(f"Updating plugin '{name}' from {current_info.source}") + return install_plugin( + source=current_info.source, + ref=None, # Get latest (don't use pinned ref) + repo_path=current_info.repo_path, + installed_dir=installed_dir, + force=True, + ) diff --git a/tests/sdk/plugin/test_installed_plugins.py b/tests/sdk/plugin/test_installed_plugins.py new file mode 100644 index 0000000000..7410260908 --- /dev/null +++ b/tests/sdk/plugin/test_installed_plugins.py @@ -0,0 +1,574 @@ +"""Tests for installed plugins management. + +This module contains both unit tests and integration tests for plugin +installation, management, and lifecycle operations. + +Unit tests use mocks for external operations (GitHub fetch). +Integration tests (marked with @pytest.mark.network) test real GitHub cloning. +""" + +import json +import shutil +from pathlib import Path +from unittest.mock import patch + +import pytest + +from openhands.sdk.plugin import ( + InstalledPluginInfo, + InstalledPluginsMetadata, + Plugin, + PluginFetchError, + get_installed_plugin, + get_installed_plugins_dir, + install_plugin, + list_installed_plugins, + load_installed_plugins, + uninstall_plugin, + update_plugin, +) +from openhands.sdk.plugin.fetch import get_cache_path, parse_plugin_source + + +# ============================================================================ +# Fixtures +# ============================================================================ + + +@pytest.fixture +def installed_dir(tmp_path: Path) -> Path: + """Create a temporary installed plugins directory.""" + installed = tmp_path / "installed" + installed.mkdir(parents=True) + return installed + + +@pytest.fixture +def sample_plugin_dir(tmp_path: Path) -> Path: + """Create a sample plugin directory structure.""" + plugin_dir = tmp_path / "sample-plugin" + plugin_dir.mkdir(parents=True) + + # Create plugin manifest + manifest_dir = plugin_dir / ".plugin" + manifest_dir.mkdir() + manifest = { + "name": "sample-plugin", + "version": "1.0.0", + "description": "A sample plugin for testing", + } + (manifest_dir / "plugin.json").write_text(json.dumps(manifest)) + + # Create a skill + skills_dir = plugin_dir / "skills" / "test-skill" + skills_dir.mkdir(parents=True) + skill_content = """--- +name: test-skill +description: A test skill +triggers: + - test +--- +# Test Skill + +This is a test skill. +""" + (skills_dir / "SKILL.md").write_text(skill_content) + + return plugin_dir + + +# ============================================================================ +# Model Tests +# ============================================================================ + + +class TestInstalledPluginInfo: + """Tests for InstalledPluginInfo model.""" + + def test_from_plugin(self, sample_plugin_dir: Path, tmp_path: Path): + """Test creating InstalledPluginInfo from a Plugin.""" + plugin = Plugin.load(sample_plugin_dir) + install_path = tmp_path / "installed" / "sample-plugin" + + info = InstalledPluginInfo.from_plugin( + plugin=plugin, + source="github:owner/sample-plugin", + resolved_ref="abc123", + repo_path=None, + install_path=install_path, + ) + + assert info.name == "sample-plugin" + assert info.version == "1.0.0" + assert info.description == "A sample plugin for testing" + assert info.source == "github:owner/sample-plugin" + assert info.resolved_ref == "abc123" + assert info.repo_path is None + assert info.installed_at is not None + assert str(install_path) in info.install_path + + +class TestInstalledPluginsMetadata: + """Tests for InstalledPluginsMetadata model.""" + + def test_load_from_dir_nonexistent(self, tmp_path: Path): + """Test loading metadata from nonexistent directory returns empty.""" + metadata = InstalledPluginsMetadata.load_from_dir(tmp_path / "nonexistent") + assert metadata.plugins == {} + + def test_load_from_dir_and_save_to_dir(self, tmp_path: Path): + """Test saving and loading metadata.""" + installed_dir = tmp_path / "installed" + installed_dir.mkdir() + + info = InstalledPluginInfo( + name="test-plugin", + version="1.0.0", + description="Test", + source="github:owner/test", + installed_at="2024-01-01T00:00:00Z", + install_path="/path/to/plugin", + ) + metadata = InstalledPluginsMetadata(plugins={"test-plugin": info}) + metadata.save_to_dir(installed_dir) + + loaded = InstalledPluginsMetadata.load_from_dir(installed_dir) + assert "test-plugin" in loaded.plugins + assert loaded.plugins["test-plugin"].name == "test-plugin" + assert loaded.plugins["test-plugin"].version == "1.0.0" + + def test_load_from_dir_invalid_json(self, tmp_path: Path): + """Test loading invalid JSON returns empty metadata.""" + installed_dir = tmp_path / "installed" + installed_dir.mkdir() + metadata_path = InstalledPluginsMetadata.get_path(installed_dir) + metadata_path.write_text("invalid json {") + + metadata = InstalledPluginsMetadata.load_from_dir(installed_dir) + assert metadata.plugins == {} + + +# ============================================================================ +# Utility Function Tests +# ============================================================================ + + +def test_get_installed_plugins_dir_returns_default_path(): + """Test that default path is under ~/.openhands/plugins/installed/.""" + path = get_installed_plugins_dir() + assert ".openhands" in str(path) + assert "plugins" in str(path) + assert "installed" in str(path) + + +# ============================================================================ +# Install Plugin Tests +# ============================================================================ + + +def test_install_from_local_path(sample_plugin_dir: Path, installed_dir: Path) -> None: + """Test installing a plugin from a local path.""" + info = install_plugin( + source=str(sample_plugin_dir), + installed_dir=installed_dir, + ) + + assert info.name == "sample-plugin" + assert info.version == "1.0.0" + assert info.source == str(sample_plugin_dir) + + # Verify plugin was copied + plugin_path = installed_dir / "sample-plugin" + assert plugin_path.exists() + assert (plugin_path / ".plugin" / "plugin.json").exists() + + # Verify metadata was updated + metadata = InstalledPluginsMetadata.load_from_dir(installed_dir) + assert "sample-plugin" in metadata.plugins + + +def test_install_already_exists_raises_error( + sample_plugin_dir: Path, installed_dir: Path +) -> None: + """Test that installing an existing plugin raises FileExistsError.""" + install_plugin(source=str(sample_plugin_dir), installed_dir=installed_dir) + + with pytest.raises(FileExistsError, match="already installed"): + install_plugin(source=str(sample_plugin_dir), installed_dir=installed_dir) + + +def test_install_with_force_overwrites( + sample_plugin_dir: Path, installed_dir: Path +) -> None: + """Test that force=True overwrites existing installation.""" + install_plugin(source=str(sample_plugin_dir), installed_dir=installed_dir) + + marker_file = installed_dir / "sample-plugin" / "marker.txt" + marker_file.write_text("original") + + install_plugin( + source=str(sample_plugin_dir), + installed_dir=installed_dir, + force=True, + ) + + assert not marker_file.exists() + + +@patch("openhands.sdk.plugin.installed.fetch_plugin_with_resolution") +def test_install_from_github_mocked( + mock_fetch, sample_plugin_dir: Path, installed_dir: Path +) -> None: + """Test installing a plugin from GitHub (mocked).""" + mock_fetch.return_value = (sample_plugin_dir, "abc123def456") + + info = install_plugin( + source="github:owner/sample-plugin", + ref="v1.0.0", + installed_dir=installed_dir, + ) + + mock_fetch.assert_called_once_with( + source="github:owner/sample-plugin", + ref="v1.0.0", + repo_path=None, + update=True, + ) + assert info.name == "sample-plugin" + assert info.source == "github:owner/sample-plugin" + assert info.resolved_ref == "abc123def456" + + +def test_install_invalid_plugin_name_raises_error( + sample_plugin_dir: Path, installed_dir: Path +) -> None: + """Test that installing a plugin with an invalid manifest name fails.""" + manifest_path = sample_plugin_dir / ".plugin" / "plugin.json" + manifest = json.loads(manifest_path.read_text()) + manifest["name"] = "bad_name" # not kebab-case + manifest_path.write_text(json.dumps(manifest)) + + with pytest.raises(ValueError, match="Invalid plugin name"): + install_plugin(source=str(sample_plugin_dir), installed_dir=installed_dir) + + +# ============================================================================ +# Uninstall Plugin Tests +# ============================================================================ + + +def test_uninstall_existing_plugin( + sample_plugin_dir: Path, installed_dir: Path +) -> None: + """Test uninstalling an existing plugin.""" + install_plugin(source=str(sample_plugin_dir), installed_dir=installed_dir) + + result = uninstall_plugin("sample-plugin", installed_dir=installed_dir) + + assert result is True + assert not (installed_dir / "sample-plugin").exists() + + metadata = InstalledPluginsMetadata.load_from_dir(installed_dir) + assert "sample-plugin" not in metadata.plugins + + +def test_uninstall_nonexistent_plugin(installed_dir: Path) -> None: + """Test uninstalling a plugin that doesn't exist.""" + result = uninstall_plugin("nonexistent", installed_dir=installed_dir) + assert result is False + + +def test_uninstall_untracked_plugin_does_not_delete( + sample_plugin_dir: Path, installed_dir: Path +) -> None: + """Test that uninstall refuses to delete untracked plugin directories.""" + dest = installed_dir / "untracked-plugin" + shutil.copytree(sample_plugin_dir, dest) + + manifest_path = dest / ".plugin" / "plugin.json" + manifest = json.loads(manifest_path.read_text()) + manifest["name"] = "untracked-plugin" + manifest_path.write_text(json.dumps(manifest)) + + result = uninstall_plugin("untracked-plugin", installed_dir=installed_dir) + + assert result is False + assert dest.exists() + + +def test_uninstall_invalid_name_raises_error(installed_dir: Path) -> None: + """Test that invalid plugin names are rejected.""" + with pytest.raises(ValueError, match="Invalid plugin name"): + uninstall_plugin("../evil", installed_dir=installed_dir) + + +# ============================================================================ +# List Installed Plugins Tests +# ============================================================================ + + +def test_list_empty_directory(installed_dir: Path) -> None: + """Test listing plugins from empty directory.""" + plugins = list_installed_plugins(installed_dir=installed_dir) + assert plugins == [] + + +def test_list_installed_plugins(sample_plugin_dir: Path, installed_dir: Path) -> None: + """Test listing installed plugins.""" + install_plugin(source=str(sample_plugin_dir), installed_dir=installed_dir) + + plugins = list_installed_plugins(installed_dir=installed_dir) + + assert len(plugins) == 1 + assert plugins[0].name == "sample-plugin" + assert plugins[0].version == "1.0.0" + + +def test_list_discovers_untracked_plugins( + sample_plugin_dir: Path, installed_dir: Path +) -> None: + """Test that list discovers plugins not in metadata.""" + dest = installed_dir / "manual-plugin" + shutil.copytree(sample_plugin_dir, dest) + + manifest_path = dest / ".plugin" / "plugin.json" + manifest = json.loads(manifest_path.read_text()) + manifest["name"] = "manual-plugin" + manifest_path.write_text(json.dumps(manifest)) + + plugins = list_installed_plugins(installed_dir=installed_dir) + + assert len(plugins) == 1 + assert plugins[0].name == "manual-plugin" + assert plugins[0].source == "local" + + +def test_list_cleans_up_missing_plugins( + sample_plugin_dir: Path, installed_dir: Path +) -> None: + """Test that list removes metadata for missing plugins.""" + install_plugin(source=str(sample_plugin_dir), installed_dir=installed_dir) + + shutil.rmtree(installed_dir / "sample-plugin") + + plugins = list_installed_plugins(installed_dir=installed_dir) + + assert len(plugins) == 0 + metadata = InstalledPluginsMetadata.load_from_dir(installed_dir) + assert "sample-plugin" not in metadata.plugins + + +# ============================================================================ +# Load Installed Plugins Tests +# ============================================================================ + + +def test_load_empty_directory(installed_dir: Path) -> None: + """Test loading plugins from empty directory.""" + plugins = load_installed_plugins(installed_dir=installed_dir) + assert plugins == [] + + +def test_load_installed_plugins(sample_plugin_dir: Path, installed_dir: Path) -> None: + """Test loading installed plugins.""" + install_plugin(source=str(sample_plugin_dir), installed_dir=installed_dir) + + plugins = load_installed_plugins(installed_dir=installed_dir) + + assert len(plugins) == 1 + assert plugins[0].name == "sample-plugin" + assert len(plugins[0].skills) == 1 + + +# ============================================================================ +# Get Installed Plugin Tests +# ============================================================================ + + +def test_get_existing_plugin(sample_plugin_dir: Path, installed_dir: Path) -> None: + """Test getting info for an existing plugin.""" + install_plugin(source=str(sample_plugin_dir), installed_dir=installed_dir) + + info = get_installed_plugin("sample-plugin", installed_dir=installed_dir) + + assert info is not None + assert info.name == "sample-plugin" + + +def test_get_nonexistent_plugin(installed_dir: Path) -> None: + """Test getting info for a nonexistent plugin.""" + info = get_installed_plugin("nonexistent", installed_dir=installed_dir) + assert info is None + + +def test_get_plugin_with_missing_directory( + sample_plugin_dir: Path, installed_dir: Path +) -> None: + """Test getting info when plugin directory is missing.""" + install_plugin(source=str(sample_plugin_dir), installed_dir=installed_dir) + + shutil.rmtree(installed_dir / "sample-plugin") + + info = get_installed_plugin("sample-plugin", installed_dir=installed_dir) + assert info is None + + +# ============================================================================ +# Update Plugin Tests +# ============================================================================ + + +def test_update_existing_plugin_local( + sample_plugin_dir: Path, installed_dir: Path +) -> None: + """Test updating an installed plugin from local source.""" + install_plugin(source=str(sample_plugin_dir), installed_dir=installed_dir) + + # Modify the source to new version + (sample_plugin_dir / ".plugin" / "plugin.json").write_text( + json.dumps( + { + "name": "sample-plugin", + "version": "1.0.1", + "description": "Updated plugin", + } + ) + ) + + updated = update_plugin("sample-plugin", installed_dir=installed_dir) + + assert updated is not None + assert updated.version == "1.0.1" + + +def test_update_existing_plugin_mocked( + sample_plugin_dir: Path, installed_dir: Path +) -> None: + """Test updating fetches with ref=None to get latest.""" + # Install first without mocking + install_plugin(source=str(sample_plugin_dir), installed_dir=installed_dir) + + # Now mock for the update call only + with patch( + "openhands.sdk.plugin.installed.fetch_plugin_with_resolution" + ) as mock_fetch: + mock_fetch.return_value = (sample_plugin_dir, "newcommit123") + + info = update_plugin("sample-plugin", installed_dir=installed_dir) + + assert info is not None + assert info.resolved_ref == "newcommit123" + + mock_fetch.assert_called_once() + call_kwargs = mock_fetch.call_args[1] + assert call_kwargs["source"] == str(sample_plugin_dir) + assert call_kwargs["ref"] is None # Get latest + + +def test_update_nonexistent_plugin(installed_dir: Path) -> None: + """Test updating a plugin that doesn't exist.""" + info = update_plugin("nonexistent", installed_dir=installed_dir) + assert info is None + + +# ============================================================================ +# Integration Tests (Real GitHub) +# ============================================================================ + + +@pytest.mark.network +def test_install_from_github_with_repo_path(installed_dir: Path) -> None: + """Test installing a plugin from GitHub using repo_path for monorepo.""" + try: + info = install_plugin( + source="github:OpenHands/agent-sdk", + repo_path=( + "examples/05_skills_and_plugins/" + "02_loading_plugins/example_plugins/code-quality" + ), + installed_dir=installed_dir, + ) + + assert info.name == "code-quality" + assert info.source == "github:OpenHands/agent-sdk" + assert info.resolved_ref is not None + assert info.repo_path is not None + + plugins = load_installed_plugins(installed_dir=installed_dir) + code_quality = next((p for p in plugins if p.name == "code-quality"), None) + assert code_quality is not None + assert len(code_quality.get_all_skills()) >= 1 + + except PluginFetchError: + pytest.skip("GitHub not accessible (network issue)") + + +@pytest.mark.network +def test_install_from_github_with_ref(installed_dir: Path) -> None: + """Test installing a plugin from GitHub with specific ref.""" + try: + info = install_plugin( + source="github:OpenHands/agent-sdk", + ref="main", + repo_path=( + "examples/05_skills_and_plugins/" + "02_loading_plugins/example_plugins/code-quality" + ), + installed_dir=installed_dir, + ) + + assert info.name == "code-quality" + assert info.resolved_ref is not None + assert len(info.resolved_ref) == 40 # SHA length + + except PluginFetchError: + pytest.skip("GitHub not accessible (network issue)") + + +@pytest.mark.network +def test_install_document_skills_plugin(installed_dir: Path) -> None: + """Test installing the document-skills plugin from anthropics/skills repository. + + This tests loading a proper Claude Code plugin which bundles multiple skills + (xlsx, docx, pptx, pdf) in the skills/ subdirectory. + """ + try: + source = "github:anthropics/skills" + info = install_plugin( + source=source, + ref="main", + installed_dir=installed_dir, + ) + + _, url = parse_plugin_source(source) + expected_name = get_cache_path(url).name + assert info.name == expected_name + assert info.source == source + + # Verify the plugin directory has the expected structure + install_path = Path(info.install_path) + skills_dir = install_path / "skills" + assert skills_dir.is_dir() + + # Check that the expected skill directories exist + for skill_name in ["pptx", "xlsx", "docx", "pdf"]: + skill_dir = skills_dir / skill_name + assert skill_dir.is_dir(), f"Expected skill directory: {skill_name}" + skill_md = skill_dir / "SKILL.md" + assert skill_md.exists(), f"Expected SKILL.md in {skill_name}" + + # Verify skills are loaded from the plugin + plugins = load_installed_plugins(installed_dir=installed_dir) + doc_plugin = next((p for p in plugins if p.name == expected_name), None) + assert doc_plugin is not None + skills = doc_plugin.get_all_skills() + # Should have at least the 4 document skills + assert len(skills) >= 4 + skill_names = {s.name for s in skills} + assert "pptx" in skill_names + assert "xlsx" in skill_names + assert "docx" in skill_names + assert "pdf" in skill_names + + except PluginFetchError: + pytest.skip("GitHub not accessible (network issue)")