Coverage for mcpgateway / services / plugin_service.py: 99%
107 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-02-11 07:10 +0000
1# -*- coding: utf-8 -*-
2"""Location: ./mcpgateway/services/plugin_service.py
3Copyright 2025
4SPDX-License-Identifier: Apache-2.0
5Authors: Mihai Criveti
7Plugin service for managing and querying MCP Gateway plugins.
8This module provides a service layer for accessing plugin information,
9statistics, and configuration from the PluginManager.
10"""
12# Standard
13from collections import defaultdict
14import logging
15from typing import Any, Dict, List, Optional
17# First-Party
18from mcpgateway.plugins.framework import PluginManager
19from mcpgateway.plugins.framework.models import PluginMode
21logger = logging.getLogger(__name__)
23# Cache import (lazy to avoid circular dependencies)
24_ADMIN_STATS_CACHE = None
27def _get_admin_stats_cache():
28 """Get admin stats cache singleton lazily.
30 Returns:
31 AdminStatsCache instance.
32 """
33 global _ADMIN_STATS_CACHE # pylint: disable=global-statement
34 if _ADMIN_STATS_CACHE is None:
35 # First-Party
36 from mcpgateway.cache.admin_stats_cache import admin_stats_cache # pylint: disable=import-outside-toplevel
38 _ADMIN_STATS_CACHE = admin_stats_cache
39 return _ADMIN_STATS_CACHE
42class PluginService:
43 """Service for managing plugin information and statistics."""
45 def __init__(self, plugin_manager: Optional[PluginManager] = None):
46 """Initialize the plugin service.
48 Args:
49 plugin_manager: Optional PluginManager instance. If not provided,
50 attempts to get from app state will be made at runtime.
51 """
52 self._plugin_manager = plugin_manager
54 def get_plugin_manager(self) -> Optional[PluginManager]:
55 """Get the plugin manager instance.
57 Returns:
58 PluginManager instance or None if plugins are disabled.
59 """
60 return self._plugin_manager
62 def set_plugin_manager(self, manager: PluginManager) -> None:
63 """Set the plugin manager instance.
65 Args:
66 manager: The PluginManager instance to use.
67 """
68 self._plugin_manager = manager
70 def get_all_plugins(self) -> List[Dict[str, Any]]:
71 """Get all registered plugins with their configuration, including disabled plugins.
73 Returns:
74 List of plugin dictionaries containing configuration and status.
75 """
76 if not self._plugin_manager:
77 return []
79 plugins = []
80 registry = self._plugin_manager._registry # pylint: disable=protected-access
81 config = self._plugin_manager._config # pylint: disable=protected-access
83 # First, add all registered (enabled) plugins from the registry
84 registered_names = set()
85 for plugin_ref in registry.get_all_plugins():
86 # Get the plugin config from the plugin reference
87 plugin_config = plugin_ref.plugin.config if hasattr(plugin_ref, "plugin") else plugin_ref._plugin.config if hasattr(plugin_ref, "_plugin") else None # pylint: disable=protected-access
89 plugin_dict = {
90 "name": plugin_ref.name,
91 "description": plugin_config.description if plugin_config and plugin_config.description else "",
92 "author": plugin_config.author if plugin_config and plugin_config.author else "Unknown",
93 "version": plugin_config.version if plugin_config and plugin_config.version else "0.0.0",
94 "mode": plugin_ref.mode if isinstance(plugin_ref.mode, str) else plugin_ref.mode.value if plugin_ref.mode else "disabled",
95 "priority": plugin_ref.priority,
96 "hooks": [hook if isinstance(hook, str) else hook.value for hook in plugin_ref.hooks] if plugin_ref.hooks else [],
97 "tags": plugin_ref.tags or [],
98 "kind": plugin_config.kind if plugin_config and plugin_config.kind else "",
99 "namespace": plugin_config.namespace if plugin_config and plugin_config.namespace else "",
100 "status": "enabled" if plugin_ref.mode != PluginMode.DISABLED else "disabled",
101 }
103 # Add implementation type if available (e.g., Rust vs Python for PII filter)
104 plugin_instance = plugin_ref.plugin if hasattr(plugin_ref, "plugin") else plugin_ref._plugin if hasattr(plugin_ref, "_plugin") else None # pylint: disable=protected-access
105 if plugin_instance and hasattr(plugin_instance, "implementation"):
106 plugin_dict["implementation"] = plugin_instance.implementation
108 # Add config summary (first few keys only for list view)
109 if plugin_config and hasattr(plugin_config, "config") and plugin_config.config:
110 config_keys = list(plugin_config.config.keys())[:5]
111 plugin_dict["config_summary"] = {k: plugin_config.config[k] for k in config_keys}
112 else:
113 plugin_dict["config_summary"] = {}
115 plugins.append(plugin_dict)
116 registered_names.add(plugin_ref.name)
118 # Then, add disabled plugins from the configuration (not in registry)
119 if config and config.plugins:
120 for plugin_config in config.plugins:
121 if plugin_config.mode == PluginMode.DISABLED and plugin_config.name not in registered_names:
122 plugin_dict = {
123 "name": plugin_config.name,
124 "description": plugin_config.description or "",
125 "author": plugin_config.author or "Unknown",
126 "version": plugin_config.version or "0.0.0",
127 "mode": plugin_config.mode if isinstance(plugin_config.mode, str) else plugin_config.mode.value,
128 "priority": plugin_config.priority or 100,
129 "hooks": [hook if isinstance(hook, str) else hook.value for hook in plugin_config.hooks] if plugin_config.hooks else [],
130 "tags": plugin_config.tags or [],
131 "kind": plugin_config.kind or "",
132 "namespace": plugin_config.namespace or "",
133 "status": "disabled",
134 "config_summary": {},
135 }
137 # Add config summary (first few keys only for list view)
138 if hasattr(plugin_config, "config") and plugin_config.config:
139 config_keys = list(plugin_config.config.keys())[:5]
140 plugin_dict["config_summary"] = {k: plugin_config.config[k] for k in config_keys}
142 plugins.append(plugin_dict)
144 return sorted(plugins, key=lambda x: x["priority"])
146 def get_plugin_by_name(self, name: str) -> Optional[Dict[str, Any]]:
147 """Get detailed information about a specific plugin.
149 Args:
150 name: The name of the plugin to retrieve.
152 Returns:
153 Plugin dictionary with full configuration or None if not found.
154 """
155 if not self._plugin_manager:
156 return None
158 registry = self._plugin_manager._registry # pylint: disable=protected-access
159 plugin_ref = registry.get_plugin(name)
161 if plugin_ref:
162 # Get the plugin config from the plugin reference
163 plugin_config = plugin_ref.plugin.config if hasattr(plugin_ref, "plugin") else plugin_ref._plugin.config if hasattr(plugin_ref, "_plugin") else None # pylint: disable=protected-access
165 plugin_dict = {
166 "name": plugin_ref.name,
167 "description": plugin_config.description if plugin_config and plugin_config.description else "",
168 "author": plugin_config.author if plugin_config and plugin_config.author else "Unknown",
169 "version": plugin_config.version if plugin_config and plugin_config.version else "0.0.0",
170 "mode": plugin_ref.mode if isinstance(plugin_ref.mode, str) else plugin_ref.mode.value if plugin_ref.mode else "disabled",
171 "priority": plugin_ref.priority,
172 "hooks": [hook if isinstance(hook, str) else hook.value for hook in plugin_ref.hooks] if plugin_ref.hooks else [],
173 "tags": plugin_ref.tags or [],
174 "kind": plugin_config.kind if plugin_config and plugin_config.kind else "",
175 "namespace": plugin_config.namespace if plugin_config and plugin_config.namespace else "",
176 "status": "enabled" if plugin_ref.mode != PluginMode.DISABLED else "disabled",
177 "conditions": plugin_ref.conditions or [],
178 "config": plugin_config.config if plugin_config and hasattr(plugin_config, "config") else {},
179 }
181 # Add manifest info if available
182 if hasattr(plugin_ref, "manifest"):
183 plugin_dict["manifest"] = {"available_hooks": plugin_ref.manifest.available_hooks, "default_config": plugin_ref.manifest.default_config}
185 return plugin_dict
187 # Fallback: check config for disabled plugins not in registry
188 config = self._plugin_manager._config # pylint: disable=protected-access
189 if config and config.plugins: 189 ↛ 208line 189 didn't jump to line 208 because the condition on line 189 was always true
190 for plugin_config in config.plugins:
191 if plugin_config.name == name:
192 return {
193 "name": plugin_config.name,
194 "description": plugin_config.description or "",
195 "author": plugin_config.author or "Unknown",
196 "version": plugin_config.version or "0.0.0",
197 "mode": plugin_config.mode if isinstance(plugin_config.mode, str) else plugin_config.mode.value,
198 "priority": plugin_config.priority or 100,
199 "hooks": [hook if isinstance(hook, str) else hook.value for hook in plugin_config.hooks] if plugin_config.hooks else [],
200 "tags": plugin_config.tags or [],
201 "kind": plugin_config.kind or "",
202 "namespace": plugin_config.namespace or "",
203 "status": "disabled",
204 "conditions": plugin_config.conditions or [],
205 "config": plugin_config.config if hasattr(plugin_config, "config") else {},
206 }
208 return None
210 async def get_plugin_statistics(self) -> Dict[str, Any]:
211 """Get statistics about all plugins.
213 Returns:
214 Dictionary containing plugin statistics by various dimensions.
215 """
216 if not self._plugin_manager:
217 return {
218 "total_plugins": 0,
219 "enabled_plugins": 0,
220 "disabled_plugins": 0,
221 "plugins_by_hook": {},
222 "plugins_by_mode": {},
223 "plugins_by_tag": {},
224 "plugins_by_author": {},
225 }
227 # Check cache first
228 cache = _get_admin_stats_cache()
229 cached = await cache.get_plugin_stats()
230 if cached is not None:
231 return cached
233 all_plugins = self.get_all_plugins()
235 # Count by status
236 enabled_count = sum(1 for p in all_plugins if p["status"] == "enabled")
237 disabled_count = sum(1 for p in all_plugins if p["status"] == "disabled")
239 # Count by hook
240 hooks_count = defaultdict(int)
241 for plugin in all_plugins:
242 for hook in plugin["hooks"]:
243 hooks_count[hook] += 1
245 # Count by mode
246 mode_count = defaultdict(int)
247 for plugin in all_plugins:
248 mode_count[plugin["mode"]] += 1
250 # Count by tag
251 tag_count = defaultdict(int)
252 for plugin in all_plugins:
253 for tag in plugin["tags"]:
254 tag_count[tag] += 1
256 # Count by author
257 author_count = defaultdict(int)
258 for plugin in all_plugins:
259 author = plugin.get("author", "Unknown")
260 author_count[author] += 1
262 stats = {
263 "total_plugins": len(all_plugins),
264 "enabled_plugins": enabled_count,
265 "disabled_plugins": disabled_count,
266 "plugins_by_hook": dict(hooks_count),
267 "plugins_by_mode": dict(mode_count),
268 "plugins_by_tag": dict(sorted(tag_count.items(), key=lambda x: x[1], reverse=True)[:10]), # Top 10 tags
269 "plugins_by_author": dict(sorted(author_count.items(), key=lambda x: x[1], reverse=True)), # All authors sorted by count
270 }
272 # Store in cache
273 await cache.set_plugin_stats(stats)
275 return stats
277 def search_plugins(self, query: Optional[str] = None, mode: Optional[str] = None, hook: Optional[str] = None, tag: Optional[str] = None) -> List[Dict[str, Any]]:
278 """Search and filter plugins based on criteria.
280 Args:
281 query: Text search in name, description, author.
282 mode: Filter by mode (enforce, permissive, disabled).
283 hook: Filter by hook type.
284 tag: Filter by tag.
286 Returns:
287 Filtered list of plugins.
288 """
289 plugins = self.get_all_plugins()
291 # Text search
292 if query:
293 query_lower = query.lower()
294 plugins = [p for p in plugins if query_lower in p["name"].lower() or query_lower in p["description"].lower() or query_lower in p["author"].lower()]
296 # Mode filter
297 if mode:
298 plugins = [p for p in plugins if p["mode"] == mode]
300 # Hook filter
301 if hook:
302 plugins = [p for p in plugins if hook in p["hooks"]]
304 # Tag filter
305 if tag:
306 plugins = [p for p in plugins if tag in p["tags"]]
308 return plugins
311# Singleton instance
312_plugin_service = PluginService()
315def get_plugin_service() -> PluginService:
316 """Get the singleton plugin service instance.
318 Returns:
319 The global PluginService instance.
320 """
321 return _plugin_service