#!/usr/bin/env python3 """PRD Plugin UTCP tool surface (REQ-082). The read-only tools over a repo's .prd_plugin method state, defined ONCE here and owned by this repo. A UTCP manual (``--utcp-manual`` / the repo-root ``utcp.json``) describes them with ``cli`` call templates that run THIS script, so the AI-Collab hub -- or any ``@utcp/mcp-bridge`` client -- mounts the manual and exposes the tools via MCP instead of re-implementing them. Nothing new is added to the installed plugin's runtime: the manual is data, and each tool just runs an existing prd-plugin reader and prints JSON. Zero dependencies, stdlib only. Tools (all read-only, idempotent): status one-screen method status (requests / tracking / health / stale) tracking tracking records (goals / active work / follow-ups), filterable decisions durable decisions (DEC-*) evidence evidence records (EV-*) wiki LLM-wiki freshness / drift drift the cheap on-Stop drift summary gate the verification gate findings reporting deterministic delegated-reporting bundle and effective policy config_inventory complete persistent and non-persistent configuration surface Writes are deliberately NOT here: they stay on the validated MCP/script path (single-writer + the consent floor). This surface is observe-only. """ from __future__ import annotations import argparse import json import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) UTCP_VERSION = "1.2.0" MANUAL_VERSION = "2.0.0" # The canonical UTCP+MCP capability catalog (REQ-103). Installed downstream at # .prd_plugin/tool-surface.json; the hub reads its own templates/ copy. _TOOL_SURFACE_PATHS = ( Path(".prd_plugin") / "tool-surface.json", Path("templates") / "tool-surface.json", ) # The server's generated tool-metadata projection (IMP-TASK-091): the manual # builder consumes it so MCP metadata is never authored twice. Regenerate with: # node mcp/server.cjs --describe > mcp/tool-metadata.json _TOOL_METADATA_PATHS = ( Path(".prd_plugin") / "mcp" / "tool-metadata.json", Path("mcp") / "tool-metadata.json", ) # Shared cross-repo band taxonomy (agreed 2026-07-03): observe/recall/act/orchestrate. _ACCESS_BANDS = {"read": "band:observe", "write": "band:act", "coordinate": "band:orchestrate"} # The command a UTCP `cli` call template runs for a tool. `repo` is the target # repo root; the hub substitutes it per workspace repo (UTCP arg placeholder). _ARG = "UTCP_ARG_repo_UTCP_END" def _read_json(path, default): try: return json.loads(Path(path).read_text(encoding="utf-8-sig")) except Exception: return default def _records(root, rel, keys=("records",)): """Best-effort list of records from a .prd_plugin/state file.""" data = _read_json(Path(root) / ".prd_plugin" / "state" / rel, {}) if isinstance(data, dict): for key in keys: if isinstance(data.get(key), list): return data[key] return data if isinstance(data, list) else [] def _limit(items, args): n = args.get("limit") try: n = int(n) except (TypeError, ValueError): n = 25 return items[:n] if n and n > 0 else items # ---- tool implementations (each returns {"text": str, ...}) ---------------- def _tool_status(root, args): import prd_status s = prd_status.build_status(root) reqs = s.get("requests_by_status", {}) if isinstance(s, dict) else {} text = (f"autonomy={s.get('autonomy_level')}, " f"active tracking={len(s.get('active_tracking', []) or [])}, " f"open health={len(s.get('open_health', []) or [])}, " f"stale={s.get('stale_count')}, requests={reqs}") return {"text": text, "status": s} def _tool_tracking(root, args): records = _records(root, "tracking.json") want = str(args.get("status", "") or "").lower() if want and want not in ("all", "any"): open_set = {"open", "in_progress", "active", "proposed", "accepted"} def keep(r): st = str(r.get("status", "")).lower() return st == want if want not in ("open", "in_progress") else st in open_set records = [r for r in records if keep(r)] records = _limit(records, args) return {"text": f"{len(records)} tracking record(s)", "records": records} def _tool_decisions(root, args): records = _limit(_records(root, "decisions.json", ("decisions", "records")), args) return {"text": f"{len(records)} decision(s)", "records": records} def _tool_evidence(root, args): records = _limit(_records(root, "evidence.json"), args) return {"text": f"{len(records)} evidence record(s)", "records": records} WIKI_PAGE_CONTRACT_VERSION = 1 def _wiki_title(markdown): for line in markdown.splitlines(): if line.startswith("# "): return line[2:].strip() return None def _wiki_updated_map(root): """Article path -> Updated date, parsed from the wiki index tables.""" import re index = Path(root) / "wiki" / "index.md" if not index.is_file(): return {} updated = {} for match in re.finditer( r"\|\s*\[[^\]]+\]\(([^)]+)\)\s*\|[^|]*\|\s*(\d{4}-\d{2}-\d{2})\s*\|", index.read_text(encoding="utf-8-sig")): rel = match.group(1).lstrip("./") updated[f"wiki/{rel}"] = match.group(2) return updated def _wiki_page_path(root, rel): """Resolve one contract path with hard boundaries (REQ-109).""" if not rel: raise ValueError("wiki read requires --path (repo-relative, under wiki/)") candidate = Path(rel) if candidate.is_absolute(): raise ValueError("wiki page paths must be repo-relative, not absolute") wiki_root = (Path(root) / "wiki").resolve() resolved = (Path(root) / candidate).resolve() if resolved != wiki_root and wiki_root not in resolved.parents: raise ValueError(f"{rel} is outside wiki/ — only canonical wiki pages are readable") if resolved.suffix.lower() != ".md": raise ValueError(f"{rel} is not a Markdown page") if not resolved.is_file(): raise ValueError(f"no wiki page at {rel}") return resolved def _wiki_read_page(root, rel): import hashlib import re resolved = _wiki_page_path(root, rel) raw = resolved.read_bytes() try: markdown = raw.decode("utf-8") except UnicodeDecodeError: raise ValueError(f"{rel} is not valid UTF-8 — refusing to export malformed content") posix = resolved.relative_to(Path(root).resolve()).as_posix() commit = re.search(r"^> Commit: (\S+)", markdown, re.M) return { "contract_version": WIKI_PAGE_CONTRACT_VERSION, "repo_id": Path(root).resolve().name, "path": posix, "title": _wiki_title(markdown), "filename": resolved.name, "content_type": "text/markdown", "markdown": markdown, "content_sha256": hashlib.sha256(raw).hexdigest(), "commit": commit.group(1) if commit else None, "updated": _wiki_updated_map(root).get(posix), } def _wiki_list_pages(root): wiki_root = Path(root) / "wiki" pages = [] if wiki_root.is_dir(): for page in sorted(wiki_root.rglob("*.md")): try: markdown = page.read_text(encoding="utf-8") except UnicodeDecodeError: continue # malformed pages are readable only via explicit errors pages.append({ "path": page.relative_to(Path(root)).as_posix(), "title": _wiki_title(markdown), "filename": page.name, }) return { "contract_version": WIKI_PAGE_CONTRACT_VERSION, "repo_id": Path(root).resolve().name, "pages": pages, } def _tool_fabric(root, args): """Read-only fabric profile resolver (REQ-120): which calibrated prediction treatment applies to a model, or the fail-safe raw/abstain default.""" sys.path.insert(0, str(Path(root).resolve() / ".prd_plugin" / "scripts")) sys.path.insert(0, str(Path(__file__).resolve().parent)) import prd_config model = str(args.get("model") or "").strip() if not model: raise ValueError("fabric: a model id is required") task_type = args.get("task_type") task_type = str(task_type).strip() if task_type else None return prd_config.resolve_fabric_profile(root, model, task_type) def _tool_wiki(root, args): action = str(args.get("action") or "drift") if action == "list": result = _wiki_list_pages(root) result["text"] = f"{len(result['pages'])} wiki page(s)" return result if action == "read": result = _wiki_read_page(root, str(args.get("path") or "")) result["text"] = f"wiki page {result['path']}" return result if action != "drift": raise ValueError(f"unknown wiki action {action!r}") import prd_wiki_backfill r = prd_wiki_backfill.wiki_drift(root) return {"text": r.get("summary", ""), "status": r.get("status"), "findings": r.get("findings", [])} def _tool_drift(root, args): import drift_monitor r = drift_monitor.stop_check(Path(root)) text = r.get("summary") or r.get("status") or "drift check complete" return {"text": str(text), "drift": r} def _tool_gate(root, args): import prd_gate r = prd_gate.run_checks(root) findings = r.get("findings", r) if isinstance(r, dict) else r findings = findings if isinstance(findings, list) else [] return {"text": f"{len(findings)} gate finding(s)", "findings": findings} def _tool_reporting(root, args): import prd_reporting task = str(args.get("task") or "report_summary") bundle = prd_reporting.build_bundle(root, task) return {"text": f"delegated reporting bundle for {task}", "bundle": bundle} def _tool_substrate(root, args): import prd_substrate action = str(args.get("action") or "handshake") repo_id = str(args.get("repo_id") or Path(root).resolve().name) if action == "handshake": result = prd_substrate.build_handshake(root, contact_runtime=bool(args.get("contact_runtime", False))) elif action == "snapshot": result = prd_substrate.build_snapshot(root, repo_id=repo_id) elif action == "graph": result = prd_substrate.build_graph_export(root, repo_id=repo_id) elif action == "events": result = prd_substrate.build_event_export(root, repo_id=repo_id) elif action == "discover": result = prd_substrate.discover_runtime(root) elif action == "diagnose": result = prd_substrate.diagnose_runtime(root) elif action == "capabilities": import prd_substrate_catalog result = prd_substrate_catalog.audit_catalog(root) elif action == "links": import prd_substrate_links result = {"registry": prd_substrate_links.load_links(root), "audit": prd_substrate_links.audit_value(prd_substrate_links.load_links(root))} elif action == "verification": import prd_test_scope result = prd_test_scope.build_plan( root, changed_files=args.get("changed_files"), base_ref=str(args.get("base_ref") or ""), release=bool(args.get("release", False)), ) else: raise ValueError(f"unknown substrate action {action!r}") return {"text": f"substrate adapter {action}", "substrate": result} def _tool_services(root, args): import prd_services action = str(args.get("action") or "list") if action == "audit": result = prd_services.audit_manifest(root) elif action == "projection": result = prd_services.manifest_projection(root) elif action == "get": result = prd_services.get_service(root, str(args.get("kind") or ""), str(args.get("id") or "")) elif action == "list": result = prd_services.list_services(root, str(args.get("kind") or "")) else: raise ValueError(f"unknown services action {action!r}") return {"text": f"repository services {action}", "services": result} def _tool_config_inventory(root, args): import prd_config result = prd_config.inventory(root) summary = result.get("summary", {}) return {"text": (f"{summary.get('persistent_settings', 0)} persistent settings; " f"{summary.get('unclassified_settings', 0)} unclassified"), "inventory": result} def _tool_workflow(root, args): import prd_workflows action = args.get("action") or "list" if action == "list": result = prd_workflows.list_workflows(root) elif action == "actions": result = prd_workflows.action_inventory() elif action == "audit": result = prd_workflows.audit_catalog(root) elif action == "plan": if not args.get("workflow"): raise ValueError("workflow plan requires --workflow") raw = args.get("inputs") or "{}" result = prd_workflows.plan_workflow(root, args["workflow"], json.loads(raw)) elif action == "status": if not args.get("run_id"): raise ValueError("workflow status requires --run-id") result = prd_workflows.get_run(root, args["run_id"]) else: raise ValueError(f"unknown workflow action {action!r}") return {"text": f"workflow {action}", "workflow": result} # ---- tool registry: single source for dispatch AND the UTCP manual --------- _OUT_RECORDS = {"type": "object", "properties": { "text": {"type": "string"}, "records": {"type": "array", "items": {"type": "object"}}}} _OUT_TEXT = {"type": "object", "properties": {"text": {"type": "string"}}} _IN_LIST = {"type": "object", "properties": { "repo": {"type": "string", "description": "Target repo root (default: this repo)."}, "limit": {"type": "number", "description": "Max records (default 25)."}}} _IN_TRACKING = {"type": "object", "properties": { "repo": {"type": "string", "description": "Target repo root (default: this repo)."}, "status": {"type": "string", "description": "open | active | proposed | accepted | resolved | all."}, "limit": {"type": "number", "description": "Max records (default 25)."}}} _IN_REPO = {"type": "object", "properties": { "repo": {"type": "string", "description": "Target repo root (default: this repo)."}}} _IN_REPORTING = {"type": "object", "properties": { "repo": {"type": "string", "description": "Target repo root (default: this repo)."}, "task": {"type": "string", "enum": ["report_summary", "session_summary", "wiki_synthesis", "triage_draft"], "description": "Eligible non-deterministic reporting task."}}, "required": ["task"]} _IN_SUBSTRATE = {"type": "object", "properties": { "repo": {"type": "string", "description": "Target repo root (default: this repo)."}, "action": {"type": "string", "enum": ["handshake", "snapshot", "graph", "events", "discover", "diagnose", "capabilities", "links", "verification"], "description": "Read the adapter handshake, projections, runtime discovery/diagnostics, or verification plan."}, "changed_files": {"type": "array", "items": {"type": "string"}, "description": "Optional exact changed files; omitted means collect local git changes."}, "base_ref": {"type": "string", "description": "Optional local base ref for committed changes."}, "release": {"type": "boolean", "description": "Force the release full-verification trigger."}, "contact_runtime": {"type": "boolean", "description": "For handshake, also negotiate live runtime identity and tool availability."}, "repo_id": {"type": "string", "description": "Registered AI-Collab repo id used to qualify projected identities."}}, "required": ["action", "repo_id"]} _IN_SERVICES = {"type": "object", "properties": { "repo": {"type": "string", "description": "Target repo root (default: this repo)."}, "action": {"type": "string", "enum": ["list", "get", "audit", "projection"]}, "kind": {"type": "string", "enum": ["consumes", "provides"]}, "id": {"type": "string", "description": "Stable service id required by get."}}, "required": ["action"]} _IN_WORKFLOW = {"type": "object", "properties": { "repo": {"type": "string", "description": "Target repo root (default: this repo)."}, "action": {"type": "string", "enum": ["list", "actions", "audit", "plan", "status"], "description": "Read the workflow catalog, action allowlist, audit, stable plan, or persistent run."}, "workflow": {"type": "string", "description": "Workflow id required by plan."}, "inputs": {"type": "string", "description": "JSON object string containing plan inputs."}, "run_id": {"type": "string", "description": "WFR id required by status."}}, "required": ["action"]} TOOLS = [ {"name": "status", "band": "observe", "impl": _tool_status, "inputs": _IN_REPO, "outputs": _OUT_TEXT, "description": "One-screen PRD Plugin method status for a repo: requests by " "state, active tracking, open health findings, stale items."}, {"name": "tracking", "band": "recall", "impl": _tool_tracking, "inputs": _IN_TRACKING, "outputs": _OUT_RECORDS, "description": "A repo's tracking records (goals / active work / follow-ups) " "from its canonical PRD-plugin state, filterable by status."}, {"name": "decisions", "band": "recall", "impl": _tool_decisions, "inputs": _IN_LIST, "outputs": _OUT_RECORDS, "description": "A repo's durable decisions (DEC-*) from its PRD-plugin ledger."}, {"name": "evidence", "band": "recall", "impl": _tool_evidence, "inputs": _IN_LIST, "outputs": _OUT_RECORDS, "description": "A repo's evidence records (EV-*) backing completion claims."}, {"name": "wiki", "band": "observe", "impl": _tool_wiki, "inputs": {"type": "object", "properties": { "repo": {"type": "string", "description": "Target repo root (default: this repo)."}, "action": {"type": "string", "enum": ["drift", "list", "read"], "description": "drift (default) = freshness check; list = enumerate canonical " "wiki pages; read = one page as exact Markdown (page contract v1)."}, "path": {"type": "string", "description": "Repo-relative page path under wiki/, required by read."}}}, "outputs": _OUT_TEXT, "cli_args": "--action UTCP_ARG_action_UTCP_END --path UTCP_ARG_path_UTCP_END", "description": "LLM-wiki freshness (drift check) plus the versioned page read/export " "contract: list canonical wiki pages, or read one page as exact UTF-8 " "Markdown with title, safe export filename, content hash, and " "Commit/Updated provenance — the surface consuming hubs use to render " "Copy-Markdown and Download-.md controls. Paths outside wiki/, traversal, " "non-Markdown, and non-UTF-8 content are rejected."}, {"name": "fabric", "band": "observe", "impl": _tool_fabric, "inputs": {"type": "object", "properties": { "repo": {"type": "string", "description": "Target repo root (default: this repo)."}, "model": {"type": "string", "description": "Model id to resolve, e.g. glm-5.2."}, "task_type": {"type": "string", "description": "Optional task type; a specific " "binding overrides the model-wide '*' default."}}, "required": ["model"]}, "outputs": _OUT_TEXT, "cli_args": "--model UTCP_ARG_model_UTCP_END --task-type UTCP_ARG_task_type_UTCP_END", "description": "Resolve the fabric prediction profile bound to a model " "(fabric.model_profiles policy, evidence-bound): returns the " "calibrated treatment to apply, or the fail-safe raw/abstain " "default for unmapped models — a profile is never guessed."}, {"name": "drift", "band": "observe", "impl": _tool_drift, "inputs": _IN_REPO, "outputs": _OUT_TEXT, "description": "The cheap on-Stop drift summary for a repo (state / docs / " "wiki / ingest-manual / fork-version drift)."}, {"name": "gate", "band": "observe", "impl": _tool_gate, "inputs": _IN_REPO, "outputs": _OUT_TEXT, "description": "The PRD Plugin verification gate findings for a repo " "(duplicate ids, version markers, stranded work, timescales)."}, {"name": "reporting", "band": "observe", "impl": _tool_reporting, "inputs": _IN_REPORTING, "outputs": _OUT_TEXT, "cli_args": "--task UTCP_ARG_task_UTCP_END", "description": "Build a deterministic, source-referenced bundle and expose " "the effective policy for an eligible delegated-reporting task. " "AI-Collab executes the configured model separately."}, {"name": "substrate", "band": "observe", "impl": _tool_substrate, "inputs": _IN_SUBSTRATE, "outputs": _OUT_TEXT, "cli_args": "--action UTCP_ARG_action_UTCP_END --repo-id UTCP_ARG_repo_id_UTCP_END", "description": "Read the versioned PRD Plugin to AI-Collab Substrate adapter handshake, " "complete record snapshot, repo-qualified traceability graph, or deterministic verification plan. " "The master config switch remains authoritative."}, {"name": "services", "band": "observe", "impl": _tool_services, "inputs": _IN_SERVICES, "outputs": _OUT_TEXT, "cli_args": "--action UTCP_ARG_action_UTCP_END --kind UTCP_ARG_kind_UTCP_END --id UTCP_ARG_id_UTCP_END", "description": "Read, audit, or project the repository-owned service consume/provide manifest. Mutations remain MCP/CLI-only."}, {"name": "config_inventory", "band": "observe", "impl": _tool_config_inventory, "inputs": _IN_REPO, "outputs": _OUT_TEXT, "description": "Inspect the unified PRD Plugin configuration inventory: every persistent " "setting with ownership, mutability, activation, dependencies, and latency, " "plus environment, install-time, host-wiring, specialized-CRUD, and " "invocation-only controls."}, {"name": "workflow", "band": "observe", "impl": _tool_workflow, "inputs": _IN_WORKFLOW, "outputs": _OUT_TEXT, "cli_args": "--action UTCP_ARG_action_UTCP_END --workflow UTCP_ARG_workflow_UTCP_END --inputs UTCP_ARG_inputs_UTCP_END --run-id UTCP_ARG_run_id_UTCP_END", "description": "Inspect deterministic workflow definitions, action metadata, catalog findings, stable dry-run plans, and persistent receipts. Mutations remain MCP-only."}, ] _BY_NAME = {t["name"]: t for t in TOOLS} def load_tool_surface_catalog(root): """The canonical tool-surface catalog every transport validates against. Resolution order: the installed .prd_plugin/tool-surface.json (downstream), then templates/tool-surface.json (the hub's canonical source).""" root = Path(root) for rel in _TOOL_SURFACE_PATHS: path = root / rel if path.is_file(): return json.loads(path.read_text(encoding="utf-8-sig")) raise FileNotFoundError( f"tool-surface.json not found under {root} " f"(looked in {', '.join(str(p) for p in _TOOL_SURFACE_PATHS)})" ) def _default_root(): """The repo root this script belongs to: /scripts/ on the hub, /.prd_plugin/scripts/ downstream.""" here = Path(__file__).resolve().parent for candidate in (here.parent, here.parent.parent): for rel in _TOOL_METADATA_PATHS + _TOOL_SURFACE_PATHS: if (candidate / rel).is_file(): return candidate return here.parent def load_tool_metadata(root): """The server's generated {name, description, inputSchema} projection.""" root = Path(root) for rel in _TOOL_METADATA_PATHS: path = root / rel if path.is_file(): return json.loads(path.read_text(encoding="utf-8-sig")) raise FileNotFoundError( f"tool-metadata.json not found under {root}; regenerate with: " "node mcp/server.cjs --describe > mcp/tool-metadata.json" ) def run_tool(name, root, args): tool = _BY_NAME.get(name) if not tool: raise KeyError(f"unknown tool {name!r}; known: {sorted(_BY_NAME)}") return tool["impl"](root, args or {}) def _call_template(tool): extra = f" {tool.get('cli_args')}" if tool.get("cli_args") else "" cmd = (f"python .prd_plugin/scripts/prd_tools.py {tool['name']} " f"--repo-root {_ARG}{extra} --format json") return {"call_template_type": "cli", "commands": [{"command": cmd, "append_to_final_output": True}]} def _mcp_call_template(): """Official @utcp/mcp call template targeting the validated state server — mutations execute inside the server's locking/validation/worker guards. The server key MUST equal the name the consumer mounts this manual under (@utcp/mcp resolves `mount.tool` by looking up mcpServers[mount]), so the manual is mounted as `prd_plugin` by contract.""" return { "call_template_type": "mcp", "config": {"mcpServers": {"prd_plugin": { "transport": "stdio", "command": "node", "args": [".prd_plugin/mcp/server.cjs"], }}}, } TOOL_SPEC_PATHS = ( Path("templates") / "tool-spec.json", Path(".prd_plugin") / "templates" / "tool-spec.json", ) def load_tool_spec(root): """The authored source of tool definitions (REQ-107). Authority used to flow the other way - the hand-written MCP server defined tools, `--describe` projected them, and this builder read that projection, making the manual a VIEW of the server. The spec is now the source: the manual is generated from it and the server reads it too, so neither can drift from the other. Returns None when absent so an older downstream install still builds a manual from the server projection rather than failing. """ root = Path(root) for rel in TOOL_SPEC_PATHS: path = root / rel try: data = json.loads(path.read_text(encoding="utf-8-sig")) except (OSError, ValueError): continue if isinstance(data, dict) and isinstance(data.get("tools"), list): return data return None def _bridged_tools(root): """Every MCP-transport tool as a UTCP manual entry, defined by the tool spec when present and falling back to the server's projection when it is not.""" catalog = load_tool_surface_catalog(root) capability_of = {} access_of = {} for capability in catalog["capabilities"]: for name in capability["transports"].get("mcp", []): capability_of[name] = capability["id"] access_of[name] = capability.get("access", "write") spec = load_tool_spec(root) if spec is not None: source = [{"name": s["name"], "description": s["description"], "inputSchema": s.get("inputSchema", {})} for s in spec["tools"] if s.get("transport") == "mcp"] else: source = load_tool_metadata(root)["tools"] tools = [] for meta in source: name = meta["name"] access = access_of.get(name, "write") tools.append({ "name": name, "description": meta["description"], "tags": [_ACCESS_BANDS.get(access, "band:act"), "cap:read-only" if access == "read" else "cap:mutating", f"capability:{capability_of.get(name, 'uncataloged')}", "host:mcp", "prd-plugin"], "inputs": meta["inputSchema"], "outputs": {"type": "object", "description": "JSON result from the validated PRD Plugin state server."}, "tool_call_template": _mcp_call_template(), }) return tools def build_manual(root=None): """The complete UTCP manual: the source of truth for every PRD Plugin tool (PRD-REQ-076, ARCH-DEC-049). Read-only observe/recall tools call this script via `cli` templates; every state-server tool is bridged with an `mcp` template so guarded mutation semantics stay in the handlers. Mount with the UTCP SDK; MCP comes generated via @utcp/mcp-bridge.""" root = Path(root) if root else _default_root() tools = [] for t in TOOLS: tools.append({ "name": t["name"], "description": t["description"], "tags": [f"band:{t['band']}", "cap:read-only", "cap:idempotent", "host:cli", "prd-plugin"], "inputs": t["inputs"], "outputs": t["outputs"], "tool_call_template": _call_template(t), }) tools.extend(_bridged_tools(root)) # Pure UtcpManual per the official SDK's strict schema — the shape a UTCP # client or @utcp/mcp-bridge mounts. Repo/hub metadata lives in the static # manifest (build_static_manifest), the workspace hub-feed convention. return { "manual_version": MANUAL_VERSION, "utcp_version": UTCP_VERSION, "tools": tools, } def build_static_manifest(root=None): """The committed repo-root utcp.json: the hub-discovery manifest (hub descriptor + the manual fields), matching the workspace convention. Mount the PURE manual (build_manual / --utcp-manual) with the official SDK; this merged shape is for hub feeds and humans.""" manifest = {"hub": { "repo_id": "prd-plugin", "title": "PRD Plugin - method/governance tools", "description": "The complete PRD Plugin tool surface: observe/recall " "cli tools over canonical .prd_plugin method state, plus " "every validated state-server tool bridged via mcp call " "templates. UTCP is the source of truth; generate MCP " "with @utcp/mcp-bridge.", "categories": ["observe", "recall", "act", "orchestrate"], }} manifest.update(build_manual(root)) return manifest def main(argv=None): parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("tool", nargs="?", help="Tool name, or omit with --utcp-manual.") parser.add_argument("--utcp-manual", action="store_true", help="Emit the pure UTCP manual (spec-valid; mountable).") parser.add_argument("--static-manifest", action="store_true", help="Emit the hub-feed manifest (hub descriptor + manual).") parser.add_argument("--repo-root", default=".") parser.add_argument("--status", default=None) parser.add_argument("--limit", default=None) parser.add_argument("--task", default=None) parser.add_argument("--action", default=None) parser.add_argument("--path", default=None) parser.add_argument("--repo-id", default=None) parser.add_argument("--model", default=None) parser.add_argument("--task-type", dest="task_type", default=None) parser.add_argument("--workflow", default=None) parser.add_argument("--inputs", default=None) parser.add_argument("--run-id", default=None) parser.add_argument("--format", choices=("json", "markdown"), default="json") parser.add_argument("--output") args = parser.parse_args(argv) if args.static_manifest: result = build_static_manifest() elif args.utcp_manual: result = build_manual() else: if not args.tool: parser.error("a tool name (or --utcp-manual / --static-manifest) is required") try: result = run_tool(args.tool, args.repo_root, {"status": args.status, "limit": args.limit, "task": args.task, "action": args.action, "path": args.path, "repo_id": args.repo_id, "workflow": args.workflow, "inputs": args.inputs, "run_id": args.run_id, "model": args.model, "task_type": args.task_type}) except KeyError as exc: print(str(exc), file=sys.stderr) return 2 text = (result.get("text", "") if args.format == "markdown" and isinstance(result, dict) else json.dumps(result, indent=2, default=str)) if args.output: out = Path(args.output) out.parent.mkdir(parents=True, exist_ok=True) # Pin the encoding and the line ending. Without newline="\n" this writes # CRLF on Windows, so a regenerated artifact churns the diff for # everyone else; without the trailing newline it differs from every # other JSON this repo writes. Prefer --output over `> file`: python's # stdout is cp1252 on a default Windows console, which is how six em # dashes shipped as mojibake in 0.16.45 (REQ-159). out.write_text(text + "\n", encoding="utf-8", newline="\n") else: print(text) return 0 if __name__ == "__main__": sys.exit(main())