Tools Telemetry Exporter plugin

docs/docs/using/plugins/plugins.md

@@ -6,6 +6,7 @@ MCP Context Forge provides a comprehensive collection of production-ready plugin
 
 - [Security & Safety](#security-safety)
 - [Reliability & Performance](#reliability-performance)
+- [Observability & Monitoring](#observability-monitoring)
 - [Content Transformation & Formatting](#content-transformation-formatting)
 - [Content Filtering & Validation](#content-filtering-validation)
 - [Compliance & Governance](#compliance-governance)
@@ -44,6 +45,14 @@ Plugins for improving system reliability, performance, and resource management.
 | [Response Cache by Prompt](https://github.com/IBM/mcp-context-forge/tree/main/plugins/response_cache_by_prompt) | Native | Advisory response cache using cosine similarity over prompt/input fields with configurable threshold |
 | [Retry with Backoff](https://github.com/IBM/mcp-context-forge/tree/main/plugins/retry_with_backoff) | Native | Annotates retry/backoff policy in metadata with exponential backoff on specific HTTP status codes |
 
+## Observability & Monitoring
+
+Plugins for telemetry, tracing, and monitoring tool invocations.
+
+| Plugin | Type | Description |
+|--------|------|-------------|
+| [Tools Telemetry Exporter](https://github.com/IBM/mcp-context-forge/tree/main/plugins/tools_telemetry_exporter) | Native | Export comprehensive tool invocation telemetry to OpenTelemetry for observability and monitoring with configurable payload export |
+
 ## Content Transformation & Formatting
 
 Plugins for transforming, formatting, and normalizing content.

plugins/config.yaml

@@ -899,3 +899,18 @@
         api_key: "" # optional, can define ANTHROPIC_API_KEY instead
       model_id: "ibm/granite-3-3-8b-instruct" # note that this changes depending on provider
       length_threshold: 100000
+
+  # Tools Telemetry Exporter - export tool invocation telemetry to OpenTelemetry
+  - name: "ToolsTelemetryExporter"
+    kind: "plugins.tools_telemetry_exporter.telemetry_exporter.ToolsTelemetryExporterPlugin"
+    description: "Export comprehensive tool invocation telemetry to OpenTelemetry"
+    version: "0.1.0"
+    author: "Bar Haim"
+    hooks: ["tool_pre_invoke", "tool_post_invoke"]
+    tags: ["telemetry", "observability", "opentelemetry", "monitoring"]
+    mode: "disabled" # enforce | permissive | disabled
+    priority: 200  # Run late to capture all context
+    conditions: []  # Apply to all tools
+    config:
+        export_full_payload: true
+        max_payload_bytes_size: 10000

plugins/tools_telemetry_exporter/README.md

@@ -0,0 +1,74 @@
+# Tools Telemetry Exporter Plugin
+
+> Author: Bar Haim
+> Version: 0.1.0
+
+Export comprehensive tool invocation telemetry to OpenTelemetry for observability and monitoring.
+
+## Hooks
+- `tool_pre_invoke`
+- `tool_post_invoke`
+
+## Config
+```yaml
+config:
+  export_full_payload: true
+  max_payload_bytes_size: 10000  # 10 KB default
+```
+
+## Features
+
+- **Pre-Invocation Telemetry**: Captures request context, tool metadata, target MCP server details, and tool arguments
+- **Post-Invocation Telemetry**: Captures request context, tool results (optional), and error status
+- **Automatic Payload Truncation**: Large results are truncated to respect size limits
+- **Graceful Degradation**: Automatically disables if OpenTelemetry is not available
+
+## Exported Attributes
+
+### Pre-Invocation (`tool.pre_invoke`)
+- Request metadata: `request_id`, `user`, `tenant_id`, `server_id`
+- Target server: `target_mcp_server.id`, `target_mcp_server.name`, `target_mcp_server.url`
+- Tool info: `tool.name`, `tool.target_tool_name`, `tool.description`
+- Invocation data: `tool.invocation.args`, `headers`
+
+### Post-Invocation (`tool.post_invoke`)
+- Request metadata: `request_id`, `user`, `tenant_id`, `server_id`
+- Results: `tool.invocation.result` (if `export_full_payload` is enabled and no error)
+- Status: `tool.invocation.has_error`
+
+## Configuration Options
+
+| Option | Default | Description |
+|--------|---------|-------------|
+| `export_full_payload` | `true` | Export full tool results in post-invocation telemetry |
+| `max_payload_bytes_size` | `10000` | Maximum payload size in bytes before truncation |
+
+## Requirements
+
+OpenTelemetry enabled on MCP context forge (see [Observability Setup](../../docs/docs/manage/observability.md#opentelemetry-external)).
+
+
+## Usage
+
+```yaml
+plugins:
+  - name: "ToolsTelemetryExporter"
+    kind: "plugins.tools_telemetry_exporter.telemetry_exporter.ToolsTelemetryExporterPlugin"
+    hooks: ["tool_pre_invoke", "tool_post_invoke"]
+    mode: "permissive"
+    priority: 200  # Run late to capture all context
+    config:
+        export_full_payload: true
+        max_payload_bytes_size: 10000
+```
+
+## Limitations
+
+- Requires active OpenTelemetry tracing to export telemetry
+- No local buffering; telemetry exported in real-time only
+
+## Security Notes
+
+- Tool arguments are always exported in pre-invocation telemetry
+- Consider running PII filter plugin before this plugin to sanitize data
+- Disable `export_full_payload` in production for sensitive workloads

plugins/tools_telemetry_exporter/__init__.py

@@ -0,0 +1,5 @@
+"""Tools Telemetry Exporter Plugin - exports tool invocation telemetry to OpenTelemetry."""
+
+from plugins.tools_telemetry_exporter.telemetry_exporter import ToolsTelemetryExporterPlugin
+
+__all__ = ["ToolsTelemetryExporterPlugin"]

plugins/tools_telemetry_exporter/plugin-manifest.yaml

@@ -0,0 +1,9 @@
+description: "Export comprehensive tool invocation telemetry to OpenTelemetry"
+author: "Bar Haim"
+version: "0.1.0"
+available_hooks:
+  - "tool_pre_invoke"
+  - "tool_post_invoke"
+default_configs:
+  export_full_payload: true
+  max_payload_bytes_size: 10000  # (10 KB default)

plugins/tools_telemetry_exporter/telemetry_exporter.py

@@ -0,0 +1,205 @@
+# -*- coding: utf-8 -*-
+"""Location: ./plugins/tools_telemetry_exporter/telemetry_exporter.py
+Copyright 2025
+SPDX-License-Identifier: Apache-2.0
+
+Tools Telemetry Exporter Plugin.
+This plugin exports comprehensive tool invocation telemetry to OpenTelemetry.
+"""
+
+# Standard
+import json
+from typing import Dict
+
+# First-Party
+from mcpgateway.common.models import Gateway, Tool
+from mcpgateway.plugins.framework import Plugin, PluginConfig, PluginContext
+from mcpgateway.plugins.framework.constants import GATEWAY_METADATA, TOOL_METADATA
+from mcpgateway.plugins.framework.hooks.tools import ToolPostInvokePayload, ToolPostInvokeResult, ToolPreInvokePayload, ToolPreInvokeResult
+from mcpgateway.services.logging_service import LoggingService
+
+# Initialize logging service first
+logging_service = LoggingService()
+logger = logging_service.get_logger(__name__)
+
+
+class ToolsTelemetryExporterPlugin(Plugin):
+    """Export comprehensive tool invocation telemetry to OpenTelemetry."""
+
+    def __init__(self, config: PluginConfig):
+        """Initialize the ToolsTelemetryExporterPlugin."""
+        super().__init__(config)
+        self.is_open_telemetry_available = self._is_open_telemetry_available()
+        self.telemetry_config = config.config
+
+    @staticmethod
+    def _is_open_telemetry_available() -> bool:
+        """Check if OpenTelemetry is available for import.
+
+        Returns:
+            True if OpenTelemetry can be imported, False otherwise.
+        """
+        try:
+            # Third-Party
+            from opentelemetry import trace  # noqa: F401  # pylint: disable=import-outside-toplevel,unused-import
+
+            return True
+        except ImportError:
+            logger.warning("ToolsTelemetryExporter: OpenTelemetry is not available. Telemetry export will be disabled.")
+            return False
+
+    @staticmethod
+    def _get_base_context_attributes(context: PluginContext) -> Dict:
+        """Extract base context attributes from plugin context.
+
+        Args:
+            context: Plugin execution context containing global context.
+
+        Returns:
+            Dictionary with base attributes (request_id, user, tenant_id, server_id).
+        """
+        global_context = context.global_context
+        return {
+            "request_id": global_context.request_id or "",
+            "user": global_context.user or "",
+            "tenant_id": global_context.tenant_id or "",
+            "server_id": global_context.server_id or "",
+        }
+
+    def _get_pre_invoke_context_attributes(self, context: PluginContext) -> Dict:
+        """Extract pre-invocation context attributes including tool and gateway metadata.
+
+        Args:
+            context: Plugin execution context containing tool and gateway metadata.
+
+        Returns:
+            Dictionary with base attributes plus tool and target MCP server details.
+        """
+        global_context = context.global_context
+        tool_metadata: Tool = global_context.metadata.get(TOOL_METADATA)
+        target_mcp_server_metadata: Gateway = global_context.metadata.get(GATEWAY_METADATA)
+
+        return {
+            **self._get_base_context_attributes(context),
+            "tool": {
+                "name": tool_metadata.name or "",
+                "target_tool_name": tool_metadata.original_name or "",
+                "description": tool_metadata.description or "",
+            },
+            "target_mcp_server": {
+                "id": target_mcp_server_metadata.id or "",
+                "name": target_mcp_server_metadata.name or "",
+                "url": str(target_mcp_server_metadata.url or ""),
+            },
+        }
+
+    def _get_post_invoke_context_attributes(self, context: PluginContext) -> Dict:
+        """Extract post-invocation context attributes.
+
+        Args:
+            context: Plugin execution context.
+
+        Returns:
+            Dictionary with base context attributes for post-invocation telemetry.
+        """
+        return {
+            **self._get_base_context_attributes(context),
+        }
+
+    async def tool_pre_invoke(self, payload: ToolPreInvokePayload, context: PluginContext) -> ToolPreInvokeResult:
+        """Capture pre-invocation telemetry for tools.
+
+        Args:
+            payload: The tool payload containing arguments.
+            context: Plugin execution context.
+
+        Returns:
+            Result with potentially modified tool arguments.
+        """
+        logger.info("ToolsTelemetryExporter: Capturing pre-invocation tool telemetry.")
+        context_attributes = self._get_pre_invoke_context_attributes(context)
+
+        export_attributes = {
+            "request_id": context_attributes["request_id"],
+            "user": context_attributes["user"],
+            "tenant_id": context_attributes["tenant_id"],
+            "server_id": context_attributes["server_id"],
+            "target_mcp_server.id": context_attributes["target_mcp_server"]["id"],
+            "target_mcp_server.name": context_attributes["target_mcp_server"]["name"],
+            "target_mcp_server.url": context_attributes["target_mcp_server"]["url"],
+            "tool.name": context_attributes["tool"]["name"],
+            "tool.target_tool_name": context_attributes["tool"]["target_tool_name"],
+            "tool.description": context_attributes["tool"]["description"],
+            "tool.invocation.args": json.dumps(payload.args),
+            "headers": payload.headers.model_dump_json(),
+        }
+
+        await self._export_telemetry(attributes=export_attributes, span_name="tool.pre_invoke")
+        return ToolPreInvokeResult(continue_processing=True)
+
+    async def tool_post_invoke(self, payload: ToolPostInvokePayload, context: PluginContext) -> ToolPostInvokeResult:
+        """Capture post-invocation telemetry.
+
+        Args:
+            payload: Tool result payload containing the tool name and execution result.
+            context: Plugin context with state from pre-invoke hook.
+
+        Returns:
+            ToolPostInvokeResult allowing execution to continue.
+        """
+        logger.info("ToolsTelemetryExporter: Capturing post-invocation tool telemetry.")
+        context_attributes = self._get_post_invoke_context_attributes(context)
+
+        export_attributes = {
+            "request_id": context_attributes["request_id"],
+            "user": context_attributes["user"],
+            "tenant_id": context_attributes["tenant_id"],
+            "server_id": context_attributes["server_id"],
+        }
+
+        result = payload.result if payload.result else {}
+        has_error = result.get("isError", True)
+        if self.telemetry_config.get("export_full_payload", False) and not has_error:
+            max_payload_bytes_size = self.telemetry_config.get("max_payload_bytes_size", 10000)
+            result_content = result.get("content")
+            if result_content:
+                result_content_str = json.dumps(result_content, default=str)
+                if len(result_content_str) <= max_payload_bytes_size:
+                    export_attributes["tool.invocation.result"] = result_content_str
+                else:
+                    truncated_content = result_content_str[:max_payload_bytes_size]
+                    export_attributes["tool.invocation.result"] = truncated_content + "...<truncated>"
+            else:
+                export_attributes["tool.invocation.result"] = "<No content in result>"
+        export_attributes["tool.invocation.has_error"] = has_error
+
+        await self._export_telemetry(attributes=export_attributes, span_name="tool.post_invoke")
+        return ToolPostInvokeResult(continue_processing=True)
+
+    async def _export_telemetry(self, attributes: Dict, span_name: str) -> None:
+        """Export telemetry attributes to OpenTelemetry.
+
+        Args:
+            attributes: Dictionary of telemetry attributes to export.
+            span_name: Name of the OpenTelemetry span to create.
+        """
+        if not self.is_open_telemetry_available:
+            logger.debug("ToolsTelemetryExporter: OpenTelemetry not available. Skipping telemetry export.")
+            return
+
+        # Third-Party
+        from opentelemetry import trace  # pylint: disable=import-outside-toplevel
+
+        try:
+            tracer = trace.get_tracer(__name__)
+            current_span = trace.get_current_span()
+            if not current_span or not current_span.is_recording():
+                logger.warning("ToolsTelemetryExporter: No active span found. Skipping telemetry export.")
+                return
+
+            with tracer.start_as_current_span(span_name) as span:
+                for key, value in attributes.items():
+                    span.set_attribute(key, value)
+                logger.debug(f"ToolsTelemetryExporter: Exported telemetry for span '{span_name}' with attributes: {attributes}")
+        except Exception as e:
+            logger.error(f"ToolsTelemetryExporter: Error creating span '{span_name}': {e}", exc_info=True)