"""Install bundled skills and wire the MCP server into an agent's config.
Framework-agnostic helpers used by the ``devops-utils setup`` CLI. Kept out of
the Click layer so the copy/merge logic stays importable and testable on its own
(mirrors how :mod:`devops_utils.agent.tools` stays free of any framework import).
The bundled skill markdown lives at ``devops_utils/agent/skills/`` — either a
flat ``*.md`` or a ``<dir>/SKILL.md`` with reference files beside it — and is
read through :mod:`importlib.resources`, so it resolves the same way from a
source checkout and from an installed wheel.
"""
from __future__ import annotations
import json
from collections.abc import Callable
from dataclasses import dataclass
from importlib.resources import files
from pathlib import Path
#: Name under which the MCP server is registered in an agent's config.
MCP_SERVER_NAME = "devops-utils"
#: Console script that launches the stdio MCP server (see ``[project.scripts]``).
MCP_COMMAND = "devops-utils-mcp"
#: Claude Code plugin name. The harness namespaces a plugin's skills/agents as
#: ``<plugin-name>:<component>``, so with this name the bundled skills/agents list
#: as ``devops-utils:azure-devops-research``, ``devops-utils:azdo-workitem-analyst``,
#: etc. — distinguishable at a glance from unrelated skills/agents.
PLUGIN_NAME = "devops-utils"
#: Marketplace name users add before installing the plugin
#: (``/plugin install devops-utils@<MARKETPLACE_NAME>``).
MARKETPLACE_NAME = "carneirofc"
[docs]
@dataclass(frozen=True)
class OverwriteRequest:
"""One pending overwrite, handed to a :data:`ConfirmOverwrite` callback.
Attributes:
path: File that would be written.
new_text: Content that would replace what is on disk.
existing_text: Content currently on disk.
label: Human-facing name for the item (usually ``str(path)``, but for a
merged JSON config it names the *entry*, not just the file).
"""
path: Path
new_text: str
existing_text: str
label: str
#: Callback deciding whether an existing target may be overwritten. Returning
#: ``True`` writes the new content; ``False`` leaves the file untouched. The
#: install helpers stay UI-free — the Click layer supplies the prompt.
ConfirmOverwrite = Callable[[OverwriteRequest], bool]
def _skills_resource():
"""Return the ``importlib.resources`` traversable for the skills package."""
return files("devops_utils.agent.skills")
def _skill_name(text: str, fallback: str) -> str:
"""Extract the ``name:`` field from a skill's YAML frontmatter.
Args:
text: Full markdown text of the skill (may start with a ``---`` block).
fallback: Name to use when no frontmatter ``name:`` is present.
Returns:
The frontmatter ``name`` value, or ``fallback``.
"""
lines = text.splitlines()
if not lines or lines[0].strip() != "---":
return fallback
for line in lines[1:]:
stripped = line.strip()
if stripped == "---":
break
if stripped.startswith("name:"):
value = stripped[len("name:") :].strip().strip("\"'")
if value:
return value
return fallback
[docs]
@dataclass(frozen=True)
class BundledSkill:
"""One skill bundled with the package.
Attributes:
name: Frontmatter ``name:`` (falls back to the file/dir stem).
source: Bundled entry name — ``foo.md`` for a single-file skill, ``foo``
for a directory skill.
files: ``{relative_path: text}``. A single-file skill has one entry,
``SKILL.md``; a directory skill carries ``SKILL.md`` plus its
reference files (e.g. ``reference/operations.md``), relative paths
always ``/``-separated.
"""
name: str
source: str
files: dict[str, str]
@property
def text(self) -> str:
"""The ``SKILL.md`` body (the part loaded when the skill triggers)."""
return self.files["SKILL.md"]
def _read_skill_dir(entry, prefix: str = "") -> dict[str, str]:
"""Collect every ``.md`` file under a skill directory, recursively."""
found: dict[str, str] = {}
for child in entry.iterdir():
rel = f"{prefix}{child.name}"
if child.is_dir():
if child.name != "__pycache__":
found.update(_read_skill_dir(child, f"{rel}/"))
elif child.name.endswith(".md"):
found[rel] = child.read_text(encoding="utf-8")
return found
[docs]
def iter_bundled_skills() -> list[BundledSkill]:
"""List the skills bundled with the package, sorted by source name.
A skill is either a flat ``<stem>.md`` or a ``<stem>/`` directory holding
``SKILL.md`` plus reference files it links to — the progressive-disclosure
shape for skills too long to load in one go.
"""
skills: list[BundledSkill] = []
for entry in _skills_resource().iterdir():
if entry.is_dir():
files_ = _read_skill_dir(entry)
if "SKILL.md" not in files_:
continue
name = _skill_name(files_["SKILL.md"], entry.name)
elif entry.name.endswith(".md"):
files_ = {"SKILL.md": entry.read_text(encoding="utf-8")}
name = _skill_name(files_["SKILL.md"], entry.name[: -len(".md")])
else:
continue
skills.append(BundledSkill(name, entry.name, files_))
return sorted(skills, key=lambda s: s.source)
def _write(
path: Path,
text: str,
force: bool,
confirm: ConfirmOverwrite | None = None,
) -> Path | None:
"""Write ``text`` to ``path``, returning the path or ``None`` if skipped.
An existing target is skipped unless ``force`` is set or ``confirm`` says
otherwise. ``errors="replace"`` on the read keeps an odd-encoding file from
crashing the callback before the user can decide what to do with it.
"""
if path.exists() and not force:
if confirm is None:
return None
existing = path.read_text(encoding="utf-8", errors="replace")
if not confirm(OverwriteRequest(path, text, existing, str(path))):
return None
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(text, encoding="utf-8")
return path
[docs]
def install_skills(
dest: Path,
layout: str = "claude",
force: bool = False,
confirm: ConfirmOverwrite | None = None,
) -> tuple[list[Path], list[Path]]:
"""Copy the bundled skills into ``dest``.
Args:
dest: Base directory to install into.
layout: ``"claude"`` writes ``dest/skills/<name>/SKILL.md`` (Claude Code
discovery layout) plus any reference files beside it; ``"flat"``
writes a single-file skill as ``dest/<filename>`` and a directory
skill as ``dest/<dir>/...``.
force: Overwrite existing files instead of skipping them.
confirm: Asked per existing file when ``force`` is not set; ``None``
skips existing files outright.
Returns:
A ``(written, skipped)`` tuple of destination paths.
"""
if layout not in ("claude", "flat"):
raise ValueError(f"unknown layout: {layout!r}")
written: list[Path] = []
skipped: list[Path] = []
for skill in iter_bundled_skills():
for rel, text in sorted(skill.files.items()):
if layout == "claude":
target = dest / "skills" / skill.name / Path(rel)
elif len(skill.files) == 1:
target = dest / skill.source
else:
target = dest / skill.source / Path(rel)
result = _write(target, text, force, confirm)
(written if result is not None else skipped).append(target)
return written, skipped
def _agents_resource():
"""Return the ``importlib.resources`` traversable for the agents package."""
return files("devops_utils.agent.agents")
[docs]
def iter_bundled_agents() -> list[tuple[str, str, str]]:
"""List the Claude Code agent files bundled with the package.
Returns:
A list of ``(agent_name, filename, text)`` tuples, sorted by filename.
``agent_name`` comes from the frontmatter ``name:`` (falling back to
the file stem).
"""
agents: list[tuple[str, str, str]] = []
for entry in _agents_resource().iterdir():
if not entry.name.endswith(".md"):
continue
text = entry.read_text(encoding="utf-8")
stem = entry.name[: -len(".md")]
agents.append((_skill_name(text, stem), entry.name, text))
return sorted(agents, key=lambda a: a[1])
[docs]
def install_agents(
dest: Path,
force: bool = False,
confirm: ConfirmOverwrite | None = None,
) -> tuple[list[Path], list[Path]]:
"""Copy the bundled Claude Code agents into ``dest/agents/<name>.md``.
Claude Code discovers subagents as single ``.md`` files in an ``agents/``
directory (user scope ``~/.claude/agents``, project scope ``.claude/agents``).
Args:
dest: Base directory to install into (e.g. ``~/.claude``).
force: Overwrite existing files instead of skipping them.
confirm: Asked per existing file when ``force`` is not set; ``None``
skips existing files outright.
Returns:
A ``(written, skipped)`` tuple of destination paths.
"""
written: list[Path] = []
skipped: list[Path] = []
for name, _filename, text in iter_bundled_agents():
target = dest / "agents" / f"{name}.md"
result = _write(target, text, force, confirm)
(written if result is not None else skipped).append(target)
return written, skipped
def _trackers_resource():
"""Return the ``importlib.resources`` traversable for the tracker templates."""
return files("devops_utils.agent.trackers")
def _tracker_substitutions(
project_name: str,
done_state: str,
org_url: str | None,
parent_epic: int | None,
area_path: str | None,
default_tags: list[str] | None,
) -> dict[str, str]:
"""Compute the placeholder → value map for the tracker templates.
Display placeholders (``{org_url}``, ``{parent_epic}``, ``{area_path}``,
``{default_tags}``) always render something readable, falling back to an
instruction when the setting was not configured. The flag placeholders
(``{create_flags}``, ``{query_flags}``) expand to ready-to-paste CLI
fragments (empty when nothing was configured) so command examples in the
template stay copy-pasteable.
"""
tags = default_tags or []
create_flags = ""
query_flags = ""
if parent_epic is not None:
create_flags += f" --parent {parent_epic}"
if area_path:
create_flags += f" --area-path '{area_path}'"
query_flags += f" --area-path '{area_path}'"
for tag in tags:
create_flags += f" --tag '{tag}'"
query_flags += f" --tag '{tag}'"
return {
"{project}": project_name,
"{done_state}": done_state,
"{org_url}": org_url or "set via `AZURE_DEVOPS_ORG_URL`",
"{parent_epic}": (
f"#{parent_epic}"
if parent_epic is not None
else "(none — find or create the proper Epic/Feature first)"
),
"{parent_epic_id}": str(parent_epic)
if parent_epic is not None
else "<epic-id>",
"{area_path}": area_path or "(project root)",
"{default_tags}": ", ".join(tags) if tags else "(none)",
"{create_flags}": create_flags,
"{query_flags}": query_flags,
}
[docs]
def install_tracker(
dest: Path,
project_name: str,
done_state: str = "Closed",
org_url: str | None = None,
parent_epic: int | None = None,
area_path: str | None = None,
default_tags: list[str] | None = None,
force: bool = False,
confirm: ConfirmOverwrite | None = None,
) -> tuple[list[Path], list[Path]]:
"""Write the Azure DevOps tracker config for mattpocock-style skills.
Renders the bundled templates (``agent/trackers/*.md``) into
``dest/docs/agents/``, substituting the placeholders computed by
:func:`_tracker_substitutions`. The resulting ``issue-tracker.md`` /
``triage-labels.md`` are the config files skills like mattpocock/skills
read to learn how to talk to the issue tracker.
Args:
dest: Repository root to install into (files go to ``docs/agents/``).
project_name: Azure DevOps team project name.
done_state: ``System.State`` value that means "closed" in the project's
process template (e.g. ``Closed``, ``Done``, ``Resolved``).
org_url: Azure DevOps organization/collection URL recorded in the
defaults table (the env var stays authoritative at runtime).
parent_epic: Work-item id of the Epic new items are parented under.
area_path: Default ``System.AreaPath`` for creates and queries.
default_tags: Tags applied to every created item and used to scope
queries.
force: Overwrite existing files instead of skipping them.
confirm: Asked per existing file when ``force`` is not set; ``None``
skips existing files outright.
Returns:
A ``(written, skipped)`` tuple of destination paths.
"""
substitutions = _tracker_substitutions(
project_name, done_state, org_url, parent_epic, area_path, default_tags
)
written: list[Path] = []
skipped: list[Path] = []
for entry in sorted(_trackers_resource().iterdir(), key=lambda e: e.name):
if not entry.name.endswith(".md"):
continue
text = entry.read_text(encoding="utf-8")
for placeholder, value in substitutions.items():
text = text.replace(placeholder, value)
target = dest / "docs" / "agents" / entry.name
result = _write(target, text, force, confirm)
(written if result is not None else skipped).append(target)
return written, skipped
[docs]
def mcp_server_entry(use_uvx: bool = True) -> dict[str, object]:
"""Return the MCP ``mcpServers`` entry for the stdio server.
Args:
use_uvx: When ``True`` (default) the entry launches the server through
``uvx --from "devops-utils[mcp]" devops-utils-mcp`` — zero-install,
nothing needs to be on ``PATH`` beyond ``uv``. When ``False`` the
entry is the bare ``devops-utils-mcp`` console script, for
environments where the package is installed (``pip install
"devops-utils[mcp]"``) and ``uv`` is not available.
"""
if use_uvx:
return {
"command": "uvx",
"args": ["--from", "devops-utils[mcp]", MCP_COMMAND],
"env": {},
}
return {"command": MCP_COMMAND, "args": [], "env": {}}
[docs]
def merge_mcp_config(
path: Path,
name: str = MCP_SERVER_NAME,
force: bool = False,
use_uvx: bool = True,
confirm: ConfirmOverwrite | None = None,
) -> tuple[Path, bool]:
"""Register the MCP server in a JSON config, preserving other servers.
Loads ``path`` if it exists, sets ``mcpServers[name]`` to
:func:`mcp_server_entry` without touching sibling servers, and writes it back
pretty-printed. Works for both ``~/.claude.json`` (user) and ``.mcp.json``
(project).
Args:
path: Path to the JSON config file (created if missing).
name: Server key to register under.
force: Overwrite an existing entry for ``name``. When ``False`` and the
entry already exists, the file is left unchanged.
use_uvx: Passed to :func:`mcp_server_entry` — ``True`` (default) writes
the zero-install ``uvx`` launcher, ``False`` the on-``PATH``
``devops-utils-mcp`` console script.
confirm: Asked before replacing an existing entry when ``force`` is not
set; ``None`` leaves the entry alone. The request compares the
*entry* JSON, not the whole file — sibling servers are never at
stake.
Returns:
A ``(path, changed)`` tuple. ``changed`` is ``False`` when an entry for
``name`` already existed and neither ``force`` nor ``confirm`` allowed
replacing it.
"""
data: dict[str, object] = {}
if path.exists():
loaded = json.loads(path.read_text(encoding="utf-8") or "{}")
if isinstance(loaded, dict):
data = loaded
servers = data.get("mcpServers")
if not isinstance(servers, dict):
servers = {}
entry = mcp_server_entry(use_uvx=use_uvx)
if name in servers and not force:
if confirm is None:
return path, False
request = OverwriteRequest(
path=path,
new_text=json.dumps(entry, indent=2, ensure_ascii=False) + "\n",
existing_text=json.dumps(servers[name], indent=2, ensure_ascii=False)
+ "\n",
label=f"{path} (mcpServers.{name})",
)
if not confirm(request):
return path, False
servers[name] = entry
data["mcpServers"] = servers
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
return path, True
[docs]
def env_template() -> str:
"""Return an ``.env`` scaffold for the Azure DevOps credentials."""
return (
"# Azure DevOps configuration for devops-utils (azdo tools / MCP).\n"
"# No machine credentials are read; only these env vars are used.\n"
"#\n"
"# Save a filled-in copy WITHOUT the .example suffix and the CLI and MCP\n"
"# server load it automatically (first existing file wins, never merged;\n"
"# real environment variables always win):\n"
"# 1. the path in DEVOPS_UTILS_ENV_FILE\n"
"# 2. ./.env.devops-utils (project; add it to .gitignore)\n"
"# 3. ~/.devops-utils.env (user; %USERPROFILE% on Windows)\n"
"\n"
"# Cloud: https://dev.azure.com/{org} | on-prem: https://server/tfs/{collection}\n"
"AZURE_DEVOPS_ORG_URL=\n"
"# Bearer token or PAT.\n"
"AZURE_DEVOPS_TOKEN=\n"
"# 'bearer' (default) or 'pat'.\n"
"AZURE_DEVOPS_AUTH_SCHEME=bearer\n"
"# Default 7.1; lower it for older on-prem servers.\n"
"AZURE_DEVOPS_API_VERSION=7.1\n"
)
[docs]
def write_env_scaffold(
path: Path,
force: bool = False,
confirm: ConfirmOverwrite | None = None,
) -> Path | None:
"""Write the env scaffold to ``path`` (skipped if it exists and not ``force``)."""
return _write(path, env_template(), force, confirm)
[docs]
def plugin_manifest() -> dict[str, object]:
"""Return the ``.claude-plugin/plugin.json`` contents for the bundled plugin.
The plugin name drives Claude Code's namespacing: skills and agents shipped
under it list as ``devops-utils:<component>`` (see :data:`PLUGIN_NAME`). The
MCP server is intentionally *not* declared here — a plugin-bundled MCP server
is re-scoped to ``mcp__plugin_<plugin>_<server>__<tool>``, which would break
the agents' ``tools:`` frontmatter that targets ``mcp__devops-utils__azdo_*``.
MCP stays wired via :func:`merge_mcp_config` / ``setup mcp`` instead.
``version`` is deliberately omitted (it is optional): the package uses VCS-derived
versioning, so a value here would change on every commit and make the committed
manifest churn and drift from a regeneration.
"""
return {
"name": PLUGIN_NAME,
"description": (
"Read-only Azure DevOps skills and subagents for work-item, build, "
"and repository research (cloud + on-prem)."
),
"author": {
"name": "Cláudio Ferreira Carneiro",
"email": "claudiofcarneiro@gmail.com",
},
"homepage": "https://carneirofc.github.io/devops-utils/",
"repository": "https://github.com/carneirofc/devops-utils",
"license": "MIT",
"keywords": ["azure-devops", "devops", "work-items", "mcp"],
}
[docs]
def marketplace_manifest() -> dict[str, object]:
"""Return the ``.claude-plugin/marketplace.json`` contents for the repo.
Declares a single-plugin marketplace whose ``source`` is the relative path
``./plugins/devops-utils`` (Claude Code resolves it from the marketplace root,
i.e. the repo root). Users install with::
/plugin marketplace add carneirofc/devops-utils
/plugin install devops-utils@carneirofc
"""
return {
"name": MARKETPLACE_NAME,
"owner": {
"name": "Cláudio Ferreira Carneiro",
"email": "claudiofcarneiro@gmail.com",
},
"plugins": [
{
"name": PLUGIN_NAME,
"source": f"./plugins/{PLUGIN_NAME}",
"description": (
"Read-only Azure DevOps skills and subagents "
"(work items, builds, repositories)."
),
}
],
}
def _write_json(
path: Path,
data: dict[str, object],
force: bool,
confirm: ConfirmOverwrite | None = None,
) -> Path | None:
"""Write ``data`` as pretty UTF-8 JSON (skipped if it exists sans ``force``)."""
text = json.dumps(data, indent=2, ensure_ascii=False) + "\n"
return _write(path, text, force, confirm)
[docs]
def install_plugin(
repo_root: Path,
force: bool = False,
confirm: ConfirmOverwrite | None = None,
) -> tuple[list[Path], list[Path]]:
"""Generate the Claude Code plugin + marketplace tree under ``repo_root``.
Lays the bundled skills/agents out where Claude Code discovers them for a
plugin named :data:`PLUGIN_NAME`, and writes the two manifests:
- ``repo_root/plugins/devops-utils/skills/<name>/SKILL.md`` (via
:func:`install_skills` with the ``"claude"`` layout)
- ``repo_root/plugins/devops-utils/agents/<name>.md`` (via
:func:`install_agents`)
- ``repo_root/plugins/devops-utils/.claude-plugin/plugin.json``
- ``repo_root/.claude-plugin/marketplace.json``
The skill/agent markdown is copied byte-for-byte from the bundled sources;
the ``<plugin>:<name>`` namespace is derived by Claude Code from this layout,
so no frontmatter rewriting is needed.
Args:
repo_root: Repository root to generate the plugin tree into.
force: Overwrite existing files instead of skipping them.
confirm: Asked per existing file when ``force`` is not set; ``None``
skips existing files outright.
Returns:
A ``(written, skipped)`` tuple of destination paths.
"""
plugin_root = repo_root / "plugins" / PLUGIN_NAME
written: list[Path] = []
skipped: list[Path] = []
for install in (install_skills, install_agents):
w, s = install(plugin_root, force=force, confirm=confirm)
written.extend(w)
skipped.extend(s)
for path, data in (
(plugin_root / ".claude-plugin" / "plugin.json", plugin_manifest()),
(repo_root / ".claude-plugin" / "marketplace.json", marketplace_manifest()),
):
result = _write_json(path, data, force, confirm)
(written if result is not None else skipped).append(path)
return written, skipped