Skip to content
Merged
Show file tree
Hide file tree
Changes from 3 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
Copy Markdown
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
115 changes: 115 additions & 0 deletions cforge/commands/resources/plugins.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,115 @@
# -*- 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 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,
)


def _handle_plugins_exception(exception: Exception) -> None:
"""Provide plugin-specific hints and raise a CLI error."""
console = get_console()

if isinstance(exception, AuthenticationError):
Comment thread
MatthewGrigsby marked this conversation as resolved.
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[str] = typer.Option(None, "--mode", help="Filter by mode (enforce, permissive, disabled)"),
Comment thread
MatthewGrigsby marked this conversation as resolved.
Outdated
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
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]] = []
if isinstance(result, dict):
if "plugins" in result:
raw_plugins = result.get("plugins", [])
if isinstance(raw_plugins, list):
plugins = raw_plugins
elif isinstance(raw_plugins, dict):
plugins = [raw_plugins]
else:
plugins = [result]
elif isinstance(result, list):
Comment thread
MatthewGrigsby marked this conversation as resolved.
Outdated
plugins = result

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
142 changes: 142 additions & 0 deletions tests/commands/resources/test_plugins.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,142 @@
# -*- 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 plugins_get, plugins_list, plugins_stats
from cforge.common import AuthenticationError, CLIError
from tests.conftest import patch_functions


class TestPluginCommands:
"""Tests for plugins commands."""

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:
plugins_list(json_output=False)
Comment thread
MatthewGrigsby marked this conversation as resolved.
Outdated
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:
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}):
plugins_list(json_output=False)

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:
plugins_list(search="pii", mode="enforce", hook="tool_pre_invoke", tag="security", json_output=False)

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):
plugins_list(json_output=False)

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:
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):
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:
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):
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):
plugins_list(json_output=False)

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):
plugins_list(json_output=False)

assert any("Admin plugin API unavailable" in str(call) for call in mock_console.print.call_args_list)