feat: initial revision of adding plugin support.

.env.example

@@ -290,3 +290,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

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

docs/docs/using/plugins/index.md

@@ -46,6 +46,10 @@ PLUGIN_CONFIG_FILE=plugins/config.yaml
 
 ### 2. Plugin Configuration
 
+The plugin configuration file is used to configure a set of plugins to run a
+set of hook points throughout the MCP Context Forge.  An example configuration
+is below.  It contains two main sections: `plugins` and `plugin_settings`.
+
 Create or modify `plugins/config.yaml`:
 
 ```yaml
@@ -78,6 +82,35 @@ plugin_settings:
   plugin_health_check_interval: 60
 ```
 
+The `plugins` section lists the set of configured plugins that will be loaded
+by the Context Forge at startup.  Each plugin contains a set of standard configurations,
+and then a `config` section designed for plugin specific configurations. The attributes
+are defined as follows:
+
+| Attribute | Description | Example Value |
+|-----------|-------------|---------------|
+| **name**  | A unique name for the plugin. | MyFirstPlugin |
+| **kind**  | A fully qualified string representing the plugin python object. | plugins.native.content_filter.ContentFilterPlugin |
+| **description** | The description of the plugin configuration. | A plugin for replacing bad words. |
+| **version** | The version of the plugin configuration. | 0.1 |
+| **author** | The team that wrote the plugin. | MCP Context Forge |
+| **hooks** | A list of hooks for which the plugin will be executed. **Note**: currently supports two hooks: "prompt_pre_fetch", "prompt_post_fetch"  | ["prompt_pre_fetch", "prompt_post_fetch"] |
+| **tags** | Descriptive keywords that make the configuration searchable. | ["security", "filter"] |
+| **mode** | Mode of operation of the plugin. - enforce (stops during a violation), permissive (audits a violation but doesn't stop), disabled (disabled) | permissive |
+| **priority** | The priority in which the plugin will run - 0 is higher priority | 100 |
+| **conditions** | A list of conditions under which a plugin is run. See section on conditions.|  |
+| **config** | Plugin specific configuration.  This is a dictionary and is passed to the plugin on initialization. |   |
+
+The `plugin_settings` are as follows:
+
+| Attribute | Description | Example Value |
+|-----------|-------------|---------------|
+| **parallel_execution_within_band** | Plugins in the same band are run in parallel (currently not implemented). | true or false |
+| **plugin_timeout** | The time in seconds before stopping plugin execution (not implemented). | 30 |
+| **fail_on_plugin_error** | Cause the execution of the task to fail if the plugin errors. | true or false |
+| **plugin_health_check_interval** | Health check interval in seconds (not implemented). | 60 |
+
+
 ### 3. Execution Modes
 
 Each plugin can operate in one of three modes:
@@ -110,6 +143,18 @@ plugins:
 
 Plugins with the same priority may execute in parallel if `parallel_execution_within_band` is enabled.
 
+### 5. Conditions of Execution
+
+Users may only want plugins to be invoked on specific servers, tools, and prompts. To address this, a set of conditionals can be applied to a plugin. The attributes in a conditional combine together in as a set of `and` operations, while each attribute list item is `ored` with other items in the list.  The attributes are defined as follows:
+
+| Attribute | Description
+|-----------|------------|
+| **server_ids** | The list of MCP servers on which the plugin will trigger |
+| **tools** | The list of tools on which the plugin will be applied. |
+| **prompts** | The list of prompts on which the plugin will be applied. |
+| **user_patterns** | The list of users on which the plugin will be applied. |
+| **content_types** | The list of content types on which the plugin will trigger. |
+
 ## Available Hooks
 
 Currently implemented hooks:

mcpgateway/config.py

@@ -311,6 +311,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

mcpgateway/main.py

@@ -75,6 +75,7 @@
     ResourceContent,
     Root,
 )
+from mcpgateway.plugins import PluginManager, PluginViolationError
 from mcpgateway.schemas import (
     GatewayCreate,
     GatewayRead,
@@ -160,6 +161,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()
@@ -214,6 +217,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()
@@ -232,6 +238,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]:
@@ -1703,9 +1716,11 @@ async def get_prompt(
         PromptExecuteArgs(args=args)
         return await prompt_service.get_prompt(db, name, args)
     except Exception as ex:
-        logger.error(f"Error retrieving prompt {name}: {ex}")
-        if isinstance(ex, ValueError):
+        logger.error(f"Could not retrieve prompt {name}: {ex}")
+        if isinstance(ex, ValueError) or isinstance(ex, PromptError):
             return JSONResponse(content={"message": "Prompt execution arguments contains HTML tags that may cause security issues"}, status_code=422)
+        if isinstance(ex, PluginViolationError):
+            return JSONResponse(content={"message": "Prompt execution arguments contains HTML tags that may cause security issues", "details": ex.message}, status_code=422)
 
 
 @prompt_router.get("/{name}")

mcpgateway/plugins/__init__.py

@@ -0,0 +1,26 @@
+# -*- coding: utf-8 -*-
+"""Services Package.
+
+Copyright 2025
+SPDX-License-Identifier: Apache-2.0
+Authors: Fred Araujo
+
+Exposes core MCP Gateway plugin components:
+- Context
+- Manager
+- Payloads
+- Models
+"""
+
+from mcpgateway.plugins.framework.manager import PluginManager
+from mcpgateway.plugins.framework.models import PluginViolation
+from mcpgateway.plugins.framework.plugin_types import GlobalContext, PluginViolationError, PromptPosthookPayload, PromptPrehookPayload
+
+__all__ = [
+    "GlobalContext",
+    "PluginManager",
+    "PluginViolation",
+    "PluginViolationError",
+    "PromptPosthookPayload",
+    "PromptPrehookPayload",
+]

mcpgateway/plugins/framework/base.py

@@ -0,0 +1,216 @@
+# -*- coding: utf-8 -*-
+"""Base plugin implementation.
+
+Copyright 2025
+SPDX-License-Identifier: Apache-2.0
+Authors: Teryl Taylor
+
+This module implements the base plugin object.
+It supports pre and post hooks AI safety, security and business processing
+for the following locations in the server:
+server_pre_register / server_post_register - for virtual server verification
+tool_pre_invoke / tool_post_invoke - for guardrails
+prompt_pre_fetch / prompt_post_fetch - for prompt filtering
+resource_pre_fetch / resource_post_fetch - for content filtering
+auth_pre_check / auth_post_check - for custom auth logic
+federation_pre_sync / federation_post_sync - for gateway federation
+"""
+
+# Standard
+import uuid
+
+# First-Party
+from mcpgateway.plugins.framework.models import HookType, PluginCondition, PluginConfig, PluginMode
+from mcpgateway.plugins.framework.plugin_types import (
+    PluginContext,
+    PromptPosthookPayload,
+    PromptPosthookResult,
+    PromptPrehookPayload,
+    PromptPrehookResult,
+)
+
+
+class Plugin:
+    """Base plugin object for pre/post processing of inputs and outputs at various locations throughout the server."""
+
+    def __init__(self, config: PluginConfig) -> None:
+        """Initialize a plugin with a configuration and context.
+
+        Args:
+            config: The plugin configuration
+        """
+        self._config = config
+
+    @property
+    def priority(self) -> int:
+        """Return the plugin's priority.
+
+        Returns:
+            Plugin's priority.
+        """
+        return self._config.priority
+
+    @property
+    def mode(self) -> PluginMode:
+        """Return the plugin's mode.
+
+        Returns:
+            Plugin's mode.
+        """
+        return self._config.mode
+
+    @property
+    def name(self) -> str:
+        """Return the plugin's name.
+
+        Returns:
+            Plugin's name.
+        """
+        return self._config.name
+
+    @property
+    def hooks(self) -> list[HookType]:
+        """Return the plugin's currently configured hooks.
+
+        Returns:
+            Plugin's configured hooks.
+        """
+        return self._config.hooks
+
+    @property
+    def tags(self) -> list[str]:
+        """Return the plugin's tags.
+
+        Returns:
+            Plugin's tags.
+        """
+        return self._config.tags
+
+    @property
+    def conditions(self) -> list[PluginCondition] | None:
+        """Return the plugin's conditions for operation.
+
+        Returns:
+            Plugin's conditions for executing.
+        """
+        return self._config.conditions
+
+    async def prompt_pre_fetch(self, payload: PromptPrehookPayload, context: PluginContext) -> PromptPrehookResult:
+        """Plugin hook run before a prompt is retrieved and rendered.
+
+        Args:
+            payload: The prompt payload to be analyzed.
+            context: contextual information about the hook call. Including why it was called.
+
+        Raises:
+            NotImplementedError: needs to be implemented by sub class.
+        """
+        raise NotImplementedError(
+            f"""'prompt_pre_fetch' not implemented for plugin {self._config.name}
+                                    of plugin type {type(self)}
+                                   """
+        )
+
+    async def prompt_post_fetch(self, payload: PromptPosthookPayload, context: PluginContext) -> PromptPosthookResult:
+        """Plugin hook run after a prompt is rendered.
+
+        Args:
+            payload: The prompt payload to be analyzed.
+            context: Contextual information about the hook call.
+
+        Raises:
+            NotImplementedError: needs to be implemented by sub class.
+        """
+        raise NotImplementedError(
+            f"""'prompt_post_fetch' not implemented for plugin {self._config.name}
+                                    of plugin type {type(self)}
+                                   """
+        )
+
+    async def shutdown(self) -> None:
+        """Plugin cleanup code."""
+
+
+class PluginRef:
+    """Plugin reference which contains a uuid."""
+
+    def __init__(self, plugin: Plugin):
+        """Initialize a plugin reference.
+
+        Args:
+            plugin: The plugin to reference.
+        """
+        self._plugin = plugin
+        self._uuid = uuid.uuid4()
+
+    @property
+    def plugin(self) -> Plugin:
+        """Return the underlying plugin.
+
+        Returns:
+            The underlying plugin.
+        """
+        return self._plugin
+
+    @property
+    def uuid(self) -> str:
+        """Return the plugin's UUID.
+
+        Returns:
+            Plugin's UUID.
+        """
+        return self._uuid.hex
+
+    @property
+    def priority(self) -> int:
+        """Returns the plugin's priority.
+
+        Returns:
+            Plugin's priority.
+        """
+        return self._plugin.priority
+
+    @property
+    def name(self) -> str:
+        """Return the plugin's name.
+
+        Returns:
+            Plugin's name.
+        """
+        return self._plugin.name
+
+    @property
+    def hooks(self) -> list[HookType]:
+        """Returns the plugin's currently configured hooks.
+
+        Returns:
+            Plugin's configured hooks.
+        """
+        return self._plugin.hooks
+
+    @property
+    def tags(self) -> list[str]:
+        """Return the plugin's tags.
+
+        Returns:
+            Plugin's tags.
+        """
+        return self._plugin.tags
+
+    @property
+    def conditions(self) -> list[PluginCondition] | None:
+        """Return the plugin's conditions for operation.
+
+        Returns:
+            Plugin's conditions for operation.
+        """
+        return self._plugin.conditions
+
+    @property
+    def mode(self) -> PluginMode:
+        """Return the plugin's mode.
+
+        Returns:
+            Plugin's mode.
+        """
+        return self.plugin.mode