prometheus_query_range

Instructions

Implementation Reference

• src/prometheus_mcp/tools.py:292-301 (registration)

  Registration of the 'prometheus_query_range' tool with FastMCP using @mcp.tool decorator, including annotations (title, readOnlyHint, idempotentHint) and structured_output=True.

  @mcp.tool(
      name="prometheus_query_range",
      annotations={
          "title": "Range Query",
          "readOnlyHint": True,
          "destructiveHint": False,
          "idempotentHint": True,
          "openWorldHint": True,
      },
      structured_output=True,

• src/prometheus_mcp/tools.py:303-444 (handler)

  Handler function 'prometheus_query_range' that executes a PromQL range query via GET /api/v1/query_range, processes the result into structured format with type QueryRangeOutput, applies a 5000-point cap with downsampling, and returns both structured data and markdown rendering.

  def prometheus_query_range(
      query: Annotated[
          str,
          Field(
              min_length=1,
              max_length=2000,
              description=(
                  "PromQL expression to evaluate over a time range. "
                  "Examples: 'rate(http_requests_total[5m])', "
                  "'node_cpu_seconds_total{mode=\"idle\"}'."
              ),
          ),
      ],
      start: Annotated[
          str,
          Field(
              min_length=1,
              max_length=50,
              description=(
                  "Start of range. RFC3339 (e.g. '2024-01-15T10:00:00Z') or Unix timestamp (e.g. '1705312800')."
              ),
          ),
      ],
      end: Annotated[
          str,
          Field(
              min_length=1,
              max_length=50,
              description=("End of range. RFC3339 (e.g. '2024-01-15T11:00:00Z') or Unix timestamp (e.g. '1705316400')."),
          ),
      ],
      step: Annotated[
          str,
          Field(
              min_length=1,
              max_length=20,
              description=(
                  "Query resolution step. Duration string (e.g. '15s', '1m', '5m') "
                  "or float seconds (e.g. '30'). "
                  "Prometheus rejects steps that produce more than 11,000 data points per series."
              ),
          ),
      ],
  ) -> QueryRangeOutput:
      """Execute a PromQL range query returning time-series data points.
  
      Wraps ``GET /api/v1/query_range``. Returns one series per matching time
      series, each with labels and a list of ``[timestamp, value]`` pairs.
      Total points across all series are capped at 5000 with a truncation hint.
  
      Prometheus may reject the query with HTTP 422 (bad_data) if the step
      produces too many data points (> 11,000 per series). Increase the step
      or narrow the time range if this happens.
  
      Note: The Prometheus API does not support filtering by branch or commit
      in this endpoint — filters are expressed purely in PromQL label matchers.
  
      Examples:
          - Use when: "Show me CPU usage over the last hour with 1-minute resolution"
            → ``query='rate(node_cpu_seconds_total[5m])'``, ``step='1m'``.
          - Use when: "Graph HTTP error rate for the last 24 hours"
            → ``query='rate(http_requests_total{status=~"5.."}[5m])'``,
            ``start='2024-01-15T00:00:00Z'``, ``end='2024-01-16T00:00:00Z'``,
            ``step='5m'``.
          - Use when: Investigating a past incident — pick the time window of the
            incident and use a fine step.
          - Don't use when: You only want the current value
            (call ``prometheus_query`` — faster and simpler).
          - Don't use when: You want alert history (call ``prometheus_list_alerts``).
  
      Returns:
          dict with ``query`` / ``start`` / ``end`` / ``step`` / ``result_type`` /
          ``series_count`` / ``total_points`` / ``truncated`` /
          ``data`` (list of series with ``labels``, ``point_count``, ``values``).
      """
      try:
          client = get_client()
          params: dict[str, Any] = {
              "query": query,
              "start": start,
              "end": end,
              "step": step,
          }
  
          raw = client.get("/query_range", params=params) or {}
          result_data = raw.get("data") or {}
          result_type: str = result_data.get("resultType", "matrix")
          raw_result: list[dict[str, Any]] = result_data.get("result") or []
  
          series: list[RangeSeries] = [_shape_range_series(item) for item in raw_result]
  
          # Count total points and enforce cap
          total_points = sum(s["point_count"] for s in series)
          truncated = total_points > _RANGE_POINTS_CAP
  
          if truncated:
              # Downsample: keep only the first _RANGE_POINTS_CAP points across series in order
              kept: list[RangeSeries] = []
              remaining = _RANGE_POINTS_CAP
              for s in series:
                  if remaining <= 0:
                      break
                  take = min(s["point_count"], remaining)
                  kept.append(
                      {
                          "labels": s["labels"],
                          "point_count": take,
                          "values": s["values"][:take],
                      }
                  )
                  remaining -= take
              series = kept
  
          result: QueryRangeOutput = {
              "query": query,
              "start": start,
              "end": end,
              "step": step,
              "result_type": result_type,
              "series_count": len(series),
              "total_points": min(total_points, _RANGE_POINTS_CAP) if truncated else total_points,
              "truncated": truncated,
              "data": series,
          }
  
          md = f"## Range Query: `{query}`\n\n"
          md += f"**Period:** {start} → {end} (step: {step})\n"
          md += f"**Series:** {len(series)} | **Points:** {result['total_points']}"
          if truncated:
              md += f" (capped at {_RANGE_POINTS_CAP})"
          md += "\n\n"
          md_series = series[:_MD_ITEM_LIMIT]
          for s in md_series:
              label_str = ", ".join(f'{k}="{v}"' for k, v in s["labels"].items()) if s["labels"] else "(no labels)"
              first_val = s["values"][0][1] if s["values"] else "—"
              last_val = s["values"][-1][1] if s["values"] else "—"
              md += f"- `{label_str}` — {s['point_count']} points, first={first_val}, last={last_val}\n"
          if len(series) > _MD_ITEM_LIMIT:
              md += _truncation_hint(len(series), _MD_ITEM_LIMIT, "series")
          return output.ok(result, md)  # type: ignore[return-value]
      except Exception as exc:
          output.fail(exc, f"executing range query {query!r}")

• src/prometheus_mcp/models.py:56-71 (schema)

  Schema definitions for RangeSeries and QueryRangeOutput TypedDicts used as the output type for the prometheus_query_range tool.

  class RangeSeries(TypedDict):
      labels: dict[str, str]
      point_count: int
      values: list[list[float | str]]
  
  
  class QueryRangeOutput(TypedDict):
      query: str
      start: str
      end: str
      step: str
      result_type: str
      series_count: int
      total_points: int
      truncated: bool
      data: list[RangeSeries]

• src/prometheus_mcp/tools.py:82-97 (helper)

  Helper function '_shape_range_series' that converts a Prometheus range vector result item into the RangeSeries TypedDict format used by the range query handler.

  def _shape_range_series(item: dict[str, Any]) -> RangeSeries:
      """Convert a Prometheus range vector result item into :class:`RangeSeries`."""
      metric = item.get("metric") or {}
      values = item.get("values") or []
      # Each value: [timestamp (float), value (str)]
      shaped: list[list[float | str]] = []
      for v in values:
          if isinstance(v, list) and len(v) == 2:
              shaped.append([float(v[0]), str(v[1])])
          else:
              shaped.append([0.0, str(v)])
      return {
          "labels": {k: str(mv) for k, mv in metric.items()},
          "point_count": len(shaped),
          "values": shaped,
      }

• src/prometheus_mcp/_mcp.py:39-55 (helper)

  Shared FastMCP instance and get_client() helper that provides the HTTP client used by the prometheus_query_range handler to make API requests.

  mcp = FastMCP("prometheus_mcp", lifespan=app_lifespan)
  
  
  def get_client() -> PrometheusClient:
      """Return a cached :class:`PrometheusClient` (thread-safe lazy-init).
  
      FastMCP runs synchronous tools in worker threads via
      ``anyio.to_thread.run_sync``; concurrent first-calls could otherwise
      race on the ``_client`` global. The lock ensures exactly one instance
      is constructed.
      """
      global _client
      if _client is None:
          with _client_lock:
              if _client is None:  # double-checked locking
                  _client = PrometheusClient()
      return _client