Coverage for mcpgateway / plugins / framework / loader / plugin.py: 100%

1# -*- coding: utf-8 -*- 

2"""Location: ./mcpgateway/plugins/framework/loader/plugin.py 

3Copyright 2025 

4SPDX-License-Identifier: Apache-2.0 

5Authors: Teryl Taylor, Mihai Criveti 

6 

7Plugin loader implementation. 

8This module implements the plugin loader. 

9""" 

10 

11# Standard 

12import logging 

13from typing import cast, Type 

14 

15# First-Party 

16from mcpgateway.plugins.framework.base import Plugin 

17from mcpgateway.plugins.framework.constants import EXTERNAL_PLUGIN_TYPE 

18from mcpgateway.plugins.framework.external.mcp.client import ExternalPlugin 

19from mcpgateway.plugins.framework.models import PluginConfig 

20from mcpgateway.plugins.framework.utils import import_module, parse_class_name 

21 

22# Use standard logging to avoid circular imports (plugins -> services -> plugins) 

23logger = logging.getLogger(__name__) 

24 

25 

26class PluginLoader: 

27 """A plugin loader object for loading and instantiating plugins. 

28 

29 Examples: 

30 >>> loader = PluginLoader() 

31 >>> isinstance(loader._plugin_types, dict) 

32 True 

33 >>> len(loader._plugin_types) 

34 0 

35 """ 

36 

37 def __init__(self) -> None: 

38 """Initialize the plugin loader. 

39 

40 Examples: 

41 >>> loader = PluginLoader() 

42 >>> loader._plugin_types 

43 {} 

44 """ 

45 self._plugin_types: dict[str, Type[Plugin]] = {} 

46 

47 def __get_plugin_type(self, kind: str) -> Type[Plugin]: 

48 """Import a plugin type from a python module. 

49 

50 Args: 

51 kind: The fully-qualified type of the plugin to be registered. 

52 

53 Raises: 

54 Exception: if unable to import a module. 

55 

56 Returns: 

57 A plugin type. 

58 """ 

59 try: 

60 (mod_name, cls_name) = parse_class_name(kind) 

61 module = import_module(mod_name) 

62 class_ = getattr(module, cls_name) 

63 return cast(Type[Plugin], class_) 

64 except Exception: 

65 logger.exception("Unable to import plugin type '%s'", kind) 

66 raise 

67 

68 def __register_plugin_type(self, kind: str) -> None: 

69 """Register a plugin type. 

70 

71 Args: 

72 kind: The fully-qualified type of the plugin to be registered. 

73 """ 

74 if kind not in self._plugin_types: 

75 plugin_type: Type[Plugin] 

76 if kind == EXTERNAL_PLUGIN_TYPE: 

77 plugin_type = ExternalPlugin 

78 else: 

79 plugin_type = self.__get_plugin_type(kind) 

80 self._plugin_types[kind] = plugin_type 

81 

82 async def load_and_instantiate_plugin(self, config: PluginConfig) -> Plugin | None: 

83 """Load and instantiate a plugin, given a configuration. 

84 

85 For external plugins, the transport type is determined by the presence 

86 of 'mcp', 'grpc', or 'unix_socket' configuration: 

87 - If 'grpc' is set, uses GrpcExternalPlugin for gRPC transport 

88 - If 'mcp' is set, uses ExternalPlugin for MCP transport 

89 - If 'unix_socket' is set, uses UnixSocketExternalPlugin for raw Unix socket transport 

90 

91 Args: 

92 config: A plugin configuration. 

93 

94 Returns: 

95 A plugin instance. 

96 

97 Raises: 

98 ValueError: If an external plugin has no transport configured. 

99 """ 

100 # Handle external plugins with transport selection 

101 if config.kind == EXTERNAL_PLUGIN_TYPE: 

102 plugin: Plugin 

103 if config.grpc: 

104 # Use gRPC transport 

105 # Import here to avoid circular dependency and to make grpc optional 

106 # First-Party 

107 from mcpgateway.plugins.framework.external.grpc.client import GrpcExternalPlugin # pylint: disable=import-outside-toplevel 

108 

109 plugin = GrpcExternalPlugin(config) 

110 logger.info("Loading external plugin '%s' with gRPC transport", config.name) 

111 elif config.unix_socket: 

112 # Use raw Unix socket transport (high-performance local IPC) 

113 # First-Party 

114 from mcpgateway.plugins.framework.external.unix.client import UnixSocketExternalPlugin # pylint: disable=import-outside-toplevel 

115 

116 plugin = UnixSocketExternalPlugin(config) 

117 logger.info("Loading external plugin '%s' with Unix socket transport", config.name) 

118 elif config.mcp: 

119 # Use MCP transport 

120 plugin = ExternalPlugin(config) 

121 logger.info("Loading external plugin '%s' with MCP transport", config.name) 

122 else: 

123 # Defensive fallback: PluginConfig validation should prevent this path. 

124 raise ValueError(f"External plugin '{config.name}' must have 'mcp', 'grpc', or 'unix_socket' configuration") # pragma: no cover 

125 

126 await plugin.initialize() 

127 return plugin 

128 

129 # Handle other plugin types 

130 if config.kind not in self._plugin_types: 

131 self.__register_plugin_type(config.kind) 

132 plugin_type = self._plugin_types[config.kind] 

133 if plugin_type: 

134 plugin = plugin_type(config) 

135 await plugin.initialize() 

136 return plugin 

137 return None 

138 

139 async def shutdown(self) -> None: 

140 """Shutdown and cleanup plugin loader. 

141 

142 Examples: 

143 >>> import asyncio 

144 >>> loader = PluginLoader() 

145 >>> asyncio.run(loader.shutdown()) 

146 >>> loader._plugin_types 

147 {} 

148 """ 

149 if self._plugin_types: 

150 self._plugin_types.clear()