diff --git a/steward/alerts/routes.py b/steward/alerts/routes.py
index ffbafa8..3090367 100644
--- a/steward/alerts/routes.py
+++ b/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"))
diff --git a/steward/alerts/service.py b/steward/alerts/service.py
index 2dbe4db..7826cb5 100644
--- a/steward/alerts/service.py
+++ b/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}/*"
diff --git a/steward/mcp/tools/alerts.py b/steward/mcp/tools/alerts.py
index 99c9c3a..da677ba 100644
--- a/steward/mcp/tools/alerts.py
+++ b/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)
diff --git a/steward/mcp/tools/monitors.py b/steward/mcp/tools/monitors.py
index db4376b..525eadf 100644
--- a/steward/mcp/tools/monitors.py
+++ b/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)
diff --git a/steward/monitors/routes.py b/steward/monitors/routes.py
index 3350173..9a3103d 100644
--- a/steward/monitors/routes.py
+++ b/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"))
 
 
diff --git a/steward/monitors/service.py b/steward/monitors/service.py
index e2653e0..9a12531 100644
--- a/steward/monitors/service.py
+++ b/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):
diff --git a/tests/core/test_monitors.py b/tests/core/test_monitors.py
index 71e10d3..3c662c9 100644
--- a/tests/core/test_monitors.py
+++ b/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)
diff --git a/tests/integration/test_mcp_monitoring_writes.py b/tests/integration/test_mcp_monitoring_writes.py
new file mode 100644
index 0000000..0df5778
--- /dev/null
+++ b/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())
diff --git a/tests/mcp/test_monitoring_writes.py b/tests/mcp/test_monitoring_writes.py
new file mode 100644
index 0000000..8fe9916
--- /dev/null
+++ b/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