Grafana MCP Server

Instructions

Implementation Reference

• src/grafana_mcp_server/processor/grafana_processor.py:172-235 (handler)

  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:

• src/grafana_mcp_server/mcp_server.py:304-318 (handler)

  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}"}