Prometheus MCP Server

Instructions

Implementation Reference

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

  The execute_query tool handler: an async function decorated with @mcp.tool that executes PromQL instant queries against Prometheus using the make_prometheus_request helper. Handles query and optional time parameters, formats results, adds optional Prometheus UI links, and includes logging.

  @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:239-248 (registration)

  Registration of the execute_query tool via FastMCP's @mcp.tool decorator, including description and annotations for MCP protocol compliance.

  @mcp.tool(
      description="Execute a PromQL instant query against Prometheus",
      annotations={
          "title": "Execute PromQL Query",
          "icon": "📊",
          "readOnlyHint": True,
          "destructiveHint": False,
          "idempotentHint": True,
          "openWorldHint": True
      }

• src/prometheus_mcp_server/server.py:154-209 (helper)

  Helper function make_prometheus_request that performs authenticated HTTP requests to the Prometheus API /query endpoint, handles errors, SSL, auth, custom headers, and returns the data field. Directly called by execute_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