feat(mcp): Traefik routers, certificates and traffic (#5838)

list_routers grades every router as the Traefik page does (the last 15
minutes, 5xx on its cutoff, 4xx and p95 against the router's normal) and
gives a troubled one its status codes, what they mean and the swarm
service behind it. get_router adds its history, codes, normal and
anomalies; list_certificates the served certificates and the routers they
cover; get_traffic the access log's clients, countries and routers.
get_overview names a router answering 5xx and a certificate near expiry.

The access-log aggregation moves into routes.summarize_access, shared by
the traffic page and get_traffic.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

docs/core/mcp.md

@@ -80,6 +80,7 @@ Each role includes the tools of the roles above it in this list.
 | Audit | `list_audit_events` |
 | Docker plugin | `list_containers`, `get_container`, `list_swarm_services`, `search_container_logs`, `get_docker_disk_usage` |
 | Host agent (built in, always on) | `list_agents`, `list_maintenance_plans` (each plan's group, weekly slot, next run and latest run: how far it got, who it skipped and why), `get_maintenance_job` (a host's update jobs or one job: state, packages updated, whether it rebooted, why it failed) |
+| Traefik plugin | `list_routers` (each router graded as the Traefik page grades it, with the status codes behind a problem, what they mean and the swarm service behind it), `get_router` (one router's history, codes, normal and anomalies), `list_certificates` (served certificates, days left, the routers they cover), `get_traffic` (the access log's clients, countries and routers, and Traefik's own figures) |
 
 ### Operator: run playbooks, change monitoring
 

plugins/traefik/__init__.py

@@ -62,3 +62,13 @@ def get_alert_metrics() -> dict[str, list[str]]:
     return {"traefik": ["request_rate", "error_rate_4xx_pct", "error_rate_5xx_pct",
                         "latency_p50_ms", "latency_p95_ms", "latency_p99_ms",
                         "response_bytes_rate", "cert_expiry_days"]}
+
+
+def register_mcp_tools(register) -> None:
+    from .mcp_tools import register_tools
+    register_tools(register)
+
+
+async def mcp_overview(db) -> list[dict]:
+    from .mcp_tools import mcp_overview as findings
+    return await findings(db)

plugins/traefik/mcp_tools.py

@@ -0,0 +1,304 @@
+"""The traefik plugin's MCP tools and its get_overview findings (#5838).
+
+Registered through register_mcp_tools() in __init__.py, so they exist only
+while the plugin is enabled. Routers are graded exactly as the Traefik page
+grades them (routes.py): over the last 15 minutes, the 5xx share against its
+fixed cutoff, the 4xx share and p95 against the router's own normal where it
+has one.
+"""
+from __future__ import annotations
+
+from datetime import timedelta
+
+from sqlalchemy import select
+
+from steward.core import anomalies as anom
+from steward.core.norms import norms_at
+from steward.mcp._context import mcp_app
+from steward.mcp.tools._shape import (
+    age_s, bounded, clamp_limit, db_session, iso, lookback, now_utc, rnd, summarize,
+)
+from steward.mcp.tools.overview import finding
+from steward.models.users import UserRole
+
+from . import routes as page
+from .models import TraefikAccessSummary, TraefikCert, TraefikGlobalStat
+from .scheduler import error_page_services, router_domains
+from .scraper import cert_names, covers
+
+STATES = ("crit", "warn", "ok")
+
+
+def _cfg():
+    app = mcp_app()
+    return app.config.get("THRESHOLDS", {}), int(app.config["PLUGINS"]["traefik"]["scrape_interval_seconds"])
+
+
+async def _graded(db, now, thresholds: dict, scrape: int) -> tuple[list[dict], list[dict]]:
+    """(current, gone) routers, graded and explained as the page does them."""
+    newest = await page.latest_per_router(db)
+    current, gone = page.gone_routers([{"name": m.router_name, "latest": m} for m in newest], scrape)
+    names = [r["name"] for r in current]
+    page.grade_routers(current, await page.recent_window(db, names, now), thresholds,
+                       await page.router_norms(db, names, now), error_page_services())
+    troubled = [r for r in current if r["state"] != "ok"]
+    codes = await page.status_codes(db, [r["name"] for r in troubled], now - page.STATE_WINDOW)
+    for r in current:
+        r["codes"] = codes.get(r["name"], [])
+    if troubled:
+        page.explain_problems(troubled, await page.swarm_service_names(db))
+    return current, gone
+
+
+def _cert_view(c, now, routers: list[str] | None) -> dict:
+    return {"cn": c.cn or c.serial, "serial": c.serial,
+            "names": sorted(cert_names(c.cn, c.sans)),
+            "not_after": iso(c.not_after),
+            "days_left": rnd((c.not_after - now).total_seconds() / 86400.0, 1),
+            "routers": routers}
+
+
+def _router_row(r: dict, certs: dict) -> dict:
+    latest = r["latest"]
+    row = {
+        "router": r["name"], "state": r["state"], "why": r["why"],
+        "requests_per_s": rnd(latest.request_rate, 3),
+        "error_4xx_pct": rnd(latest.error_rate_4xx_pct, 2),
+        "error_5xx_pct": rnd(latest.error_rate_5xx_pct, 2),
+        "p50_ms": rnd(latest.latency_p50_ms, 1), "p95_ms": rnd(latest.latency_p95_ms, 1),
+        "p99_ms": rnd(latest.latency_p99_ms, 1),
+        "graded_on_normal": r.get("normed", False),
+        "last_scraped_at": iso(latest.scraped_at),
+    }
+    if r["state"] != "ok":
+        row["codes"] = [[c, n] for c, n in r.get("codes", [])]
+        row["meaning"] = r.get("meaning")
+        row["swarm_service"] = r.get("service")
+    if r["name"] in certs:
+        row["certificate"] = {"cn": certs[r["name"]]["cn"], "days_left": rnd(certs[r["name"]]["days"], 1)}
+    return row
+
+
+async def list_routers(state: str = "", limit: int = 100) -> dict:
+    """Traefik's routers, graded as the Traefik page grades them: problems
+    first ("crit" before "warn"), then healthy ones, each by name.
+
+    state: "crit" (5xx share past its cutoff: failing), "warn" (p95 or 4xx
+      share past its cutoff and outside the router's usual for this hour),
+      "ok"; empty for all.
+    limit: default 100, max 500.
+
+    The grade is over the last 15 minutes, weighted by traffic; the figures
+    shown are the latest scrape. A troubled router carries `codes` (the 4xx
+    and 5xx codes it answered in that window, most first, e.g. 404 means
+    scanners, 502/503 the app unreachable), `meaning` (what the codes point
+    to, whether it is usually a fault, and `action`: "act" on the app or
+    "tune" the cutoff), and `swarm_service` (the service behind it, for
+    list_swarm_services and search_container_logs). `graded_on_normal` says
+    the 4xx/p95 were judged against the router's own normal. `certificate`
+    is the soonest-expiring served certificate for its hosts. `gone` lists
+    routers Traefik stopped reporting.
+    """
+    if state and state not in STATES:
+        raise ValueError(f"state must be one of {', '.join(STATES)}, or empty for all.")
+    thresholds, scrape = _cfg()
+    now = now_utc()
+    async with db_session() as db:
+        current, gone = await _graded(db, now, thresholds, scrape)
+        served, _old = page.current_certs(
+            (await db.execute(select(TraefikCert))).scalars().all(), scrape)
+    domains = router_domains() or {}
+    certs = page.certs_by_router([r["name"] for r in current], domains, served, now)
+    problems, healthy = page.split_routers(current)
+    rows = [_router_row(r, certs) for r in problems + healthy if not state or r["state"] == state]
+    error_5xx, error_4xx = page.error_cutoffs(thresholds)
+    out = bounded(rows, clamp_limit(limit, 100, 500), key="routers",
+                  more="Pass a larger `limit`, or a `state`.")
+    out["cutoffs"] = {"error_5xx_pct": error_5xx, "error_4xx_pct": error_4xx,
+                      "min_requests_15m": page.min_requests(thresholds)}
+    out["gone"] = [{"router": r["name"], "last_scraped_at": iso(r["latest"].scraped_at)} for r in gone[:20]]
+    return out
+
+
+def _match(want: str, names: list[str]) -> str:
+    want = (want or "").strip()
+    for test in (lambda n: n == want, lambda n: n.lower() == want.lower(),
+                 lambda n: n.split("@")[0].lower() == want.lower()):
+        found = [n for n in names if test(n)]
+        if len(found) == 1:
+            return found[0]
+        if len(found) > 1:
+            raise ValueError(f"{want!r} matches {', '.join(sorted(found))}; name one with its @provider.")
+    raise ValueError(f"No router {want!r}; list_routers shows them.")
+
+
+async def get_router(router: str, range: str = "24h") -> dict:
+    """One router: its grade now, its history over `range`, the status codes
+    it answered, what is normal for it at this hour, and its anomalies.
+
+    router: its name, with or without the @provider ("radarr@swarm",
+      "radarr").
+    range: "1h", "6h", "24h" (default), "7d" or "30d".
+
+    `series` summarizes requests/s, p95 ms and the 4xx and 5xx shares over the
+    range (min/avg/max/last, ≤60 points each). `codes` are the 4xx and 5xx
+    codes answered over the range, most first. `normal` is its learned usual
+    for this hour (4xx share, p95), null until it has enough history.
+    `anomalies` are its metric anomalies of the last 30 days (get_anomaly
+    explains one).
+    """
+    since, key = lookback(range)
+    thresholds, scrape = _cfg()
+    now = now_utc()
+    async with db_session() as db:
+        current, gone = await _graded(db, now, thresholds, scrape)
+        everyone = {r["name"]: r for r in current + gone}
+        name = _match(router, list(everyone))
+        r = everyone[name]
+        history = (await page.router_history(db, [name], since, max(60, int((now - since).total_seconds() // 60))))[name]
+        codes = (await page.status_codes(db, [name], since)).get(name, [])
+        norms = await norms_at(db, [("traefik", name, m) for m in page.NORMED], now)
+        found = await anom.list_anomalies(db, source="traefik", resource=name,
+                                          since=now - timedelta(days=30), limit=10)
+    if name not in {c["name"] for c in current}:
+        r = {**r, "state": "gone", "why": "Traefik no longer reports this router"}
+
+    def series(attr):
+        return summarize([(p.bucket, float(getattr(p, attr))) for p in history if getattr(p, attr) is not None])
+
+    out = {
+        "router": name, "state": r["state"], "why": r["why"],
+        "meaning": r.get("meaning"), "swarm_service": r.get("service"),
+        "range": key,
+        "series": {"requests_per_s": series("request_rate"), "p95_ms": series("latency_p95_ms"),
+                   "error_4xx_pct": series("error_rate_4xx_pct"),
+                   "error_5xx_pct": series("error_rate_5xx_pct")},
+        "codes": [[c, n] for c, n in codes],
+        "normal": {m: {"usual_hold": rnd(n.usual_hold, 3), "usual_peak": rnd(n.usual_peak, 3),
+                       "kind": n.kind, "days_seen": n.days_seen}
+                   for (_s, _r, m), n in norms.items()} or None,
+        "anomalies": [{"anomaly_id": a.id, "metric": a.metric_name, "review": a.review,
+                       "started_at": iso(a.started_at), "ongoing": a.ended_at is None,
+                       "peak_level": rnd(a.peak_level, 3), "usual_hold": rnd(a.usual_hold, 3)}
+                      for a in found],
+    }
+    if r["state"] == "ok" and r.get("why"):
+        out["note"] = r["why"]
+    return out
+
+
+async def list_certificates(include_old: bool = False) -> dict:
+    """The TLS certificates Traefik serves, soonest expiry first: names
+    covered, expiry, days left, and the routers whose hosts each covers
+    (null when Traefik's API couldn't say which hosts a router serves).
+
+    include_old: also list certificates Traefik stopped serving (renewed or
+      retired), which Steward keeps on record.
+    `warn_days` / `crit_days` are the Settings → Thresholds cutoffs
+    get_overview reports a served certificate at.
+    """
+    thresholds, scrape = _cfg()
+    now = now_utc()
+    async with db_session() as db:
+        certs = (await db.execute(select(TraefikCert).order_by(TraefikCert.not_after))).scalars().all()
+        newest = await page.latest_per_router(db)
+    served, old = page.current_certs(certs, scrape)
+    domains = router_domains()
+    names = [m.router_name for m in newest]
+
+    def routers_for(c):
+        if domains is None:
+            return None
+        covered = cert_names(c.cn, c.sans)
+        return sorted({n for n in names for h in page.router_hosts(n, domains) if covers(covered, h)})
+
+    tls = thresholds.get("tls_days", {})
+    out = {"certificates": [_cert_view(c, now, routers_for(c)) for c in served],
+           "warn_days": tls.get("warn", 14), "crit_days": tls.get("crit", 3)}
+    if include_old:
+        out["old"] = [_cert_view(c, now, None) | {"last_seen_at": iso(c.scraped_at)} for c in old[:50]]
+    return out
+
+
+async def get_traffic(range: str = "24h") -> dict:
+    """Who reached Traefik over `range`, from its access log: requests in
+    total, internal (private networks) and external, bytes, and the top
+    client IPs (with country when the GeoIP database is set), countries and
+    routers. Also Traefik's own process figures (open connections, memory,
+    when its configuration last loaded).
+
+    range: "1h", "6h", "24h" (default), "7d" or "30d".
+    `access_log: false` means the access log isn't being read (Settings →
+    Plugins → Traefik), so there are no traffic figures.
+    """
+    since, key = lookback(range)
+    now = now_utc()
+    async with db_session() as db:
+        rows = (await db.execute(select(TraefikAccessSummary)
+                                 .where(TraefikAccessSummary.period_end >= since)
+                                 .order_by(TraefikAccessSummary.period_end))).scalars().all()
+        stat = (await db.execute(select(TraefikGlobalStat)
+                                 .order_by(TraefikGlobalStat.scraped_at.desc()).limit(1))).scalar_one_or_none()
+    cfg = mcp_app().config["PLUGINS"]["traefik"].get("access_log", {})
+    out = {"range": key, "access_log": bool(isinstance(cfg, dict) and cfg.get("enabled")),
+           "traffic": None, "traefik": None}
+    if rows:
+        summary = page.summarize_access(rows)
+        summary["external_pct"] = rnd(summary["external_pct"], 1)
+        summary["from"], summary["to"] = iso(rows[0].period_start), iso(rows[-1].period_end)
+        out["traffic"] = summary
+    if stat is not None:
+        out["traefik"] = {"open_connections": stat.open_conns_total, "memory_bytes": stat.process_memory_bytes,
+                          "config_reloads": stat.config_reloads_total,
+                          "config_loaded_at": iso(stat.config_last_reload_at),
+                          "config_loaded_age_s": age_s(stat.config_last_reload_at, now),
+                          "scraped_at": iso(stat.scraped_at)}
+    return out
+
+
+# ── get_overview ─────────────────────────────────────────────────────────────
+
+def cert_findings(certs, now, thresholds: dict) -> list[dict]:
+    """A served certificate within its warning days (critical within its
+    critical days, or expired)."""
+    tls = thresholds.get("tls_days", {})
+    warn, crit = float(tls.get("warn", 14)), float(tls.get("crit", 3))
+    out = []
+    for c in certs:
+        days = (c.not_after - now).total_seconds() / 86400.0
+        if days > warn:
+            continue
+        name = c.cn or c.serial
+        left = "expired" if days < 0 else f"expires in {days:.1f} days"
+        out.append(finding("critical" if days <= crit else "warning", "traefik_cert_expiring", name,
+                           f"TLS certificate {name} served by Traefik {left} ({iso(c.not_after)})",
+                           ids={"serial": c.serial}, next="list_certificates()"))
+    return out
+
+
+def router_findings(routers: list[dict]) -> list[dict]:
+    """A router answering 5xx past its cutoff: failing. 4xx and latency are
+    left to the metric anomalies, which judge them against the router's
+    normal (System 182)."""
+    out = []
+    for r in routers:
+        if r["state"] != "crit":
+            continue
+        said = (r.get("meaning") or {}).get("meaning")
+        out.append(finding("critical", "traefik_router_failing", r["name"],
+                           f"Traefik router {r['name']}: {r['why']}" + (f". {said}" if said else ""),
+                           ids={"router": r["name"]}, next=f'get_router(router="{r["name"]}")'))
+    return out
+
+
+async def mcp_overview(db) -> list[dict]:
+    thresholds, scrape = _cfg()
+    now = now_utc()
+    current, _gone = await _graded(db, now, thresholds, scrape)
+    served, _old = page.current_certs((await db.execute(select(TraefikCert))).scalars().all(), scrape)
+    return router_findings(current) + cert_findings(served, now, thresholds)
+
+
+def register_tools(register) -> None:
+    for fn in (list_routers, get_router, list_certificates, get_traffic):
+        register(fn, role=UserRole.viewer, read_only=True)

plugins/traefik/routes.py

@@ -709,6 +709,42 @@ async def widget_request_chart():
     )
 
 
+def summarize_access(rows, top: int = 10) -> dict:
+    """The access-log summaries in a range added up: totals, the internal and
+    external split, and the top client IPs, countries and routers."""
+    total = sum(r.total_requests for r in rows)
+    external = sum(r.external_requests for r in rows)
+    ip_counts: dict = defaultdict(lambda: {"count": 0, "internal": True, "country": None})
+    country_counts: dict[str, int] = defaultdict(int)
+    router_counts: dict[str, int] = defaultdict(int)
+    for row in rows:
+        for entry in json.loads(row.top_ips_json):
+            k = entry["ip"]
+            ip_counts[k]["count"] += entry["count"]
+            ip_counts[k]["internal"] = entry["internal"]
+            ip_counts[k]["country"] = entry.get("country")
+        for entry in json.loads(row.top_countries_json):
+            country_counts[entry["country"]] += entry["count"]
+        for entry in json.loads(row.top_routers_json):
+            router_counts[entry["router"]] += entry["count"]
+
+    def ranked(counts: dict, label: str) -> list[dict]:
+        return sorted(({label: k, "count": v} for k, v in counts.items()),
+                      key=lambda x: x["count"], reverse=True)[:top]
+
+    return {
+        "total": total,
+        "internal": sum(r.internal_requests for r in rows),
+        "external": external,
+        "total_bytes": sum(r.total_bytes for r in rows),
+        "external_pct": (external / total * 100.0) if total else 0.0,
+        "top_ips": sorted(({"ip": k, **v} for k, v in ip_counts.items()),
+                          key=lambda x: x["count"], reverse=True)[:top],
+        "top_countries": ranked(country_counts, "country"),
+        "top_routers": ranked(router_counts, "router"),
+    }
+
+
 @traefik_bp.get("/traffic")
 @require_role(UserRole.viewer)
 async def traffic():
@@ -732,38 +768,7 @@ async def traffic():
             access_log_enabled=bool(al.get("enabled", False)), access_log_path=al.get("path", ""),
         )
 
-    total = sum(r.total_requests for r in rows)
+    summary = summarize_access(rows)
-    internal = sum(r.internal_requests for r in rows)
-    external = sum(r.external_requests for r in rows)
-    total_bytes = sum(r.total_bytes for r in rows)
-
-    ip_counts: dict = defaultdict(lambda: {"count": 0, "internal": True, "country": None})
-    country_counts: dict[str, int] = defaultdict(int)
-    router_counts: dict[str, int] = defaultdict(int)
-
-    for row in rows:
-        for entry in json.loads(row.top_ips_json):
-            k = entry["ip"]
-            ip_counts[k]["count"] += entry["count"]
-            ip_counts[k]["internal"] = entry["internal"]
-            ip_counts[k]["country"] = entry.get("country")
-        for entry in json.loads(row.top_countries_json):
-            country_counts[entry["country"]] += entry["count"]
-        for entry in json.loads(row.top_routers_json):
-            router_counts[entry["router"]] += entry["count"]
-
-    top_ips = sorted(
-        [{"ip": k, **v} for k, v in ip_counts.items()],
-        key=lambda x: x["count"], reverse=True
-    )[:10]
-    top_countries = sorted(
-        [{"country": k, "count": v} for k, v in country_counts.items()],
-        key=lambda x: x["count"], reverse=True
-    )[:10]
-    top_routers = sorted(
-        [{"router": k, "count": v} for k, v in router_counts.items()],
-        key=lambda x: x["count"], reverse=True
-    )[:10]
 
     history = [
         {
@@ -774,19 +779,7 @@ async def traffic():
         for r in rows
     ]
 
-    summary = {
+    summary["sparkline_external"] = _sparkline([h["external_pct"] for h in subsample(history, 80)])
-        "total": total,
-        "internal": internal,
-        "external": external,
-        "total_bytes": total_bytes,
-        "external_pct": (external / total * 100.0) if total else 0.0,
-        "top_ips": top_ips,
-        "top_countries": top_countries,
-        "top_routers": top_routers,
-        "sparkline_external": _sparkline(
-            [h["external_pct"] for h in subsample(history, 80)]
-        ),
-    }
 
     return await render_template(
         "traefik/traffic.html", summary=summary, history=history, range_key=range_key

tests/plugins/traefik/test_mcp_tools.py

@@ -0,0 +1,74 @@
+"""The traefik plugin's MCP tools (#5838): which certificates and routers the
+overview names, how a router is found by name, and what a row says."""
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace as NS
+
+import pytest
+
+import plugins.traefik as traefik_plugin
+from plugins.traefik import mcp_tools as t
+from steward.models.users import UserRole
+
+NOW = datetime(2026, 10, 15, 12, 0, tzinfo=timezone.utc)
+THRESHOLDS = {"tls_days": {"warn": 14, "crit": 3}}
+
+
+def _cert(cn, days):
+    return NS(cn=cn, serial=f"s-{cn}", sans=None, not_after=NOW + timedelta(days=days), scraped_at=NOW)
+
+
+def test_a_served_certificate_is_a_finding_inside_its_warning_days():
+    found = {f["subject"]: f for f in t.cert_findings(
+        [_cert("fine.example", 40), _cert("soon.example", 10), _cert("now.example", 2),
+         _cert("gone.example", -1)], NOW, THRESHOLDS)}
+    assert set(found) == {"soon.example", "now.example", "gone.example"}
+    assert found["soon.example"]["severity"] == "warning"
+    assert found["now.example"]["severity"] == "critical"
+    assert "expired" in found["gone.example"]["summary"]
+
+
+def test_only_a_failing_router_is_an_overview_finding():
+    routers = [
+        {"name": "app@swarm", "state": "crit", "why": "12.0% of responses are 5xx over the last 15 min",
+         "meaning": {"meaning": "Mostly 502: Traefik couldn't reach the app."}},
+        {"name": "radarr@swarm", "state": "warn", "why": "7.2% of responses are 4xx"},
+        {"name": "ok@swarm", "state": "ok", "why": None},
+    ]
+    [f] = t.router_findings(routers)
+    assert f["severity"] == "critical" and f["subject"] == "app@swarm"
+    assert "Mostly 502" in f["summary"] and f["next"] == 'get_router(router="app@swarm")'
+
+
+def test_a_router_is_found_with_or_without_its_provider():
+    names = ["radarr@swarm", "sonarr@swarm", "api@internal", "web@docker", "web@file"]
+    assert t._match("radarr@swarm", names) == "radarr@swarm"
+    assert t._match("RADARR", names) == "radarr@swarm"
+    with pytest.raises(ValueError, match="matches web@docker, web@file"):
+        t._match("web", names)
+    with pytest.raises(ValueError, match="list_routers shows them"):
+        t._match("nope", names)
+
+
+def test_a_row_names_codes_and_meaning_only_when_troubled():
+    latest = NS(request_rate=2.5, error_rate_4xx_pct=7.25, error_rate_5xx_pct=0.0, latency_p50_ms=12.0,
+                latency_p95_ms=40.0, latency_p99_ms=80.0, scraped_at=NOW)
+    troubled = t._router_row({"name": "radarr@swarm", "state": "warn", "why": "4xx", "latest": latest,
+                              "codes": [("404", 30)], "meaning": {"action": "tune"}, "service": "arr_radarr",
+                              "normed": True}, {"radarr@swarm": {"cn": "radarr.example", "days": 50.0}})
+    assert troubled["codes"] == [["404", 30]] and troubled["swarm_service"] == "arr_radarr"
+    assert troubled["graded_on_normal"] is True and troubled["certificate"]["days_left"] == 50.0
+    healthy = t._router_row({"name": "ok@swarm", "state": "ok", "why": None, "latest": latest}, {})
+    assert "codes" not in healthy and "meaning" not in healthy and "certificate" not in healthy
+
+
+def test_the_plugin_registers_its_tools_for_viewers():
+    got = {}
+    traefik_plugin.register_mcp_tools(lambda fn, role, read_only: got.update({fn.__name__: (role, read_only)}))
+    assert got == {name: (UserRole.viewer, True)
+                   for name in ("list_routers", "get_router", "list_certificates", "get_traffic")}
+    assert callable(traefik_plugin.mcp_overview)
+
+
+async def test_list_routers_refuses_an_unknown_state():
+    with pytest.raises(ValueError, match="state must be one of"):
+        await t.list_routers(state="broken")