Grafana MCP Server

Instructions

Implementation Reference

• src/grafana_mcp_server/processor/grafana_processor.py:358-441 (handler)

  The core logic for querying Grafana dashboard panels, which fetches the dashboard definition, filters for requested panels, and executes panel-specific queries.

  def grafana_query_dashboard_panels(
      self,
      dashboard_uid: str,
      panel_ids: list[int],
      template_variables: Optional[dict[str, str]] = None,
  ) -> dict[str, Any]:
      """
      Executes queries for specific dashboard panels.
  
      Args:
          dashboard_uid: Dashboard UID
          panel_ids: List of panel IDs to query (max 4)
          template_variables: Template variables for the dashboard
  
      Returns:
          Dict containing panel data with optimized metrics
      """
      try:
          if len(panel_ids) > 4:
              raise ValueError("Maximum 4 panels can be queried at once")
  
          logger.info(f"Querying dashboard panels: {dashboard_uid}, panel_ids: {panel_ids}")
  
          # First get dashboard configuration
          dashboard_url = f"{self.__host}/api/dashboards/uid/{dashboard_uid}"
          dashboard_response = requests.get(
              dashboard_url,
              headers=self.headers,
              verify=self.__ssl_verify,
              timeout=20,
          )
  
          if dashboard_response.status_code != 200:
              raise Exception(f"Failed to fetch dashboard. Status: {dashboard_response.status_code}")
  
          dashboard_data = dashboard_response.json()
          dashboard = dashboard_data.get("dashboard", {})
  
          # Handle both old and new dashboard structures
          panels = dashboard.get("panels", [])
          if not panels:
              # Try to get panels from rows (newer dashboard structure)
              rows = dashboard.get("rows", [])
              for row in rows:
                  row_panels = row.get("panels", [])
                  panels.extend(row_panels)
  
          logger.info(f"Found {len(panels)} panels in dashboard")
  
          # Filter panels by requested IDs
          target_panels = [panel for panel in panels if panel.get("id") in panel_ids]
  
          if not target_panels:
              logger.warning(f"No panels found with IDs: {panel_ids}")
              logger.info(f"Available panel IDs: {[panel.get('id') for panel in panels]}")
              raise Exception(f"No panels found with IDs: {panel_ids}")
  
          logger.info(f"Found {len(target_panels)} target panels")
  
          # Execute queries for each panel
          panel_results = []
          for panel in target_panels:
              logger.info(f"Processing panel {panel.get('id')}: {panel.get('title', 'Unknown')}")
              panel_result = self._execute_panel_query(panel, template_variables or {})
              panel_results.append(
                  {
                      "panel_id": panel.get("id"),
                      "title": panel.get("title"),
                      "type": panel.get("type"),
                      "data": panel_result,
                  }
              )
  
          return {
              "status": "success",
              "dashboard_uid": dashboard_uid,
              "panel_ids": panel_ids,
              "template_variables": template_variables,
              "results": panel_results,
          }
  
      except Exception as e:
          logger.error(f"Error querying dashboard panels: {e!s}")
          raise e