feat(mcp): write, test and delete this instance's own playbooks (#5748)
CI / lint (push) Successful in 2s
CI / integration (push) Successful in 1m29s
CI / unit (push) Successful in 1m39s
CI / publish (push) Successful in 1m6s

#5748 "MCP can write, edit and test this instance's own playbooks
(steward-local)": two admin tools beside run_playbook, so an agent can
build automation for these hosts that isn't part of Steward itself.

- save_playbook(playbook, content, source="steward-local", replace=false)
  goes through the editor's own validate_playbook_yaml and
  write_playbook. A playbook that isn't a list of plays comes back with
  the YAML error and its line, and nothing is written. An existing file
  is overwritten only with replace=true. steward-builtin is refused
  (Steward's playbooks change in its repository), as are git sources
  (their repo owns them).
- After writing, `ansible-playbook --syntax-check` runs against a
  throwaway localhost inventory (no host contacted, 60 s), so a module
  name Ansible can't resolve shows before any run. The result rides the
  reply; the file is kept either way.
- delete_playbook(playbook, source="steward-local").
- Both audit as ansible.playbook.created/updated/deleted, "<user> via
  MCP", with the size and never the content.

The testing loop is the existing one: run_playbook defaults to check
mode, on one target, then get_ansible_run. The integration test does
that loop with the real ansible-playbook.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-15 12:01:39 -04:00
co-authored by Claude Opus 5.5
parent 39685585cd
commit e122860cdc
6 changed files with 299 additions and 6 deletions
+102 -2
View File
@@ -25,6 +25,7 @@ VIEWER_TOOLS = {"list_playbooks", "get_playbook", "list_inventory", "list_ansibl
"get_ansible_run", "list_schedules", "wait_for_run"}
OPERATOR_TOOLS = {"run_playbook", "cancel_ansible_run", "create_schedule", "update_schedule",
"delete_schedule", "update_group_members"}
ADMIN_TOOLS = {"set_alert_action", "save_playbook", "delete_playbook"}
def test_split_vars_keeps_secrets_apart_from_what_is_stored():
@@ -96,8 +97,10 @@ def test_automation_tools_are_registered_with_their_roles(app):
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
assert mcp.tool_roles["set_alert_action"] is UserRole.admin
for name in VIEWER_TOOLS | OPERATOR_TOOLS | {"set_alert_action"}:
for name in ADMIN_TOOLS:
assert mcp.tool_roles[name] is UserRole.admin, name
assert mcp._tool_manager.get_tool(name).annotations.read_only_hint is False, name
for name in VIEWER_TOOLS | OPERATOR_TOOLS | ADMIN_TOOLS:
assert (mcp._tool_manager.get_tool(name).description or "").strip(), name
@@ -107,6 +110,9 @@ def test_a_viewer_token_cannot_start_a_run(app):
assert "needs the operator role" in app.mcp_server.refusal("run_playbook", viewer)
assert app.mcp_server.refusal("run_playbook", operator) is None
assert "needs the admin role" in app.mcp_server.refusal("set_alert_action", operator)
for name in ("save_playbook", "delete_playbook"):
assert "needs the admin role" in app.mcp_server.refusal(name, operator)
assert "needs the admin role" in app.mcp_server.refusal(name, viewer)
def test_builtin_playbooks_get_steward_url_from_steward():
@@ -168,3 +174,97 @@ def test_group_members_are_named_by_target_or_by_linked_host():
assert [t.name for t in got] == ["web01", "db01"]
with pytest.raises(ValueError, match="no target or linked host named 'nas'.*Nothing was changed"):
automation.member_targets(["web01", "nas"], [web, db], hosts)
# ── writing this instance's own playbooks (#5748) ───────────────────────────
PLAY = "- hosts: all\n gather_facts: false\n tasks:\n - ansible.builtin.ping:\n"
@pytest.fixture
def writable(monkeypatch, tmp_path):
"""steward-local in a temp folder beside the read-only sources, with the
audit rows and the syntax check captured."""
from steward.ansible import sources as src_module
local = tmp_path / "local"
srcs = [{"name": src_module.BUILTIN_SOURCE_NAME, "type": "local", "path": str(tmp_path / "builtin")},
{"name": src_module.LOCAL_SOURCE_NAME, "type": "local", "path": str(local)},
{"name": "infra", "type": "git", "path": str(tmp_path / "git")}]
monkeypatch.setattr(automation, "_sources", lambda: srcs)
seen = {"audits": [], "checked": []}
async def audit(action, entity_type=None, entity_id=None, detail=None):
seen["audits"].append((action, entity_id, detail))
async def syntax(path, rel):
seen["checked"].append(rel)
return seen.get("syntax", (True, ""))
monkeypatch.setattr(automation, "audit_mcp", audit)
monkeypatch.setattr(automation.executor, "syntax_check", syntax)
seen["dir"] = local
return seen
async def test_a_new_playbook_is_written_checked_and_audited_without_its_content(writable):
out = await automation.save_playbook("upkeep/ping.yml", PLAY)
assert (writable["dir"] / "upkeep/ping.yml").read_text() == PLAY
assert out["source"] == "steward-local" and out["replaced"] is False
assert out["syntax"] == {"ok": True} and writable["checked"] == ["upkeep/ping.yml"]
[(action, entity, detail)] = writable["audits"]
assert action == "ansible.playbook.created" and entity == "steward-local:upkeep/ping.yml"
assert detail == {"source": "steward-local", "playbook": "upkeep/ping.yml", "bytes": len(PLAY)}
assert PLAY not in str(writable["audits"])
async def test_an_existing_playbook_is_replaced_only_when_asked(writable):
await automation.save_playbook("ping.yml", PLAY)
with pytest.raises(ValueError, match="already has 'ping.yml'.*replace=true.*Nothing was saved"):
await automation.save_playbook("ping.yml", "- hosts: web\n tasks: []\n")
assert (writable["dir"] / "ping.yml").read_text() == PLAY
out = await automation.save_playbook("ping.yml", "- hosts: web\n tasks: []\n", replace=True)
assert out["replaced"] is True
assert [a for a, _, _ in writable["audits"]] == ["ansible.playbook.created", "ansible.playbook.updated"]
@pytest.mark.parametrize("content, why", [
("- hosts: all\n tasks: [unclosed\n", "(?s)YAML error.*line \\d"),
("hosts: all\n", "list of plays"),
("", "empty"),
])
async def test_a_playbook_that_isnt_one_says_why_and_writes_nothing(writable, content, why):
with pytest.raises(ValueError, match=why):
await automation.save_playbook("bad.yml", content)
assert not (writable["dir"] / "bad.yml").exists()
assert writable["audits"] == [] and writable["checked"] == []
async def test_stewards_own_and_git_sources_are_refused(writable):
with pytest.raises(ValueError, match="Steward's own playbooks.*repository.*steward-local"):
await automation.save_playbook("x.yml", PLAY, source="steward-builtin")
with pytest.raises(ValueError, match="git repository"):
await automation.save_playbook("x.yml", PLAY, source="infra")
with pytest.raises(ValueError, match="Steward's own playbooks"):
await automation.delete_playbook("x.yml", source="steward-builtin")
for bad in ("../escape.yml", "notes.txt"):
with pytest.raises(ValueError, match="Nothing was saved"):
await automation.save_playbook(bad, PLAY)
assert writable["audits"] == []
async def test_a_failed_syntax_check_is_reported_and_the_file_kept(writable):
writable["syntax"] = (False, "ERROR! couldn't resolve module/action 'ansible.builtin.pingg'")
out = await automation.save_playbook("ping.yml", PLAY)
assert out["syntax"] == {"ok": False, "output": writable["syntax"][1]}
assert (writable["dir"] / "ping.yml").exists()
writable["syntax"] = (None, "ansible-playbook isn't installed here, so only the YAML was checked.")
out = await automation.save_playbook("ping.yml", PLAY, replace=True)
assert out["syntax"] == {"ok": None, "note": writable["syntax"][1]}
async def test_a_playbook_is_deleted_and_audited(writable):
await automation.save_playbook("ping.yml", PLAY)
assert await automation.delete_playbook("ping.yml") == {"deleted": "ping.yml", "source": "steward-local"}
assert not (writable["dir"] / "ping.yml").exists()
assert writable["audits"][-1][:2] == ("ansible.playbook.deleted", "steward-local:ping.yml")
with pytest.raises(ValueError, match="has no playbook 'ping.yml'"):
await automation.delete_playbook("ping.yml")