feat(mcp): monitoring-config write tools — monitors, alert rules, acks, maintenance

MCP-7 #4971 "Monitoring-config write tools: monitors, alert rules, acks,
maintenance".

The validation the forms did inline (or not at all) moves to the services,
and both the routes and the tools call it, so bad input is refused the same
way with the same reason:
- monitors/service: monitor_fields, build_config, normalise_target. Port
  1-65535, status 100-599, timeout 1-60, method GET/HEAD/POST, headers an
  object of strings, a key the type does not take is named. The forms now
  answer 400 with the reason instead of redirecting silently or a 500.
- alerts/service: METRIC_CATALOG and SOURCE_MODULES, resources_for,
  metrics_for, rule_fields (source, resource and metric from the same
  choices the rule form offers), add_rule, acknowledge, window_fields.

Tools: create_monitor, update_monitor (config merged over the stored one;
a rename warns about rules still watching the old name),
set_monitor_enabled, create_alert_rule, update_alert_rule,
acknowledge_alert, create_maintenance_window (duration or end),
end_maintenance_window (ends one in progress, cancels an upcoming one) —
operator. delete_monitor, delete_alert_rule — admin. get_alert_rule_options
— viewer. Every write is audited as "<user> via MCP".

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

steward/alerts/routes.py

@@ -6,63 +6,32 @@ from sqlalchemy import select
 from steward.auth.middleware import require_role
 from steward.alerts import service as alert_svc
 from steward.ansible import sources as src_module
-from steward.models.alerts import AlertRule, AlertState, AlertStateEnum, AlertOperator, MaintenanceWindow
-from steward.models.metrics import PluginMetric
+from steward.models.alerts import AlertRule, AlertOperator, MaintenanceWindow
 from steward.models.users import UserRole
 
 alerts_bp = Blueprint("alerts", __name__, url_prefix="/alerts")
 
 _OPERATORS = [(op.value, op.value) for op in AlertOperator]
 
-# Static catalog: source_module → available metric names.
-# Unified monitors emit metrics under their TYPE (icmp/tcp/dns/http), each with
-# is_up (1.0 up / 0.0 down) and response_ms (where the probe times it).
-METRIC_CATALOG: dict[str, list[str]] = {
-    "icmp":    ["is_up", "response_ms"],
-    "tcp":     ["is_up", "response_ms"],
-    "dns":     ["is_up", "response_ms"],
-    "http":    ["is_up", "response_ms"],
-    "traefik": ["request_rate", "error_rate", "latency_p50_ms", "latency_p95_ms",
-                "latency_p99_ms", "response_bytes_rate", "cert_expiry_days"],
-    "unifi":   ["is_up", "latency_ms", "total_clients"],
-    # restart_count = cumulative restarts (alert on crash-looping); is_healthy =
-    # 1.0 healthy / 0.0 unhealthy from the container HEALTHCHECK (alert on <1).
-    "docker":  ["cpu_pct", "mem_pct", "restart_count", "is_healthy"],
-    "host_agent": [
-        "cpu_pct", "mem_used_pct", "mem_available_bytes", "swap_used_bytes",
-        "disk_used_pct_worst", "load_1m", "load_5m", "load_15m", "uptime_secs",
-    ],
-    # snmp metrics are user-defined OID labels — populated dynamically from plugin_metrics
-    "snmp":    [],
-}
-
-# Ordered list for source module picker
-_SOURCE_MODULES = list(METRIC_CATALOG.keys())
+def _rule_form(form) -> dict:
+    """The rule form's fields, as alert_svc.rule_fields takes them."""
+    return {
+        "name": form.get("name", ""),
+        "source_module": form.get("source_module", ""),
+        "resource_name": form.get("resource_name", ""),
+        "metric_name": form.get("metric_name", ""),
+        "operator": form.get("operator", ""),
+        "threshold": form.get("threshold"),
+        "consecutive_failures_required": form.get("consecutive_failures_required"),
+    }
 
 
-async def _get_resources(db, source_module: str) -> list[str]:
-    """Return sorted distinct resource names for a source module.
-
-    For monitor types (icmp/tcp/dns/http), supplements with monitor names so
-    the list is populated even before any metrics have been recorded.
-    """
-    result = await db.execute(
-        select(PluginMetric.resource_name)
-        .where(PluginMetric.source_module == source_module)
-        .distinct()
-        .order_by(PluginMetric.resource_name)
-    )
-    resources: set[str] = {r for (r,) in result}
-
-    if source_module in ("icmp", "tcp", "dns", "http"):
-        from steward.models.monitors import Monitor
-        mon_result = await db.execute(
-            select(Monitor.name).where(Monitor.type == source_module).order_by(Monitor.name)
-        )
-        for (name,) in mon_result:
-            resources.add(name)
-
-    return sorted(resources)
+def _form_time(value) -> datetime:
+    """A datetime-local field, read as UTC."""
+    try:
+        return datetime.fromisoformat((value or "").strip()).replace(tzinfo=timezone.utc)
+    except ValueError:
+        raise ValueError(f"{value!r} is not a date and time.") from None
 
 
 def _is_admin() -> bool:
@@ -137,18 +106,9 @@ async def rule_fields_api():
     current_resource = request.args.get("current_resource", "")
     current_metric = request.args.get("current_metric", "")
     async with current_app.db_sessionmaker() as db:
-        resources = await _get_resources(db, source) if source else []
-    metrics = METRIC_CATALOG.get(source, [])
-    # Dynamic sources (e.g. snmp) have empty static catalog — query live metric names
-    if source and not metrics:
-        async with current_app.db_sessionmaker() as db:
-            m_result = await db.execute(
-                select(PluginMetric.metric_name)
-                .where(PluginMetric.source_module == source)
-                .distinct()
-                .order_by(PluginMetric.metric_name)
-            )
-            metrics = [r for (r,) in m_result]
+        resources = await alert_svc.resources_for(db, source) if source else []
+    async with current_app.db_sessionmaker() as db:
+        metrics = await alert_svc.metrics_for(db, source)
     return await render_template(
         "alerts/_rule_fields.html",
         resources=resources,
@@ -161,17 +121,17 @@ async def rule_fields_api():
 @alerts_bp.get("/rules/new")
 @require_role(UserRole.operator)
 async def new_rule():
-    first_source = _SOURCE_MODULES[0]
+    first_source = alert_svc.SOURCE_MODULES[0]
     async with current_app.db_sessionmaker() as db:
-        resources = await _get_resources(db, first_source)
+        resources = await alert_svc.resources_for(db, first_source)
     return await render_template(
         "alerts/rules_form.html",
         rule=None,
         operators=_OPERATORS,
-        source_modules=_SOURCE_MODULES,
+        source_modules=alert_svc.SOURCE_MODULES,
         initial_source=first_source,
         resources=resources,
-        metrics=METRIC_CATALOG.get(first_source, []),
+        metrics=alert_svc.METRIC_CATALOG.get(first_source, []),
         ansible_sources=_ansible_source_names(),
     )
 
@@ -184,15 +144,15 @@ async def edit_rule(rule_id: str):
         rule = result.scalar_one_or_none()
         if rule is None:
             return "Not found", 404
-        resources = await _get_resources(db, rule.source_module)
+        resources = await alert_svc.resources_for(db, rule.source_module)
     return await render_template(
         "alerts/rules_form.html",
         rule=rule,
         operators=_OPERATORS,
-        source_modules=_SOURCE_MODULES,
+        source_modules=alert_svc.SOURCE_MODULES,
         initial_source=rule.source_module,
         resources=resources,
-        metrics=METRIC_CATALOG.get(rule.source_module, []),
+        metrics=alert_svc.METRIC_CATALOG.get(rule.source_module, []),
         ansible_sources=_ansible_source_names(),
     )
 
@@ -202,36 +162,20 @@ async def edit_rule(rule_id: str):
 async def create_rule():
     form = await request.form
     user_id = session.get("user_id")
-    now = datetime.now(timezone.utc)
     # Ansible action is admin-only; non-admins simply can't set one.
     ansible_action = None
     if _is_admin():
         ansible_action, err = _parse_ansible_action(form)
         if err:
             return err, 400
-    rule = AlertRule(
-        name=form["name"].strip(),
-        source_module=form["source_module"].strip(),
-        resource_name=form["resource_name"].strip(),
-        metric_name=form["metric_name"].strip(),
-        operator=AlertOperator(form["operator"]),
-        threshold=float(form["threshold"]),
-        consecutive_failures_required=int(form.get("consecutive_failures_required") or 1),
-        enabled=True,
-        ansible_action=ansible_action,
-        created_by=user_id,
-    )
-    state = AlertState(
-        rule_id=rule.id,
-        state=AlertStateEnum.inactive,
-        last_transition_at=now,
-    )
     async with current_app.db_sessionmaker() as db:
         async with db.begin():
-            db.add(rule)
-            await db.flush()
-            state.rule_id = rule.id
-            db.add(state)
+            try:
+                fields = await alert_svc.rule_fields(db, **_rule_form(form))
+            except ValueError as exc:
+                return str(exc), 400
+            rule = await alert_svc.add_rule(db, fields, created_by=user_id,
+                                            ansible_action=ansible_action)
     await log_audit(current_app, user_id, session.get("username", ""), "alert_rule.created",
                     entity_type="alert_rule", entity_id=rule.name,
                     detail={"source": rule.source_module, "resource": rule.resource_name,
@@ -256,13 +200,12 @@ async def update_rule(rule_id: str):
             rule = result.scalar_one_or_none()
             if rule is None:
                 return "Not found", 404
-            rule.name = form["name"].strip()
-            rule.source_module = form["source_module"].strip()
-            rule.resource_name = form["resource_name"].strip()
-            rule.metric_name = form["metric_name"].strip()
-            rule.operator = AlertOperator(form["operator"])
-            rule.threshold = float(form["threshold"])
-            rule.consecutive_failures_required = int(form.get("consecutive_failures_required") or 1)
+            try:
+                fields = await alert_svc.rule_fields(db, **_rule_form(form))
+            except ValueError as exc:
+                return str(exc), 400
+            for k, v in fields.items():
+                setattr(rule, k, v)
             if apply_action:
                 rule.ansible_action = ansible_action
     await log_audit(current_app, session.get("user_id"), session.get("username", ""),
@@ -313,7 +256,7 @@ async def list_maintenance():
         "alerts/maintenance.html",
         windows=windows,
         now=now,
-        source_modules=_SOURCE_MODULES,
+        source_modules=alert_svc.SOURCE_MODULES,
     )
 
 
@@ -323,7 +266,7 @@ async def new_maintenance():
     return await render_template(
         "alerts/maintenance_form.html",
         window=None,
-        source_modules=_SOURCE_MODULES,
+        source_modules=alert_svc.SOURCE_MODULES,
     )
 
 
@@ -332,26 +275,18 @@ async def new_maintenance():
 async def create_maintenance():
     form = await request.form
     user_id = session.get("user_id")
-    start_at = datetime.fromisoformat(form["start_at"]).replace(tzinfo=timezone.utc)
-    end_at = datetime.fromisoformat(form["end_at"]).replace(tzinfo=timezone.utc)
-    scope_source = form.get("scope_source", "").strip() or None
-    scope_resource = form.get("scope_resource", "").strip() or None
-    if scope_source is None:
-        scope_resource = None  # can't have resource without source
-    window = MaintenanceWindow(
-        name=form["name"].strip(),
-        start_at=start_at,
-        end_at=end_at,
-        scope_source=scope_source,
-        scope_resource=scope_resource,
-        created_by=user_id,
-    )
+    try:
+        fields = alert_svc.window_fields(
+            name=form.get("name", ""),
+            start_at=_form_time(form.get("start_at")), end_at=_form_time(form.get("end_at")),
+            scope_source=form.get("scope_source"), scope_resource=form.get("scope_resource"))
+    except ValueError as exc:
+        return str(exc), 400
+    window = MaintenanceWindow(created_by=user_id, **fields)
     async with current_app.db_sessionmaker() as db:
         async with db.begin():
             db.add(window)
-    scope_desc = "all" if not scope_source else (
-        f"{scope_source}/{scope_resource}" if scope_resource else f"{scope_source}/*"
-    )
+    scope_desc = alert_svc.window_scope(window)
     await log_audit(current_app, user_id, session.get("username", ""),
                     "maintenance_window.created", entity_type="maintenance_window",
                     entity_id=window.name, detail={"scope": scope_desc})
@@ -382,19 +317,12 @@ async def delete_maintenance(window_id: str):
 @require_role(UserRole.operator)
 async def acknowledge(rule_id: str):
     user_id = session.get("user_id")
-    now = datetime.now(timezone.utc)
     async with current_app.db_sessionmaker() as db:
         async with db.begin():
-            result = await db.execute(
-                select(AlertState).where(AlertState.rule_id == rule_id)
-            )
-            state = result.scalar_one_or_none()
-            if state is None:
+            try:
+                await alert_svc.acknowledge(db, rule_id, user_id)
+            except LookupError:
                 return "Not found", 404
-            if state.state != AlertStateEnum.firing:
-                return "Conflict: alert is not in FIRING state", 409
-            state.state = AlertStateEnum.acknowledged
-            state.acknowledged_by = user_id
-            state.acknowledged_at = now
-            state.last_transition_at = now
+            except ValueError as exc:
+                return f"Conflict: {exc}", 409
     return redirect(url_for("alerts.list_alerts"))

steward/alerts/service.py

@@ -1,14 +1,45 @@
-"""Alert queries shared by the alerts pages and the MCP tools.
+"""Alert queries and validation shared by the alerts pages and the MCP tools.
 
-Each function takes an open session and returns rows, with no Quart request
-context, so the routes and the /mcp tools (which run outside Quart's pipeline)
-call the same code.
+Each function takes an open session (or plain values), with no Quart request
+context, so the routes and the /mcp tools (which run outside Quart's
+pipeline) call the same code and refuse the same input for the same reasons.
 """
 from __future__ import annotations
 
+from datetime import datetime, timezone
+
 from sqlalchemy import select
 
-from steward.models.alerts import AlertRule, AlertState, AlertStateEnum, MaintenanceWindow
+from steward.models.alerts import (
+    AlertOperator, AlertRule, AlertState, AlertStateEnum, MaintenanceWindow,
+)
+from steward.models.metrics import PluginMetric
+from steward.models.monitors import MONITOR_TYPE_VALUES, Monitor
+
+# Static catalog: source_module → available metric names.
+# Unified monitors emit metrics under their TYPE (icmp/tcp/dns/http), each with
+# is_up (1.0 up / 0.0 down) and response_ms (where the probe times it).
+METRIC_CATALOG: dict[str, list[str]] = {
+    "icmp":    ["is_up", "response_ms"],
+    "tcp":     ["is_up", "response_ms"],
+    "dns":     ["is_up", "response_ms"],
+    "http":    ["is_up", "response_ms"],
+    "traefik": ["request_rate", "error_rate", "latency_p50_ms", "latency_p95_ms",
+                "latency_p99_ms", "response_bytes_rate", "cert_expiry_days"],
+    "unifi":   ["is_up", "latency_ms", "total_clients"],
+    # restart_count = cumulative restarts (alert on crash-looping); is_healthy =
+    # 1.0 healthy / 0.0 unhealthy from the container HEALTHCHECK (alert on <1).
+    "docker":  ["cpu_pct", "mem_pct", "restart_count", "is_healthy"],
+    "host_agent": [
+        "cpu_pct", "mem_used_pct", "mem_available_bytes", "swap_used_bytes",
+        "disk_used_pct_worst", "load_1m", "load_5m", "load_15m", "uptime_secs",
+    ],
+    # snmp metrics are user-defined OID labels — populated dynamically from plugin_metrics
+    "snmp":    [],
+}
+
+# Ordered list for the source module picker.
+SOURCE_MODULES = list(METRIC_CATALOG.keys())
 
 # States that need someone's attention: pending (breaching, not yet for long
 # enough), firing, and acknowledged-but-still-breaching.
@@ -34,3 +65,144 @@ async def list_maintenance_windows(db) -> list[MaintenanceWindow]:
     return list((await db.execute(
         select(MaintenanceWindow).order_by(MaintenanceWindow.start_at.desc())
     )).scalars())
+
+
+async def resources_for(db, source_module: str) -> list[str]:
+    """Sorted distinct resource names for a source module.
+
+    For monitor types (icmp/tcp/dns/http), supplements with monitor names so
+    the list is populated even before any metrics have been recorded.
+    """
+    result = await db.execute(
+        select(PluginMetric.resource_name)
+        .where(PluginMetric.source_module == source_module)
+        .distinct()
+        .order_by(PluginMetric.resource_name)
+    )
+    resources: set[str] = {r for (r,) in result}
+    if source_module in MONITOR_TYPE_VALUES:
+        mon_result = await db.execute(
+            select(Monitor.name).where(Monitor.type == source_module).order_by(Monitor.name))
+        resources.update(name for (name,) in mon_result)
+    return sorted(resources)
+
+
+async def metrics_for(db, source_module: str) -> list[str]:
+    """The metric names a rule on this source can watch: the static catalog, or
+    for dynamic sources (snmp) the names recorded so far."""
+    metrics = METRIC_CATALOG.get(source_module, [])
+    if source_module and not metrics:
+        result = await db.execute(
+            select(PluginMetric.metric_name)
+            .where(PluginMetric.source_module == source_module)
+            .distinct()
+            .order_by(PluginMetric.metric_name))
+        metrics = [r for (r,) in result]
+    return metrics
+
+
+def _choices(values: list[str], limit: int = 20) -> str:
+    if not values:
+        return "none yet"
+    shown = ", ".join(values[:limit])
+    return shown + (f" (and {len(values) - limit} more)" if len(values) > limit else "")
+
+
+async def rule_fields(db, *, name: str, source_module: str, resource_name: str, metric_name: str,
+                      operator: str, threshold, consecutive_failures_required=1) -> dict:
+    """Validated column values for a new or edited alert rule.
+
+    The same choices the rule form offers: a known source, a resource that
+    source has reported (or, for monitors, a monitor of that type), and a
+    metric it emits. ValueError says what was wrong and what is valid.
+    """
+    name = (name or "").strip()
+    if not name:
+        raise ValueError("an alert rule needs a name.")
+    source_module = (source_module or "").strip()
+    if source_module not in METRIC_CATALOG:
+        raise ValueError(f"source_module {source_module!r} is not one of {', '.join(SOURCE_MODULES)}.")
+    resource_name = (resource_name or "").strip()
+    resources = await resources_for(db, source_module)
+    if resource_name not in resources:
+        raise ValueError(f"{source_module} has no resource {resource_name!r}. "
+                         f"Known: {_choices(resources)}.")
+    metric_name = (metric_name or "").strip()
+    metrics = await metrics_for(db, source_module)
+    if metric_name not in metrics:
+        raise ValueError(f"{source_module} does not report {metric_name!r}. "
+                         f"Its metrics: {_choices(metrics)}.")
+    try:
+        op = AlertOperator((operator or "").strip())
+    except ValueError:
+        raise ValueError(f"operator {operator!r} is not one of "
+                         f"{', '.join(o.value for o in AlertOperator)}.") from None
+    try:
+        threshold = float(threshold)
+    except (TypeError, ValueError):
+        raise ValueError(f"threshold must be a number, got {threshold!r}.") from None
+    try:
+        consecutive = (1 if consecutive_failures_required in (None, "")
+                       else int(consecutive_failures_required))
+    except (TypeError, ValueError):
+        raise ValueError("consecutive checks must be a whole number.") from None
+    if consecutive < 1:
+        raise ValueError("consecutive checks must be at least 1.")
+    return {"name": name[:128], "source_module": source_module, "resource_name": resource_name,
+            "metric_name": metric_name, "operator": op, "threshold": threshold,
+            "consecutive_failures_required": consecutive}
+
+
+async def add_rule(db, fields: dict, *, created_by: str, enabled: bool = True,
+                   ansible_action: dict | None = None) -> AlertRule:
+    """Insert a rule and its inactive state row, in the caller's transaction."""
+    rule = AlertRule(enabled=enabled, ansible_action=ansible_action, created_by=created_by, **fields)
+    db.add(rule)
+    await db.flush()
+    db.add(AlertState(rule_id=rule.id, state=AlertStateEnum.inactive,
+                      last_transition_at=datetime.now(timezone.utc)))
+    return rule
+
+
+async def acknowledge(db, rule_id: str, user_id: str) -> AlertState:
+    """Acknowledge a firing alert, in the caller's transaction.
+
+    LookupError when the rule has no state; ValueError when it is not firing.
+    """
+    state = (await db.execute(
+        select(AlertState).where(AlertState.rule_id == rule_id))).scalar_one_or_none()
+    if state is None:
+        raise LookupError(rule_id)
+    if state.state != AlertStateEnum.firing:
+        raise ValueError(f"the alert is {state.state.value}, not firing; only a firing alert "
+                         "can be acknowledged.")
+    now = datetime.now(timezone.utc)
+    state.state = AlertStateEnum.acknowledged
+    state.acknowledged_by = user_id
+    state.acknowledged_at = now
+    state.last_transition_at = now
+    return state
+
+
+def window_fields(*, name: str, start_at: datetime, end_at: datetime,
+                  scope_source: str | None, scope_resource: str | None) -> dict:
+    """Validated columns for a maintenance window. A resource needs a source."""
+    name = (name or "").strip()
+    if not name:
+        raise ValueError("a maintenance window needs a name.")
+    if end_at <= start_at:
+        raise ValueError("the window must end after it starts.")
+    scope_source = (scope_source or "").strip() or None
+    scope_resource = (scope_resource or "").strip() or None
+    if scope_source is not None and scope_source not in METRIC_CATALOG:
+        raise ValueError(f"source {scope_source!r} is not one of {', '.join(SOURCE_MODULES)}.")
+    if scope_source is None:
+        scope_resource = None  # can't have resource without source
+    return {"name": name[:128], "start_at": start_at, "end_at": end_at,
+            "scope_source": scope_source, "scope_resource": scope_resource}
+
+
+def window_scope(w) -> str:
+    if not w.scope_source:
+        return "all"
+    return f"{w.scope_source}/{w.scope_resource}" if w.scope_resource else f"{w.scope_source}/*"

steward/mcp/tools/alerts.py

@@ -1,12 +1,16 @@
 """Alert tools: what is firing, what fired before, the rules, and maintenance windows."""
 from __future__ import annotations
 
+from datetime import datetime, timedelta, timezone
+
 from sqlalchemy import select
 
 from steward.alerts import service as alert_svc
-from steward.models.alerts import AlertEvent, AlertRule, AlertState
+from steward.mcp._context import current_caller
+from steward.models.alerts import AlertEvent, AlertOperator, AlertRule, AlertState, MaintenanceWindow
 from steward.models.users import User, UserRole
 
+from ._audit import audit_mcp
 from ._shape import (
     age_s, bounded, clamp_limit, db_session, iso, lookback, now_utc, pick, rnd,
 )
@@ -184,8 +188,211 @@ async def list_maintenance_windows(include_past: bool = False, limit: int = 50)
     return bounded(rows, limit, key="windows")
 
 
+# ── changes ──────────────────────────────────────────────────────────────────
+
+async def get_alert_rule_options(source_module: str = "") -> dict:
+    """What an alert rule can watch: the choices create_alert_rule accepts.
+
+    source_module: empty lists every source and its metrics; a source's name
+      also lists its resources (hosts, containers, monitors… whatever that
+      source reports under).
+
+    A monitor reports under its type (icmp/tcp/dns/http) with its name as the
+    resource: is_up (1 up, 0 down) and response_ms.
+    """
+    async with db_session() as db:
+        if not source_module:
+            return {"sources": {src: await alert_svc.metrics_for(db, src)
+                                for src in alert_svc.SOURCE_MODULES},
+                    "operators": [o.value for o in AlertOperator]}
+        if source_module not in alert_svc.METRIC_CATALOG:
+            raise ValueError(f"source_module {source_module!r} is not one of "
+                             f"{', '.join(alert_svc.SOURCE_MODULES)}.")
+        return {"source_module": source_module,
+                "metrics": await alert_svc.metrics_for(db, source_module),
+                "resources": await alert_svc.resources_for(db, source_module),
+                "operators": [o.value for o in AlertOperator]}
+
+
+async def _rule(db, rule: str) -> AlertRule:
+    return pick(rule, await alert_svc.list_rules(db), kind="alert rule", list_tool="list_alert_rules")
+
+
+def _rule_row(r: AlertRule) -> dict:
+    return {"rule_id": r.id, "rule_name": r.name, "enabled": r.enabled,
+            "source_module": r.source_module, "resource": r.resource_name,
+            "metric": r.metric_name, "condition": _condition(r), "ansible_action": _action(r)}
+
+
+async def create_alert_rule(name: str, source_module: str, resource: str, metric: str,
+                            operator: str, threshold: float, consecutive_checks: int = 1,
+                            enabled: bool = True) -> dict:
+    """Add an alert rule: notify when `metric` of `resource` crosses `threshold`.
+
+    source_module, resource, metric: get_alert_rule_options lists the valid
+      choices; a resource must have reported (or, for a monitor, exist).
+    operator: ">", "<", ">=", "<=", "==" or "!=".
+    consecutive_checks: how many checks in a row must breach before it fires
+      (default 1); higher values ride out a single blip.
+
+    Notifications go to the channels set in Steward's Settings. The playbook a
+    rule runs when it fires is set separately (set_alert_action, admin).
+    Recorded in the audit log as "<you> via MCP".
+    """
+    caller = current_caller()
+    async with db_session() as db, db.begin():
+        fields = await alert_svc.rule_fields(
+            db, name=name, source_module=source_module, resource_name=resource,
+            metric_name=metric, operator=operator, threshold=threshold,
+            consecutive_failures_required=consecutive_checks)
+        r = await alert_svc.add_rule(db, fields, created_by=caller.user_id, enabled=enabled)
+    await audit_mcp("alert_rule.created", "alert_rule", r.name,
+                    {"rule_id": r.id, "source": r.source_module, "resource": r.resource_name,
+                     "metric": r.metric_name})
+    return _rule_row(r)
+
+
+async def update_alert_rule(rule: str, name: str | None = None, source_module: str | None = None,
+                            resource: str | None = None, metric: str | None = None,
+                            operator: str | None = None, threshold: float | None = None,
+                            consecutive_checks: int | None = None,
+                            enabled: bool | None = None) -> dict:
+    """Change an alert rule. Only the arguments you pass change; the rest stay.
+
+    rule: its name or id (list_alert_rules). Pass enabled=false to silence it
+    without deleting it. The changed rule is validated as create_alert_rule
+    validates a new one. Recorded in the audit log.
+    """
+    async with db_session() as db, db.begin():
+        r = await _rule(db, rule)
+        fields = await alert_svc.rule_fields(
+            db,
+            name=r.name if name is None else name,
+            source_module=r.source_module if source_module is None else source_module,
+            resource_name=r.resource_name if resource is None else resource,
+            metric_name=r.metric_name if metric is None else metric,
+            operator=r.operator.value if operator is None else operator,
+            threshold=r.threshold if threshold is None else threshold,
+            consecutive_failures_required=(r.consecutive_failures_required
+                                           if consecutive_checks is None else consecutive_checks))
+        for k, v in fields.items():
+            setattr(r, k, v)
+        if enabled is not None:
+            r.enabled = bool(enabled)
+    changed = sorted(k for k, v in {"name": name, "source_module": source_module,
+                                    "resource": resource, "metric": metric, "operator": operator,
+                                    "threshold": threshold, "consecutive_checks": consecutive_checks,
+                                    "enabled": enabled}.items() if v is not None)
+    await audit_mcp("alert_rule.updated", "alert_rule", r.name, {"rule_id": r.id, "changed": changed})
+    return _rule_row(r)
+
+
+async def delete_alert_rule(rule: str) -> dict:
+    """Delete an alert rule and its history. To silence it but keep it,
+    update_alert_rule(enabled=false). Admin only; recorded in the audit log."""
+    async with db_session() as db, db.begin():
+        r = await _rule(db, rule)
+        await db.delete(r)
+    await audit_mcp("alert_rule.deleted", "alert_rule", r.name, {"rule_id": r.id})
+    return {"deleted": r.name, "rule_id": r.id}
+
+
+async def acknowledge_alert(rule: str) -> dict:
+    """Acknowledge a firing alert: you have seen it. It stays "acknowledged"
+    while the condition holds and resolves when it clears.
+
+    rule: the firing rule's name or id (list_alerts). Recorded in the audit log.
+    """
+    caller = current_caller()
+    async with db_session() as db, db.begin():
+        r = await _rule(db, rule)
+        try:
+            state = await alert_svc.acknowledge(db, r.id, caller.user_id)
+        except LookupError:
+            raise ValueError(f"{r.name} has never been evaluated, so there is nothing to "
+                             "acknowledge.") from None
+    await audit_mcp("alert_rule.acknowledged", "alert_rule", r.name, {"rule_id": r.id})
+    return {"rule_id": r.id, "rule_name": r.name, "state": state.state.value,
+            "acknowledged_at": iso(state.acknowledged_at)}
+
+
+def _when(value: str, what: str) -> datetime:
+    try:
+        dt = datetime.fromisoformat(value.strip().replace("Z", "+00:00"))
+    except ValueError:
+        raise ValueError(f"{what} {value!r} is not an ISO-8601 time, e.g. 2026-10-06T22:00:00Z.") from None
+    return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
+
+
+def _window_row(w: MaintenanceWindow, now: datetime) -> dict:
+    return {"window_id": w.id, "name": w.name, "start_at": iso(w.start_at), "end_at": iso(w.end_at),
+            "active": w.start_at <= now <= w.end_at,
+            "scope": {"source_module": w.scope_source, "resource": w.scope_resource}}
+
+
+async def create_maintenance_window(name: str, duration_minutes: int = 0, end_at: str = "",
+                                    start_at: str = "", source_module: str = "",
+                                    resource: str = "") -> dict:
+    """Hold back alert notifications while you work on something.
+
+    Give the end as duration_minutes OR end_at (ISO-8601; a time with no zone
+    is UTC). start_at defaults to now.
+    source_module, resource: limit the window to one source, or to one
+      resource of it; both empty silences everything.
+
+    Alerts still change state during a window, so list_alerts shows them;
+    only notifications are held. Recorded in the audit log.
+    """
+    now = now_utc()
+    start = _when(start_at, "start_at") if (start_at or "").strip() else now
+    if bool(duration_minutes) == bool((end_at or "").strip()):
+        raise ValueError("give exactly one of duration_minutes or end_at.")
+    if duration_minutes:
+        if duration_minutes < 0:
+            raise ValueError("duration_minutes must be positive.")
+        end = start + timedelta(minutes=duration_minutes)
+    else:
+        end = _when(end_at, "end_at")
+    fields = alert_svc.window_fields(name=name, start_at=start, end_at=end,
+                                     scope_source=source_module, scope_resource=resource)
+    w = MaintenanceWindow(created_by=current_caller().user_id, **fields)
+    async with db_session() as db, db.begin():
+        db.add(w)
+    await audit_mcp("maintenance_window.created", "maintenance_window", w.name,
+                    {"window_id": w.id, "scope": alert_svc.window_scope(w),
+                     "start_at": iso(w.start_at), "end_at": iso(w.end_at)})
+    return _window_row(w, now)
+
+
+async def end_maintenance_window(window: str) -> dict:
+    """End a maintenance window now, so notifications resume.
+
+    window: its name or id (list_maintenance_windows). A window in progress
+    ends now and stays on record; one that has not started yet is removed.
+    Recorded in the audit log.
+    """
+    now = now_utc()
+    async with db_session() as db, db.begin():
+        windows = await alert_svc.list_maintenance_windows(db)
+        w = pick(window, windows, kind="maintenance window", list_tool="list_maintenance_windows")
+        if w.end_at <= now:
+            raise ValueError(f"{w.name} already ended at {iso(w.end_at)}.")
+        if w.start_at > now:
+            await db.delete(w)
+            outcome = "cancelled"
+        else:
+            w.end_at = now
+            outcome = "ended"
+    await audit_mcp(f"maintenance_window.{outcome}", "maintenance_window", w.name,
+                    {"window_id": w.id})
+    return {**_window_row(w, now), "outcome": outcome}
+
+
 def register(register_tool) -> None:
-    register_tool(list_alerts, role=UserRole.viewer, read_only=True)
-    register_tool(get_alert_history, role=UserRole.viewer, read_only=True)
-    register_tool(list_alert_rules, role=UserRole.viewer, read_only=True)
-    register_tool(list_maintenance_windows, role=UserRole.viewer, read_only=True)
+    for fn in (list_alerts, get_alert_history, list_alert_rules, list_maintenance_windows,
+               get_alert_rule_options):
+        register_tool(fn, role=UserRole.viewer, read_only=True)
+    for fn in (create_alert_rule, update_alert_rule, acknowledge_alert,
+               create_maintenance_window, end_maintenance_window):
+        register_tool(fn, role=UserRole.operator, read_only=False)
+    register_tool(delete_alert_rule, role=UserRole.admin, read_only=False)

steward/mcp/tools/monitors.py

@@ -5,11 +5,13 @@ import json
 
 from sqlalchemy import select
 
+from steward.models.alerts import AlertRule
 from steward.models.hosts import Host
 from steward.models.monitors import MONITOR_TYPE_VALUES, Monitor
 from steward.models.users import UserRole
 from steward.monitors import service as monitor_svc
 
+from ._audit import audit_mcp
 from ._shape import (
     age_s, bounded, clamp_limit, db_session, iso, now_utc, pick, redact_url, rnd,
 )
@@ -137,6 +139,138 @@ async def get_monitor(monitor: str, failures: int = 10) -> dict:
     }
 
 
+# ── changes ──────────────────────────────────────────────────────────────────
+
+async def _host_id(db, host: str) -> str | None:
+    if not (host or "").strip():
+        return None
+    hosts = (await db.execute(select(Host))).scalars()
+    return pick(host, hosts, kind="host", list_tool="list_hosts").id
+
+
+def _saved(m: Monitor, host_names: dict[str, str]) -> dict:
+    return {
+        "monitor_id": m.id,
+        "monitor_name": m.name,
+        "type": m.type,
+        "target": redact_url(m.target),
+        "host_name": host_names.get(m.host_id) if m.host_id else None,
+        "enabled": m.enabled,
+        "check_interval_s": m.check_interval_seconds or None,
+        "config": _config(m),
+        "next": f'get_monitor(monitor="{m.id}")',
+    }
+
+
+async def create_monitor(name: str, type: str, target: str, host: str = "",
+                         config: dict | None = None, check_interval_s: int = 0,
+                         enabled: bool = True) -> dict:
+    """Add an uptime check. Recorded in the audit log as "<you> via MCP".
+
+    type: "icmp" (ping), "tcp" (connect to a port), "dns" (resolve a name)
+      or "http" (request a URL).
+    target: an address or hostname; for http a URL (a bare host gets https://).
+    host: a Steward host's name or id to file the check under; empty for a
+      free-standing destination.
+    config, by type (every key optional, defaults shown):
+      tcp:  {"port": 80}
+      dns:  {"expected_ip": null}  (set to fail when the name resolves elsewhere)
+      http: {"method": "GET" | "HEAD" | "POST", "expected_status": 200,
+             "content_match": "", "headers": {}, "timeout_seconds": 10 (1-60),
+             "follow_redirects": true, "verify_ssl": true}
+      icmp takes none.
+    check_interval_s: seconds between checks; 0 (default) uses the global
+      interval from Settings.
+
+    The check's metrics are reported under source_module = its type and
+    resource = its name, which is what an alert rule on it watches.
+    """
+    fields = monitor_svc.monitor_fields(type, target=target, name=name,
+                                        check_interval_seconds=check_interval_s, opts=config or {})
+    async with db_session() as db, db.begin():
+        m = Monitor(host_id=await _host_id(db, host), enabled=enabled, **fields)
+        db.add(m)
+        _monitors, host_names = await _load(db)
+    await audit_mcp("monitor.created", "monitor", m.name,
+                    {"monitor_id": m.id, "type": m.type, "target": redact_url(m.target)})
+    return _saved(m, host_names)
+
+
+async def update_monitor(monitor: str, name: str | None = None, target: str | None = None,
+                         host: str | None = None, config: dict | None = None,
+                         check_interval_s: int | None = None) -> dict:
+    """Change a monitor. Only the arguments you pass change; the rest stay.
+
+    monitor: its name or id (list_monitors).
+    host: a host's name or id to file it under, or "" to unlink it.
+    config: the keys to change, merged over its current config (see
+      create_monitor for the keys of each type). A monitor's type cannot
+      change; delete it and create another.
+
+    Alert rules watch a monitor by its name, so a rename returns `warning`
+    naming the rules still on the old name.
+
+    Recorded in the audit log.
+    """
+    async with db_session() as db, db.begin():
+        monitors, host_names = await _load(db)
+        m = pick(monitor, monitors, kind="monitor", list_tool="list_monitors")
+        current = {k: v for k, v in m.config.items() if k in monitor_svc.CONFIG_KEYS[m.type]}
+        fields = monitor_svc.monitor_fields(
+            m.type, target=m.target if target is None else target,
+            name=m.name if name is None else name,
+            check_interval_seconds=(m.check_interval_seconds if check_interval_s is None
+                                    else check_interval_s),
+            opts={**current, **(config or {})})
+        old_name = m.name
+        for k, v in fields.items():
+            setattr(m, k, v)
+        if host is not None:
+            m.host_id = await _host_id(db, host)
+        stranded = []
+        if m.name != old_name:
+            stranded = list((await db.execute(
+                select(AlertRule.name).where(AlertRule.source_module == m.type,
+                                             AlertRule.resource_name == old_name))).scalars())
+    changed = sorted(k for k, v in {"name": name, "target": target, "host": host, "config": config,
+                                    "check_interval_s": check_interval_s}.items() if v is not None)
+    await audit_mcp("monitor.updated", "monitor", m.name, {"monitor_id": m.id, "changed": changed})
+    out = _saved(m, host_names)
+    if stranded:
+        out["warning"] = (f"Alert rules watch this monitor by name and still watch {old_name!r}: "
+                          f"{', '.join(stranded)}. update_alert_rule(resource=\"{m.name}\") "
+                          "points each at the new name.")
+    return out
+
+
+async def set_monitor_enabled(monitor: str, enabled: bool) -> dict:
+    """Pause (enabled=false) or resume a monitor. A paused monitor runs no checks
+    and raises no alerts. Recorded in the audit log."""
+    async with db_session() as db, db.begin():
+        monitors, host_names = await _load(db)
+        m = pick(monitor, monitors, kind="monitor", list_tool="list_monitors")
+        m.enabled = bool(enabled)
+    await audit_mcp("monitor.enabled" if m.enabled else "monitor.disabled", "monitor", m.name,
+                    {"monitor_id": m.id})
+    return _saved(m, host_names)
+
+
+async def delete_monitor(monitor: str) -> dict:
+    """Delete a monitor and its check history. To stop it but keep its history,
+    set_monitor_enabled(enabled=false) instead. Admin only; recorded in the
+    audit log."""
+    async with db_session() as db, db.begin():
+        monitors, _host_names = await _load(db)
+        m = pick(monitor, monitors, kind="monitor", list_tool="list_monitors")
+        await db.delete(m)
+    await audit_mcp("monitor.deleted", "monitor", m.name,
+                    {"monitor_id": m.id, "type": m.type, "target": redact_url(m.target)})
+    return {"deleted": m.name, "monitor_id": m.id}
+
+
 def register(register_tool) -> None:
     register_tool(list_monitors, role=UserRole.viewer, read_only=True)
     register_tool(get_monitor, role=UserRole.viewer, read_only=True)
+    for fn in (create_monitor, update_monitor, set_monitor_enabled):
+        register_tool(fn, role=UserRole.operator, read_only=False)
+    register_tool(delete_monitor, role=UserRole.admin, read_only=False)

steward/monitors/routes.py

@@ -5,7 +5,6 @@ type-aware add/edit form, one dashboard widget. A monitor may be linked to a
 Host (host_id) or watch a free-standing custom destination (host_id = NULL).
 """
 from __future__ import annotations
-import json
 import uuid
 from datetime import datetime, timezone
 
@@ -20,35 +19,24 @@ from steward.monitors import service as monitor_svc
 monitors_bp = Blueprint("monitors", __name__, url_prefix="/monitors")
 
 
-def _build_config(form, mtype: str) -> dict:
-    """Assemble the type-specific config dict from submitted form fields."""
+def _form_opts(form, mtype: str) -> dict:
+    """The submitted fields of `mtype`'s config, for monitor_svc.build_config.
+    An unchecked checkbox is absent from the form, so it reads as False."""
     if mtype == MonitorType.tcp.value:
-        return {"port": int(form.get("port") or 80)}
+        return {"port": form.get("port")}
     if mtype == MonitorType.dns.value:
-        return {"expected_ip": (form.get("expected_ip", "").strip() or None)}
+        return {"expected_ip": form.get("expected_ip")}
     if mtype == MonitorType.http.value:
-        try:
-            headers = json.loads(form.get("headers_json") or "{}")
-        except (ValueError, TypeError):
-            headers = {}
         return {
-            "method": form.get("method", "GET").upper(),
-            "expected_status": int(form.get("expected_status", 200) or 200),
-            "content_match": form.get("content_match", "").strip(),
-            "headers": headers,
-            "timeout_seconds": int(form.get("timeout_seconds", 10) or 10),
+            "method": form.get("method"),
+            "expected_status": form.get("expected_status"),
+            "content_match": form.get("content_match"),
+            "headers": form.get("headers_json"),
+            "timeout_seconds": form.get("timeout_seconds"),
             "follow_redirects": "follow_redirects" in form,
             "verify_ssl": "verify_ssl" in form,
         }
-    return {}  # icmp
-
-
-def _normalise_target(mtype: str, target: str) -> str:
-    """For http, default a bare host to https://; others pass through."""
-    target = target.strip()
-    if mtype == MonitorType.http.value and target and not target.startswith(("http://", "https://")):
-        return "https://" + target
-    return target
+    return {}
 
 
 @monitors_bp.get("/")
@@ -80,22 +68,17 @@ async def rows():
 async def add_monitor():
     form = await request.form
     mtype = form.get("type", "").strip()
-    target = _normalise_target(mtype, form.get("target", ""))
-    if mtype not in MONITOR_TYPE_VALUES or not target:
-        return redirect(url_for("monitors.index"))
+    try:
+        fields = monitor_svc.monitor_fields(
+            mtype, target=form.get("target", ""), name=form.get("name", ""),
+            check_interval_seconds=form.get("check_interval_seconds"),
+            opts=_form_opts(form, mtype))
+    except ValueError as exc:
+        return str(exc), 400
 
     host_id = form.get("host_id", "").strip() or None
-    monitor = Monitor(
-        id=str(uuid.uuid4()),
-        name=form.get("name", "").strip() or target,
-        type=mtype,
-        target=target,
-        host_id=host_id,
-        config_json=json.dumps(_build_config(form, mtype)),
-        enabled=True,
-        check_interval_seconds=int(form.get("check_interval_seconds", 0) or 0),
-        created_at=datetime.now(timezone.utc),
-    )
+    monitor = Monitor(id=str(uuid.uuid4()), host_id=host_id, enabled=True,
+                      created_at=datetime.now(timezone.utc), **fields)
     async with current_app.db_sessionmaker() as db:
         async with db.begin():
             db.add(monitor)
@@ -129,10 +112,15 @@ async def edit_monitor(monitor_id: str):
                 return redirect(url_for("monitors.index"))
             # Type is immutable on edit (changes the meaning of config/target);
             # to switch type, delete and re-add.
-            m.name = form.get("name", "").strip() or m.target
-            m.target = _normalise_target(m.type, form.get("target", m.target))
-            m.config_json = json.dumps(_build_config(form, m.type))
-            m.check_interval_seconds = int(form.get("check_interval_seconds", 0) or 0)
+            try:
+                fields = monitor_svc.monitor_fields(
+                    m.type, target=form.get("target", m.target), name=form.get("name", ""),
+                    check_interval_seconds=form.get("check_interval_seconds"),
+                    opts=_form_opts(form, m.type))
+            except ValueError as exc:
+                return str(exc), 400
+            for k, v in fields.items():
+                setattr(m, k, v)
     return redirect(url_for("monitors.index"))
 
 

steward/monitors/service.py

@@ -1,18 +1,117 @@
-"""Monitor queries shared by the monitors pages and the MCP tools.
+"""Monitor queries and validation shared by the monitors pages and the MCP tools.
 
-Each function takes an open session, with no Quart request context, so the
-routes and the /mcp tools (which run outside Quart's pipeline) call the same
-code.
+Each function takes an open session (or plain values), with no Quart request
+context, so the routes and the /mcp tools (which run outside Quart's
+pipeline) call the same code and refuse the same input for the same reasons.
 """
 from __future__ import annotations
 
+import json
+from collections.abc import Mapping
 from datetime import datetime, timezone
 
 from sqlalchemy import func, select
 
 from steward.core.status import _heartbeat_title, _last_n_results, _uptime_by_monitor
 from steward.core.time_range import parse_range
-from steward.models.monitors import Monitor, MonitorResult
+from steward.models.monitors import MONITOR_TYPE_VALUES, Monitor, MonitorResult, MonitorType
+
+HTTP_METHODS = ("GET", "HEAD", "POST")
+# The config keys each type takes (models/monitors.py documents their meaning).
+CONFIG_KEYS: dict[str, tuple[str, ...]] = {
+    MonitorType.icmp.value: (),
+    MonitorType.tcp.value: ("port",),
+    MonitorType.dns.value: ("expected_ip",),
+    MonitorType.http.value: ("method", "expected_status", "content_match", "headers",
+                             "timeout_seconds", "follow_redirects", "verify_ssl"),
+}
+
+
+def _whole(name: str, value, default: int, lo: int, hi: int | None = None) -> int:
+    if value is None or value == "":
+        return default
+    if isinstance(value, bool):
+        raise ValueError(f"{name} must be a whole number, got {value!r}.")
+    try:
+        n = int(value)
+    except (TypeError, ValueError):
+        raise ValueError(f"{name} must be a whole number, got {value!r}.") from None
+    if n < lo or (hi is not None and n > hi):
+        bounds = f"between {lo} and {hi}" if hi is not None else f"at least {lo}"
+        raise ValueError(f"{name} must be {bounds}, got {n}.")
+    return n
+
+
+def _headers(value) -> dict:
+    if value is None or value == "":
+        return {}
+    if isinstance(value, str):
+        try:
+            value = json.loads(value)
+        except ValueError:
+            raise ValueError("headers must be a JSON object of name: value strings.") from None
+    if not isinstance(value, dict) or not all(isinstance(k, str) and isinstance(v, str)
+                                              for k, v in value.items()):
+        raise ValueError("headers must be an object of name: value strings.")
+    return value
+
+
+def normalise_target(mtype: str, target: str) -> str:
+    """For http, default a bare host to https://; others pass through."""
+    target = (target or "").strip()
+    if mtype == MonitorType.http.value and target and not target.startswith(("http://", "https://")):
+        return "https://" + target
+    return target
+
+
+def build_config(mtype: str, opts: Mapping) -> dict:
+    """The type-specific config from already-typed or form-string values.
+
+    Missing keys take their defaults; a key the type does not take, or a
+    value out of range, raises ValueError saying which and why.
+    """
+    unknown = sorted(set(opts) - set(CONFIG_KEYS[mtype]))
+    if unknown:
+        allowed = ", ".join(CONFIG_KEYS[mtype]) or "no config"
+        raise ValueError(f"a {mtype} monitor does not take {', '.join(unknown)}; it takes {allowed}.")
+    if mtype == MonitorType.tcp.value:
+        return {"port": _whole("port", opts.get("port"), 80, 1, 65535)}
+    if mtype == MonitorType.dns.value:
+        return {"expected_ip": (str(opts.get("expected_ip") or "").strip() or None)}
+    if mtype == MonitorType.http.value:
+        method = str(opts.get("method") or "GET").strip().upper()
+        if method not in HTTP_METHODS:
+            raise ValueError(f"method {method!r} is not one of {', '.join(HTTP_METHODS)}.")
+        return {
+            "method": method,
+            "expected_status": _whole("expected_status", opts.get("expected_status"), 200, 100, 599),
+            "content_match": str(opts.get("content_match") or "").strip(),
+            "headers": _headers(opts.get("headers")),
+            "timeout_seconds": _whole("timeout_seconds", opts.get("timeout_seconds"), 10, 1, 60),
+            "follow_redirects": bool(opts.get("follow_redirects", True)),
+            "verify_ssl": bool(opts.get("verify_ssl", True)),
+        }
+    return {}  # icmp
+
+
+def monitor_fields(mtype: str, *, target: str, name: str = "", check_interval_seconds=0,
+                   opts: Mapping | None = None) -> dict:
+    """Validated column values for a new or edited monitor; ValueError says why not."""
+    mtype = (mtype or "").strip().lower()
+    if mtype not in MONITOR_TYPE_VALUES:
+        raise ValueError(f"type {mtype!r} is not one of {', '.join(MONITOR_TYPE_VALUES)}.")
+    target = normalise_target(mtype, target)
+    if not target:
+        raise ValueError("a monitor needs a target: an address, or a URL for http.")
+    if len(target) > 2048:
+        raise ValueError("target is longer than 2048 characters.")
+    return {
+        "type": mtype,
+        "target": target,
+        "name": ((name or "").strip() or target)[:128],
+        "config_json": json.dumps(build_config(mtype, opts or {})),
+        "check_interval_seconds": _whole("check_interval_seconds", check_interval_seconds, 0, 0),
+    }
 
 
 async def monitor_rows(db, range_key: str):

tests/core/test_monitors.py

@@ -8,7 +8,10 @@ import asyncio
 
 from steward.models.monitors import Monitor, MONITOR_TYPE_VALUES
 from steward.monitors import runner as runner_mod
-from steward.monitors.routes import _build_config, _normalise_target
+import pytest
+
+from steward.monitors.routes import _form_opts
+from steward.monitors.service import build_config, monitor_fields, normalise_target
 
 
 def _monitor(mtype, target="example.com", cfg=None):
@@ -85,26 +88,64 @@ def test_run_monitor_dns_passes_expected_ip(monkeypatch):
     assert res["resolved_ip"] == "192.0.2.1"
 
 
-# ── route form helpers ────────────────────────────────────────────────────────
+# ── config validation (shared by the form and the MCP tools) ─────────────────
 
 def test_normalise_target_adds_https_for_bare_http_host():
-    assert _normalise_target("http", "example.com") == "https://example.com"
-    assert _normalise_target("http", "http://x.com") == "http://x.com"
-    assert _normalise_target("tcp", "example.com") == "example.com"
+    assert normalise_target("http", "example.com") == "https://example.com"
+    assert normalise_target("http", "http://x.com") == "http://x.com"
+    assert normalise_target("tcp", "example.com") == "example.com"
 
 
 class _Form(dict):
     """Minimal stand-in for a Quart form (dict + 'in' membership for checkboxes)."""
 
 
-def test_build_config_per_type():
-    assert _build_config(_Form(port="8080"), "tcp") == {"port": 8080}
-    assert _build_config(_Form(), "tcp") == {"port": 80}
-    assert _build_config(_Form(expected_ip="10.0.0.1"), "dns") == {"expected_ip": "10.0.0.1"}
-    assert _build_config(_Form(), "dns") == {"expected_ip": None}
-    assert _build_config(_Form(), "icmp") == {}
+def _from_form(form, mtype):
+    return build_config(mtype, _form_opts(form, mtype))
 
-    http = _build_config(_Form(method="get", expected_status="201", follow_redirects="on"), "http")
+
+def test_build_config_from_the_form_per_type():
+    assert _from_form(_Form(port="8080"), "tcp") == {"port": 8080}
+    assert _from_form(_Form(), "tcp") == {"port": 80}
+    assert _from_form(_Form(expected_ip="10.0.0.1"), "dns") == {"expected_ip": "10.0.0.1"}
+    assert _from_form(_Form(), "dns") == {"expected_ip": None}
+    assert _from_form(_Form(), "icmp") == {}
+
+    http = _from_form(_Form(method="get", expected_status="201", follow_redirects="on"), "http")
     assert http["method"] == "GET" and http["expected_status"] == 201
     assert http["follow_redirects"] is True       # present in form → checked
     assert http["verify_ssl"] is False            # absent from form → unchecked
+
+
+def test_build_config_from_typed_values_defaults_to_safe_checks():
+    http = build_config("http", {"headers": {"X-Key": "v"}})
+    assert http["follow_redirects"] is True and http["verify_ssl"] is True
+    assert http["headers"] == {"X-Key": "v"} and http["timeout_seconds"] == 10
+
+
+@pytest.mark.parametrize("mtype, opts, why", [
+    ("tcp", {"port": 70000}, "port must be between 1 and 65535"),
+    ("tcp", {"port": "http"}, "port must be a whole number"),
+    ("tcp", {"expected_ip": "1.2.3.4"}, "does not take expected_ip; it takes port"),
+    ("icmp", {"port": 22}, "takes no config"),
+    ("http", {"method": "DELETE"}, "not one of GET, HEAD, POST"),
+    ("http", {"expected_status": 42}, "between 100 and 599"),
+    ("http", {"timeout_seconds": 0}, "between 1 and 60"),
+    ("http", {"headers": "{not json"}, "JSON object"),
+    ("http", {"headers": {"X": 1}}, "name: value strings"),
+])
+def test_build_config_refuses_with_a_reason(mtype, opts, why):
+    with pytest.raises(ValueError, match=why):
+        build_config(mtype, opts)
+
+
+def test_monitor_fields():
+    f = monitor_fields("HTTP", target="nas.lan", opts={"expected_status": 204})
+    assert (f["type"], f["target"], f["name"]) == ("http", "https://nas.lan", "https://nas.lan")
+    assert f["check_interval_seconds"] == 0
+    with pytest.raises(ValueError, match="'smtp' is not one of icmp, tcp, dns, http"):
+        monitor_fields("smtp", target="x")
+    with pytest.raises(ValueError, match="needs a target"):
+        monitor_fields("icmp", target="  ")
+    with pytest.raises(ValueError, match="check_interval_seconds must be at least 0"):
+        monitor_fields("icmp", target="x", check_interval_seconds=-5)

tests/integration/test_mcp_monitoring_writes.py

@@ -0,0 +1,146 @@
+"""Integration: the monitoring-config write tools against real Postgres.
+
+A monitor is created on a host, edited (config merged, a rename warning for
+the rule that watches it by name), paused and deleted; a rule is created on
+that monitor, edited, acknowledged once firing, and deleted; a maintenance
+window in progress is ended and an upcoming one cancelled. Every write leaves
+an audit row as "<user> via MCP".
+"""
+from __future__ import annotations
+
+import asyncio
+import os
+import uuid
+
+import pytest
+
+pytestmark = pytest.mark.integration
+
+_NEEDS_DB = pytest.mark.skipif(
+    not os.environ.get("STEWARD_DATABASE_URL"),
+    reason="integration test needs a live Postgres (STEWARD_DATABASE_URL)",
+)
+
+
+@pytest.fixture
+def app():
+    if not os.environ.get("STEWARD_DATABASE_URL"):
+        pytest.skip("needs Postgres")
+    from steward.app import create_app
+    return create_app(testing=False)
+
+
+@_NEEDS_DB
+def test_monitor_rule_ack_and_maintenance_writes(app):
+    from sqlalchemy import select
+    from steward.core.api_tokens import Caller
+    from steward.mcp import _context
+    from steward.mcp.tools import alerts as a
+    from steward.mcp.tools import monitors as m
+    from steward.models.alerts import AlertRule, AlertState, AlertStateEnum, MaintenanceWindow
+    from steward.models.audit import AuditEvent
+    from steward.models.hosts import Host
+    from steward.models.monitors import Monitor
+    from steward.models.users import User, UserRole
+
+    tag = uuid.uuid4().hex[:8]
+    uid, username, host_name = str(uuid.uuid4()), f"mcp-cfg-{tag}", f"cfg-host-{tag}"
+    mon_name, rule_name = f"web-{tag}", f"web-down-{tag}"
+
+    async def _seed():
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                s.add(User(id=uid, username=username, email=f"{tag}@e.test",
+                           password_hash="x", role=UserRole.admin, is_active=True))
+                s.add(Host(id=str(uuid.uuid4()), name=host_name, address="10.7.7.7"))
+
+    async def _actions():
+        async with app.db_sessionmaker() as s:
+            rows = (await s.execute(select(AuditEvent).where(AuditEvent.user_id == uid))).scalars()
+            return [(e.action, e.username) for e in rows]
+
+    async def _go():
+        await _seed()
+        _context._app_ctx.set(app)
+        _context._caller_ctx.set(Caller(user_id=uid, username=username,
+                                        role=UserRole.admin, token_id=7))
+
+        # ── monitors ───────────────────────────────────────────────────────
+        with pytest.raises(ValueError, match="does not take port"):
+            await m.create_monitor(mon_name, "http", "nas.lan", config={"port": 80})
+        with pytest.raises(ValueError, match="no host named"):
+            await m.create_monitor(mon_name, "http", "nas.lan", host=f"ghost-{tag}")
+        made = await m.create_monitor(mon_name, "http", "nas.lan", host=host_name,
+                                      config={"expected_status": 204,
+                                              "headers": {"Authorization": "Bearer s3cret"}})
+        assert made["target"] == "https://nas.lan" and made["host_name"] == host_name
+        assert made["config"]["headers"] == {"Authorization": "<redacted>"}
+        assert made["config"]["verify_ssl"] is True
+
+        # ── rules ──────────────────────────────────────────────────────────
+        opts = await a.get_alert_rule_options("http")
+        assert mon_name in opts["resources"] and "is_up" in opts["metrics"]
+        with pytest.raises(ValueError, match="does not report 'up'"):
+            await a.create_alert_rule(rule_name, "http", mon_name, "up", "<", 1)
+        rule = await a.create_alert_rule(rule_name, "http", mon_name, "is_up", "<", 1,
+                                         consecutive_checks=3)
+        assert rule["condition"] == "is_up < 1 for 3 consecutive checks"
+        rule = await a.update_alert_rule(rule_name, threshold=0.5, enabled=False)
+        assert rule["condition"].startswith("is_up < 0.5") and rule["enabled"] is False
+
+        # Editing the config merges: the stored header survives a timeout change.
+        upd = await m.update_monitor(mon_name, config={"timeout_seconds": 5})
+        assert upd["config"]["timeout_seconds"] == 5 and upd["config"]["expected_status"] == 204
+        async with app.db_sessionmaker() as s:
+            row = await s.get(Monitor, made["monitor_id"])
+        assert row.config["headers"] == {"Authorization": "Bearer s3cret"}
+
+        # A rename leaves the rule on the old name, and says so.
+        renamed = await m.update_monitor(mon_name, name=f"{mon_name}-2")
+        assert rule_name in renamed["warning"]
+        await a.update_alert_rule(rule_name, resource=f"{mon_name}-2")
+        assert "warning" not in await m.update_monitor(f"{mon_name}-2", check_interval_s=30)
+        assert (await m.set_monitor_enabled(f"{mon_name}-2", False))["enabled"] is False
+
+        # ── acknowledge ────────────────────────────────────────────────────
+        with pytest.raises(ValueError, match="not firing"):
+            await a.acknowledge_alert(rule_name)
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                st = (await s.execute(select(AlertState).where(
+                    AlertState.rule_id == rule["rule_id"]))).scalar_one()
+                st.state = AlertStateEnum.firing
+        ack = await a.acknowledge_alert(rule_name)
+        assert ack["state"] == "acknowledged"
+
+        # ── maintenance ────────────────────────────────────────────────────
+        win = await a.create_maintenance_window(f"patch-{tag}", duration_minutes=30,
+                                                source_module="http", resource=f"{mon_name}-2")
+        assert win["active"] is True
+        ended = await a.end_maintenance_window(f"patch-{tag}")
+        assert ended["outcome"] == "ended" and ended["active"] is True
+        with pytest.raises(ValueError, match="already ended"):
+            await a.end_maintenance_window(win["window_id"])
+        later = await a.create_maintenance_window(f"later-{tag}", start_at="2099-01-01T00:00:00Z",
+                                                  end_at="2099-01-01T02:00:00Z")
+        assert (await a.end_maintenance_window(later["window_id"]))["outcome"] == "cancelled"
+        async with app.db_sessionmaker() as s:
+            assert await s.get(MaintenanceWindow, later["window_id"]) is None
+            assert await s.get(MaintenanceWindow, win["window_id"]) is not None
+
+        # ── deletes ────────────────────────────────────────────────────────
+        await a.delete_alert_rule(rule_name)
+        await m.delete_monitor(f"{mon_name}-2")
+        async with app.db_sessionmaker() as s:
+            assert await s.get(AlertRule, rule["rule_id"]) is None
+            assert await s.get(Monitor, made["monitor_id"]) is None
+
+        actions = await _actions()
+        assert all(user == f"{username} via MCP" for _, user in actions)
+        assert sorted({act for act, _ in actions}) == sorted({
+            "monitor.created", "monitor.updated", "monitor.disabled", "monitor.deleted",
+            "alert_rule.created", "alert_rule.updated", "alert_rule.acknowledged",
+            "alert_rule.deleted", "maintenance_window.created", "maintenance_window.ended",
+            "maintenance_window.cancelled"})
+
+    asyncio.run(_go())

tests/mcp/test_monitoring_writes.py

@@ -0,0 +1,97 @@
+"""The monitoring-config write tools: their validation refusals and their roles.
+
+Monitor-config validation is covered in tests/core/test_monitors.py (the form
+and the tools share it). The writes themselves, their audit rows and the
+rename warning run against Postgres in
+tests/integration/test_mcp_monitoring_writes.py.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from steward.alerts import service as alert_svc
+from steward.mcp.tools import alerts
+from steward.models.alerts import AlertOperator
+from steward.models.users import UserRole
+
+NOW = datetime(2026, 10, 6, 12, 0, tzinfo=timezone.utc)
+
+OPERATOR_TOOLS = {"create_monitor", "update_monitor", "set_monitor_enabled", "create_alert_rule",
+                  "update_alert_rule", "acknowledge_alert", "create_maintenance_window",
+                  "end_maintenance_window"}
+ADMIN_TOOLS = {"delete_monitor", "delete_alert_rule"}
+
+
+@pytest.fixture
+def known(monkeypatch):
+    """A source with one resource and its catalog metrics, without a database."""
+    async def resources(db, src):
+        return ["nas"] if src == "host_agent" else []
+    monkeypatch.setattr(alert_svc, "resources_for", resources)
+
+
+async def _fields(**over):
+    args = dict(name="cpu hot", source_module="host_agent", resource_name="nas",
+                metric_name="cpu_pct", operator=">", threshold="90", consecutive_failures_required=2)
+    args.update(over)
+    return await alert_svc.rule_fields(None, **args)
+
+
+async def test_rule_fields_accepts_the_form_choices(known):
+    f = await _fields()
+    assert f["operator"] is AlertOperator.gt and f["threshold"] == 90.0
+    assert f["consecutive_failures_required"] == 2
+
+
+@pytest.mark.parametrize("over, why", [
+    ({"name": " "}, "needs a name"),
+    ({"source_module": "zabbix"}, "'zabbix' is not one of icmp, tcp"),
+    ({"resource_name": "web01"}, "host_agent has no resource 'web01'. Known: nas"),
+    ({"metric_name": "cpu"}, "does not report 'cpu'. Its metrics: cpu_pct"),
+    ({"operator": "=>"}, "'=>' is not one of >, <"),
+    ({"threshold": "high"}, "threshold must be a number"),
+    ({"consecutive_failures_required": 0}, "at least 1"),
+])
+async def test_rule_fields_refuses_with_a_reason(known, over, why):
+    with pytest.raises(ValueError, match=why):
+        await _fields(**over)
+
+
+def test_window_fields():
+    f = alert_svc.window_fields(name="patching", start_at=NOW, end_at=NOW + timedelta(hours=1),
+                                scope_source="", scope_resource="nas")
+    assert f["scope_source"] is None and f["scope_resource"] is None  # resource needs a source
+    with pytest.raises(ValueError, match="must end after it starts"):
+        alert_svc.window_fields(name="x", start_at=NOW, end_at=NOW, scope_source=None,
+                                scope_resource=None)
+    with pytest.raises(ValueError, match="'nope' is not one of"):
+        alert_svc.window_fields(name="x", start_at=NOW, end_at=NOW + timedelta(minutes=1),
+                                scope_source="nope", scope_resource=None)
+
+
+async def test_maintenance_window_needs_exactly_one_end():
+    with pytest.raises(ValueError, match="exactly one of duration_minutes or end_at"):
+        await alerts.create_maintenance_window("x")
+    with pytest.raises(ValueError, match="exactly one of duration_minutes or end_at"):
+        await alerts.create_maintenance_window("x", duration_minutes=30, end_at="2026-10-06T13:00:00Z")
+    with pytest.raises(ValueError, match="not an ISO-8601 time"):
+        await alerts.create_maintenance_window("x", end_at="tonight")
+
+
+def test_when_reads_a_zoneless_time_as_utc():
+    assert alerts._when("2026-10-06T22:00:00", "end_at") == datetime(2026, 10, 6, 22, tzinfo=timezone.utc)
+    assert alerts._when("2026-10-06T22:00:00Z", "end_at").utcoffset() == timedelta(0)
+
+
+def test_write_tools_are_registered_with_their_roles(app):
+    mcp = app.mcp_server
+    for name in OPERATOR_TOOLS:
+        assert mcp.tool_roles[name] is UserRole.operator, name
+        assert mcp._tool_manager.get_tool(name).annotations.read_only_hint is False, name
+    for name in ADMIN_TOOLS:
+        assert mcp.tool_roles[name] is UserRole.admin, name
+    assert mcp.tool_roles["get_alert_rule_options"] is UserRole.viewer
+    for name in OPERATOR_TOOLS | ADMIN_TOOLS | {"get_alert_rule_options"}:
+        assert (mcp._tool_manager.get_tool(name).description or "").strip(), name