grafana_promql_query
Execute PromQL queries to fetch and analyze metrics data from Grafana's Prometheus datasource, optimizing time series responses for efficient data retrieval.
Instructions
Executes PromQL queries against Grafana's Prometheus datasource. Fetches metrics data using PromQL expressions, optimizes time series responses to reduce token size.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| datasource_uid | Yes | Prometheus datasource UID | |
| query | Yes | PromQL query string | |
| start_time | No | Start time in RFC3339 or relative string (e.g., 'now-2h', '2023-01-01T00:00:00Z') | |
| end_time | No | End time in RFC3339 or relative string (e.g., 'now-2h', '2023-01-01T00:00:00Z') | |
| duration | No | Duration string for the time window (e.g., '2h', '90m') |
Implementation Reference
- The core logic for executing the PromQL query using the Grafana /api/ds/query endpoint.
def grafana_promql_query( self, datasource_uid: str, query: str, start_time: Optional[str] = None, end_time: Optional[str] = None, duration: Optional[str] = None, ) -> dict[str, Any]: """ Executes PromQL queries against Grafana's Prometheus datasource. Args: datasource_uid: Prometheus datasource UID query: PromQL query string start_time: Start time in RFC3339 or relative string (e.g., 'now-2h', '2023-01-01T00:00:00Z') end_time: End time in RFC3339 or relative string (e.g., 'now-2h', '2023-01-01T00:00:00Z') duration: Duration string for the time window (e.g., '2h', '90m') Returns: Dict containing query results with optimized time series data """ try: # Use standardized time range logic start_dt, end_dt = self._get_time_range(start_time, end_time, duration, default_hours=3) # Convert to milliseconds since epoch (Grafana format) start_ms = int(start_dt.timestamp() * 1000) end_ms = int(end_dt.timestamp() * 1000) payload = { "queries": [ { "refId": "A", "expr": query, "editorMode": "code", "legendFormat": "__auto", "range": True, "exemplar": False, "requestId": "A", "utcOffsetSec": 0, "scopes": [], "adhocFilters": [], "interval": "", "datasource": {"type": "prometheus", "uid": datasource_uid}, "intervalMs": 30000, "maxDataPoints": 1000, } ], "from": str(start_ms), "to": str(end_ms), } url = f"{self.__host}/api/ds/query" logger.info(f"Executing PromQL query: {query} from {start_dt.isoformat()} to {end_dt.isoformat()}") response = requests.post( url, headers=self.headers, json=payload, verify=self.__ssl_verify, timeout=30, ) if response.status_code == 200: - The MCP server wrapper function that calls the grafana_processor logic.
def grafana_promql_query(datasource_uid, query, start_time=None, end_time=None, duration=None): """Execute PromQL query against Grafana's Prometheus datasource""" try: grafana_processor = current_app.config.get("grafana_processor") if not grafana_processor: return { "status": "error", "message": "Grafana processor not initialized. Check configuration.", } result = grafana_processor.grafana_promql_query(datasource_uid, query, start_time, end_time, duration) return result except Exception as e: logger.error(f"Error executing PromQL query: {e!s}") return {"status": "error", "message": f"PromQL query failed: {e!s}"}