Coverage for mcpgateway / plugins / framework / base.py: 99%

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

2"""Location: ./mcpgateway/plugins/framework/base.py 

3Copyright 2025 

4SPDX-License-Identifier: Apache-2.0 

5-Authors: Teryl Taylor, Mihai Criveti 

6 

7Base plugin implementation. 

8This module implements the base plugin object. 

9""" 

10 

11# Standard 

12from abc import ABC 

13from typing import Awaitable, Callable, Optional, Union 

14import uuid 

15 

16# First-Party 

17from mcpgateway.plugins.framework.errors import PluginError 

18from mcpgateway.plugins.framework.models import PluginCondition, PluginConfig, PluginContext, PluginErrorModel, PluginMode, PluginPayload, PluginResult 

19 

20# pylint: disable=import-outside-toplevel 

21 

22 

23class Plugin(ABC): 

24 """Base plugin object for pre/post processing of inputs and outputs at various locations throughout the server. 

25 

26 Examples: 

27 >>> from mcpgateway.plugins.framework import PluginConfig, PluginMode 

28 >>> from mcpgateway.plugins.framework.hooks.prompts import PromptHookType 

29 >>> config = PluginConfig( 

30 ... name="test_plugin", 

31 ... description="Test plugin", 

32 ... author="test", 

33 ... kind="mcpgateway.plugins.framework.Plugin", 

34 ... version="1.0.0", 

35 ... hooks=[PromptHookType.PROMPT_PRE_FETCH], 

36 ... tags=["test"], 

37 ... mode=PluginMode.ENFORCE, 

38 ... priority=50 

39 ... ) 

40 >>> plugin = Plugin(config) 

41 >>> plugin.name 

42 'test_plugin' 

43 >>> plugin.priority 

44 50 

45 >>> plugin.mode 

46 <PluginMode.ENFORCE: 'enforce'> 

47 >>> PromptHookType.PROMPT_PRE_FETCH in plugin.hooks 

48 True 

49 """ 

50 

51 def __init__( 

52 self, 

53 config: PluginConfig, 

54 hook_payloads: Optional[dict[str, PluginPayload]] = None, 

55 hook_results: Optional[dict[str, PluginResult]] = None, 

56 ) -> None: 

57 """Initialize a plugin with a configuration and context. 

58 

59 Args: 

60 config: The plugin configuration 

61 hook_payloads: optional mapping of hookpoints to payloads for the plugin. 

62 Used for external plugins for converting json to pydantic. 

63 hook_results: optional mapping of hookpoints to result types for the plugin. 

64 Used for external plugins for converting json to pydantic. 

65 

66 Examples: 

67 >>> from mcpgateway.plugins.framework import PluginConfig 

68 >>> from mcpgateway.plugins.framework.hooks.prompts import PromptHookType 

69 >>> config = PluginConfig( 

70 ... name="simple_plugin", 

71 ... description="Simple test", 

72 ... author="test", 

73 ... kind="test.Plugin", 

74 ... version="1.0.0", 

75 ... hooks=[PromptHookType.PROMPT_POST_FETCH], 

76 ... tags=["simple"] 

77 ... ) 

78 >>> plugin = Plugin(config) 

79 >>> plugin._config.name 

80 'simple_plugin' 

81 """ 

82 self._config = config 

83 self._hook_payloads = hook_payloads 

84 self._hook_results = hook_results 

85 

86 @property 

87 def priority(self) -> int: 

88 """Return the plugin's priority. 

89 

90 Returns: 

91 Plugin's priority. 

92 """ 

93 return self._config.priority 

94 

95 @property 

96 def config(self) -> PluginConfig: 

97 """Return the plugin's configuration. 

98 

99 Returns: 

100 Plugin's configuration. 

101 """ 

102 return self._config 

103 

104 @property 

105 def mode(self) -> PluginMode: 

106 """Return the plugin's mode. 

107 

108 Returns: 

109 Plugin's mode. 

110 """ 

111 return self._config.mode 

112 

113 @property 

114 def name(self) -> str: 

115 """Return the plugin's name. 

116 

117 Returns: 

118 Plugin's name. 

119 """ 

120 return self._config.name 

121 

122 @property 

123 def hooks(self) -> list[str]: 

124 """Return the plugin's currently configured hooks. 

125 

126 Returns: 

127 Plugin's configured hooks. 

128 """ 

129 return self._config.hooks 

130 

131 @property 

132 def tags(self) -> list[str]: 

133 """Return the plugin's tags. 

134 

135 Returns: 

136 Plugin's tags. 

137 """ 

138 return self._config.tags 

139 

140 @property 

141 def conditions(self) -> list[PluginCondition] | None: 

142 """Return the plugin's conditions for operation. 

143 

144 Returns: 

145 Plugin's conditions for executing. 

146 """ 

147 return self._config.conditions 

148 

149 async def initialize(self) -> None: 

150 """Initialize the plugin.""" 

151 

152 async def shutdown(self) -> None: 

153 """Plugin cleanup code.""" 

154 

155 def json_to_payload(self, hook: str, payload: Union[str | dict]) -> PluginPayload: 

156 """Converts a json payload to the proper pydantic payload object given a hook type. Used 

157 mainly for serialization/deserialization of external plugin payloads. 

158 

159 Args: 

160 hook: the hook type for which the payload needs converting. 

161 payload: the payload as a string or dict. 

162 

163 Returns: 

164 A pydantic payload object corresponding to the hook type. 

165 

166 Raises: 

167 PluginError: if no payload type is defined. 

168 """ 

169 hook_payload_type: type[PluginPayload] | None = None 

170 

171 # First try instance-level hook_payloads 

172 if self._hook_payloads: 

173 hook_payload_type = self._hook_payloads.get(hook, None) # type: ignore[assignment] 

174 

175 # Fall back to global registry 

176 if not hook_payload_type: 

177 # First-Party 

178 from mcpgateway.plugins.framework.hooks.registry import get_hook_registry 

179 

180 registry = get_hook_registry() 

181 hook_payload_type = registry.get_payload_type(hook) 

182 

183 if not hook_payload_type: 

184 raise PluginError(error=PluginErrorModel(message=f"No payload defined for hook {hook}.", plugin_name=self.name)) 

185 

186 if isinstance(payload, str): 

187 return hook_payload_type.model_validate_json(payload) 

188 return hook_payload_type.model_validate(payload) 

189 

190 def json_to_result(self, hook: str, result: Union[str | dict]) -> PluginResult: 

191 """Converts a json result to the proper pydantic result object given a hook type. Used 

192 mainly for serialization/deserialization of external plugin results. 

193 

194 Args: 

195 hook: the hook type for which the result needs converting. 

196 result: the result as a string or dict. 

197 

198 Returns: 

199 A pydantic result object corresponding to the hook type. 

200 

201 Raises: 

202 PluginError: if no result type is defined. 

203 """ 

204 hook_result_type: type[PluginResult] | None = None 

205 

206 # First try instance-level hook_results 

207 if self._hook_results: 

208 hook_result_type = self._hook_results.get(hook, None) # type: ignore[assignment] 

209 

210 # Fall back to global registry 

211 if not hook_result_type: 

212 # First-Party 

213 from mcpgateway.plugins.framework.hooks.registry import get_hook_registry 

214 

215 registry = get_hook_registry() 

216 hook_result_type = registry.get_result_type(hook) 

217 

218 if not hook_result_type: 

219 raise PluginError(error=PluginErrorModel(message=f"No result defined for hook {hook}.", plugin_name=self.name)) 

220 

221 if isinstance(result, str): 

222 return hook_result_type.model_validate_json(result) 

223 return hook_result_type.model_validate(result) 

224 

225 

226class PluginRef: 

227 """Plugin reference which contains a uuid. 

228 

229 Examples: 

230 >>> from mcpgateway.plugins.framework import PluginConfig, PluginMode 

231 >>> from mcpgateway.plugins.framework.hooks.prompts import PromptHookType 

232 >>> config = PluginConfig( 

233 ... name="ref_test", 

234 ... description="Reference test", 

235 ... author="test", 

236 ... kind="test.Plugin", 

237 ... version="1.0.0", 

238 ... hooks=[PromptHookType.PROMPT_PRE_FETCH], 

239 ... tags=["ref", "test"], 

240 ... mode=PluginMode.PERMISSIVE, 

241 ... priority=100 

242 ... ) 

243 >>> plugin = Plugin(config) 

244 >>> ref = PluginRef(plugin) 

245 >>> ref.name 

246 'ref_test' 

247 >>> ref.priority 

248 100 

249 >>> ref.mode 

250 <PluginMode.PERMISSIVE: 'permissive'> 

251 >>> len(ref.uuid) # UUID is a 32-character hex string 

252 32 

253 >>> ref.tags 

254 ['ref', 'test'] 

255 """ 

256 

257 def __init__(self, plugin: Plugin): 

258 """Initialize a plugin reference. 

259 

260 Args: 

261 plugin: The plugin to reference. 

262 

263 Examples: 

264 >>> from mcpgateway.plugins.framework import PluginConfig 

265 >>> from mcpgateway.plugins.framework.hooks.prompts import PromptHookType 

266 >>> config = PluginConfig( 

267 ... name="plugin_ref", 

268 ... description="Test", 

269 ... author="test", 

270 ... kind="test.Plugin", 

271 ... version="1.0.0", 

272 ... hooks=[PromptHookType.PROMPT_POST_FETCH], 

273 ... tags=[] 

274 ... ) 

275 >>> plugin = Plugin(config) 

276 >>> ref = PluginRef(plugin) 

277 >>> ref._plugin.name 

278 'plugin_ref' 

279 >>> isinstance(ref._uuid, uuid.UUID) 

280 True 

281 """ 

282 self._plugin = plugin 

283 self._uuid = uuid.uuid4() 

284 

285 @property 

286 def plugin(self) -> Plugin: 

287 """Return the underlying plugin. 

288 

289 Returns: 

290 The underlying plugin. 

291 """ 

292 return self._plugin 

293 

294 @property 

295 def uuid(self) -> str: 

296 """Return the plugin's UUID. 

297 

298 Returns: 

299 Plugin's UUID. 

300 """ 

301 return self._uuid.hex 

302 

303 @property 

304 def priority(self) -> int: 

305 """Returns the plugin's priority. 

306 

307 Returns: 

308 Plugin's priority. 

309 """ 

310 return self._plugin.priority 

311 

312 @property 

313 def name(self) -> str: 

314 """Return the plugin's name. 

315 

316 Returns: 

317 Plugin's name. 

318 """ 

319 return self._plugin.name 

320 

321 @property 

322 def hooks(self) -> list[str]: 

323 """Returns the plugin's currently configured hooks. 

324 

325 Returns: 

326 Plugin's configured hooks. 

327 """ 

328 return self._plugin.hooks 

329 

330 @property 

331 def tags(self) -> list[str]: 

332 """Return the plugin's tags. 

333 

334 Returns: 

335 Plugin's tags. 

336 """ 

337 return self._plugin.tags 

338 

339 @property 

340 def conditions(self) -> list[PluginCondition] | None: 

341 """Return the plugin's conditions for operation. 

342 

343 Returns: 

344 Plugin's conditions for operation. 

345 """ 

346 return self._plugin.conditions 

347 

348 @property 

349 def mode(self) -> PluginMode: 

350 """Return the plugin's mode. 

351 

352 Returns: 

353 Plugin's mode. 

354 """ 

355 return self.plugin.mode 

356 

357 

358class HookRef: 

359 """A Hook reference point with plugin and function.""" 

360 

361 def __init__(self, hook: str, plugin_ref: PluginRef): 

362 """Initialize a hook reference point. 

363 

364 Discovers the hook method using either: 

365 1. Convention-based naming (method name matches hook type) 

366 2. Decorator-based (@hook decorator with matching hook_type) 

367 

368 Args: 

369 hook: name of the hook point (e.g., 'tool_pre_invoke'). 

370 plugin_ref: The reference to the plugin to hook. 

371 

372 Raises: 

373 PluginError: If no method is found for the specified hook. 

374 

375 Examples: 

376 >>> from mcpgateway.plugins.framework import PluginConfig 

377 >>> config = PluginConfig(name="test", kind="test", version="1.0", author="test", hooks=["tool_pre_invoke"]) 

378 >>> plugin = Plugin(config) 

379 >>> plugin_ref = PluginRef(plugin) 

380 >>> # This would work if plugin has tool_pre_invoke method or @hook("tool_pre_invoke") decorator 

381 """ 

382 # Standard 

383 import inspect 

384 

385 # First-Party 

386 from mcpgateway.plugins.framework.decorator import get_hook_metadata 

387 

388 self._plugin_ref = plugin_ref 

389 self._hook = hook 

390 

391 # Try convention-based lookup first (method name matches hook type) 

392 self._func: Callable[[PluginPayload, PluginContext], Awaitable[PluginResult]] | None = getattr(plugin_ref.plugin, hook, None) 

393 

394 # If not found by convention, scan for @hook decorated methods 

395 if self._func is None: 

396 for name, method in inspect.getmembers(plugin_ref.plugin, predicate=inspect.ismethod): 

397 # Skip private/magic methods 

398 if name.startswith("_"): 

399 continue 

400 

401 # Check for @hook decorator metadata 

402 metadata = get_hook_metadata(method) 

403 if metadata and metadata.hook_type == hook: 

404 self._func = method 

405 break 

406 

407 # Raise error if hook method not found by either approach 

408 if not self._func: 

409 raise PluginError( 

410 error=PluginErrorModel( 

411 message=f"Plugin '{plugin_ref.plugin.name}' has no hook: '{hook}'. " f"Method must either be named '{hook}' or decorated with @hook('{hook}')", 

412 plugin_name=plugin_ref.plugin.name, 

413 ) 

414 ) 

415 

416 # Validate hook method signature (parameter count and async) 

417 self._validate_hook_signature(hook, self._func, plugin_ref.plugin.name) 

418 

419 def _validate_hook_signature(self, hook: str, func: Callable, plugin_name: str) -> None: 

420 """Validate that the hook method has the correct signature. 

421 

422 Checks: 

423 1. Method accepts correct number of parameters (self, payload, context) 

424 2. Method is async (returns coroutine) 

425 

426 Args: 

427 hook: The hook type being validated 

428 func: The hook method to validate 

429 plugin_name: Name of the plugin (for error messages) 

430 

431 Raises: 

432 PluginError: If the signature is invalid 

433 """ 

434 # Standard 

435 import inspect 

436 

437 sig = inspect.signature(func) 

438 params = list(sig.parameters.values()) 

439 

440 # Check parameter count (should be: payload, context) 

441 # Note: 'self' is not included in bound method signatures 

442 if len(params) != 2: 

443 raise PluginError( 

444 error=PluginErrorModel( 

445 message=f"Plugin '{plugin_name}' hook '{hook}' has invalid signature. " 

446 f"Expected 2 parameters (payload, context), got {len(params)}: {list(sig.parameters.keys())}. " 

447 f"Correct signature: async def {hook}(self, payload: PayloadType, context: PluginContext) -> ResultType", 

448 plugin_name=plugin_name, 

449 ) 

450 ) 

451 

452 # Check that method is async 

453 if not inspect.iscoroutinefunction(func): 

454 raise PluginError( 

455 error=PluginErrorModel( 

456 message=f"Plugin '{plugin_name}' hook '{hook}' must be async. " 

457 f"Method '{func.__name__}' is not a coroutine function. " 

458 f"Use 'async def {func.__name__}(...)' instead of 'def {func.__name__}(...)'.", 

459 plugin_name=plugin_name, 

460 ) 

461 ) 

462 

463 # ========== OPTIONAL: Type Hint Validation ========== 

464 # Uncomment to enable strict type checking of payload and return types. 

465 # This validates that type hints match the expected types from the hook registry. 

466 # Pros: Catches type errors at plugin load time instead of runtime 

467 # Cons: Requires all plugins to have type hints, adds validation overhead 

468 # 

469 # self._validate_type_hints(hook, func, params, plugin_name) 

470 

471 def _validate_type_hints(self, hook: str, func: Callable, params: list, plugin_name: str) -> None: 

472 """Validate that type hints match expected payload and result types. 

473 

474 This is an optional validation that can be enabled to enforce type safety. 

475 

476 Args: 

477 hook: The hook type being validated 

478 func: The hook method to validate 

479 params: List of function parameters 

480 plugin_name: Name of the plugin (for error messages) 

481 

482 Raises: 

483 PluginError: If type hints are missing or don't match expected types 

484 """ 

485 # Standard 

486 from typing import get_type_hints 

487 

488 # First-Party 

489 from mcpgateway.plugins.framework.hooks.registry import get_hook_registry 

490 

491 # Get expected types from registry 

492 registry = get_hook_registry() 

493 expected_payload_type = registry.get_payload_type(hook) 

494 expected_result_type = registry.get_result_type(hook) 

495 

496 # If hook is not registered in global registry, we can't validate types 

497 if not expected_payload_type or not expected_result_type: 

498 return 

499 

500 # Get type hints from the function 

501 try: 

502 hints = get_type_hints(func) 

503 except Exception as e: 

504 # Type hints might use forward references or unavailable types 

505 # We'll skip validation rather than fail 

506 # Standard 

507 import logging 

508 

509 logger = logging.getLogger(__name__) 

510 logger.debug("Could not extract type hints for plugin '%s' hook '%s': %s", plugin_name, hook, e) 

511 return 

512 

513 # Validate payload parameter type (first parameter, since 'self' is not in params) 

514 payload_param_name = params[0].name 

515 if payload_param_name not in hints: 

516 raise PluginError( 

517 error=PluginErrorModel( 

518 message=f"Plugin '{plugin_name}' hook '{hook}' missing type hint for parameter '{payload_param_name}'. " f"Expected: {payload_param_name}: {expected_payload_type.__name__}", 

519 plugin_name=plugin_name, 

520 ) 

521 ) 

522 

523 actual_payload_type = hints[payload_param_name] 

524 

525 # Check if types match (exact match or subclass) 

526 if actual_payload_type != expected_payload_type: 

527 # Check for generic types or complex type hints 

528 actual_type_str = str(actual_payload_type) 

529 expected_type_str = expected_payload_type.__name__ 

530 

531 # If the expected type name is in the string representation, it's probably OK 

532 if expected_type_str not in actual_type_str: 532 ↛ 541line 532 didn't jump to line 541 because the condition on line 532 was always true

533 raise PluginError( 

534 error=PluginErrorModel( 

535 message=f"Plugin '{plugin_name}' hook '{hook}' parameter '{payload_param_name}' " f"has incorrect type hint. Expected: {expected_type_str}, Got: {actual_type_str}", 

536 plugin_name=plugin_name, 

537 ) 

538 ) 

539 

540 # Validate return type 

541 if "return" not in hints: 

542 raise PluginError( 

543 error=PluginErrorModel( 

544 message=f"Plugin '{plugin_name}' hook '{hook}' missing return type hint. " f"Expected: -> {expected_result_type.__name__}", 

545 plugin_name=plugin_name, 

546 ) 

547 ) 

548 

549 actual_return_type = hints["return"] 

550 return_type_str = str(actual_return_type) 

551 expected_return_str = expected_result_type.__name__ 

552 

553 # For async functions, the return type might be wrapped in Coroutine or Awaitable 

554 # We just check if the expected type is mentioned in the return type 

555 if expected_return_str not in return_type_str and actual_return_type != expected_result_type: 555 ↛ exitline 555 didn't return from function '_validate_type_hints' because the condition on line 555 was always true

556 raise PluginError( 

557 error=PluginErrorModel( 

558 message=f"Plugin '{plugin_name}' hook '{hook}' has incorrect return type hint. " f"Expected: {expected_return_str}, Got: {return_type_str}", 

559 plugin_name=plugin_name, 

560 ) 

561 ) 

562 

563 @property 

564 def plugin_ref(self) -> PluginRef: 

565 """The reference to the plugin object. 

566 

567 Returns: 

568 A plugin reference. 

569 """ 

570 return self._plugin_ref 

571 

572 @property 

573 def name(self) -> str: 

574 """The name of the hooking function. 

575 

576 Returns: 

577 A plugin name. 

578 """ 

579 return self._hook 

580 

581 @property 

582 def hook(self) -> Callable[[PluginPayload, PluginContext], Awaitable[PluginResult]] | None: 

583 """The hooking function that can be invoked within the reference. 

584 

585 Returns: 

586 An awaitable hook function reference. 

587 """ 

588 return self._func