Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice, thanks for adding this!

- `MCPGATEWAY_ADMIN_API_ENABLED=true` on the gateway
- A token with `admin.plugins` permission

### Server Operations

```bash
Expand Down
133 changes: 133 additions & 0 deletions cforge/commands/resources/plugins.py
Original file line number Diff line number Diff line change
@@ -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):
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Interesting! I think we might want to hoist this to common.py and make it a standard practice for all places that we use an enum for explicit choices. That can happen later though.

"""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)
16 changes: 16 additions & 0 deletions cforge/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -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
# ---------------------------------------------------------------------------
Expand Down
154 changes: 154 additions & 0 deletions tests/commands/resources/test_plugins.py
Original file line number Diff line number Diff line change
@@ -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)