feat: initial revision of adding plugin support.

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

Author: Teryl Taylor

mcpgateway/plugins/framework/base.py

@@ -0,0 +1,212 @@
+# -*- 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.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)}
+                                   """)
+
+    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

mcpgateway/plugins/framework/loader/config.py

@@ -0,0 +1,92 @@
+# -*- coding: utf-8 -*-
+"""Configuration loader implementation.
+
+Copyright 2025
+SPDX-License-Identifier: Apache-2.0
+Authors: Teryl Taylor
+
+This module loads configurations for plugins.
+"""
+
+# Standard
+import os
+
+# Third-Party
+import jinja2
+import yaml
+
+# First-Party
+from mcpgateway.plugins.framework.models import Config, PluginConfig, PluginManifest
+
+
+class ConfigLoader:
+    """A configuration loader."""
+
+    @staticmethod
+    def load_config(config: str, use_jinja: bool = True) -> Config:
+        """Load the plugin configuration from a file path.
+
+        Args:
+            config: the configuration path.
+            use_jinja: use jinja to replace env variables if true.
+
+        Returns:
+            The plugin configuration object.
+        """
+        with open(os.path.normpath(config), "r", encoding="utf-8") as file:
+            template = file.read()
+            if use_jinja:
+                jinja_env = jinja2.Environment(loader=jinja2.BaseLoader())
+                rendered_template = jinja_env.from_string(template).render(env=os.environ)
+            else:
+                rendered_template = template
+            config_data = yaml.safe_load(rendered_template)
+        return Config(**config_data)
+
+    @staticmethod
+    def dump_config(path: str, config: Config) -> None:
+        """Dump plugin configuration to a file.
+
+        Args:
+            path: configuration file path
+            config: the plugin configuration path
+        """
+        with open(os.path.normpath(path), "w", encoding="utf-8") as file:
+            yaml.safe_dump(config.model_dump(exclude_none=True), file)
+
+    @staticmethod
+    def load_plugin_config(config: str) -> PluginConfig:
+        """Load a plugin configuration from a file path.
+
+        This function autoescapes curly brackets in the 'instruction'
+        and 'examples' keys under the config attribute.
+
+        Args:
+            config: the plugin configuration path
+
+        Returns:
+            The plugin configuration object
+        """
+        with open(os.path.normpath(config), "r", encoding="utf8") as file:
+            template = file.read()
+            jinja_env = jinja2.Environment(loader=jinja2.BaseLoader(), autoescape=True)
+            rendered_template = jinja_env.from_string(template).render(env=os.environ)
+            config_data = yaml.safe_load(rendered_template)
+        return PluginConfig(**config_data)
+
+    @staticmethod
+    def load_plugin_manifest(manifest: str) -> PluginManifest:
+        """Load a plugin manifest from a file path.
+
+        Args:
+            manifest: the plugin manifest path
+
+        Returns:
+            The plugin manifest object
+        """
+        with open(os.path.normpath(manifest), "r", encoding="utf8") as file:
+            template = file.read()
+            jinja_env = jinja2.Environment(loader=jinja2.BaseLoader(), autoescape=True)
+            rendered_template = jinja_env.from_string(template).render(env=os.environ)
+            config_data = yaml.safe_load(rendered_template)
+        return PluginManifest(**config_data)

mcpgateway/plugins/framework/loader/plugin.py

@@ -0,0 +1,59 @@
+# -*- coding: utf-8 -*-
+"""Plugin loader implementation.
+
+Copyright 2025
+SPDX-License-Identifier: Apache-2.0
+Authors: Teryl Taylor
+
+This module implements the plugin loader.
+"""
+
+# Standard
+import logging
+from typing import cast, Type
+
+# First-Party
+from mcpgateway.plugins.framework.base import Plugin
+from mcpgateway.plugins.framework.models import PluginConfig
+from mcpgateway.plugins.framework.utils import import_module, parse_class_name
+
+logger = logging.getLogger(__name__)
+
+
+class PluginLoader(object):
+    """A plugin loader object for loading and instantiating plugins."""
+
+    def __init__(self) -> None:
+        """Initialize the plugin loader."""
+        self._plugin_types: dict[str, Type[Plugin]] = {}
+
+    def __get_plugin_type(self, kind: str) -> Type[Plugin]:
+        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)
+            raise
+
+    def __register_plugin_type(self, kind: str) -> None:
+        if kind not in self._plugin_types:
+            plugin_type = self.__get_plugin_type(kind)
+            self._plugin_types[kind] = plugin_type
+
+    async def load_and_instantiate_plugin(self, config: PluginConfig) -> Plugin | None:
+        """Load and instantiate a plugin, given a configuration.
+
+        Args:
+            config: A plugin configuration.
+
+        Returns:
+            A plugin instance.
+        """
+        if config.kind not in self._plugin_types:
+            self.__register_plugin_type(config.kind)
+        plugin_type = self._plugin_types[config.kind]
+        if plugin_type:
+            return plugin_type(config)
+        return None