Execute PromQL Query

Instructions

Implementation Reference

• src/prometheus_mcp_server/server.py:278-289 (registration)

  Registration of the execute_query tool with FastMCP decorator, including metadata annotations (readOnly, idempotent, etc.) and the prefixed tool name via _tool_name() helper.

  @mcp.tool(
      name=_tool_name("execute_query"),
      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:290-329 (handler)

  Handler function for execute_query. Accepts a PromQL query string and optional time parameter, calls make_prometheus_request to the /api/v1/query endpoint, extracts resultType and result, optionally appends a Prometheus UI link, and returns the result dictionary.

  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:187-249 (helper)

  Helper function that makes HTTP requests to the Prometheus API. Handles authentication (bearer token, basic auth), SSL verification, custom headers, OrgID, client TLS certs, request timeout, and error handling. Called by execute_query with endpoint='query' and the PromQL params.

  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)
  
      # Build client certificate tuple for mutual TLS authentication
      client_cert = None
      if config.client_cert:
          if config.client_key:
              client_cert = (config.client_cert, config.client_key)
          else:
              client_cert = config.client_cert
  
      try:
          logger.debug("Making Prometheus API request", endpoint=endpoint, url=url, params=params, headers=headers, timeout=config.request_timeout)
  
          # Make the request with appropriate headers, auth, and timeout (DDoS protection)
          response = requests.get(url, params=params, auth=auth, headers=headers, verify=url_ssl_verify, cert=client_cert, timeout=config.request_timeout)
  
          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

• src/prometheus_mcp_server/server.py:21-23 (helper)

  Helper function that adds an optional TOOL_PREFIX environment variable prefix to the tool name. Used by execute_query and all other tools to support namespacing.

  def _tool_name(name: str) -> str:
      """Build tool name with optional prefix."""
      return f"{TOOL_PREFIX}_{name}" if TOOL_PREFIX else name

• src/prometheus_mcp_server/server.py:159-177 (helper)

  Configuration initialization that feeds into execute_query's behavior, particularly the disable_prometheus_links setting which controls whether the Prometheus UI link is appended to results.

  config = PrometheusConfig(
      url=os.environ.get("PROMETHEUS_URL", ""),
      url_ssl_verify=os.environ.get("PROMETHEUS_URL_SSL_VERIFY", "True").lower() in ("true", "1", "yes"),
      disable_prometheus_links=os.environ.get("PROMETHEUS_DISABLE_LINKS", "False").lower() in ("true", "1", "yes"),
      username=os.environ.get("PROMETHEUS_USERNAME", ""),
      password=os.environ.get("PROMETHEUS_PASSWORD", ""),
      token=os.environ.get("PROMETHEUS_TOKEN", ""),
      org_id=os.environ.get("ORG_ID", ""),
      mcp_server_config=MCPServerConfig(
          mcp_server_transport=os.environ.get("PROMETHEUS_MCP_SERVER_TRANSPORT", "stdio").lower(),
          mcp_bind_host=os.environ.get("PROMETHEUS_MCP_BIND_HOST", "127.0.0.1"),
          mcp_bind_port=int(os.environ.get("PROMETHEUS_MCP_BIND_PORT", "8080")),
          stateless_http=os.environ.get("PROMETHEUS_MCP_STATELESS_HTTP", "False").lower() in ("true", "1", "yes"),
      ),
      client_cert=os.environ.get("PROMETHEUS_CLIENT_CERT", "") or None,
      client_key=os.environ.get("PROMETHEUS_CLIENT_KEY", "") or None,
      custom_headers=json.loads(os.environ.get("PROMETHEUS_CUSTOM_HEADERS")) if os.environ.get("PROMETHEUS_CUSTOM_HEADERS") else None,
      request_timeout=int(os.environ.get("PROMETHEUS_REQUEST_TIMEOUT", "30")),
  )