mcp_helpers.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Shared helpers for knowtation MCP tools and resources.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | import pathlib |
| 7 | import urllib.error |
| 8 | import urllib.parse |
| 9 | import urllib.request |
| 10 | from datetime import datetime, timezone |
| 11 | from typing import Any |
| 12 | |
| 13 | from muse.core.object_store import read_object |
| 14 | from muse.core.repo import read_repo_id |
| 15 | from muse.core.store import ( |
| 16 | get_commit_snapshot_manifest, |
| 17 | read_commit, |
| 18 | read_current_branch, |
| 19 | resolve_commit_ref, |
| 20 | ) |
| 21 | from muse.core.validation import clamp_int, sanitize_display |
| 22 | from muse.plugins.code._query import walk_commits_bfs |
| 23 | from muse.plugins.knowtation._query import NOTE_EXTENSIONS, group_by_project, is_note |
| 24 | from muse.plugins.knowtation.code_intel import describe_link |
| 25 | from muse.plugins.knowtation.events import MemoryEventKind, is_valid_event_kind |
| 26 | from muse.plugins.knowtation.hooks import DEFAULT_PORT, _resolve_port |
| 27 | from muse.plugins.knowtation.link_index import LinkIndex, _normalise_rel_path, build_link_index |
| 28 | from muse.plugins.knowtation.parser import parse_frontmatter |
| 29 | from muse.plugins.knowtation.plugin import _ALWAYS_IGNORE_DIRS |
| 30 | from muse.plugins.knowtation.prime_context import build_prime_context |
| 31 | from muse.plugins.knowtation.stats import collect_vault_stats |
| 32 | from muse.plugins.knowtation.symbols import extract_symbols |
| 33 | |
| 34 | _MAX_HUB_RESPONSE_BYTES = 256 * 1024 |
| 35 | _STRUCTURAL_DIRS = frozenset({"templates", "meta"}) |
| 36 | _IGNORE_DIRS = _ALWAYS_IGNORE_DIRS | _STRUCTURAL_DIRS |
| 37 | |
| 38 | |
| 39 | def _note_body_and_frontmatter(raw: bytes) -> tuple[dict[str, Any], str]: |
| 40 | """Return ``(frontmatter_dict, body_text)`` from raw note bytes.""" |
| 41 | text = raw.decode("utf-8", errors="replace") |
| 42 | parsed = parse_frontmatter(raw) |
| 43 | if parsed is None: |
| 44 | return {}, text |
| 45 | if text.startswith("---"): |
| 46 | parts = text.split("---", 2) |
| 47 | body = parts[2].lstrip("\n") if len(parts) >= 3 else text |
| 48 | else: |
| 49 | body = text |
| 50 | return parsed.to_dict(), body |
| 51 | |
| 52 | |
| 53 | class PathValidationError(ValueError): |
| 54 | """Raised when a vault-relative path fails grammar checks.""" |
| 55 | |
| 56 | def __init__(self, reason: str, detail: str | None = None) -> None: |
| 57 | self.reason = reason |
| 58 | super().__init__(detail or reason) |
| 59 | |
| 60 | |
| 61 | def validate_note_path(path: str) -> str: |
| 62 | """Validate vault-relative note path grammar (§KD-5a.0).""" |
| 63 | if not path or not isinstance(path, str): |
| 64 | raise PathValidationError("PATH_INVALID", "path must be a non-empty string") |
| 65 | if path.startswith("/") or "\\" in path: |
| 66 | raise PathValidationError("PATH_INVALID", "absolute paths are not allowed") |
| 67 | if ".." in path.split("/"): |
| 68 | raise PathValidationError("PATH_INVALID", "path must not contain '..' segments") |
| 69 | normalised = _normalise_rel_path("", path.replace("\\", "/")) |
| 70 | if normalised is None: |
| 71 | raise PathValidationError("PATH_INVALID", "path escapes vault root") |
| 72 | lower = normalised.lower() |
| 73 | if not any(lower.endswith(ext) for ext in NOTE_EXTENSIONS): |
| 74 | raise PathValidationError("NOT_A_NOTE", "path is not a markdown note") |
| 75 | return normalised |
| 76 | |
| 77 | |
| 78 | def resolve_ref(root: pathlib.Path, ref: str | None) -> Any: |
| 79 | """Resolve a commit ref via the store layer.""" |
| 80 | repo_id = read_repo_id(root) |
| 81 | branch = read_current_branch(root) |
| 82 | commit = resolve_commit_ref(root, repo_id, branch, ref if ref and ref != "HEAD" else None) |
| 83 | if commit is None: |
| 84 | raise PathValidationError("REF_NOT_FOUND", f"ref not found: {ref or 'HEAD'}") |
| 85 | return commit |
| 86 | |
| 87 | |
| 88 | def read_note_at_ref( |
| 89 | root: pathlib.Path, |
| 90 | path: str, |
| 91 | ref: str = "HEAD", |
| 92 | *, |
| 93 | fmt: str = "json", |
| 94 | ) -> dict[str, Any]: |
| 95 | """Read a note at *ref* — backing for get-note and vault resources.""" |
| 96 | normalised = validate_note_path(path) |
| 97 | commit = resolve_ref(root, ref) |
| 98 | manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {} |
| 99 | content_hash = manifest.get(normalised) |
| 100 | if not content_hash: |
| 101 | return { |
| 102 | "path": normalised, |
| 103 | "ref": ref or "HEAD", |
| 104 | "exists": False, |
| 105 | "frontmatter": {}, |
| 106 | "body": "", |
| 107 | "symbols": [], |
| 108 | "content_hash": "", |
| 109 | } |
| 110 | |
| 111 | raw = read_object(root, content_hash) |
| 112 | if raw is None: |
| 113 | raise PathValidationError("NOTE_NOT_FOUND", "note object missing from store") |
| 114 | |
| 115 | parsed_fm, body_text = _note_body_and_frontmatter(raw) |
| 116 | symbols = [s.to_dict() for s in extract_symbols(normalised, raw)] |
| 117 | payload = { |
| 118 | "path": normalised, |
| 119 | "ref": ref or "HEAD", |
| 120 | "exists": True, |
| 121 | "frontmatter": parsed_fm, |
| 122 | "body": body_text if fmt == "json" else raw.decode("utf-8", errors="replace"), |
| 123 | "symbols": symbols, |
| 124 | "content_hash": content_hash, |
| 125 | } |
| 126 | return payload |
| 127 | |
| 128 | |
| 129 | def link_index_to_json(index: LinkIndex) -> dict[str, Any]: |
| 130 | """Serialise a :class:`LinkIndex` for ``knowtation://graph``.""" |
| 131 | forward: dict[str, list[dict[str, Any]]] = {} |
| 132 | for src, links in index.forward.items(): |
| 133 | forward[src] = [describe_link(link) for link in links] |
| 134 | |
| 135 | backward: dict[str, list[dict[str, Any]]] = {} |
| 136 | for tgt, links in index.backward.items(): |
| 137 | backward[tgt] = [describe_link(link) for link in links] |
| 138 | |
| 139 | entities = {label: sorted(paths) for label, paths in index.entity_index.items()} |
| 140 | return { |
| 141 | "notes_indexed": index.notes_indexed, |
| 142 | "forward": forward, |
| 143 | "backward": backward, |
| 144 | "broken_links": [describe_link(link) for link in index.broken_links], |
| 145 | "entities": entities, |
| 146 | } |
| 147 | |
| 148 | |
| 149 | def list_vault_notes(root: pathlib.Path) -> dict[str, Any]: |
| 150 | """Build ``knowtation://vault`` listing payload.""" |
| 151 | stats = collect_vault_stats(root) |
| 152 | notes: list[dict[str, Any]] = [] |
| 153 | for dirpath, dirnames, filenames in __import__("os").walk(root): |
| 154 | dirnames[:] = [d for d in dirnames if d not in _IGNORE_DIRS] |
| 155 | for name in filenames: |
| 156 | if not is_note(name): |
| 157 | continue |
| 158 | full = pathlib.Path(dirpath) / name |
| 159 | try: |
| 160 | rel = full.relative_to(root).as_posix() |
| 161 | except ValueError: |
| 162 | continue |
| 163 | try: |
| 164 | raw = full.read_bytes() |
| 165 | fm_dict, _body = _note_body_and_frontmatter(raw) |
| 166 | except OSError: |
| 167 | continue |
| 168 | notes.append( |
| 169 | { |
| 170 | "path": rel, |
| 171 | "title": str(fm_dict.get("title") or pathlib.Path(rel).stem), |
| 172 | "project": fm_dict.get("project"), |
| 173 | "tags": fm_dict.get("tags"), |
| 174 | "source_type": fm_dict.get("source_type") or fm_dict.get("source"), |
| 175 | } |
| 176 | ) |
| 177 | notes.sort(key=lambda n: n["path"]) |
| 178 | return {"notes": notes, "stats": stats} |
| 179 | |
| 180 | |
| 181 | def recent_memory_events(root: pathlib.Path, *, limit: int = 20) -> dict[str, Any]: |
| 182 | """Last *limit* commits carrying ``metadata.event_type``.""" |
| 183 | branch = read_current_branch(root) |
| 184 | head = resolve_ref(root, "HEAD") |
| 185 | commits, _ = walk_commits_bfs(root, head.commit_id, 500) |
| 186 | events: list[dict[str, Any]] = [] |
| 187 | for commit in commits: |
| 188 | event_type = commit.metadata.get("event_type") |
| 189 | if not event_type: |
| 190 | continue |
| 191 | if not is_valid_event_kind(str(event_type)): |
| 192 | continue |
| 193 | events.append( |
| 194 | { |
| 195 | "commit_id": commit.commit_id, |
| 196 | "committed_at": commit.committed_at.isoformat(), |
| 197 | "message": sanitize_display(commit.message), |
| 198 | "event_type": str(event_type), |
| 199 | "agent_id": sanitize_display(commit.agent_id or "") or None, |
| 200 | "model_id": sanitize_display(commit.model_id or "") or None, |
| 201 | } |
| 202 | ) |
| 203 | if len(events) >= limit: |
| 204 | break |
| 205 | return {"events": events, "count": len(events)} |
| 206 | |
| 207 | |
| 208 | def projects_summary(root: pathlib.Path) -> dict[str, Any]: |
| 209 | """Aggregate project note counts + last modified mtimes.""" |
| 210 | head = resolve_ref(root, "HEAD") |
| 211 | manifest = get_commit_snapshot_manifest(root, head.commit_id) or {} |
| 212 | groups = group_by_project(manifest) |
| 213 | stats = collect_vault_stats(root) |
| 214 | projects: list[dict[str, Any]] = [] |
| 215 | for project in stats.get("projects") or sorted(groups): |
| 216 | paths = groups.get(project, []) |
| 217 | last_mod = 0.0 |
| 218 | for p in paths: |
| 219 | try: |
| 220 | last_mod = max(last_mod, (root / p).stat().st_mtime) |
| 221 | except OSError: |
| 222 | continue |
| 223 | lm_iso = ( |
| 224 | datetime.fromtimestamp(last_mod, tz=timezone.utc).isoformat() |
| 225 | if last_mod |
| 226 | else None |
| 227 | ) |
| 228 | projects.append( |
| 229 | { |
| 230 | "project": project, |
| 231 | "note_count": len(paths) if paths else 0, |
| 232 | "last_modified": lm_iso, |
| 233 | } |
| 234 | ) |
| 235 | return {"projects": projects} |
| 236 | |
| 237 | |
| 238 | def enrich_prime( |
| 239 | root: pathlib.Path, |
| 240 | *, |
| 241 | max_commits: int = 100, |
| 242 | hot_note_count: int = 10, |
| 243 | ) -> dict[str, Any]: |
| 244 | """``build_prime_context`` plus ``active_projects`` / ``suggested_next``.""" |
| 245 | ctx = build_prime_context(root, max_commits=max_commits, hot_note_count=hot_note_count) |
| 246 | payload = ctx.to_dict() |
| 247 | stats = collect_vault_stats(root) |
| 248 | payload["active_projects"] = list(stats.get("projects") or []) |
| 249 | suggested: list[str] = [] |
| 250 | for hot in payload.get("hot_notes") or []: |
| 251 | path = hot.get("path") if isinstance(hot, dict) else None |
| 252 | if isinstance(path, str): |
| 253 | suggested.append(path) |
| 254 | payload["suggested_next"] = suggested[:5] |
| 255 | return payload |
| 256 | |
| 257 | |
| 258 | def create_hub_proposal( |
| 259 | root: pathlib.Path, |
| 260 | *, |
| 261 | path: str, |
| 262 | title: str, |
| 263 | proposed_body: str | None = None, |
| 264 | proposed_frontmatter: dict[str, Any] | None = None, |
| 265 | base_ref: str = "HEAD", |
| 266 | scope: str = "personal", |
| 267 | rationale: str | None = None, |
| 268 | ) -> dict[str, Any]: |
| 269 | """POST a proposal to the local Knowtation hub (§KD-5a.B tool 8).""" |
| 270 | normalised = validate_note_path(path) |
| 271 | port, port_err = _resolve_port() |
| 272 | if port is None: |
| 273 | raise PathValidationError("HUB_UNREACHABLE", port_err or "invalid hub port") |
| 274 | |
| 275 | note = read_note_at_ref(root, normalised, base_ref) |
| 276 | if not note.get("exists"): |
| 277 | raise PathValidationError("NOTE_NOT_FOUND", "note not found at base_ref") |
| 278 | |
| 279 | intent_parts = [title.strip()] |
| 280 | if rationale: |
| 281 | intent_parts.append(rationale.strip()) |
| 282 | body_payload: dict[str, Any] = { |
| 283 | "path": normalised, |
| 284 | "body": proposed_body if proposed_body is not None else note.get("body", ""), |
| 285 | "frontmatter": proposed_frontmatter if proposed_frontmatter is not None else note.get("frontmatter", {}), |
| 286 | "intent": "\n\n".join(intent_parts), |
| 287 | "base_state_id": note.get("content_hash") or "", |
| 288 | "labels": [f"scope:{scope}"], |
| 289 | "source": "muse-mcp", |
| 290 | } |
| 291 | data = json.dumps(body_payload).encode("utf-8") |
| 292 | url = f"http://localhost:{port}/api/v1/proposals" |
| 293 | request = urllib.request.Request( |
| 294 | url=url, |
| 295 | data=data, |
| 296 | method="POST", |
| 297 | headers={ |
| 298 | "Content-Type": "application/json", |
| 299 | "User-Agent": "muse-mcp-propose/1.0", |
| 300 | "Content-Length": str(len(data)), |
| 301 | }, |
| 302 | ) |
| 303 | try: |
| 304 | with urllib.request.urlopen(request, timeout=10.0) as resp: # noqa: S310 |
| 305 | raw = resp.read(_MAX_HUB_RESPONSE_BYTES + 1) |
| 306 | if len(raw) > _MAX_HUB_RESPONSE_BYTES: |
| 307 | raise PathValidationError("HUB_UNREACHABLE", "hub response too large") |
| 308 | result = json.loads(raw.decode("utf-8")) |
| 309 | except urllib.error.HTTPError as exc: |
| 310 | if exc.code in (401, 403): |
| 311 | raise PathValidationError("SCOPE_UNAUTHORIZED", "hub rejected scope") from exc |
| 312 | if exc.code == 400: |
| 313 | raise PathValidationError("VALIDATION_FAILED", "hub validation failed") from exc |
| 314 | raise PathValidationError("HUB_UNREACHABLE", "hub request failed") from exc |
| 315 | except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, OSError) as exc: |
| 316 | raise PathValidationError("HUB_UNREACHABLE", "hub unreachable") from exc |
| 317 | |
| 318 | proposal_id = str(result.get("id") or result.get("proposal_id") or result.get("proposalId") or "") |
| 319 | if not proposal_id: |
| 320 | raise PathValidationError("VALIDATION_FAILED", "hub returned no proposal id") |
| 321 | |
| 322 | return { |
| 323 | "proposal_id": proposal_id, |
| 324 | "status": "created", |
| 325 | "hub_url": f"/proposals/{proposal_id}", |
| 326 | "path": normalised, |
| 327 | "base_ref": base_ref, |
| 328 | "scope": scope, |
| 329 | } |
| 330 | |
| 331 | |
| 332 | def consolidate_receipt(root: pathlib.Path, **_kwargs: Any) -> dict[str, Any]: |
| 333 | """Dispatch consolidation hook and return MCP receipt (§KD-5a.A(b)).""" |
| 334 | from muse.plugins.knowtation.hooks import KnowtationMergeHook |
| 335 | |
| 336 | hook = KnowtationMergeHook() |
| 337 | result = hook.pre_merge_hook( |
| 338 | root, |
| 339 | ours_commit_id="", |
| 340 | theirs_commit_id="", |
| 341 | branch="", |
| 342 | consolidate=True, |
| 343 | ) |
| 344 | port, _ = _resolve_port() |
| 345 | return { |
| 346 | "hook_name": result.hook_name, |
| 347 | "fired": result.fired, |
| 348 | "out_of_band": result.out_of_band, |
| 349 | "message": sanitize_display(result.message or ""), |
| 350 | "error": sanitize_display(result.error) if result.error else None, |
| 351 | "dispatched_at": datetime.now(tz=timezone.utc).isoformat(), |
| 352 | "hub_port": port if port is not None else DEFAULT_PORT, |
| 353 | } |
| 354 | |
| 355 | |
| 356 | def clamp_tool_int(value: Any, default: int, lo: int, hi: int, name: str) -> int: |
| 357 | try: |
| 358 | n = int(value) |
| 359 | except (TypeError, ValueError): |
| 360 | n = default |
| 361 | return clamp_int(n, lo, hi, name) |
| 362 | |
| 363 | |
| 364 | __all__ = [ |
| 365 | "PathValidationError", |
| 366 | "clamp_tool_int", |
| 367 | "consolidate_receipt", |
| 368 | "create_hub_proposal", |
| 369 | "enrich_prime", |
| 370 | "link_index_to_json", |
| 371 | "list_vault_notes", |
| 372 | "projects_summary", |
| 373 | "read_note_at_ref", |
| 374 | "recent_memory_events", |
| 375 | "resolve_ref", |
| 376 | "validate_note_path", |
| 377 | ] |