grafana_loki_query

Instructions

Implementation Reference

• src/grafana_mcp_server/processor/grafana_processor.py:254-324 (handler)

  The `grafana_loki_query` function in the GrafanaProcessor class executes Loki queries against a specified datasource using the Grafana API.

  def grafana_loki_query(
      self,
      datasource_uid: str,
      query: str,
      duration: Optional[str] = None,
      start_time: Optional[str] = None,
      end_time: Optional[str] = None,
      limit: int = 100,
  ) -> dict[str, Any]:
      """
      Queries Grafana Loki for log data.
  
      Args:
          query: Loki query string
          duration: Time duration (e.g., '5m', '1h', '2d') - overrides start_time/end_time if provided
          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')
          limit: Maximum number of log entries to return
  
      Returns:
          Dict containing log data from Loki datasource
      """
      try:
          # Use standardized time range logic
          start_dt, end_dt = self._get_time_range(start_time, end_time, duration, default_hours=1)
  
          # 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,
                      "datasource": {"type": "loki", "uid": datasource_uid},
                      "maxLines": limit,
                  }
              ],
              "from": str(start_ms),
              "to": str(end_ms),
          }
  
          url = f"{self.__host}/api/ds/query"
          logger.info(f"Executing Loki 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:
              data = response.json()
              return {
                  "status": "success",
                  "query": query,
                  "start_time": start_dt.isoformat(),
                  "end_time": end_dt.isoformat(),
                  "duration": duration,
                  "limit": limit,
                  "results": data,
              }
          else:
              raise Exception(f"Loki query failed. Status: {response.status_code}, Response: {response.text}")
  
      except Exception as e:
          logger.error(f"Error executing Loki query: {e!s}")
          raise e

• src/grafana_mcp_server/mcp_server.py:150-151 (registration)

  The tool 'grafana_loki_query' is defined in the tools list within the MCP server setup.

  "name": "grafana_loki_query",
  "description": "Queries Grafana Loki for log data. Fetches logs for a specified duration "

• src/grafana_mcp_server/mcp_server.py:470-470 (registration)

  The tool 'grafana_loki_query' is registered in the tool handler mapping in the MCP server.

  "grafana_loki_query": grafana_loki_query,

• src/grafana_mcp_server/mcp_server.py:321-333 (handler)

  The tool handler function in `mcp_server.py` that calls the underlying processor method.

  def grafana_loki_query(datasource_uid, query, duration=None, start_time=None, end_time=None, limit=100):
      """Query Grafana Loki for log data"""
      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_loki_query(datasource_uid, query, duration, start_time, end_time, limit)
          return result
      except Exception as e: