execute_range_query
Retrieve Prometheus metrics over a time range using PromQL queries with specified start, end, and step intervals for time-series analysis.
Instructions
Execute a PromQL range query with start time, end time, and step interval
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | ||
| start | Yes | ||
| end | Yes | ||
| step | Yes |
Implementation Reference
- The handler function that executes the PromQL range query. It constructs the query parameters, makes the API request to Prometheus using make_prometheus_request, handles progress reporting via context if provided, adds Prometheus UI links, and returns the formatted result.async def execute_range_query(query: str, start: str, end: str, step: str, ctx: Context | None = None) -> Dict[str, Any]: """Execute a range query against Prometheus. Args: query: PromQL query string start: Start time as RFC3339 or Unix timestamp end: End time as RFC3339 or Unix timestamp step: Query resolution step width (e.g., '15s', '1m', '1h') Returns: Range query result with type (usually matrix) and values over time """ params = { "query": query, "start": start, "end": end, "step": step } logger.info("Executing range query", query=query, start=start, end=end, step=step) # Report progress if context available if ctx: await ctx.report_progress(progress=0, total=100, message="Initiating range query...") data = make_prometheus_request("query_range", params=params) # Report progress if ctx: await ctx.report_progress(progress=50, total=100, message="Processing query results...") result = { "resultType": data["resultType"], "result": data["result"] } if not config.disable_prometheus_links: from urllib.parse import urlencode ui_params = { "g0.expr": query, "g0.tab": "0", "g0.range_input": f"{start} to {end}", "g0.step_input": step } prometheus_ui_link = f"{config.url.rstrip('/')}/graph?{urlencode(ui_params)}" result["links"] = [{ "href": prometheus_ui_link, "rel": "prometheus-ui", "title": "View in Prometheus UI" }] # Report completion if ctx: await ctx.report_progress(progress=100, total=100, message="Range query completed") logger.info("Range query completed", query=query, result_type=data["resultType"], result_count=len(data["result"]) if isinstance(data["result"], list) else 1) return result
- src/prometheus_mcp_server/server.py:291-301 (registration)Registers the execute_range_query tool with the FastMCP server, providing description and annotations for the tool schema.@mcp.tool( description="Execute a PromQL range query with start time, end time, and step interval", annotations={ "title": "Execute PromQL Range Query", "icon": "📈", "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True } )
- Input parameter types, descriptions via docstring Args section, and return type definition serving as the tool schema.async def execute_range_query(query: str, start: str, end: str, step: str, ctx: Context | None = None) -> Dict[str, Any]: """Execute a range query against Prometheus. Args: query: PromQL query string start: Start time as RFC3339 or Unix timestamp end: End time as RFC3339 or Unix timestamp step: Query resolution step width (e.g., '15s', '1m', '1h') Returns: Range query result with type (usually matrix) and values over time """
- Shared helper function that performs the actual HTTP request to the Prometheus API endpoint '/api/v1/query_range', handles authentication, error checking, and returns the data. Called by execute_range_query.def make_prometheus_request(endpoint, params=None): """Make a request to the Prometheus API with proper authentication and headers.""" if not config.url: logger.error("Prometheus configuration missing", error="PROMETHEUS_URL not set") raise ValueError("Prometheus configuration is missing. Please set PROMETHEUS_URL environment variable.") if not config.url_ssl_verify: logger.warning("SSL certificate verification is disabled. This is insecure and should not be used in production environments.", endpoint=endpoint) url = f"{config.url.rstrip('/')}/api/v1/{endpoint}" url_ssl_verify = config.url_ssl_verify auth = get_prometheus_auth() headers = {} if isinstance(auth, dict): # Token auth is passed via headers headers.update(auth) auth = None # Clear auth for requests.get if it's already in headers # Add OrgID header if specified if config.org_id: headers["X-Scope-OrgID"] = config.org_id if config.custom_headers: headers.update(config.custom_headers) try: logger.debug("Making Prometheus API request", endpoint=endpoint, url=url, params=params, headers=headers) # Make the request with appropriate headers and auth response = requests.get(url, params=params, auth=auth, headers=headers, verify=url_ssl_verify) response.raise_for_status() result = response.json() if result["status"] != "success": error_msg = result.get('error', 'Unknown error') logger.error("Prometheus API returned error", endpoint=endpoint, error=error_msg, status=result["status"]) raise ValueError(f"Prometheus API error: {error_msg}") data_field = result.get("data", {}) if isinstance(data_field, dict): result_type = data_field.get("resultType") else: result_type = "list" logger.debug("Prometheus API request successful", endpoint=endpoint, result_type=result_type) return result["data"] except requests.exceptions.RequestException as e: logger.error("HTTP request to Prometheus failed", endpoint=endpoint, url=url, error=str(e), error_type=type(e).__name__) raise except json.JSONDecodeError as e: logger.error("Failed to parse Prometheus response as JSON", endpoint=endpoint, url=url, error=str(e)) raise ValueError(f"Invalid JSON response from Prometheus: {str(e)}") except Exception as e: logger.error("Unexpected error during Prometheus request", endpoint=endpoint, url=url, error=str(e), error_type=type(e).__name__) raise