grafana_loki_query

Query Grafana Loki log data with time-based filtering and structured matchers to retrieve specific log entries for analysis.

Instructions

Queries Grafana Loki for log data. Fetches logs for a specified duration (e.g., '5m', '1h', '2d'), converts relative time to absolute timestamps. Note: Loki queries require at least one non-empty matcher. Use patterns like '{job=~".+"}' instead of '{job=~".*"}' or '{}' to avoid syntax errors.

Input Schema

TableJSON Schema

Name	Required	Description
`datasource_uid`	Yes	Loki datasource UID
`query`	Yes	Loki query string (e.g., '{job=~".+"}' or '{app="myapp"}')
`duration`	No	Time duration (e.g., '5m', '1h', '2d') - overrides start_time/end_time if provided
`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')
`limit`	No	Maximum number of log entries to return

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:

Grafana MCP Server