tests(plugins): added client hook tests for external plugins.

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

Author: Teryl Taylor

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

@@ -32,7 +32,23 @@
     PYTHON,
     PYTHON_SUFFIX,
 )
-from mcpgateway.plugins.framework.models import HookType, PluginConfig, PluginContext, PromptPosthookPayload, PromptPosthookResult, PromptPrehookPayload, PromptPrehookResult
+from mcpgateway.plugins.framework.models import (
+    HookType,
+    PluginConfig,
+    PluginContext,
+    PromptPosthookPayload,
+    PromptPosthookResult,
+    PromptPrehookPayload,
+    PromptPrehookResult,
+    ResourcePostFetchPayload,
+    ResourcePostFetchResult,
+    ResourcePreFetchPayload,
+    ResourcePreFetchResult,
+    ToolPostInvokePayload,
+    ToolPostInvokeResult,
+    ToolPreInvokePayload,
+    ToolPreInvokeResult,
+)
 from mcpgateway.schemas import TransportType
 
 logger = logging.getLogger(__name__)
@@ -162,6 +178,74 @@ async def prompt_post_fetch(self, payload: PromptPosthookPayload, context: Plugi
             return PromptPosthookResult.model_validate(res)
         return PromptPosthookResult(continue_processing=True)
 
+    async def tool_pre_invoke(self, payload: ToolPreInvokePayload, context: PluginContext) -> ToolPreInvokeResult:
+        """Plugin hook run before a tool is invoked.
+
+        Args:
+            payload: The tool payload to be analyzed.
+            context: contextual information about the hook call. Including why it was called.
+
+        Returns:
+            The tool prehook with name and arguments as modified or blocked by the plugin.
+        """
+
+        result = await self._session.call_tool(HookType.TOOL_PRE_INVOKE, {PLUGIN_NAME: self.name, PAYLOAD: payload, CONTEXT: context})
+        for content in result.content:
+            res = json.loads(content.text)
+            return ToolPreInvokeResult.model_validate(res)
+        return ToolPreInvokeResult(continue_processing=True)
+
+    async def tool_post_invoke(self, payload: ToolPostInvokePayload, context: PluginContext) -> ToolPostInvokeResult:
+        """Plugin hook run after a tool is invoked.
+
+        Args:
+            payload: The tool payload to be analyzed.
+            context: contextual information about the hook call. Including why it was called.
+
+        Returns:
+            The tool posthook with name and arguments as modified or blocked by the plugin.
+        """
+
+        result = await self._session.call_tool(HookType.TOOL_POST_INVOKE, {PLUGIN_NAME: self.name, PAYLOAD: payload, CONTEXT: context})
+        for content in result.content:
+            res = json.loads(content.text)
+            return ToolPostInvokeResult.model_validate(res)
+        return ToolPostInvokeResult(continue_processing=True)
+
+    async def resource_pre_fetch(self, payload: ResourcePreFetchPayload, context: PluginContext) -> ResourcePreFetchResult:
+        """Plugin hook run before a resource is fetched.
+
+        Args:
+            payload: The resource payload to be analyzed.
+            context: contextual information about the hook call. Including why it was called.
+
+        Returns:
+            The resource prehook with name and arguments as modified or blocked by the plugin.
+        """
+
+        result = await self._session.call_tool(HookType.RESOURCE_PRE_FETCH, {PLUGIN_NAME: self.name, PAYLOAD: payload, CONTEXT: context})
+        for content in result.content:
+            res = json.loads(content.text)
+            return ResourcePreFetchResult.model_validate(res)
+        return ResourcePreFetchResult(continue_processing=True)
+
+    async def resource_post_fetch(self, payload: ResourcePostFetchPayload, context: PluginContext) -> ResourcePostFetchResult:
+        """Plugin hook run after a resource is fetched.
+
+        Args:
+            payload: The resource payload to be analyzed.
+            context: contextual information about the hook call. Including why it was called.
+
+        Returns:
+            The resource posthook with name and arguments as modified or blocked by the plugin.
+        """
+
+        result = await self._session.call_tool(HookType.RESOURCE_POST_FETCH, {PLUGIN_NAME: self.name, PAYLOAD: payload, CONTEXT: context})
+        for content in result.content:
+            res = json.loads(content.text)
+            return ResourcePostFetchResult.model_validate(res)
+        return ResourcePostFetchResult(continue_processing=True)
+
     async def __get_plugin_config(self) -> PluginConfig | None:
         """Retrieve plugin configuration for the current plugin on the remote MCP server.
 

tests/unit/mcpgateway/plugins/fixtures/configs/valid_single_plugin_passthrough.yaml

@@ -0,0 +1,30 @@
+plugins:
+  # Self-contained Search Replace Plugin
+  - name: "PassThroughPlugin"
+    kind: "tests.unit.mcpgateway.plugins.fixtures.plugins.passthrough.PassThroughPlugin"
+    description: "A simple passthrough plugin."
+    version: "0.1"
+    author: "MCP Context Forge Team"
+    hooks: ["prompt_pre_fetch", "prompt_post_fetch", "tool_pre_invoke", "tool_post_invoke", "resource_pre_fetch", "resource_post_fetch"]
+    tags: ["plugin", "passthrough"]
+    mode: "enforce"  # enforce | permissive | disabled
+    priority: 150
+    conditions:
+      # Apply to specific tools/servers
+      - prompts: []
+        server_ids: []  # Apply to all servers
+        tenant_ids: []  # Apply to all tenants
+
+# Plugin directories to scan
+plugin_dirs:
+  - "plugins/native"      # Built-in plugins
+  - "plugins/custom"      # Custom organization plugins
+  - "/etc/mcpgateway/plugins"  # System-wide plugins
+
+# Global plugin settings
+plugin_settings:
+  parallel_execution_within_band: true
+  plugin_timeout: 30
+  fail_on_plugin_error: false
+  enable_plugin_api: true
+  plugin_health_check_interval: 60

tests/unit/mcpgateway/plugins/fixtures/configs/valid_stdio_external_plugin_passthrough.yaml

@@ -0,0 +1,22 @@
+# plugins/config.yaml - Main plugin configuration file
+
+plugins:
+  - name: "PassThroughPlugin"
+    kind: "external"
+    mcp:
+      proto: STDIO
+      script: mcpgateway/plugins/framework/external/mcp/server/runtime.py
+
+# Plugin directories to scan
+plugin_dirs:
+  - "plugins/native"      # Built-in plugins
+  - "plugins/custom"      # Custom organization plugins
+  - "/etc/mcpgateway/plugins"  # System-wide plugins
+
+# Global plugin settings
+plugin_settings:
+  parallel_execution_within_band: true
+  plugin_timeout: 30
+  fail_on_plugin_error: false
+  enable_plugin_api: true
+  plugin_health_check_interval: 60

tests/unit/mcpgateway/plugins/fixtures/plugins/__init__.py

@@ -0,0 +1,8 @@
+# -*- coding: utf-8 -*-
+"""
+
+Copyright 2025
+SPDX-License-Identifier: Apache-2.0
+Authors: Teryl Taylor
+
+"""

tests/unit/mcpgateway/plugins/fixtures/plugins/passthrough.py

@@ -0,0 +1,99 @@
+"""
+Passthrough plugin.
+
+Copyright 2025
+SPDX-License-Identifier: Apache-2.0
+"""
+
+
+from mcpgateway.plugins.framework import (
+    Plugin,
+    PluginContext,
+    PromptPosthookPayload,
+    PromptPosthookResult,
+    PromptPrehookPayload,
+    PromptPrehookResult,
+    ResourcePostFetchPayload,
+    ResourcePostFetchResult,
+    ResourcePreFetchPayload,
+    ResourcePreFetchResult,
+    ToolPostInvokePayload,
+    ToolPostInvokeResult,
+    ToolPreInvokePayload,
+    ToolPreInvokeResult,
+)
+
+class PassThroughPlugin(Plugin):
+    """A simple pass through plugin."""
+
+    async def prompt_pre_fetch(self, payload: PromptPrehookPayload, context: PluginContext) -> PromptPrehookResult:
+        """The 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.
+
+        Returns:
+            The result of the plugin's analysis, including whether the prompt can proceed.
+        """
+        return PromptPrehookResult(continue_processing=True)
+
+    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.
+
+        Returns:
+            The result of the plugin's analysis, including whether the prompt can proceed.
+        """
+        return PromptPosthookResult(continue_processing=True)
+
+    async def tool_pre_invoke(self, payload: ToolPreInvokePayload, context: PluginContext) -> ToolPreInvokeResult:
+        """Plugin hook run before a tool is invoked.
+
+        Args:
+            payload: The tool payload to be analyzed.
+            context: Contextual information about the hook call.
+
+        Returns:
+            The result of the plugin's analysis, including whether the tool can proceed.
+        """
+        return ToolPreInvokeResult(continue_processing=True)
+
+    async def tool_post_invoke(self, payload: ToolPostInvokePayload, context: PluginContext) -> ToolPostInvokeResult:
+        """Plugin hook run after a tool is invoked.
+
+        Args:
+            payload: The tool result payload to be analyzed.
+            context: Contextual information about the hook call.
+
+        Returns:
+            The result of the plugin's analysis, including whether the tool result should proceed.
+        """
+        return ToolPostInvokeResult(continue_processing=True)
+    
+    async def resource_post_fetch(self, payload: ResourcePostFetchPayload, context: PluginContext) -> ResourcePostFetchResult:
+        """Plugin hook run after a resource was fetched.
+
+        Args:
+            payload: The resource result payload to be analyzed.
+            context: Contextual information about the hook call.
+
+        Returns:
+            The result of the plugin's analysis, including whether the resource result should proceed.
+        """
+        return ResourcePostFetchResult(continue_processing=True)
+    
+    async def resource_pre_fetch(self, payload: ResourcePreFetchPayload, context: PluginContext) -> ResourcePreFetchResult:
+        """Plugin hook run before a resource was fetched.
+
+        Args:
+            payload: The resource result payload to be analyzed.
+            context: Contextual information about the hook call.
+
+        Returns:
+            The result of the plugin's analysis, including whether the resource result should proceed.
+        """
+        return ResourcePreFetchResult(continue_processing=True)

tests/unit/mcpgateway/plugins/framework/external/mcp/test_client_stdio.py

@@ -13,8 +13,21 @@
 from mcp import ClientSession, StdioServerParameters
 from mcp.client.stdio import stdio_client
 
-from mcpgateway.models import Message, PromptResult, Role, TextContent
-from mcpgateway.plugins.framework import ConfigLoader, PluginConfig, PluginLoader, PluginContext, PromptPrehookPayload, PromptPosthookPayload
+from mcpgateway.models import Message, PromptResult, ResourceContent, Role, TextContent
+from mcpgateway.plugins.framework import (
+    ConfigLoader,
+    GlobalContext,
+    PluginConfig,
+    PluginLoader,
+    PluginManager,
+    PluginContext,
+    PromptPrehookPayload,
+    PromptPosthookPayload,
+    ResourcePostFetchPayload,
+    ResourcePreFetchPayload,
+    ToolPostInvokePayload,
+    ToolPreInvokePayload,
+)
 from plugins.regex_filter.search_replace import SearchReplaceConfig
 
 
@@ -143,3 +156,51 @@ async def test_client_get_plugin_configs():
     assert srconfig.words[0].search == "crap"
     assert srconfig.words[0].replace == "crud"
     assert len(all_configs) == 2
+
+@pytest.mark.asyncio
+async def test_hooks():
+    os.environ["PLUGINS_CONFIG_PATH"] = "tests/unit/mcpgateway/plugins/fixtures/configs/valid_single_plugin_passthrough.yaml"
+    os.environ["PYTHONPATH"] = "."
+    pm = PluginManager()
+    if pm.initialized:
+        await pm.shutdown()
+    plugin_manager = PluginManager(config="tests/unit/mcpgateway/plugins/fixtures/configs/valid_stdio_external_plugin_passthrough.yaml")
+    await plugin_manager.initialize()
+    payload = PromptPrehookPayload(name="test_prompt", args={"arg0": "This is a crap argument"})
+    global_context = GlobalContext(request_id="1")
+    result, _ = await plugin_manager.prompt_pre_fetch(payload, global_context)
+    # Assert expected behaviors
+    assert result.continue_processing
+    """Test prompt post hook across all registered plugins."""
+    # Customize payload for testing
+    message = Message(content=TextContent(type="text", text="prompt"), role=Role.USER)
+    prompt_result = PromptResult(messages=[message])
+    payload = PromptPosthookPayload(name="test_prompt", result=prompt_result)
+    result, _ = await plugin_manager.prompt_post_fetch(payload, global_context)
+    # Assert expected behaviors
+    assert result.continue_processing
+    """Test tool pre hook across all registered plugins."""
+    # Customize payload for testing
+    payload = ToolPreInvokePayload(name="test_prompt", args={"arg0": "This is an argument"})
+    result, _ = await plugin_manager.tool_pre_invoke(payload, global_context)
+    # Assert expected behaviors
+    assert result.continue_processing
+    """Test tool post hook across all registered plugins."""
+    # Customize payload for testing
+    payload = ToolPostInvokePayload(name="test_tool", result={"output0": "output value"})
+    result, _ = await plugin_manager.tool_post_invoke(payload, global_context)
+    # Assert expected behaviors
+    assert result.continue_processing
+
+    payload = ResourcePreFetchPayload(uri="file:///data.txt")
+    result, _ = await plugin_manager.resource_pre_fetch(payload, global_context)
+    # Assert expected behaviors
+    assert result.continue_processing
+
+    content = ResourceContent(type="resource", uri="file:///data.txt",
+           text="Hello World")
+    payload = ResourcePostFetchPayload(uri="file:///data.txt", content=content)
+    result, _ = await plugin_manager.resource_post_fetch(payload, global_context)
+    # Assert expected behaviors
+    assert result.continue_processing
+    await plugin_manager.shutdown()