Merge pull request #18 from MatthewGrigsby/feature/plugin-support

gabe-l-hart · web-flow · commit 720046895ebb · 2026-02-11T06:47:15.000-07:00
feat: add read-only plugins commands (list/get/stats)
diff --git a/README.md b/README.md
@@ -96,8 +96,17 @@ cforge prompts execute <prompt-id>
 # MCP Servers
 cforge mcp-servers list
 cforge mcp-servers update <mcp-server-id> [file.json]
+
+# Plugins (read-only admin API)
+cforge plugins list [--search text] [--mode MODE] [--hook HOOK] [--tag TAG] [--json]
+cforge plugins get <plugin-name>
+cforge plugins stats
 ```
 
+Plugin commands call `/admin/plugins` endpoints and require:
+- `MCPGATEWAY_ADMIN_API_ENABLED=true` on the gateway
+- A token with `admin.plugins` permission
+
 ### Server Operations
 
 ```bash
diff --git a/cforge/commands/resources/plugins.py b/cforge/commands/resources/plugins.py
@@ -0,0 +1,133 @@
+# -*- coding: utf-8 -*-
+"""Location: ./cforge/commands/resources/plugins.py
+Copyright 2025
+SPDX-License-Identifier: Apache-2.0
+Authors: Matthew Grigsby
+
+CLI command group: plugins
+
+Note:
+    The CLI currently exposes read-only operations (list/get/stats) for plugins.
+    This matches the current capabilities of the gateway admin API: plugin
+    configuration is loaded from a YAML file at gateway startup, and the gateway
+    does not yet provide write endpoints for plugin CRUD/management. When
+    mcp-context-forge adds server-side write operations, this CLI can be extended
+    to support them.
+"""
+
+# Standard
+from enum import Enum
+from typing import Any, Dict, Optional
+
+# Third-Party
+import typer
+
+# First-Party
+from cforge.common import (
+    AuthenticationError,
+    CLIError,
+    get_console,
+    handle_exception,
+    make_authenticated_request,
+    print_json,
+    print_table,
+)
+
+
+class _CaseInsensitiveEnum(str, Enum):
+    """Enum that supports case-insensitive parsing for CLI options."""
+
+    @classmethod
+    def _missing_(cls, value: object) -> Optional["_CaseInsensitiveEnum"]:
+        """Resolve unknown values by matching enum values case-insensitively.
+
+        Typer converts CLI strings into Enum members. Implementing `_missing_`
+        allows `--mode EnFoRcE` to resolve to `PluginMode.ENFORCE`, while still
+        rejecting unknown values.
+        """
+        if not isinstance(value, str):
+            return None
+        value_folded = value.casefold()
+        for member in cls:
+            if member.value.casefold() == value_folded:
+                return member
+        return None
+
+
+class PluginMode(_CaseInsensitiveEnum):
+    """Valid plugin mode filters supported by the gateway admin API."""
+
+    ENFORCE = "enforce"
+    PERMISSIVE = "permissive"
+    DISABLED = "disabled"
+
+
+def _handle_plugins_exception(exception: Exception) -> None:
+    """Provide plugin-specific hints and raise a CLI error."""
+    console = get_console()
+
+    if isinstance(exception, AuthenticationError):
+        console.print("[yellow]Access denied. Requires admin.plugins permission.[/yellow]")
+    elif isinstance(exception, CLIError) and "(404)" in str(exception):
+        console.print("[yellow]Admin plugin API unavailable. Ensure MCPGATEWAY_ADMIN_API_ENABLED=true and gateway version supports /admin/plugins.[/yellow]")
+
+    handle_exception(exception)
+
+
+def plugins_list(
+    search: Optional[str] = typer.Option(None, "--search", help="Search by plugin name, description, or author"),
+    mode: Optional[PluginMode] = typer.Option(None, "--mode", help="Filter by mode"),
+    hook: Optional[str] = typer.Option(None, "--hook", help="Filter by hook type"),
+    tag: Optional[str] = typer.Option(None, "--tag", help="Filter by plugin tag"),
+    json_output: bool = typer.Option(False, "--json", help="Output as JSON"),
+) -> None:
+    """List all plugins with optional filtering."""
+    console = get_console()
+
+    try:
+        params: Dict[str, Any] = {}
+        if search:
+            params["search"] = search
+        if mode:
+            params["mode"] = mode.value
+        if hook:
+            params["hook"] = hook
+        if tag:
+            params["tag"] = tag
+
+        result = make_authenticated_request("GET", "/admin/plugins", params=params if params else None)
+
+        if json_output:
+            print_json(result, "Plugins")
+        else:
+            plugins: list[dict[str, Any]] = result["plugins"]
+
+            if plugins:
+                print_table(plugins, "Plugins", ["name", "version", "author", "mode", "status", "priority", "hooks", "tags"])
+            else:
+                console.print("[yellow]No plugins found[/yellow]")
+
+    except Exception as e:
+        _handle_plugins_exception(e)
+
+
+def plugins_get(
+    name: str = typer.Argument(..., help="Plugin name"),
+) -> None:
+    """Get details for a specific plugin."""
+    try:
+        result = make_authenticated_request("GET", f"/admin/plugins/{name}")
+        print_json(result, f"Plugin {name}")
+
+    except Exception as e:
+        _handle_plugins_exception(e)
+
+
+def plugins_stats() -> None:
+    """Get plugin statistics."""
+    try:
+        result = make_authenticated_request("GET", "/admin/plugins/stats")
+        print_json(result, "Plugin Statistics")
+
+    except Exception as e:
+        _handle_plugins_exception(e)
diff --git a/cforge/main.py b/cforge/main.py
@@ -97,6 +97,11 @@
     a2a_toggle,
     a2a_invoke,
 )
+from cforge.commands.resources.plugins import (
+    plugins_get,
+    plugins_list,
+    plugins_stats,
+)
 
 # Get the main app singleton
 app = get_app()
@@ -232,6 +237,17 @@
 a2a_app.command("toggle")(a2a_toggle)
 a2a_app.command("invoke")(a2a_invoke)
 
+# ---------------------------------------------------------------------------
+# Plugins command group
+# ---------------------------------------------------------------------------
+
+plugins_app = typer.Typer(help="Manage gateway plugins (read-only)")
+app.add_typer(plugins_app, name="plugins", rich_help_panel="Resources")
+
+plugins_app.command("list")(plugins_list)
+plugins_app.command("get")(plugins_get)
+plugins_app.command("stats")(plugins_stats)
+
 # ---------------------------------------------------------------------------
 # Metrics command group
 # ---------------------------------------------------------------------------
diff --git a/tests/commands/resources/test_plugins.py b/tests/commands/resources/test_plugins.py
@@ -0,0 +1,154 @@
+# -*- coding: utf-8 -*-
+"""Location: ./tests/commands/resources/test_plugins.py
+Copyright 2025
+SPDX-License-Identifier: Apache-2.0
+Authors: Matthew Grigsby
+
+Tests for the plugins commands.
+"""
+
+# Third-Party
+import pytest
+import typer
+
+# First-Party
+from cforge.commands.resources.plugins import PluginMode, plugins_get, plugins_list, plugins_stats
+from cforge.common import AuthenticationError, CLIError
+from tests.conftest import invoke_typer_command, patch_functions
+
+
+class TestPluginCommands:
+    """Tests for plugins commands."""
+
+    def test_plugin_mode_enum_is_case_insensitive(self) -> None:
+        """Typer Enum choices should accept case-insensitive values."""
+        assert PluginMode("EnFoRcE") == PluginMode.ENFORCE
+
+    def test_plugin_mode_enum_missing_non_string(self) -> None:
+        """Non-string values should not be coerced into Enum members."""
+        assert PluginMode._missing_(123) is None
+
+    def test_plugin_mode_enum_missing_unknown_value(self) -> None:
+        """Unknown strings should not be coerced into Enum members."""
+        assert PluginMode._missing_("nope") is None
+
+    def test_plugins_list_success(self, mock_console) -> None:
+        """Test plugins list command with table output."""
+        mock_response = {
+            "plugins": [
+                {"name": "pii_filter", "version": "1.0.0", "author": "ContextForge", "mode": "enforce", "status": "enabled", "priority": 10, "hooks": ["tool_pre_invoke"], "tags": ["security"]}
+            ],
+            "total": 1,
+            "enabled_count": 1,
+            "disabled_count": 0,
+        }
+
+        with patch_functions(
+            "cforge.commands.resources.plugins",
+            get_console=mock_console,
+            make_authenticated_request={"return_value": mock_response},
+            print_table=None,
+        ) as mocks:
+            invoke_typer_command(plugins_list)
+            mocks.print_table.assert_called_once()
+
+    def test_plugins_list_json_output(self, mock_console) -> None:
+        """Test plugins list with JSON output."""
+        mock_response = {"plugins": [], "total": 0, "enabled_count": 0, "disabled_count": 0}
+        with patch_functions(
+            "cforge.commands.resources.plugins",
+            get_console=mock_console,
+            make_authenticated_request={"return_value": mock_response},
+            print_json=None,
+        ) as mocks:
+            invoke_typer_command(plugins_list, json_output=True)
+            mocks.print_json.assert_called_once()
+
+    def test_plugins_list_no_results(self, mock_console) -> None:
+        """Test plugins list with no results."""
+        mock_response = {"plugins": [], "total": 0, "enabled_count": 0, "disabled_count": 0}
+        with patch_functions("cforge.commands.resources.plugins", get_console=mock_console, make_authenticated_request={"return_value": mock_response}):
+            invoke_typer_command(plugins_list)
+
+        assert any("No plugins found" in str(call) for call in mock_console.print.call_args_list)
+
+    def test_plugins_list_with_filters(self, mock_console) -> None:
+        """Test plugins list with all filters."""
+        with patch_functions(
+            "cforge.commands.resources.plugins",
+            get_console=mock_console,
+            make_authenticated_request={"return_value": {"plugins": [], "total": 0, "enabled_count": 0, "disabled_count": 0}},
+            print_table=None,
+        ) as mocks:
+            invoke_typer_command(plugins_list, search="pii", mode=PluginMode.ENFORCE, hook="tool_pre_invoke", tag="security")
+
+            call_args = mocks.make_authenticated_request.call_args
+            assert call_args[0][0] == "GET"
+            assert call_args[0][1] == "/admin/plugins"
+            assert call_args[1]["params"] == {"search": "pii", "mode": "enforce", "hook": "tool_pre_invoke", "tag": "security"}
+
+    def test_plugins_list_error(self, mock_console) -> None:
+        """Test plugins list error handling."""
+        with patch_functions("cforge.commands.resources.plugins", get_console=mock_console, make_authenticated_request={"side_effect": Exception("API error")}):
+            with pytest.raises(typer.Exit):
+                invoke_typer_command(plugins_list)
+
+    def test_plugins_get_success(self, mock_console) -> None:
+        """Test plugins get command."""
+        mock_plugin = {"name": "pii_filter", "version": "1.0.0"}
+        with patch_functions(
+            "cforge.commands.resources.plugins",
+            get_console=mock_console,
+            make_authenticated_request={"return_value": mock_plugin},
+            print_json=None,
+        ) as mocks:
+            invoke_typer_command(plugins_get, name="pii_filter")
+            mocks.print_json.assert_called_once()
+
+    def test_plugins_get_error(self, mock_console) -> None:
+        """Test plugins get error handling."""
+        with patch_functions("cforge.commands.resources.plugins", get_console=mock_console, make_authenticated_request={"side_effect": Exception("API error")}):
+            with pytest.raises(typer.Exit):
+                invoke_typer_command(plugins_get, name="pii_filter")
+
+    def test_plugins_stats_success(self, mock_console) -> None:
+        """Test plugins stats command."""
+        mock_stats = {"total_plugins": 4, "enabled_plugins": 3, "disabled_plugins": 1, "plugins_by_hook": {"tool_pre_invoke": 3}, "plugins_by_mode": {"enforce": 3, "disabled": 1}}
+        with patch_functions(
+            "cforge.commands.resources.plugins",
+            get_console=mock_console,
+            make_authenticated_request={"return_value": mock_stats},
+            print_json=None,
+        ) as mocks:
+            invoke_typer_command(plugins_stats)
+            mocks.print_json.assert_called_once()
+
+    def test_plugins_stats_error(self, mock_console) -> None:
+        """Test plugins stats error handling."""
+        with patch_functions("cforge.commands.resources.plugins", get_console=mock_console, make_authenticated_request={"side_effect": Exception("API error")}):
+            with pytest.raises(typer.Exit):
+                invoke_typer_command(plugins_stats)
+
+    def test_plugins_list_forbidden_shows_permission_hint(self, mock_console) -> None:
+        """Test plugins list shows a targeted hint on forbidden/admin failures."""
+        with patch_functions(
+            "cforge.commands.resources.plugins",
+            get_console=mock_console,
+            make_authenticated_request={"side_effect": AuthenticationError("Authentication required but not configured")},
+        ):
+            with pytest.raises(typer.Exit):
+                invoke_typer_command(plugins_list)
+
+        assert any("Requires admin.plugins permission" in str(call) for call in mock_console.print.call_args_list)
+
+    def test_plugins_list_not_found_shows_admin_api_hint(self, mock_console) -> None:
+        """Test plugins list shows an admin-api hint on 404."""
+        with patch_functions(
+            "cforge.commands.resources.plugins",
+            get_console=mock_console,
+            make_authenticated_request={"side_effect": CLIError("API request failed (404): Not Found")},
+        ):
+            with pytest.raises(typer.Exit):
+                invoke_typer_command(plugins_list)
+
+        assert any("Admin plugin API unavailable" in str(call) for call in mock_console.print.call_args_list)
