fix: pylint issues

Signed-off-by: Frederico Araujo <[email protected]>

Author: araujof

mcpgateway/plugins/framework/base.py

@@ -188,7 +188,7 @@ def json_to_payload(self, hook: str, payload: Union[str | dict]) -> PluginPayloa
         # Fall back to global registry
         if not hook_payload_type:
             # First-Party
-            from mcpgateway.plugins.framework.hook_registry import get_hook_registry
+            from mcpgateway.plugins.framework.hook_registry import get_hook_registry  # pylint: disable=import-outside-toplevel
 
             registry = get_hook_registry()
             hook_payload_type = registry.get_payload_type(hook)
@@ -223,7 +223,7 @@ def json_to_result(self, hook: str, result: Union[str | dict]) -> PluginResult:
         # Fall back to global registry
         if not hook_result_type:
             # First-Party
-            from mcpgateway.plugins.framework.hook_registry import get_hook_registry
+            from mcpgateway.plugins.framework.hook_registry import get_hook_registry  # pylint: disable=import-outside-toplevel
 
             registry = get_hook_registry()
             hook_result_type = registry.get_result_type(hook)

mcpgateway/plugins/framework/external/mcp/client.py

@@ -316,9 +316,12 @@ async def shutdown(self) -> None:
 class ExternalHookRef(HookRef):
     """A Hook reference point for external plugins."""
 
-    def __init__(self, hook: str, plugin_ref: PluginRef):
+    def __init__(self, hook: str, plugin_ref: PluginRef):  # pylint: disable=super-init-not-called
         """Initialize a hook reference point for an external plugin.
 
+        Note: We intentionally don't call super().__init__() because external plugins
+        use invoke_hook() rather than direct method attributes.
+
         Args:
             hook: name of the hook point.
             plugin_ref: The reference to the plugin to hook.

mcpgateway/plugins/framework/external/mcp/server/runtime.py

@@ -157,12 +157,12 @@ async def _start_health_check_server(self, health_port: int) -> None:
             health_port: Port number for the health check server.
         """
         # Third-Party
-        from starlette.applications import Starlette
-        from starlette.requests import Request
-        from starlette.responses import JSONResponse
-        from starlette.routing import Route
+        from starlette.applications import Starlette  # pylint: disable=import-outside-toplevel
+        from starlette.requests import Request  # pylint: disable=import-outside-toplevel
+        from starlette.responses import JSONResponse  # pylint: disable=import-outside-toplevel
+        from starlette.routing import Route  # pylint: disable=import-outside-toplevel
 
-        async def health_check(request: Request):
+        async def health_check(_request: Request):
             """Health check endpoint for container orchestration.
 
             Args:
@@ -192,11 +192,11 @@ async def run_streamable_http_async(self) -> None:
 
         # Add health check endpoint to main app
         # Third-Party
-        from starlette.requests import Request
-        from starlette.responses import JSONResponse
-        from starlette.routing import Route
+        from starlette.requests import Request  # pylint: disable=import-outside-toplevel
+        from starlette.responses import JSONResponse  # pylint: disable=import-outside-toplevel
+        from starlette.routing import Route  # pylint: disable=import-outside-toplevel
 
-        async def health_check(request: Request):
+        async def health_check(_request: Request):
             """Health check endpoint for container orchestration.
 
             Args:
@@ -254,7 +254,7 @@ async def run():
     Raises:
         Exception: If plugin server initialization or execution fails.
     """
-    global SERVER
+    global SERVER  # pylint: disable=global-statement
 
     # Initialize plugin server
     SERVER = ExternalPluginServer()

mcpgateway/plugins/framework/manager.py

@@ -612,8 +612,7 @@ async def invoke_hook_for_plugin(
             if isinstance(payload, (str, dict)):
                 pydantic_payload = plugin.json_to_payload(hook_type, payload)
                 return await self._executor.execute_plugin(hook_ref, pydantic_payload, context, violations_as_exceptions)
-            else:
-                raise ValueError(f"When payload_as_json=True, payload must be str or dict, got {type(payload)}")
+            raise ValueError(f"When payload_as_json=True, payload must be str or dict, got {type(payload)}")
         # When payload_as_json=False, payload should already be a PluginPayload
         if not isinstance(payload, PluginPayload):
             raise ValueError(f"When payload_as_json=False, payload must be a PluginPayload, got {type(payload)}")

mcpgateway/plugins/framework/registry.py

@@ -98,7 +98,7 @@ def register(self, plugin: Plugin) -> None:
             self._priority_cache.pop(hook_type, None)
         self._hooks_by_name[plugin.name] = plugin_hooks
 
-        logger.info(f"Registered plugin: {plugin.name} with hooks: {[h for h in plugin.hooks]}")
+        logger.info(f"Registered plugin: {plugin.name} with hooks: {list(plugin.hooks)}")
 
     def unregister(self, plugin_name: str) -> None:
         """Unregister a plugin given its name.

mcpgateway/plugins/mcp/entities/base.py

@@ -45,7 +45,7 @@ def _register_mcp_hooks():
     """
     # Import here to avoid circular dependency at module load time
     # First-Party
-    from mcpgateway.plugins.framework.hook_registry import get_hook_registry
+    from mcpgateway.plugins.framework.hook_registry import get_hook_registry  # pylint: disable=import-outside-toplevel
 
     registry = get_hook_registry()