Prometheus MCP Server

Instructions

Implementation Reference

• src/prometheus_mcp_server/server.py:239-289 (handler)

  The execute_query tool handler: decorated with @mcp.tool for registration, executes PromQL instant query via make_prometheus_request helper, formats results, adds Prometheus UI link if enabled.
  @mcp.tool( description="Execute a PromQL instant query against Prometheus", annotations={ "title": "Execute PromQL Query", "icon": "📊", "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True } ) async def execute_query(query: str, time: Optional[str] = None) -> Dict[str, Any]: """Execute an instant query against Prometheus. Args: query: PromQL query string time: Optional RFC3339 or Unix timestamp (default: current time) Returns: Query result with type (vector, matrix, scalar, string) and values """ params = {"query": query} if time: params["time"] = time logger.info("Executing instant query", query=query, time=time) data = make_prometheus_request("query", params=params) 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"} if time: ui_params["g0.moment_input"] = time 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" }] logger.info("Instant 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:154-209 (helper)

  Core helper function used by execute_query to make authenticated HTTP requests to the Prometheus /api/v1/query endpoint.
  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