Prometheus MCP Server

Instructions

Implementation Reference

• src/prometheus_mcp_server/server.py:302-362 (handler)

  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
      }
  )

• src/prometheus_mcp_server/server.py:302-313 (schema)

  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
      """

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

  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