feat(plugins): integrated plugins into prompt service, fixed linting and type issues.

Signed-off-by: Teryl Taylor <[email protected]>

Author: Teryl Taylor

.env.example

@@ -285,3 +285,8 @@ DEBUG=false
 # Gateway tool name separator
 GATEWAY_TOOL_NAME_SEPARATOR=-
 VALID_SLUG_SEPARATOR_REGEXP= r"^(-{1,2}|[_.])$"
+
+#####################################
+# Plugins Settings
+#####################################
+PLUGINS_ENABLED=false

MANIFEST.in

@@ -55,6 +55,8 @@ recursive-include alembic *.md
 recursive-include alembic *.py
 # recursive-include deployment *
 # recursive-include mcp-servers *
+recursive-include plugins *.py
+recursive-include plugins *.yaml
 
 # 5️⃣  (Optional) include MKDocs-based docs in the sdist
 # graft docs

mcpgateway/config.py

@@ -309,6 +309,10 @@ def _parse_federation_peers(cls, v):
     use_stateful_sessions: bool = False  # Set to False to use stateless sessions without event store
     json_response_enabled: bool = True  # Enable JSON responses instead of SSE streams
 
+    # Core plugin settings
+    plugins_enabled: bool = Field(default=False, description="Enable the plugin framework")
+    plugin_config_file: str = Field(default="plugins/config.yaml", description="Path to main plugin configuration file")
+
     # Development
     dev_mode: bool = False
     reload: bool = False
@@ -495,9 +499,7 @@ def validate_database(self) -> None:
                 db_dir.mkdir(parents=True)
 
     # Validation patterns for safe display (configurable)
-    validation_dangerous_html_pattern: str = (
-        r"<(script|iframe|object|embed|link|meta|base|form|img|svg|video|audio|source|track|area|map|canvas|applet|frame|frameset|html|head|body|style)\b|</*(script|iframe|object|embed|link|meta|base|form|img|svg|video|audio|source|track|area|map|canvas|applet|frame|frameset|html|head|body|style)>"
-    )
+    validation_dangerous_html_pattern: str = r"<(script|iframe|object|embed|link|meta|base|form|img|svg|video|audio|source|track|area|map|canvas|applet|frame|frameset|html|head|body|style)\b|</*(script|iframe|object|embed|link|meta|base|form|img|svg|video|audio|source|track|area|map|canvas|applet|frame|frameset|html|head|body|style)>"
 
     validation_dangerous_js_pattern: str = r"(?i)(?:^|\s|[\"'`<>=])(javascript:|vbscript:|data:\s*[^,]*[;\s]*(javascript|vbscript)|\bon[a-z]+\s*=|<\s*script\b)"
 

mcpgateway/main.py

@@ -73,6 +73,7 @@
     ResourceContent,
     Root,
 )
+from mcpgateway.plugins.framework.manager import PluginManager
 from mcpgateway.schemas import (
     GatewayCreate,
     GatewayRead,
@@ -158,6 +159,8 @@
 else:
     loop.create_task(bootstrap_db())
 
+# Initialize plugin manager as a singleton.
+plugin_manager: PluginManager | None = PluginManager(settings.plugin_config_file) if settings.plugins_enabled else None
 
 # Initialize services
 tool_service = ToolService()
@@ -212,6 +215,9 @@ async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
     """
     logger.info("Starting MCP Gateway services")
     try:
+        if plugin_manager:
+            await plugin_manager.initialize()
+            logger.info(f"Plugin manager initialized with {plugin_manager.plugin_count} plugins")
         await tool_service.initialize()
         await resource_service.initialize()
         await prompt_service.initialize()
@@ -230,6 +236,13 @@ async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
         logger.error(f"Error during startup: {str(e)}")
         raise
     finally:
+        # Shutdown plugin manager
+        if plugin_manager:
+            try:
+                await plugin_manager.shutdown()
+                logger.info("Plugin manager shutdown complete")
+            except Exception as e:
+                logger.error(f"Error shutting down plugin manager: {str(e)}")
         logger.info("Shutting down MCP Gateway services")
         # await stop_streamablehttp()
         for service in [resource_cache, sampling_handler, logging_service, completion_service, root_service, gateway_service, prompt_service, resource_service, tool_service, streamable_http_session]:

mcpgateway/plugins/framework/base.py

@@ -123,7 +123,7 @@ async def prompt_post_fetch(self, payload: PromptPosthookPayload, context: Plugi
                                     of plugin type {type(self)}
                                    """)
 
-    def shutdown(self) -> None:
+    async def shutdown(self) -> None:
         """Plugin cleanup code."""
 
 

mcpgateway/plugins/framework/loader/plugin.py

@@ -28,16 +28,32 @@ def __init__(self) -> None:
         self._plugin_types: dict[str, Type[Plugin]] = {}
 
     def __get_plugin_type(self, kind: str) -> Type[Plugin]:
+        """Import a plugin type from a python module.
+
+        Args:
+            kind: The fully-qualified type of the plugin to be registered.
+
+        Raises:
+            Exception: if unable to import a module.
+
+        Returns:
+            A plugin type.
+        """
         try:
             (mod_name, cls_name) = parse_class_name(kind)
             module = import_module(mod_name)
             class_ = getattr(module, cls_name)
             return cast(Type[Plugin], class_)
         except Exception:
-            logger.exception("Unable to instantiate class '%s'", kind)
+            logger.exception("Unable to import plugin type '%s'", kind)
             raise
 
     def __register_plugin_type(self, kind: str) -> None:
+        """Register a plugin type.
+
+        Args:
+            kind: The fully-qualified type of the plugin to be registered.
+        """
         if kind not in self._plugin_types:
             plugin_type = self.__get_plugin_type(kind)
             self._plugin_types[kind] = plugin_type

mcpgateway/plugins/framework/manager.py

@@ -32,11 +32,12 @@
 
 logger = logging.getLogger(__name__)
 
-T = TypeVar('T')
+T = TypeVar("T")
 
 
 class PluginExecutor(Generic[T]):
     """Executes a list of plugins."""
+
     async def execute(
         self,
         plugins: list[PluginRef],
@@ -45,15 +46,15 @@ async def execute(
         plugin_run: Callable[[PluginRef, T, PluginContext], Coroutine[Any, Any, PluginResult[T]]],
         compare: Callable[[T, list[PluginCondition], GlobalContext], bool],
         local_contexts: Optional[PluginContextTable] = None,
-    ) -> tuple[PluginResult[T] | None, PluginContextTable | None]:
+    ) -> tuple[PluginResult[T], PluginContextTable | None]:
         """Execute a plugins hook run before a prompt is retrieved and rendered.
 
         Args:
             plugins: the list of plugins to execute.
             payload: the payload to be analyzed.
             global_context: contextual information for all plugins.
             plugin_run: async function for executing plugin hook.
-            compare: function for comparing conditional information with context and payload 
+            compare: function for comparing conditional information with context and payload.
             local_contexts: context local to a single plugin.
 
         Returns:
@@ -85,11 +86,11 @@ async def execute(
             if not result.continue_processing:
                 # Check execution mode
                 if pluginref.plugin.mode == PluginMode.ENFORCE:
-                    return (PluginResult[T](continue_processing=False, modified_payload=current_payload, error=result.error, metadata=combined_metadata), None)
+                    return (PluginResult[T](continue_processing=False, modified_payload=current_payload, violation=result.violation, metadata=combined_metadata), None)
                 elif pluginref.plugin.mode == PluginMode.PERMISSIVE:
-                    logger.warning(f"Plugin {pluginref.plugin.name} would block (permissive mode): {result.error}")
+                    logger.warning(f"Plugin {pluginref.plugin.name} would block (permissive mode): {result.violation.description if result.violation else ''}")
 
-        return (PluginResult[T](continue_processing=True, modified_payload=current_payload, error=None, metadata=combined_metadata), res_local_contexts)
+        return (PluginResult[T](continue_processing=True, modified_payload=current_payload, violation=None, metadata=combined_metadata), res_local_contexts)
 
 
 async def pre_prompt_fetch(plugin: PluginRef, payload: PromptPrehookPayload, context: PluginContext) -> PromptPrehookResult:
@@ -176,7 +177,7 @@ async def initialize(self) -> None:
         """
         if self._initialized:
             return
-        
+
         plugins = self._config.plugins if self._config else []
 
         for plugin_config in plugins:
@@ -204,7 +205,7 @@ async def prompt_pre_fetch(
         payload: PromptPrehookPayload,
         global_context: GlobalContext,
         local_contexts: Optional[PluginContextTable] = None,
-    ) -> tuple[PromptPrehookResult | None, PluginContextTable | None]:
+    ) -> tuple[PromptPrehookResult, PluginContextTable | None]:
         """Plugin hook run before a prompt is retrieved and rendered.
 
         Args:
@@ -218,11 +219,9 @@ async def prompt_pre_fetch(
         plugins = self._registry.get_plugins_for_hook(HookType.PROMPT_PRE_FETCH)
         return await self._pre_prompt_executor.execute(plugins, payload, global_context, pre_prompt_fetch, pre_prompt_matches, local_contexts)
 
-
-
     async def prompt_post_fetch(
         self, payload: PromptPosthookPayload, global_context: GlobalContext, local_contexts: Optional[PluginContextTable] = None
-    ) -> tuple[PromptPosthookResult | None, PluginContextTable | None]:
+    ) -> tuple[PromptPosthookResult, PluginContextTable | None]:
         """Plugin hook run after a prompt is rendered.
 
         Args:

mcpgateway/plugins/framework/models.py

@@ -157,19 +157,19 @@ class PluginManifest(BaseModel):
     default_config: dict[str, Any]
 
 
-class PluginError(BaseModel):  # (ErrorResponse):  # Inherits from MCP error format
-    """A plugin error.
+class PluginViolation(BaseModel):
+    """A plugin filter violation.
 
     Attributes:
         plugin_name (str): The name of the plugin.
-        error_description (str): the error in text.
-        error_code (str): an error code.
+        description (str): the violation in text.
+        violation_code (str): a violation code.
         details: (dict[str, Any])
     """
 
     plugin_name: str
-    error_description: str
-    error_code: str
+    description: str
+    violation_code: str
     details: dict[str, Any]
 
 

mcpgateway/plugins/framework/registry.py

@@ -106,6 +106,7 @@ def get_all_plugins(self) -> list[PluginRef]:
         """
         return list(self._plugins.values())
 
+    @property
     def plugin_count(self) -> int:
         """Return the number of plugins registered.
 

mcpgateway/plugins/framework/types.py

@@ -14,7 +14,7 @@
 
 # First-Party
 from mcpgateway.models import PromptResult
-from mcpgateway.plugins.framework.models import PluginError
+from mcpgateway.plugins.framework.models import PluginViolation
 
 T = TypeVar("T")
 
@@ -50,18 +50,18 @@ def __init__(self, name: str, result: PromptResult):
 class PluginResult(Generic[T]):
     """A plugin result."""
 
-    def __init__(self, continue_processing: bool = True, modified_payload: Optional[T] = None, error: Optional[PluginError] = None, metadata: Optional[dict[str, Any]] = None):
+    def __init__(self, continue_processing: bool = True, modified_payload: Optional[T] = None, violation: Optional[PluginViolation] = None, metadata: Optional[dict[str, Any]] = None):
         """Initialize a plugin result object.
 
         Args:
             continue_processing (bool): Whether to stop processing.
             modified_payload (Optional[Any]): The modified payload if the plugin is a transformer.
-            error (Optional[PluginError]): error object.
+            violation (Optional[PluginViolation]): violation object.
             metadata (Optional[dict[str, Any]]): additional metadata.
         """
         self.continue_processing = continue_processing
         self.modified_payload = modified_payload
-        self.error = error
+        self.violation = violation
         self.metadata = metadata or {}
 
 
@@ -72,13 +72,7 @@ def __init__(self, continue_processing: bool = True, modified_payload: Optional[
 class GlobalContext:
     """The global context, which shared across all plugins."""
 
-    def __init__(
-        self,
-        request_id: str,
-        user: Optional[str] = None,
-        tenant_id: Optional[str] = None,
-        server_id: Optional[str] = None
-    ) -> None:
+    def __init__(self, request_id: str, user: Optional[str] = None, tenant_id: Optional[str] = None, server_id: Optional[str] = None) -> None:
         """Initialize a global context.
 
         Args:
@@ -92,6 +86,7 @@ def __init__(
         self.tenant_id = tenant_id
         self.server_id = server_id
 
+
 class PluginContext(GlobalContext):
     """The plugin's context, which lasts a request lifecycle.