Json mixin for tools

docs/data-sources/builtin-toolsets/grafanadashboards.md

@@ -58,6 +58,16 @@ A [Grafana service account token](https://grafana.com/docs/grafana/latest/admini
 | grafana_get_home_dashboard | Get the home dashboard configuration |
 | grafana_get_dashboard_tags | List all tags used across dashboards for categorization |
 
+### Controlling response size
+
+Large dashboards can produce very big JSON payloads. To keep tool outputs readable you can:
+
+- **Limit depth:** `max_depth: 1` returns only the first level of keys, replacing deeper structures with a truncated marker.
+- **Filter with jsonpath:** `jsonpath: $.dashboard.panels[*].title` extracts just the panel titles.
+- **Filter with jq:** `jq: .dashboard.panels[].id` extracts panel IDs using familiar jq syntax.
+
+Filters run before depth truncation, so you can combine them to precisely scope the response.
+
 ## How it Works
 
 ### Dashboard Query Extraction

evals_per_branch.json

@@ -0,0 +1,5 @@
+{
+  "work": [
+    "tests/llm/fixtures/test_ask_holmes/170_grafana_dashboard_filters"
+  ]
+}

holmes/plugins/toolsets/coralogix/toolset_coralogix.py

@@ -13,6 +13,7 @@
     ToolParameter,
     Toolset,
 )
+from holmes.plugins.toolsets.json_filter_mixin import JsonFilterMixin
 from holmes.plugins.toolsets.consts import TOOLSET_CONFIG_MISSING_ERROR
 from holmes.plugins.toolsets.coralogix.api import (
     health_check,
@@ -58,7 +59,7 @@ def _build_coralogix_query_url(
         return None
 
 
-class ExecuteDataPrimeQuery(Tool):
+class ExecuteDataPrimeQuery(JsonFilterMixin, Tool):
     def __init__(self, toolset: "CoralogixToolset"):
         super().__init__(
             name="coralogix_execute_dataprime_query",
@@ -166,14 +167,13 @@ def _invoke(self, params: dict, context: ToolInvokeContext) -> StructuredToolRes
             tier=tier,
         )
 
-        # Return a pretty-printed JSON string for readability by the model/user.
-        final_result = json.dumps(result_dict, indent=2, sort_keys=False)
-        return StructuredToolResult(
+        wrapped = StructuredToolResult(
             status=status,
-            data=final_result,
+            data=result_dict,
             params=params,
             url=explore_url,
         )
+        return self.filter_result(wrapped, params)
 
     def get_parameterized_one_liner(self, params) -> str:
         description = params.get("description", "")

holmes/plugins/toolsets/grafana/toolset_grafana.py

@@ -18,6 +18,7 @@
     build_headers,
 )
 from holmes.plugins.toolsets.utils import toolset_name_for_one_liner
+from holmes.plugins.toolsets.json_filter_mixin import JsonFilterMixin
 
 
 class GrafanaDashboardConfig(GrafanaConfig):
@@ -237,19 +238,21 @@ def get_parameterized_one_liner(self, params: Dict) -> str:
         return f"{toolset_name_for_one_liner(self._toolset.name)}: Search Dashboards"
 
 
-class GetDashboardByUID(BaseGrafanaTool):
+class GetDashboardByUID(JsonFilterMixin, BaseGrafanaTool):
     def __init__(self, toolset: GrafanaToolset):
         super().__init__(
             toolset=toolset,
             name="grafana_get_dashboard_by_uid",
             description="Get a dashboard by its UID using the /api/dashboards/uid/:uid endpoint",
-            parameters={
-                "uid": ToolParameter(
-                    description="The unique identifier of the dashboard",
-                    type="string",
-                    required=True,
-                ),
-            },
+            parameters=self.extend_parameters(
+                {
+                    "uid": ToolParameter(
+                        description="The unique identifier of the dashboard",
+                        type="string",
+                        required=True,
+                    )
+                }
+            ),
         )
 
     def _invoke(self, params: dict, context: ToolInvokeContext) -> StructuredToolResult:
@@ -260,42 +263,35 @@ def _invoke(self, params: dict, context: ToolInvokeContext) -> StructuredToolRes
             self._toolset.grafana_config, uid=uid
         )
 
-        return StructuredToolResult(
-            status=result.status,
-            data=result.data,
-            params=result.params,
-            url=dashboard_url if dashboard_url else result.url,
-        )
+        filtered_result = self.filter_result(result, params)
+        filtered_result.url = dashboard_url if dashboard_url else result.url
+        return filtered_result
 
     def get_parameterized_one_liner(self, params: Dict) -> str:
         return f"{toolset_name_for_one_liner(self._toolset.name)}: Get Dashboard {params.get('uid', '')}"
 
 
-class GetHomeDashboard(BaseGrafanaTool):
+class GetHomeDashboard(JsonFilterMixin, BaseGrafanaTool):
     def __init__(self, toolset: GrafanaToolset):
         super().__init__(
             toolset=toolset,
             name="grafana_get_home_dashboard",
             description="Get the home dashboard using the /api/dashboards/home endpoint",
-            parameters={},
+            parameters=self.extend_parameters({}),
         )
 
     def _invoke(self, params: dict, context: ToolInvokeContext) -> StructuredToolResult:
         result = self._make_grafana_request("/api/dashboards/home", params)
-
         config = self._toolset.grafana_config
         dashboard_url = None
         if isinstance(result.data, dict):
             uid = result.data.get("dashboard", {}).get("uid")
             if uid:
                 dashboard_url = _build_grafana_dashboard_url(config, uid=uid)
 
-        return StructuredToolResult(
-            status=result.status,
-            data=result.data,
-            params=result.params,
-            url=dashboard_url if dashboard_url else None,
-        )
+        filtered_result = self.filter_result(result, params)
+        filtered_result.url = dashboard_url if dashboard_url else None
+        return filtered_result
 
     def get_parameterized_one_liner(self, params: Dict) -> str:
         return f"{toolset_name_for_one_liner(self._toolset.name)}: Get Home Dashboard"

holmes/plugins/toolsets/json_filter_mixin.py

@@ -0,0 +1,149 @@
+import json
+import logging
+from typing import Any, Dict, Optional, Tuple
+
+import jq  # type: ignore
+from jsonpath_ng import parse as jsonpath_parse  # type: ignore
+
+from holmes.core.tools import StructuredToolResult, StructuredToolResultStatus, ToolParameter
+
+logger = logging.getLogger(__name__)
+
+
+def _truncate_to_depth(value: Any, max_depth: Optional[int], current_depth: int = 0):
+    """Recursively truncate dictionaries/lists beyond the requested depth."""
+    if max_depth is None or max_depth < 0:
+        return value
+
+    if current_depth >= max_depth:
+        if isinstance(value, (dict, list)):
+            return f"...truncated at depth {max_depth}"
+        return value
+
+    if isinstance(value, dict):
+        return {
+            key: _truncate_to_depth(sub_value, max_depth, current_depth + 1)
+            for key, sub_value in value.items()
+        }
+    if isinstance(value, list):
+        return [
+            _truncate_to_depth(item, max_depth, current_depth + 1) for item in value
+        ]
+
+    return value
+
+
+def _apply_jsonpath_filter(data: Any, expression: str) -> Tuple[Optional[Any], Optional[str]]:
+    try:
+        jsonpath_expr = jsonpath_parse(expression)
+        matches = [match.value for match in jsonpath_expr.find(data)]
+        if len(matches) == 1:
+            return matches[0], None
+        return matches, None
+    except Exception as exc:  # pragma: no cover - defensive
+        logger.debug("Failed to apply jsonpath filter", exc_info=exc)
+        return None, f"Invalid jsonpath expression: {exc}"
+
+
+def _apply_jq_filter(data: Any, expression: str) -> Tuple[Optional[Any], Optional[str]]:
+    try:
+        compiled = jq.compile(expression)
+        results = compiled.input(data).all()
+        if len(results) == 1:
+            return results[0], None
+        return results, None
+    except Exception as exc:  # pragma: no cover - defensive
+        logger.debug("Failed to apply jq filter", exc_info=exc)
+        return None, f"Invalid jq filter: {exc}"
+
+
+class JsonFilterMixin:
+    """Opt-in mixin for tools that return JSON and want filtering controls."""
+
+    filter_parameters: Dict[str, ToolParameter] = {
+        "max_depth": ToolParameter(
+            description="Maximum nesting depth to return from the JSON (0 returns only top-level keys). Leave empty for full response.",
+            type="integer",
+            required=False,
+        ),
+        "jsonpath": ToolParameter(
+            description="Optional jsonpath expression to extract specific parts of the JSON.",
+            type="string",
+            required=False,
+        ),
+        "jq": ToolParameter(
+            description="Optional jq filter to apply to the JSON.",
+            type="string",
+            required=False,
+        ),
+    }
+
+    @classmethod
+    def extend_parameters(cls, existing: Dict[str, ToolParameter]) -> Dict[str, ToolParameter]:
+        merged = dict(cls.filter_parameters)
+        merged.update(existing)
+        return merged
+
+    def _filter_result_data(self, data: Any, params: Dict) -> Tuple[Any, Optional[str]]:
+        parsed_data = data
+        if isinstance(data, str):
+            try:
+                parsed_data = json.loads(data)
+            except Exception:
+                # Not JSON, leave as-is
+                return data, None
+
+        if params.get("jsonpath"):
+            parsed_data, error = _apply_jsonpath_filter(parsed_data, params["jsonpath"])
+            if error:
+                return None, error
+
+        if params.get("jq"):
+            parsed_data, error = _apply_jq_filter(parsed_data, params["jq"])
+            if error:
+                return None, error
+
+        parsed_data = _truncate_to_depth(parsed_data, params.get("max_depth"))
+        return parsed_data, None
+
+    @staticmethod
+    def _safe_string(value: Any) -> Optional[str]:
+        if value is None:
+            return None
+        if isinstance(value, str):
+            return value
+        try:
+            return str(value)
+        except Exception:
+            return None
+
+    def filter_result(self, result: StructuredToolResult, params: Dict) -> StructuredToolResult:
+        base_result = result if isinstance(result, StructuredToolResult) else None
+        if base_result is None:
+            base_result = StructuredToolResult(
+                status=getattr(result, "status", StructuredToolResultStatus.SUCCESS),
+                data=getattr(result, "data", None),
+                params=getattr(result, "params", params),
+                url=self._safe_string(getattr(result, "url", None)),
+                invocation=self._safe_string(getattr(result, "invocation", None)),
+                icon_url=self._safe_string(getattr(result, "icon_url", None)),
+            )
+        else:
+            # Normalize string fields to avoid MagicMock validation failures
+            base_result.url = self._safe_string(base_result.url)
+            base_result.invocation = self._safe_string(base_result.invocation)
+            base_result.icon_url = self._safe_string(base_result.icon_url)
+
+        filtered_data, error = self._filter_result_data(base_result.data, params)
+        if error:
+            return StructuredToolResult(
+                status=StructuredToolResultStatus.ERROR,
+                error=error,
+                params=params,
+                url=base_result.url,
+                invocation=base_result.invocation,
+                icon_url=base_result.icon_url,
+            )
+
+        base_result.data = filtered_data
+        return base_result