"""Automation tools: playbooks, inventory, runs, schedules and alert actions. Reads are viewer tools. Anything that starts a run or changes what runs is an operator tool (set_alert_action is admin, as on the alert rule form), and each of those writes an audit row as " via MCP". Playbook runs default to Ansible check mode (`--check --diff`): the agent sees what would change before anything does, and applies only by passing check=false. Variables the playbook marks secret reach the run but are never stored or echoed back. """ from __future__ import annotations import asyncio import os from collections import deque from datetime import timedelta from sqlalchemy import select from sqlalchemy.orm import selectinload from steward.ansible import executor, sources as src_module from steward.ansible.runner import trigger_run from steward.mcp._context import current_caller, mcp_app from steward.models.alerts import AlertRule from steward.models.ansible import AnsibleRun, AnsibleRunStatus from steward.models.ansible_inventory import AnsibleGroup, AnsibleTarget from steward.models.ansible_schedule import AnsibleSchedule from steward.models.hosts import Host from steward.models.users import User, UserRole from ._audit import audit_mcp from ._shape import age_s, bounded, clamp_limit, db_session, iso, lookback, pick ACTIVE = (AnsibleRunStatus.running, AnsibleRunStatus.queued) RECAP_KEYS = ("ok", "changed", "failed", "unreachable", "skipped") CONTENT_CHARS = 20000 LINE_CHARS = 500 WAIT_MAX_S = 120 POLL_S = 2.0 MASK = "" # ── sources and playbooks ──────────────────────────────────────────────────── def _sources() -> list[dict]: return src_module.get_sources(mcp_app().config.get("ANSIBLE", {})) def _source(name: str) -> dict: srcs = _sources() for s in srcs: if s["name"] == (name or "").strip(): return s raise ValueError(f"no playbook source named {name!r}. Sources: " f"{', '.join(s['name'] for s in srcs)} (list_playbooks shows their playbooks).") def _playbook(source: dict, path: str) -> str: """The playbook's content; refuses a path the source does not have.""" path = (path or "").strip() if path not in src_module.discover_playbooks(source["path"]): raise ValueError(f"source {source['name']!r} has no playbook {path!r}; " f'list_playbooks(source="{source["name"]}") lists them.') content = src_module.read_playbook(source["path"], path) if content is None: raise ValueError(f"playbook {path!r} in {source['name']!r} could not be read.") return content def public_variables(variables: list[dict]) -> list[dict]: """A playbook's run-time variables with the defaults of secret ones masked.""" return [{**v, "default": MASK if v["secret"] and v["default"] not in ("", None) else v["default"]} for v in variables] def split_vars(variables: list[dict], extra_vars: dict | None) -> tuple[dict, dict]: """(plain, secret): the values the run stores, and those it must not. Refuses a vars_prompt variable that has no default and no value, since a run cannot stop to ask for it. """ extra_vars = dict(extra_vars or {}) for k, v in extra_vars.items(): if not isinstance(k, str) or not k.strip(): raise ValueError("extra_vars keys must be non-empty variable names.") if not isinstance(v, (str, int, float, bool)): raise ValueError(f"extra_vars[{k!r}] must be a string, number or boolean.") secret_names = {v["name"] for v in variables if v["secret"]} missing = [v["name"] for v in variables if v["required"] and v["name"] not in extra_vars] if missing: raise ValueError(f"this playbook prompts for {', '.join(missing)} and has no default; " "pass a value in extra_vars. Nothing was run.") plain = {k: v for k, v in extra_vars.items() if k not in secret_names} secret = {k: v for k, v in extra_vars.items() if k in secret_names} return plain, secret def build_params(plain: dict, limit: str, tags: str, check: bool) -> dict: params: dict = {} if plain: params["extra_vars_map"] = plain if (limit or "").strip(): params["limit"] = limit.strip() if (tags or "").strip(): params["tags"] = tags.strip() if check: params["check"] = True return params async def _scope(db, source: dict, scope: str) -> tuple[str, str]: """(stored scope, human description) from the forms the docstrings offer.""" s = (scope or "all").strip() if s in ("all", "steward:all"): n = len((await db.execute(select(AnsibleTarget.id))).all()) if not n: raise ValueError("Steward's inventory has no targets yet; add one under Ansible → " "Inventory, or pass scope=\"inventory:\" for a repo inventory.") return "steward:all", f"all {n} inventory targets" kind, _, want = s.removeprefix("steward:").partition(":") if kind == "target": rows = (await db.execute(select(AnsibleTarget))).scalars() t = pick(want, rows, kind="target", list_tool="list_inventory") return f"steward:target:{t.id}", f"target {t.name} ({t.address})" if kind == "group": rows = (await db.execute(select(AnsibleGroup).options(selectinload(AnsibleGroup.targets)))).scalars() g = pick(want, rows, kind="group", list_tool="list_inventory") if not g.targets: raise ValueError(f"group {g.name!r} has no targets. Nothing was run.") return f"steward:group:{g.id}", f"group {g.name} ({len(g.targets)} targets)" if kind in ("inventory", "repo"): inv = want.rsplit(":", 1)[-1] if kind == "repo" else want found = src_module.discover_inventories(source["path"]) if inv not in found: raise ValueError(f"source {source['name']!r} has no inventory file {inv!r}" + (f"; it has {', '.join(found)}." if found else "; it has none.")) return f"repo:{source['name']}:{inv}", f"{source['name']}/{inv}" raise ValueError(f"scope {scope!r} is not one of: all, target:, " "group:, inventory:. list_inventory shows the choices.") async def list_playbooks(source: str = "", category: str = "", limit: int = 200) -> dict: """The playbooks Steward can run, by source, with what each one says it does. source: one source's name; empty for all. "steward-builtin" holds Steward's own maintenance playbooks, "steward-local" those written in Steward's editor; the rest come from configured directories or git repos. category: only playbooks whose `# steward:category:` matches; empty for all. limit: default 200, max 1000. `confirm: true` marks a playbook its author flagged as destructive: run_playbook will apply it only with confirmed=true. get_playbook shows a playbook's variables and content. """ limit = clamp_limit(limit, 200, 1000) srcs = [_source(source)] if source else _sources() rows, out_sources = [], [] for s in srcs: inventories = src_module.discover_inventories(s["path"]) out_sources.append({"source": s["name"], "type": s["type"], "editable": src_module.is_editable_source(s), "inventories": inventories}) for path in src_module.discover_playbooks(s["path"]): meta = src_module.discover_playbook_meta(src_module.read_playbook(s["path"], path) or "") if category and meta["category"].lower() != category.strip().lower(): continue rows.append({"source": s["name"], "playbook": path, **meta}) return {"sources": out_sources, **bounded(rows, limit, key="playbooks", more="Filter by `source` or `category`.")} async def get_playbook(source: str, playbook: str) -> dict: """One playbook: its description, the variables it takes, and its content. `variables` are what run_playbook's extra_vars can set: `required` ones (a vars_prompt with no default) must be given; `secret` ones are passed to the run but never stored, and their defaults are shown as "". `content` is the YAML itself, cut at 20000 characters. """ s = _source(source) content = _playbook(s, playbook) out = { "source": s["name"], "playbook": playbook.strip(), **src_module.discover_playbook_meta(content), "variables": public_variables(src_module.discover_playbook_variables(content)), "content": content[:CONTENT_CHARS], } if len(content) > CONTENT_CHARS: out["content_truncated"] = f"{len(content) - CONTENT_CHARS} more characters not shown." return out async def list_inventory() -> dict: """Steward's Ansible inventory: targets, groups, and how to scope a run to them. Each target names the Steward host it is linked to (prune_docker needs that link), its groups and the names of its Ansible variables — not their values, which can hold credentials. `scopes` lists what run_playbook's `scope` accepts. """ async with db_session() as db: targets = list((await db.execute( select(AnsibleTarget).options(selectinload(AnsibleTarget.groups)) .order_by(AnsibleTarget.name))).scalars()) groups = list((await db.execute( select(AnsibleGroup).options(selectinload(AnsibleGroup.targets)) .order_by(AnsibleGroup.name))).scalars()) host_names = dict((await db.execute(select(Host.id, Host.name))).all()) return { "targets": [ {"target_id": t.id, "name": t.name, "address": t.address, "host_id": t.host_id, "host_name": host_names.get(t.host_id), "groups": sorted(g.name for g in t.groups), "var_names": sorted(t.ansible_vars or {})} for t in targets ], "groups": [ {"group_id": g.id, "name": g.name, "targets": sorted(t.name for t in g.targets), "var_names": sorted(g.ansible_vars or {})} for g in groups ], "repo_inventories": [{"source": s["name"], "files": inv} for s in _sources() if (inv := src_module.discover_inventories(s["path"]))], "scopes": { "all": "every target above", "target:": "one target", "group:": "every target in a group", "inventory:": "an inventory file in the playbook's own source", }, } # ── runs ───────────────────────────────────────────────────────────────────── def recap_totals(results: dict | None) -> dict | None: hosts = (results or {}).get("hosts") or {} if not hosts: return None return {"hosts": len(hosts), **{k: sum(int((h or {}).get(k) or 0) for h in hosts.values()) for k in RECAP_KEYS}} def run_row(r: AnsibleRun, users: dict) -> dict: end = r.finished_at return { "run_id": r.id, "source": r.source_name, "playbook": r.playbook_path, "scope": r.inventory_scope, "status": r.status.value, "check": bool((r.params or {}).get("check")), "triggered_by": users.get(r.triggered_by, "(deleted user)") if r.triggered_by else "system", "started_at": iso(r.started_at), "age_s": age_s(r.started_at), "finished_at": iso(end), "duration_s": int((end - r.started_at).total_seconds()) if end and r.started_at else None, "recap": recap_totals(r.results), } async def _users(db, ids) -> dict[str, str]: ids = {i for i in ids if i} if not ids: return {} return dict((await db.execute(select(User.id, User.username).where(User.id.in_(ids)))).all()) async def list_ansible_runs(status: str = "", playbook: str = "", range: str = "7d", limit: int = 50) -> dict: """Ansible runs, newest first: manual, scheduled, alert-triggered and MCP-started. status: queued, running, success, failed, interrupted or cancelled; empty for all. playbook: a substring of the playbook path; empty for all. range: "1h", "6h", "24h", "7d" (default) or "30d". limit: default 50, max 500. `check: true` is a dry run that changed nothing. `recap` sums the play recap over hosts; get_ansible_run shows per-host results and the log. """ since, key = lookback(range) limit = clamp_limit(limit, 50, 500) stmt = (select(AnsibleRun).where(AnsibleRun.started_at >= since) .order_by(AnsibleRun.started_at.desc()).limit(limit + 1)) if status: try: stmt = stmt.where(AnsibleRun.status == AnsibleRunStatus(status.strip())) except ValueError: raise ValueError(f"status {status!r} is not one of " f"{', '.join(s.value for s in AnsibleRunStatus)}.") from None if playbook: esc = playbook.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") stmt = stmt.where(AnsibleRun.playbook_path.ilike(f"%{esc}%", escape="\\")) async with db_session() as db: runs = list((await db.execute(stmt)).scalars()) users = await _users(db, [r.triggered_by for r in runs]) out = {"range": key, **bounded([run_row(r, users) for r in runs], limit, key="runs")} if out["truncated"]: out["total"] = None out["more"] = (f"More than {limit} runs in range. Pass a larger `limit` (max 500), " "a shorter `range`, or filter by `status` or `playbook`.") return out def _cut(line: str) -> str: return line if len(line) <= LINE_CHARS else line[:LINE_CHARS] + f" [… {len(line) - LINE_CHARS} more]" def read_log(run: AnsibleRun, tail: int, grep: str, max_matches: int = 100) -> dict: """The last `tail` lines of a run's log and the lines containing `grep`.""" path = executor.artifact_path(run.id) if os.path.exists(path): with open(path, encoding="utf-8", errors="replace") as f: lines = f.read().splitlines() else: lines = (run.output or "").splitlines() out: dict = {"log_lines": len(lines), "tail": [_cut(x) for x in deque(lines, maxlen=tail)]} if grep: needle = grep.lower() hits = [{"line_no": i + 1, "line": _cut(x)} for i, x in enumerate(lines) if needle in x.lower()] out["matches"] = hits[:max_matches] out["matches_total"] = len(hits) return out async def _load_run(db, run_id: str) -> AnsibleRun: run = await db.get(AnsibleRun, (run_id or "").strip()) if run is None: raise ValueError(f"no Ansible run with id {run_id!r}; list_ansible_runs lists them.") return run async def _run_detail(run_id: str, tail: int, grep: str) -> dict: async with db_session() as db: run = await _load_run(db, run_id) users = await _users(db, [run.triggered_by]) params = dict(run.params or {}) return { **run_row(run, users), "limit": params.get("limit"), "tags": params.get("tags"), "extra_vars": params.get("extra_vars_map") or {}, "hosts": (run.results or {}).get("hosts") or {}, "failures": ((run.results or {}).get("failures") or [])[:20], **read_log(run, tail, grep), } async def get_ansible_run(run_id: str, tail: int = 200, grep: str = "") -> dict: """One Ansible run: its parameters, per-host recap, failures and log. tail: how many of the log's last lines to return; default 200, max 2000. grep: also return every log line containing this text (case-insensitive), with its line number, up to 100. `failures` are the task failures Steward parsed out of the log; start there when status is failed. Secret variables were never stored, so they do not appear in `extra_vars`. """ return await _run_detail(run_id, clamp_limit(tail, 200, 2000), (grep or "").strip()) async def wait_for_run(run_id: str, timeout_s: int = 60) -> dict: """Wait for a run to finish, then return it as get_ansible_run does (last 50 lines). timeout_s: how long to wait; default 60, max 120. If the run is still going when it runs out, `finished` is false: call again to keep waiting. """ deadline = asyncio.get_running_loop().time() + max(1, min(int(timeout_s or 60), WAIT_MAX_S)) while True: async with db_session() as db: status = (await _load_run(db, run_id)).status if status not in ACTIVE or asyncio.get_running_loop().time() >= deadline: break await asyncio.sleep(POLL_S) out = await _run_detail(run_id, 50, "") out["finished"] = status not in ACTIVE return out async def run_playbook(source: str, playbook: str, scope: str = "all", extra_vars: dict | None = None, limit: str = "", tags: str = "", check: bool = True, confirmed: bool = False) -> dict: """Start an Ansible playbook run. A dry run (check mode) unless check=false. source, playbook: as list_playbooks shows them. scope: which hosts — "all" (default), "target:", "group:", or "inventory:" for an inventory file in the same source. list_inventory shows the choices. extra_vars: {"name": value}. Variables the playbook marks secret are passed to the run but never stored or returned. limit, tags: Ansible's --limit and --tags; empty for none. check: true (default) runs with --check --diff, which reports what would change and changes nothing. Pass false to apply — after a check run looked right. confirmed: a playbook with `confirm: true` (destructive, by its author's flag) applies only when this is also true. Returns at once with `run_id`; wait_for_run follows it to the end. Recorded in the audit log as " via MCP". """ s = _source(source) content = _playbook(s, playbook) meta = src_module.discover_playbook_meta(content) if not check and meta["confirm"] and not confirmed: raise ValueError(f"{playbook} is flagged destructive by its author (steward:confirm). " "Run it with check=true first; to apply, pass check=false and " "confirmed=true. Nothing was run.") plain, secret = split_vars(src_module.discover_playbook_variables(content), extra_vars) async with db_session() as db: stored_scope, scope_desc = await _scope(db, s, scope) params = build_params(plain, limit, tags, check) caller = current_caller() run, _src, err = await trigger_run( mcp_app(), source_name=s["name"], playbook_path=playbook.strip(), inventory_scope=stored_scope, params=params or None, secret_vars=secret or None, triggered_by=caller.user_id) if err: raise ValueError(f"the run could not start: {err}.") await audit_mcp("ansible.run.started", "ansible_run", run.id, { "source": s["name"], "playbook": playbook.strip(), "scope": stored_scope, "check": check, "limit": params.get("limit"), "tags": params.get("tags"), "extra_vars": sorted(plain), "secret_vars": sorted(secret)}) return { "run_id": run.id, "check": check, "scope": scope_desc, "status": run.status.value, "secret_vars_passed": sorted(secret), "summary": (f"{'Check-mode (dry) run' if check else 'Run'} of {playbook.strip()} " f"on {scope_desc} started." + (" It changes nothing; pass check=false to apply." if check else "")), "next": f'wait_for_run(run_id="{run.id}")', } async def cancel_ansible_run(run_id: str) -> dict: """Stop a queued or running Ansible run. Recorded in the audit log.""" async with db_session() as db: run = await _load_run(db, run_id) if run.status not in ACTIVE: raise ValueError(f"run {run.id} already ended ({run.status.value}); nothing to cancel.") executor.cancel_run(run.id) await audit_mcp("ansible.run.cancelled", "ansible_run", run.id, {"playbook": run.playbook_path}) return {"run_id": run.id, "cancelling": True, "next": f'wait_for_run(run_id="{run.id}", timeout_s=30)'} # ── schedules ──────────────────────────────────────────────────────────────── def schedule_row(sc: AnsibleSchedule, last: AnsibleRun | None) -> dict: params = sc.params or {} nxt = sc.last_run_at + timedelta(seconds=sc.interval_seconds) if sc.last_run_at else None return { "schedule_id": sc.id, "name": sc.name, "source": sc.source_name, "playbook": sc.playbook_path, "scope": sc.inventory_scope, "interval_s": sc.interval_seconds, "enabled": sc.enabled, "check": bool(params.get("check")), "extra_vars": params.get("extra_vars_map") or {}, "limit": params.get("limit"), "tags": params.get("tags"), "last_run_at": iso(sc.last_run_at), "last_run_id": sc.last_run_id, "last_run_status": last.status.value if last else None, "last_error": sc.last_error, "next_run_at": iso(nxt) if sc.enabled else None, } async def _schedules(db) -> list[AnsibleSchedule]: return list((await db.execute(select(AnsibleSchedule).order_by(AnsibleSchedule.name))).scalars()) async def list_schedules() -> dict: """Recurring playbook runs: what runs, how often, and how its last run went. `check: true` schedules are dry runs that never change anything. `last_error` is set when a firing could not start at all (a source or playbook that went away); a run that started and failed shows in `last_run_status`, and get_ansible_run(last_run_id) has its log. """ async with db_session() as db: scheds = await _schedules(db) ids = [s.last_run_id for s in scheds if s.last_run_id] runs = {r.id: r for r in (await db.execute( select(AnsibleRun).where(AnsibleRun.id.in_(ids)))).scalars()} if ids else {} rows = [schedule_row(s, runs.get(s.last_run_id)) for s in scheds] return {"schedules": rows, "total": len(rows)} async def _schedule_fields(db, source: str, playbook: str, scope: str, extra_vars: dict | None, limit: str, tags: str, check: bool) -> dict: s = _source(source) content = _playbook(s, playbook) plain, secret = split_vars(src_module.discover_playbook_variables(content), extra_vars) if secret: raise ValueError(f"{', '.join(sorted(secret))} {'is' if len(secret) == 1 else 'are'} secret " "and a schedule would have to store it. Put it in the target's or group's " "Ansible variables instead. Nothing was saved.") stored_scope, _desc = await _scope(db, s, scope) return {"source_name": s["name"], "playbook_path": playbook.strip(), "inventory_scope": stored_scope, "params": build_params(plain, limit, tags, check) or None} def _interval(interval_s: int) -> int: if not isinstance(interval_s, int) or interval_s <= 0: raise ValueError("interval_s must be a positive number of seconds, e.g. 86400 for daily.") return interval_s async def create_schedule(name: str, source: str, playbook: str, interval_s: int, scope: str = "all", extra_vars: dict | None = None, limit: str = "", tags: str = "", check: bool = True, enabled: bool = True) -> dict: """Run a playbook on a recurring interval. Each firing is a "system" run. interval_s: seconds between runs, e.g. 3600 hourly, 86400 daily. scope, extra_vars, limit, tags: as run_playbook. Secret variables cannot be scheduled (they would have to be stored); set them as inventory vars. check: true (default) schedules dry runs. Pass false only for a playbook already proven by a check run. Recorded in the audit log as " via MCP". """ if not (name or "").strip(): raise ValueError("a schedule needs a name.") async with db_session() as db, db.begin(): fields = await _schedule_fields(db, source, playbook, scope, extra_vars, limit, tags, check) sc = AnsibleSchedule(name=name.strip()[:128], interval_seconds=_interval(interval_s), enabled=enabled, **fields) db.add(sc) await audit_mcp("ansible.schedule.created", "ansible_schedule", sc.id, {"name": sc.name, "playbook": sc.playbook_path, "interval_s": sc.interval_seconds, "check": check}) return schedule_row(sc, None) async def update_schedule(schedule: str, name: str | None = None, source: str | None = None, playbook: str | None = None, interval_s: int | None = None, scope: str | None = None, extra_vars: dict | None = None, limit: str | None = None, tags: str | None = None, check: bool | None = None, enabled: bool | None = None) -> dict: """Change a schedule. Only the arguments you pass change; the rest stay. schedule: its name or id (list_schedules). Pass enabled=false to pause it. extra_vars, when passed, replaces the whole set. Recorded in the audit log. """ async with db_session() as db, db.begin(): sc = pick(schedule, await _schedules(db), kind="schedule", list_tool="list_schedules") params = dict(sc.params or {}) cur_scope = sc.inventory_scope if cur_scope.startswith("repo:"): cur_scope = "inventory:" + cur_scope.rsplit(":", 1)[-1] fields = await _schedule_fields( db, source if source is not None else sc.source_name, playbook if playbook is not None else sc.playbook_path, scope if scope is not None else cur_scope, extra_vars if extra_vars is not None else params.get("extra_vars_map"), limit if limit is not None else params.get("limit", ""), tags if tags is not None else params.get("tags", ""), check if check is not None else bool(params.get("check"))) for k, v in fields.items(): setattr(sc, k, v) if name is not None and name.strip(): sc.name = name.strip()[:128] if interval_s is not None: sc.interval_seconds = _interval(interval_s) if enabled is not None: sc.enabled = enabled row = schedule_row(sc, None) changed = sorted(k for k, v in {"name": name, "source": source, "playbook": playbook, "interval_s": interval_s, "scope": scope, "extra_vars": extra_vars, "limit": limit, "tags": tags, "check": check, "enabled": enabled}.items() if v is not None) await audit_mcp("ansible.schedule.updated", "ansible_schedule", row["schedule_id"], {"name": row["name"], "changed": changed}) return row async def delete_schedule(schedule: str) -> dict: """Delete a schedule (its past runs stay). Recorded in the audit log. schedule: its name or id. To pause instead, update_schedule(enabled=false). """ async with db_session() as db, db.begin(): sc = pick(schedule, await _schedules(db), kind="schedule", list_tool="list_schedules") await db.delete(sc) await audit_mcp("ansible.schedule.deleted", "ansible_schedule", sc.id, {"name": sc.name, "playbook": sc.playbook_path}) return {"deleted": sc.name, "schedule_id": sc.id} # ── alert actions ──────────────────────────────────────────────────────────── async def set_alert_action(rule: str, source: str = "", playbook: str = "", inventory: str = "", extra_vars: dict | None = None, limit: str = "", tags: str = "", check: bool = True) -> dict: """Set, replace or clear the playbook an alert rule runs when it starts firing. rule: the alert rule's name or id (list_alert_rules). playbook: empty clears the rule's action. Otherwise source, playbook and inventory (an inventory file in that source, as list_playbooks shows) are all required. extra_vars: {"name": value}, stored on the rule. The run also receives the alert's own context variables. Secret variables cannot be stored here. check: true (default) makes the action a dry run each time it fires; pass false for a remediation already proven by a check run. Admin only, as on the alert rule form. Recorded in the audit log. """ async with db_session() as db, db.begin(): r = pick(rule, (await db.execute(select(AlertRule))).scalars(), kind="alert rule", list_tool="list_alert_rules") action = None if (playbook or "").strip(): s = _source(source) content = _playbook(s, playbook) inv = (inventory or "").strip() found = src_module.discover_inventories(s["path"]) if inv not in found: raise ValueError(f"source {s['name']!r} has no inventory file {inv!r}" + (f"; it has {', '.join(found)}." if found else "; it has none.") + " Nothing was saved.") plain, secret = split_vars(src_module.discover_playbook_variables(content), extra_vars) if secret: raise ValueError(f"{', '.join(sorted(secret))} is secret and an alert action would " "have to store it. Nothing was saved.") action = {"source": s["name"], "playbook": playbook.strip(), "inventory": inv} if plain: action["extra_vars"] = [f"{k}={v}" for k, v in plain.items()] if (limit or "").strip(): action["limit"] = limit.strip() if (tags or "").strip(): action["tags"] = tags.strip() if check: action["check"] = True r.ansible_action = action name, rid = r.name, r.id await audit_mcp("alert_rule.action_set" if action else "alert_rule.action_cleared", "alert_rule", name, {"rule_id": rid, **({k: v for k, v in action.items() if k != "extra_vars"} if action else {})}) return {"rule_id": rid, "rule_name": name, "action": action, "summary": (f"When {name} starts firing it runs {action['playbook']}" f"{' in check mode' if action.get('check') else ''}." if action else f"{name} no longer runs a playbook when it fires.")} def register(register_tool) -> None: for fn in (list_playbooks, get_playbook, list_inventory, list_ansible_runs, get_ansible_run, list_schedules, wait_for_run): register_tool(fn, role=UserRole.viewer, read_only=True) for fn in (run_playbook, cancel_ansible_run, create_schedule, update_schedule, delete_schedule): register_tool(fn, role=UserRole.operator, read_only=False) register_tool(set_alert_action, role=UserRole.admin, read_only=False)