Coverage for mcpgateway / plugins / framework / external / mcp / server / server.py: 98%

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

2"""Location: ./mcpgateway/plugins/framework/external/mcp/server/server.py 

3Copyright 2025 

4SPDX-License-Identifier: Apache-2.0 

5Authors: Fred Araujo, Teryl Taylor 

6 

7Module that contains plugin MCP server code to serve external plugins. 

8 

9Examples: 

10 Create an external plugin server with a configuration file: 

11 

12 >>> server = ExternalPluginServer(config_path="./tests/unit/mcpgateway/plugins/fixtures/configs/valid_single_plugin.yaml") 

13 >>> server is not None 

14 True 

15 >>> isinstance(server._config_path, str) 

16 True 

17 

18 Get server configuration with defaults: 

19 

20 >>> server = ExternalPluginServer(config_path="./tests/unit/mcpgateway/plugins/fixtures/configs/valid_single_plugin.yaml") 

21 >>> config = server.get_server_config() 

22 >>> config.host == '127.0.0.1' 

23 True 

24 >>> config.port == 8000 

25 True 

26 

27 Verify plugin manager is initialized: 

28 

29 >>> server = ExternalPluginServer(config_path="./tests/unit/mcpgateway/plugins/fixtures/configs/valid_single_plugin.yaml") 

30 >>> server._plugin_manager is not None 

31 True 

32 >>> server._config is not None 

33 True 

34 

35 Multiple servers can be created: 

36 

37 >>> server1 = ExternalPluginServer(config_path="./tests/unit/mcpgateway/plugins/fixtures/configs/valid_single_plugin.yaml") 

38 >>> server2 = ExternalPluginServer(config_path="./tests/unit/mcpgateway/plugins/fixtures/configs/valid_multiple_plugins_filter.yaml") 

39 >>> server1._config_path != server2._config_path 

40 True 

41 

42 Configuration is loaded from file: 

43 

44 >>> import asyncio 

45 >>> server = ExternalPluginServer(config_path="./tests/unit/mcpgateway/plugins/fixtures/configs/valid_single_plugin.yaml") 

46 >>> plugins = asyncio.run(server.get_plugin_configs()) 

47 >>> isinstance(plugins, list) 

48 True 

49 >>> len(plugins) >= 1 

50 True 

51 

52 Server configuration defaults are sensible: 

53 

54 >>> server = ExternalPluginServer(config_path="./tests/unit/mcpgateway/plugins/fixtures/configs/valid_single_plugin.yaml") 

55 >>> config = server.get_server_config() 

56 >>> isinstance(config.host, str) 

57 True 

58 >>> isinstance(config.port, int) 

59 True 

60 >>> config.port > 0 

61 True 

62""" 

63 

64# Standard 

65import logging 

66import os 

67from typing import Any, Dict, TypeVar 

68 

69# Third-Party 

70from pydantic import BaseModel 

71 

72# First-Party 

73from mcpgateway.plugins.framework.constants import CONTEXT, ERROR, PLUGIN_NAME, RESULT 

74from mcpgateway.plugins.framework.errors import convert_exception_to_error, PluginError 

75from mcpgateway.plugins.framework.loader.config import ConfigLoader 

76from mcpgateway.plugins.framework.manager import PluginManager 

77from mcpgateway.plugins.framework.models import GRPCServerConfig, MCPServerConfig, PluginContext 

78from mcpgateway.plugins.framework.settings import get_config_path_settings 

79 

80P = TypeVar("P", bound=BaseModel) 

81 

82logger = logging.getLogger(__name__) 

83 

84 

85class ExternalPluginServer: 

86 """External plugin server, providing methods for invoking plugin hooks.""" 

87 

88 def __init__(self, config_path: str | None = None) -> None: 

89 """Create an external plugin server. 

90 

91 Args: 

92 config_path: The configuration file path for loading plugins. 

93 If set, this attribute overrides the value in PLUGINS_CONFIG_PATH. 

94 

95 Examples: 

96 >>> server = ExternalPluginServer(config_path="./tests/unit/mcpgateway/plugins/fixtures/configs/valid_single_plugin.yaml") 

97 >>> server is not None 

98 True 

99 """ 

100 self._config_path = config_path or get_config_path_settings().config_path or os.path.join(".", "resources", "plugins", "config.yaml") 

101 self._config = ConfigLoader.load_config(self._config_path, use_jinja=False) 

102 self._plugin_manager = PluginManager(self._config_path) 

103 

104 async def get_plugin_configs(self) -> list[dict]: 

105 """Return a list of plugin configurations for plugins currently installed on the MCP server. 

106 

107 Returns: 

108 A list of plugin configurations. 

109 

110 Examples: 

111 >>> import asyncio 

112 >>> server = ExternalPluginServer(config_path="./tests/unit/mcpgateway/plugins/fixtures/configs/valid_single_plugin.yaml") 

113 >>> plugins = asyncio.run(server.get_plugin_configs()) 

114 >>> len(plugins) > 0 

115 True 

116 

117 Returns empty list when no plugins configured: 

118 

119 >>> server = ExternalPluginServer(config_path="./tests/unit/mcpgateway/plugins/fixtures/configs/valid_single_plugin.yaml") 

120 >>> server._config.plugins = None 

121 >>> plugins = asyncio.run(server.get_plugin_configs()) 

122 >>> plugins 

123 [] 

124 

125 Each plugin config is a dictionary: 

126 

127 >>> server = ExternalPluginServer(config_path="./tests/unit/mcpgateway/plugins/fixtures/configs/valid_single_plugin.yaml") 

128 >>> plugins = asyncio.run(server.get_plugin_configs()) 

129 >>> all(isinstance(p, dict) for p in plugins) 

130 True 

131 """ 

132 plugins: list[dict] = [] 

133 if self._config.plugins: 

134 for plug in self._config.plugins: 

135 plugins.append(plug.model_dump()) 

136 return plugins 

137 

138 async def get_plugin_config(self, name: str) -> dict | None: 

139 """Return a plugin configuration give a plugin name. 

140 

141 Args: 

142 name: The name of the plugin of which to return the plugin configuration. 

143 

144 Returns: 

145 A plugin configuration dict, or None if not found. 

146 

147 Examples: 

148 >>> import asyncio 

149 >>> server = ExternalPluginServer(config_path="./tests/unit/mcpgateway/plugins/fixtures/configs/valid_single_plugin.yaml") 

150 >>> c = asyncio.run(server.get_plugin_config(name = "ReplaceBadWordsPlugin")) 

151 >>> c is not None 

152 True 

153 >>> c["name"] == "ReplaceBadWordsPlugin" 

154 True 

155 

156 Returns None when plugin not found: 

157 

158 >>> server = ExternalPluginServer(config_path="./tests/unit/mcpgateway/plugins/fixtures/configs/valid_single_plugin.yaml") 

159 >>> c = asyncio.run(server.get_plugin_config(name="NonExistentPlugin")) 

160 >>> c is None 

161 True 

162 

163 Case-insensitive plugin name lookup: 

164 

165 >>> server = ExternalPluginServer(config_path="./tests/unit/mcpgateway/plugins/fixtures/configs/valid_single_plugin.yaml") 

166 >>> c1 = asyncio.run(server.get_plugin_config(name="ReplaceBadWordsPlugin")) 

167 >>> c2 = asyncio.run(server.get_plugin_config(name="replacebadwordsplugin")) 

168 >>> c1 == c2 

169 True 

170 """ 

171 if self._config.plugins: 

172 for plug in self._config.plugins: 

173 if plug.name.lower() == name.lower(): 

174 return plug.model_dump() 

175 return None 

176 

177 async def invoke_hook(self, hook_type: str, plugin_name: str, payload: Dict[str, Any], context: Dict[str, Any] | PluginContext) -> dict: 

178 """Invoke a plugin hook. 

179 

180 Args: 

181 hook_type: The type of hook function to be invoked. 

182 plugin_name: The name of the plugin to execute. 

183 payload: The prompt name and arguments to be analyzed. 

184 context: The contextual and state information required for the execution of the hook. 

185 Can be a dict (for MCP transport) or PluginContext (for gRPC/Unix socket). 

186 

187 Raises: 

188 ValueError: If unable to retrieve a plugin. 

189 

190 Returns: 

191 The transformed or filtered response from the plugin hook. 

192 

193 Examples: 

194 >>> import asyncio 

195 >>> import os 

196 >>> os.environ["PYTHONPATH"] = "." 

197 >>> from mcpgateway.plugins.framework import GlobalContext, Plugin, PromptHookType, PromptPrehookPayload, PluginContext, PromptPrehookResult, PluginManager 

198 >>> PluginManager.reset() 

199 >>> server = ExternalPluginServer(config_path="./tests/unit/mcpgateway/plugins/fixtures/configs/valid_single_plugin.yaml") 

200 >>> payload = PromptPrehookPayload(prompt_id="123", name="test_prompt", args={"user": "This is a crap app"}) 

201 >>> context = PluginContext(global_context=GlobalContext(request_id="1", server_id="2")) 

202 >>> initialized = asyncio.run(server.initialize()) 

203 >>> initialized 

204 True 

205 >>> result = asyncio.run(server.invoke_hook(PromptHookType.PROMPT_PRE_FETCH, "ReplaceBadWordsPlugin", payload.model_dump(), context.model_dump())) 

206 >>> result is not None 

207 True 

208 >>> result["result"]["continue_processing"] 

209 True 

210 >>> "yikes" in result["result"]["modified_payload"]["args"]["user"] 

211 True 

212 """ 

213 result_payload: dict[str, Any] = {PLUGIN_NAME: plugin_name} 

214 try: 

215 # Track if input was Pydantic (for optimized response path) 

216 context_is_pydantic = isinstance(context, PluginContext) 

217 _context = context if context_is_pydantic else PluginContext.model_validate(context) 

218 

219 result = await self._plugin_manager.invoke_hook_for_plugin(plugin_name, hook_type, payload, _context, payload_as_json=True) 

220 

221 result_payload[RESULT] = result.model_dump() 

222 if not _context.is_empty(): 

223 # Return Pydantic directly if input was Pydantic (avoids extra serialization) 

224 result_payload[CONTEXT] = _context if context_is_pydantic else _context.model_dump() 

225 return result_payload 

226 except PluginError as pe: 

227 result_payload[ERROR] = pe.error 

228 return result_payload 

229 except Exception as ex: 

230 logger.exception(ex) 

231 result_payload[ERROR] = convert_exception_to_error(ex, plugin_name=plugin_name).model_dump() 

232 return result_payload 

233 

234 async def initialize(self) -> bool: 

235 """Initialize the plugin server. 

236 

237 Returns: 

238 A boolean indicating the intialization status of the server. 

239 

240 Examples: 

241 >>> import asyncio 

242 >>> server = ExternalPluginServer(config_path="./tests/unit/mcpgateway/plugins/fixtures/configs/valid_single_plugin.yaml") 

243 >>> result = asyncio.run(server.initialize()) 

244 >>> result 

245 True 

246 >>> asyncio.run(server.shutdown()) 

247 """ 

248 await self._plugin_manager.initialize() 

249 return self._plugin_manager.initialized 

250 

251 async def shutdown(self) -> None: 

252 """Shutdown the plugin server. 

253 

254 Examples: 

255 >>> import asyncio 

256 >>> server = ExternalPluginServer(config_path="./tests/unit/mcpgateway/plugins/fixtures/configs/valid_single_plugin.yaml") 

257 >>> asyncio.run(server.initialize()) 

258 True 

259 >>> asyncio.run(server.shutdown()) 

260 """ 

261 if self._plugin_manager.initialized: 

262 await self._plugin_manager.shutdown() 

263 

264 def get_server_config(self) -> MCPServerConfig: 

265 """Return the configuration for the plugin server. 

266 

267 Returns: 

268 A server configuration including host, port, and TLS information. 

269 

270 Examples: 

271 >>> server = ExternalPluginServer(config_path="./tests/unit/mcpgateway/plugins/fixtures/configs/valid_single_plugin.yaml") 

272 >>> config = server.get_server_config() 

273 >>> isinstance(config, MCPServerConfig) 

274 True 

275 >>> config.host 

276 '127.0.0.1' 

277 >>> config.port 

278 8000 

279 """ 

280 return self._config.server_settings or MCPServerConfig.from_env() or MCPServerConfig() 

281 

282 def get_grpc_server_config(self) -> GRPCServerConfig | None: 

283 """Return the gRPC server configuration if defined. 

284 

285 Returns: 

286 The gRPC server configuration from the config file, or None if not defined. 

287 

288 Examples: 

289 >>> server = ExternalPluginServer(config_path="./tests/unit/mcpgateway/plugins/fixtures/configs/valid_single_plugin.yaml") 

290 >>> config = server.get_grpc_server_config() 

291 >>> config is None or isinstance(config, GRPCServerConfig) 

292 True 

293 """ 

294 return self._config.grpc_server_settings