Build an MCP Server for Prometheus

Build an MCP Server for Prometheus

Give an agent read-only Kubernetes access and it gets good at a narrow thing: reading the current state of the cluster. It can tell you a pod is in CrashLoopBackOff, that the last exit code was 137, and that the events say the container was OOM killed.

What it cannot tell you is that memory had been climbing steadily for six hours, or that the error rate on the upstream service spiked twenty minutes before the first restart. Kubernetes objects describe now. The interesting part of almost every incident is the shape of the last hour.

That's a metrics question, and Prometheus already has the answer. It just isn't reachable from where the agent is standing.

This is the sibling build to the Kubernetes MCP server — same Python SDK, same deployment story, a different and considerably more dangerous data source. Kubernetes reads are cheap and bounded. A PromQL query is arbitrary compute against a process that has no isolation between queries, and an agent will happily write one that takes the whole thing down.

The client layer

Everything goes through the Prometheus HTTP API. No client library — you need a handful of its endpoints and you want direct control over the parameters.

pip install "mcp[cli]" httpx

1# server.py
2import os
3import re
4import time
5from typing import Any
6
7import httpx
8from mcp.server import MCPServer
9
10PROM_URL = os.environ["PROMETHEUS_URL"].rstrip("/")
11QUERY_TIMEOUT = os.getenv("PROM_QUERY_TIMEOUT", "10s")
12
13MAX_SERIES = 50            # series returned by any one tool call
14MAX_POINTS_PER_SERIES = 60 # points per series after downsampling
15MAX_RANGE_SECONDS = 24 * 3600
16
17mcp = MCPServer("prometheus")
18http = httpx.Client(base_url=PROM_URL, timeout=30.0)
19
20
21class PromError(RuntimeError):
22    """Raised for anything the model should see and can act on."""
23
24
25def _api(path: str, params: dict[str, Any] | None = None) -> Any:
26    r = http.get(path, params=params or {})
27    if r.status_code in (400, 422):
28        body = r.json()
29        # A bad query is normal input from a model. Hand back the parser
30        # error so it can fix the expression instead of giving up.
31        raise PromError(f"{body.get('errorType')}: {body.get('error')}")
32    r.raise_for_status()
33    body = r.json()
34    if body.get("status") != "success":
35        raise PromError(body.get("error", "unknown error"))
36    return body["data"]
37
38
39def _labels(metric: dict[str, str]) -> str:
40    name = metric.get("__name__", "")
41    rest = {k: v for k, v in metric.items() if k != "__name__"}
42    if not rest:
43        return name or "{}"
44    inner = ",".join(f'{k}="{v}"' for k, v in sorted(rest.items()))
45    return f"{name}{{{inner}}}"

Two decisions are already made here. timeout is sent on every query — it's capped by the server's -query.timeout flag, which defaults to 2m, and two minutes is far too long to let an agent's guess run. And a 400 becomes a PromError carrying the parser message rather than a stack trace, because a model that gets told parse error: unexpected character will usually fix its own query on the next turn.

Guarding the query before it runs

Prometheus has no per-query cost accounting you can consult before execution. The cheap disasters are easy to enumerate, so reject them by pattern:

1_BARE_SELECTOR = re.compile(r"\{\s*\}")
2_ANY_NAME = re.compile(r'__name__\s*=~\s*"\.[*+]"')
3_RANGE = re.compile(r"\[(\d+)([smhdwy])\]")
4_UNITS = {"s": 1, "m": 60, "h": 3600, "d": 86400, "w": 604800, "y": 31536000}
5
6
7def _reject_expensive(promql: str) -> None:
8    if _BARE_SELECTOR.search(promql) or _ANY_NAME.search(promql):
9        raise PromError(
10            "Refusing to run a selector that matches every series. "
11            'Name a metric, e.g. http_requests_total{job="api"}. '
12            "Use list_metrics to find one."
13        )
14    for amount, unit in _RANGE.findall(promql):
15        if int(amount) * _UNITS[unit] > MAX_RANGE_SECONDS:
16            raise PromError(
17                f"Range [{amount}{unit}] exceeds the 24h limit. "
18                "Query a shorter window, or use a recording rule."
19            )

Be clear about what this is. It is a footgun reducer, not a security boundary — regex over a query language is trivially evadable, and it is not trying to stop an adversary. The actual controls are on the Prometheus side and you should set them regardless: --query.timeout short, --query.max-samples (default 50,000,000) tuned to your instance, and --query.max-concurrency (default 20) low enough that a runaway agent can't starve your alerting rules of evaluation slots.

The strongest control is architectural: point the MCP server at a read replica, or at a Thanos Querier, and not at the Prometheus your alerts evaluate against. Then the worst case is an agent degrading its own view.

Instant queries

1@mcp.tool()
2def query(promql: str) -> str:
3    """Evaluate a PromQL expression at the current time, one value per series.
4
5    Use this for "what is X right now". To see how a value changed over a
6    window — which is what you want when correlating with an incident — use
7    query_range instead.
8    """
9    _reject_expensive(promql)
10    data = _api("/api/v1/query", {
11        "query": promql,
12        "timeout": QUERY_TIMEOUT,
13        "limit": MAX_SERIES,
14    })
15
16    kind = data["resultType"]
17    if kind in ("scalar", "string"):
18        return str(data["result"][1])
19    if not data["result"]:
20        return "Empty result. The metric may not exist, or no series match."
21
22    lines = [f"{_labels(s['metric'])} = {s['value'][1]}" for s in data["result"]]
23    if len(lines) == MAX_SERIES:
24        lines.append(f"[truncated at {MAX_SERIES} series — aggregate with sum by (...)]")
25    return "\n".join(lines)

limit truncates the returned series for vectors and matrices. Note what it does not do: the query still executes in full, so this bounds the context window, not Prometheus. That's why the truncation notice tells the model to aggregate — pushing sum by (pod) into the query is the only version that's cheaper for both sides.

Range queries, and the context window

This is where a naive implementation fails. up over one hour at a 15-second step is 240 points per series; across 200 targets that's 48,000 values, each a [timestamp, "value"] pair. You have blown the context window on a single tool call, and the model will read approximately none of it.

Downsample server-side by deriving the step from the window:

1@mcp.tool()
2def query_range(promql: str, minutes: int = 60) -> str:
3    """Evaluate a PromQL expression over the last N minutes and return a
4    downsampled series per result, plus min/max/first/last.
5
6    The step is chosen so each series returns at most 60 points, whatever the
7    window. A short spike between two steps will not appear: to hunt spikes,
8    wrap the expression in max_over_time(expr[5m]) so every sample is covered.
9    """
10    _reject_expensive(promql)
11    minutes = max(1, min(minutes, MAX_RANGE_SECONDS // 60))
12    end = time.time()
13    start = end - minutes * 60
14    step = max(15, int(minutes * 60 / MAX_POINTS_PER_SERIES))
15
16    data = _api("/api/v1/query_range", {
17        "query": promql,
18        "start": start,
19        "end": end,
20        "step": f"{step}s",
21        "timeout": QUERY_TIMEOUT,
22        "limit": MAX_SERIES,
23    })
24    if not data["result"]:
25        return "Empty result over that window."
26
27    out = [f"window={minutes}m step={step}s"]
28    for series in data["result"]:
29        values = [float(v[1]) for v in series["values"]]
30        if not values:
31            continue
32        out.append(
33            f"\n{_labels(series['metric'])}\n"
34            f"  first={values[0]:.4g} last={values[-1]:.4g} "
35            f"min={min(values):.4g} max={max(values):.4g}\n"
36            f"  series: {', '.join(f'{v:.4g}' for v in values)}"
37        )
38    return "\n".join(out)

The summary line is doing real work. "Climbed from 0.2 to 14" is the fact the model needs to correlate; the point list is there so it can see when the shape changed. Both fit comfortably in a few hundred tokens.

The docstring warning is not decorative. query_range evaluates the expression at each step timestamp, and an instant vector selector resolves to the most recent sample within lookback_delta (5 minutes by default). Widen the step to 60 seconds and a 20-second spike is genuinely gone from the result. Tell the model the workaround in the description, because it will not infer it.

Metric discovery, so the model stops guessing

A model asked for "the HTTP error rate" will confidently write http_requests_total{status=~"5.."}. In your cluster the metric might be http_server_requests_seconds_count, with the label outcome rather than status. The query parses, returns empty, and the agent concludes the service is healthy.

Empty results are the dangerous failure mode, because they look like an answer. Fix it by making discovery a first-class tool instead of hoping the model already knows your schema:

1@mcp.tool()
2def list_metrics(substring: str = "", limit: int = 40) -> str:
3    """Find metric names that actually exist here, optionally filtered by
4    substring. Call this before writing a query against an unfamiliar metric.
5    """
6    names = _api("/api/v1/label/__name__/values")
7    if substring:
8        names = [n for n in names if substring.lower() in n.lower()]
9    if not names:
10        return f"No metric names contain {substring!r}."
11
12    meta = _api("/api/v1/metadata", {"limit_per_metric": 1})
13    lines = []
14    for name in names[:limit]:
15        entry = (meta.get(name) or [{}])[0]
16        help_text = (entry.get("help") or "").split("\n")[0][:110]
17        lines.append(f"{name}  [{entry.get('type', 'unknown')}]  {help_text}")
18    if len(names) > limit:
19        lines.append(f"[{len(names) - limit} more — narrow the substring]")
20    return "\n".join(lines)
21
22
23@mcp.tool()
24def describe_metric(name: str) -> str:
25    """Show the label names on a metric and example values for each. Use this
26    to build a correct selector instead of guessing label names.
27    """
28    label_names = _api("/api/v1/labels", {"match[]": name})
29    if not label_names:
30        return f"No series found for {name}. Check the name with list_metrics."
31
32    lines = [f"{name} labels:"]
33    for label in [l for l in label_names if l != "__name__"][:10]:
34        values = _api(f"/api/v1/label/{label}/values", {"match[]": name})
35        shown = ", ".join(values[:8])
36        more = f" (+{len(values) - 8} more)" if len(values) > 8 else ""
37        lines.append(f"  {label}: {shown}{more}")
38    return "\n".join(lines)

describe_metric is the highest-value tool in the file. Two calls — list_metrics("http") then describe_metric("http_server_requests_seconds_count") — and the model is writing selectors against labels it has read rather than labels it remembers from someone else's codebase.

There's a cost: it doubles the round trips before any real query, which is tokens and latency on every investigation. Worth it. A wrong-but-empty query costs a whole investigation.

Alerts and targets

Two more tools, both trivial, both worth more than they look.

1@mcp.tool()
2def list_alerts(state: str = "") -> str:
3    """Currently firing and pending alerts with their labels and annotations.
4    Filter with state="firing" or state="pending".
5    """
6    alerts = _api("/api/v1/alerts")["alerts"]
7    if state:
8        alerts = [a for a in alerts if a["state"] == state]
9    if not alerts:
10        return "No matching alerts."
11
12    lines = []
13    for a in alerts[:MAX_SERIES]:
14        summary = a["annotations"].get("summary", "")
15        lines.append(
16            f"{a['labels'].get('alertname')}  [{a['state']}]  since={a['activeAt']}\n"
17            f"  labels: {_labels(a['labels'])}\n"
18            f"  {summary}"
19        )
20    return "\n\n".join(lines)
21
22
23@mcp.tool()
24def list_unhealthy_targets() -> str:
25    """Scrape targets that are down or erroring. If a metric is unexpectedly
26    missing, check here before concluding the service is fine.
27    """
28    targets = _api("/api/v1/targets", {"state": "active"})["activeTargets"]
29    broken = [t for t in targets if t["health"] != "up"]
30    if not broken:
31        return f"All {len(targets)} active targets are healthy."
32    return "\n".join(
33        f"{t['scrapePool']}  {t['scrapeUrl']}  health={t['health']}\n"
34        f"  lastScrape={t['lastScrape']}  error={t['lastError']}"
35        for t in broken[:MAX_SERIES]
36    )

list_alerts gives the agent the on-call view for free — your alert rules already encode which conditions matter, and reading them is cheaper than rediscovering the same conditions with ad-hoc queries. If you've invested in burn-rate alerting, that judgement is sitting right there in the annotations.

list_unhealthy_targets closes the empty-result loop from the other direction. A missing metric because the exporter is down looks identical to a missing metric because the name was wrong, and only one of those is the agent's fault.

Running it

mcp dev server.py                              # Inspector, for poking by hand
mcp run server.py --transport streamable-http  # HTTP, on 127.0.0.1:8000

Run the Inspector first and try the queries you'd expect an agent to write. Half the tuning here is discovering that your own step and series limits are wrong for your cardinality.

With the tools wired up, a realistic investigation looks like this. The agent reads the Kubernetes side and finds checkout-api restarting with exit code 137. It calls list_metrics("memory"), finds container_memory_working_set_bytes, calls describe_metric on it and learns the labels are pod and container rather than the pod_name it was about to guess. Then one query_range over 180 minutes returns twelve lines: working set climbing from 180M to 512M in a straight line, flat before that.

That's the answer — a leak, not a traffic spike — and it took four tool calls and maybe 3,000 tokens. The same conclusion from raw sample data would have been forty thousand numbers the model skims and misreads.

Deployment is the same shape as the Kubernetes server: a pod, its own ServiceAccount, a NetworkPolicy that lets it reach Prometheus and nothing else. That includes calling mcp.run(transport="streamable-http", host="0.0.0.0") yourself in the container, since the CLI's loopback default leaves the pod unreachable from its Service. Because it holds no cluster credentials, the identity story is simpler — but if your Prometheus sits behind an auth proxy, the multi-tenancy problem is identical. One server, one identity, every caller sees everything that identity sees.

What this does not solve

It does not do the correlation. The server hands over numbers. Noticing that the memory curve inflects at the same moment as the deploy is the model's job, and it is mediocre at it without being told what to compare. Pairing this with the Kubernetes troubleshooting agent is what makes it useful — one tool set for what changed, one for what it did to the system.

Downsampling is lossy, and the model won't remember that. The docstring says so; a model deep in a long investigation will still read max=0.4 from a 60-second step and conclude nothing spiked. If spike detection matters, expose a separate tool that hardcodes max_over_time rather than relying on the model to wrap its own expression.

PromQL generation stays unreliable at the edges. Discovery fixes metric and label names. It does not fix rate versus irate, counter resets, or the classic mistake of averaging a histogram quantile across pods. The queries will parse and return plausible numbers that are wrong.

Metrics are one signal. No logs, no traces, no profiles. A metric tells you the error rate tripled; the exception is in the logs. And once your agent is calling three MCP servers, you want tool calls traced end to end or debugging the agent becomes harder than debugging the incident.

Cardinality is still your problem. This server reads whatever Prometheus has. If your metrics carry per-request IDs, discovery gets slow and expensive and no client-side limit helps — that's an instrumentation fix, and it starts at the Prometheus setup itself.

Where to start

Build list_metrics, describe_metric, and query first, and use them by hand in the Inspector for a day against a real incident you've already resolved. You'll find out quickly whether the model can navigate your metric names, which is the thing that decides if the rest is worth building.

Add query_range once discovery works, and tune MAX_POINTS_PER_SERIES against your own context budget rather than the 60 here. Then set the Prometheus-side flags, because the first time an agent writes a subquery over a week of data you will want them already in place.