#!/usr/bin/env python3 """ mcp_enrich.py -- Phase 6f detail-page enrichment for MCP entities. Walks every MCP entity file (``/entities/mcp-servers//.md``) and calls the source's ``fetch_details(slug)`` to pull ``github_url`` and ``stars`` from the detail page. The scraped values are written back into the entity's YAML frontmatter. Why this exists (from Phase 6e close-out): - The listing pages scraped by Phase 6c (``mcp_ingest``) expose slug / name / description / classification but NOT a repo URL. - Without ``github_url``, the quality scorer's popularity and freshness signals stay neutral for every MCP, flattening the A/B/C/D distribution to 100% B+C. - ``ctx-mcp-install`` needs a URL to pass to ``claude mcp add``. - The knowledge graph has no way to link pulsemcp entries to the same repo surfaced elsewhere (awesome-mcp, mcp-get, ...). Checkpoint: ``/.enrich-checkpoint/.json`` — same shape as the Phase 6c ingest checkpoint. Slugs in ``processed`` skip on resume. ``failures`` retry on the next run unless ``--skip-failures``. All fetches go through the existing SSRF-hardened ``fetch_text`` in ``mcp_sources/base.py`` with the pulsemcp-level date-keyed cache, so a second run over the same day is near-instant (cache hit) and a re-run tomorrow refreshes everything naturally. Usage: ctx-mcp-enrich --source pulsemcp # all entities ctx-mcp-enrich --source pulsemcp --limit 50 # first 50 only ctx-mcp-enrich --source pulsemcp --slug foo-bar # single entity ctx-mcp-enrich --source pulsemcp --status # checkpoint report ctx-mcp-enrich --source pulsemcp --reset # delete checkpoint ctx-mcp-enrich --source pulsemcp --dry-run # fetch but don't write """ from __future__ import annotations import argparse import json import logging import os import re import signal import sys import time from datetime import datetime, timezone from pathlib import Path from typing import Any, Iterable from ctx.core.wiki.wiki_packs import load_merged_wiki_pages, write_active_wiki_overlay_pack from ctx.utils._fs_utils import atomic_write_json, reject_symlink_path, safe_atomic_write_text from ctx_config import cfg from mcp_sources import SOURCES _logger = logging.getLogger(__name__) CHECKPOINT_SUBDIR = ".enrich-checkpoint" CHECKPOINT_VERSION = 1 DEFAULT_FLUSH_EVERY = 10 DEFAULT_SLEEP_SECONDS = 0.5 # polite pacing between live fetches _MCP_ENTITY_SUBDIR = Path("entities") / "mcp-servers" # Line-break codepoints that Python's str.splitlines() treats as # boundaries. The renderer neutralises all five so that a quoted # scalar cannot be split into new frontmatter lines by the project's # splitlines-based parser (_parse_entity_frontmatter). Strix # vuln-0001 HIGH. \x85 is NEL, 
 is LINE SEPARATOR, 
 is # PARAGRAPH SEPARATOR — written with \u escapes so editor autonorm # cannot silently convert them to plain space (U+0020). _LINE_SEP_TRANSLATE = str.maketrans({ "\r": " ", "\n": " ", "\x85": " ", "\u2028": " ", "\u2029": " ", }) # ── Graceful exit ──────────────────────────────────────────────────────────── class _GracefulExit: """Mirror of mcp_ingest._GracefulExit — see that module for the rationale (cooperative signal handling between records).""" def __init__(self) -> None: self.requested = False self._prev_int = signal.getsignal(signal.SIGINT) self._prev_term = signal.getsignal(signal.SIGTERM) def install(self) -> None: signal.signal(signal.SIGINT, self._handle) try: signal.signal(signal.SIGTERM, self._handle) except (ValueError, OSError): pass def uninstall(self) -> None: signal.signal(signal.SIGINT, self._prev_int) try: signal.signal(signal.SIGTERM, self._prev_term) except (ValueError, OSError): pass def _handle(self, signum: int, frame: object) -> None: # noqa: ARG002 self.requested = True # ── Checkpoint ─────────────────────────────────────────────────────────────── def _now_iso() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def _checkpoint_path(wiki_path: Path, source: str) -> Path: # Delegate to shared validator. Prior impl only rejected ``/``, ``\\`` # and leading ``.``, which let Windows drive-relative names like # ``C:evil`` through — those resolve against drive C's CWD, not the # wiki, which an attacker could use to overwrite ~/.claude/settings.json # via ``--source "C:...\\settings"``. Security-auditor H-3, fixed here. from ctx.utils._safe_name import validate_source_name # noqa: PLC0415 validate_source_name(source, field="source") return wiki_path / CHECKPOINT_SUBDIR / f"{source}.json" def _empty_checkpoint(source: str) -> dict: now = _now_iso() return { "version": CHECKPOINT_VERSION, "source": source, "started_at": now, "updated_at": now, "total_seen": 0, "processed": {}, "failures": {}, } def load_checkpoint(wiki_path: Path, source: str) -> dict: """Tolerant load — any error resets to a fresh checkpoint. The authoritative state lives in the entity frontmatter itself; a lost or corrupt checkpoint just means the next run re-fetches the detail pages (which hit the cache, so cheap). """ path = _checkpoint_path(wiki_path, source) if not path.is_file(): return _empty_checkpoint(source) try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return _empty_checkpoint(source) if not isinstance(data, dict) or data.get("version") != CHECKPOINT_VERSION: return _empty_checkpoint(source) if data.get("source") != source: return _empty_checkpoint(source) processed = data.get("processed") or {} failures = data.get("failures") or {} if not isinstance(processed, dict) or not isinstance(failures, dict): return _empty_checkpoint(source) return { "version": CHECKPOINT_VERSION, "source": source, "started_at": str(data.get("started_at") or _now_iso()), "updated_at": str(data.get("updated_at") or _now_iso()), "total_seen": int(data.get("total_seen") or 0), "processed": processed, "failures": failures, } def save_checkpoint(wiki_path: Path, checkpoint: dict) -> None: checkpoint["updated_at"] = _now_iso() atomic_write_json(_checkpoint_path(wiki_path, checkpoint["source"]), checkpoint) # ── Entity discovery ───────────────────────────────────────────────────────── def _iter_entities(wiki_path: Path) -> Iterable[Path]: """Yield every ``/entities/mcp-servers//.md`` file. Sort globally for determinism — otherwise a resume after a crash at entity #5,000 might skip ahead or rewind depending on platform shard-iteration order. """ packs_dir = wiki_path / "wiki-packs" if packs_dir.is_dir(): prefix = f"{_MCP_ENTITY_SUBDIR.as_posix()}/" return [ wiki_path / relpath for relpath in sorted(load_merged_wiki_pages(packs_dir)) if relpath.startswith(prefix) and relpath.endswith(".md") ] root = wiki_path / _MCP_ENTITY_SUBDIR if not root.is_dir(): return [] return sorted(root.rglob("*.md")) def _extract_slug_from_path(path: Path) -> str: """Entity files are named ``.md`` — stem is the WIKI slug. Note this is not necessarily the slug used by the upstream source (e.g. pulsemcp): during ingest, non-ASCII names can cause the wiki slug to get mangled (``ignfab-geoportail`` → ``g-oportail``) while the authoritative ``homepage_url`` preserves the real URL. Use ``_source_slug_from_entity`` when you need the slug to pass to ``source.fetch_details``. """ return path.stem # Map {source_name: regex that pulls the upstream slug out of a # homepage_url}. Each regex has one named group ``slug``. _SOURCE_SLUG_PATTERNS: dict[str, re.Pattern[str]] = { "pulsemcp": re.compile( r"^https?://(?:www\.)?pulsemcp\.com/servers/(?P[^/?#\s]+)" ), } def _entity_relpath(wiki_path: Path, entity_path: Path) -> str: return entity_path.relative_to(wiki_path).as_posix() def _load_active_wiki_pack_pages(wiki_path: Path) -> dict[str, str] | None: packs_dir = wiki_path / "wiki-packs" if not packs_dir.is_dir(): return None return load_merged_wiki_pages(packs_dir) def _read_entity_text( wiki_path: Path, entity_path: Path, *, pages: dict[str, str] | None = None, ) -> str | None: relpath = _entity_relpath(wiki_path, entity_path) packs_dir = wiki_path / "wiki-packs" if packs_dir.is_dir(): page_map = pages if pages is not None else load_merged_wiki_pages(packs_dir) if relpath in page_map: return page_map[relpath] if entity_path.exists(): reject_symlink_path(entity_path) return entity_path.read_text(encoding="utf-8", errors="replace") return None def _write_entity_text( wiki_path: Path, entity_path: Path, text: str, *, pages: dict[str, str] | None = None, ) -> None: relpath = _entity_relpath(wiki_path, entity_path) packs_dir = wiki_path / "wiki-packs" if entity_path.exists() or not packs_dir.is_dir(): safe_atomic_write_text(entity_path, text, encoding="utf-8") if packs_dir.is_dir(): write_active_wiki_overlay_pack( packs_dir=packs_dir, pages={relpath: text}, tombstones=[], ) if pages is not None: pages[relpath] = text def _source_slug_from_entity( entity_path: Path, source_name: str, *, wiki_path: Path | None = None, pages: dict[str, str] | None = None, ) -> str | None: """Pull the upstream slug out of the entity's frontmatter. For pulsemcp, parses ``homepage_url``: ``https://www.pulsemcp.com/servers/`` → ````. Returns ``None`` when the entity has no homepage_url, the URL doesn't match the expected shape, or the source isn't registered in ``_SOURCE_SLUG_PATTERNS``. The caller treats None as "skip with a clear failure message" rather than as an error — a partner like glama/mcp-get won't have pulsemcp URLs. """ pattern = _SOURCE_SLUG_PATTERNS.get(source_name) if pattern is None: return None text: str | None if wiki_path is None: try: reject_symlink_path(entity_path) text = entity_path.read_text(encoding="utf-8", errors="replace") except OSError: return None else: text = _read_entity_text(wiki_path, entity_path, pages=pages) if text is None: return None fm_match = _FRONTMATTER_RE.match(text) if fm_match is None: return None url_match = re.search( r"^homepage_url:[ \t]*(?P.+)$", fm_match.group(1), flags=re.MULTILINE, ) if url_match is None: return None url = url_match.group("url").strip().strip('"').strip("'") slug_match = pattern.match(url) return slug_match.group("slug") if slug_match else None # ── Frontmatter update ─────────────────────────────────────────────────────── _FRONTMATTER_RE = re.compile(r"^---\n(.*?)\n---\n", re.DOTALL) def _set_frontmatter_field(text: str, field: str, value: Any) -> str: """Replace or insert ``field: value`` in the YAML frontmatter. Lossy for complex YAML (lists, mappings) but we only touch simple scalar fields (``github_url``, ``stars``, ``updated``), and the frontmatter on these entity files is flat. Uses line-anchored regex so a ``github_url: null`` in the middle doesn't match a hypothetical ``sub_github_url`` key on a following line. """ escaped = re.escape(field) rendered = _render_scalar(value) # Try to replace an existing key. pattern = rf"^{escaped}:[ \t]*.*$" repl = f"{field}: {rendered}" new_text, n = re.subn(pattern, repl, text, count=1, flags=re.MULTILINE) if n: return new_text # Key didn't exist — insert after the opening delimiter. We only # do this inside the frontmatter block to avoid polluting bodies. fm_match = _FRONTMATTER_RE.match(text) if fm_match is None: return text # no frontmatter at all; skip rather than fabricate insert_at = fm_match.end(1) return text[:insert_at] + f"\n{field}: {rendered}" + text[insert_at:] def _render_scalar(value: Any) -> str: """YAML-scalar rendering for the handful of types we write. ``None`` → ``null`` (matches the existing frontmatter convention for unset fields). Strings are double-quoted when they could be misparsed as another YAML type (start with ``-``, contain ``:``); github URLs are alnum+``/.-_`` so they fall in the safe-bare range, but we quote anyway for consistency. SECURITY NOTE: strings are stripped of ``\\n`` / ``\\r`` before rendering to prevent frontmatter injection. Security-auditor H-1: a Source implementation that returned a multi-line URL value (e.g. scraped from a poisoned detail page) would otherwise inject fake YAML keys like ``status: installed`` or ``install_cmd: ...`` which the next ``ctx-mcp-install --force`` would dutifully pick up from the now-poisoned frontmatter. The executable allowlist (commit b79be55) catches most of the blast radius today, but the injection vector must be shut at the writer too. """ if value is None: return "null" if isinstance(value, bool): return "true" if value else "false" if isinstance(value, int): return str(value) if isinstance(value, str): # Neutralise ASCII line breaks (\r\n) AND Unicode line separators # (U+0085 NEL, U+2028 LS, U+2029 PS). Python's str.splitlines() # treats all five as line boundaries, so downstream splitlines- # based parsers (mcp_install._parse_entity_frontmatter, # wiki_utils) would otherwise see a quoted scalar as multiple # frontmatter lines — Strix vuln-0001 HIGH (CWE-116). Mirrors # install_utils._render_scalar so the two stay aligned. sanitised = value.translate(_LINE_SEP_TRANSLATE) # Conservative: quote on the full YAML 1.1 reserved-indicator set, # leading block indicators, or leading/trailing whitespace. The # unquoted path is reserved for simple alphanumeric-style values # that YAML's plain-scalar scanner parses unambiguously. ``<`` is # included so bare ``<<`` is never resolved as YAML's merge tag. # Mirrors install_utils._render_scalar — the two must stay aligned. yaml_structural = set(",[]{}:?#&*!|>%@`=\"'\\<") needs_quote = ( any(ch in sanitised for ch in yaml_structural) or ( sanitised and ( sanitised[0] == "-" or sanitised[0].isspace() or sanitised[-1].isspace() ) ) or sanitised.startswith(("?", "[", "{")) ) if needs_quote: escaped = sanitised.replace("\\", "\\\\").replace('"', '\\"') return f'"{escaped}"' return sanitised # Defensive fallback — stringify and quote. return f'"{str(value)}"' def apply_enrichment( entity_path: Path, enrichment: dict, *, dry_run: bool, wiki_path: Path | None = None, pages: dict[str, str] | None = None, ) -> dict: """Write ``enrichment`` fields into the entity's frontmatter. Returns a diff dict ``{field: (old, new)}`` describing what changed, or an empty dict when nothing needed updating. Updates ``updated`` with today's ISO date only when at least one other field changed — a no-op run shouldn't bump the mtime. """ if not enrichment: return {} if wiki_path is None: reject_symlink_path(entity_path) text = entity_path.read_text(encoding="utf-8", errors="replace") else: read_text = _read_entity_text(wiki_path, entity_path, pages=pages) if read_text is None: return {} text = read_text fm_match = _FRONTMATTER_RE.match(text) if fm_match is None: return {} diff: dict[str, tuple[Any, Any]] = {} for field, new_val in enrichment.items(): # Parse current value cheaply with a line grep rather than # pulling a yaml dependency for three scalar reads. pat = re.compile(rf"^{re.escape(field)}:[ \t]*(?P.*)$", re.MULTILINE) cur_match = pat.search(fm_match.group(0)) current: Any = None if cur_match is not None: raw = cur_match.group("val").strip() if raw in ("", "null", "~"): current = None elif raw.isdigit(): current = int(raw) else: current = raw.strip('"').strip("'") if current != new_val: diff[field] = (current, new_val) text = _set_frontmatter_field(text, field, new_val) if diff and not dry_run: today = datetime.now(timezone.utc).strftime("%Y-%m-%d") text = _set_frontmatter_field(text, "updated", today) if wiki_path is None: safe_atomic_write_text(entity_path, text, encoding="utf-8") else: _write_entity_text(wiki_path, entity_path, text, pages=pages) return diff # ── Core loop ──────────────────────────────────────────────────────────────── def enrich_entities( entity_paths: list[Path], *, source_name: str, wiki_path: Path, checkpoint: dict, refresh: bool = False, dry_run: bool = False, limit: int | None = None, skip_failures: bool = False, flush_every: int = DEFAULT_FLUSH_EVERY, sleep_seconds: float = DEFAULT_SLEEP_SECONDS, graceful: _GracefulExit | None = None, report_progress: bool = True, ) -> dict: """Enrich each entity via ``source.fetch_details(slug)``. Returns the same checkpoint (mutated). """ source = SOURCES.get(source_name) if source is None: raise ValueError( f"unknown source {source_name!r}; known: {sorted(SOURCES)}" ) if not hasattr(source, "fetch_details"): raise NotImplementedError( f"source {source_name!r} does not implement fetch_details()" ) processed = checkpoint["processed"] failures = checkpoint["failures"] pages = _load_active_wiki_pack_pages(wiki_path) attempted = enriched = unchanged = failed = skipped = 0 for path in entity_paths: if limit is not None and attempted >= limit: break if graceful and graceful.requested: break wiki_slug = _extract_slug_from_path(path) if wiki_slug in processed: skipped += 1 continue if wiki_slug in failures and skip_failures: skipped += 1 continue attempted += 1 checkpoint["total_seen"] += 1 source_slug = _source_slug_from_entity( path, source_name, wiki_path=wiki_path, pages=pages, ) if source_slug is None: # Entity has no homepage_url for this source (e.g. ingested # from a different source). Record a skip so we don't # retry; it's not a failure. processed[wiki_slug] = { "result": "no-source-url", "at": _now_iso(), "fields": [], } if report_progress: print( f" [{attempted}] [no-source-url] {wiki_slug}", flush=True, ) if attempted % flush_every == 0: save_checkpoint(wiki_path, checkpoint) continue try: enrichment = source.fetch_details(source_slug, refresh=refresh) except Exception as exc: # noqa: BLE001 — batch must continue failed += 1 failures[wiki_slug] = { "error": f"{type(exc).__name__}: {exc}", "at": _now_iso(), "source_slug": source_slug, } if report_progress: print( f" [{attempted}] [FAIL] {wiki_slug} " f"(source={source_slug}): {type(exc).__name__}", flush=True, ) if attempted % flush_every == 0: save_checkpoint(wiki_path, checkpoint) continue try: diff = apply_enrichment( path, enrichment, dry_run=dry_run, wiki_path=wiki_path, pages=pages, ) except Exception as exc: # noqa: BLE001 failed += 1 failures[wiki_slug] = { "error": f"apply: {type(exc).__name__}: {exc}", "at": _now_iso(), "source_slug": source_slug, } if report_progress: print( f" [{attempted}] [APPLY-FAIL] {wiki_slug}: {exc}", flush=True, ) continue if diff: enriched += 1 outcome = "enriched" failures.pop(wiki_slug, None) elif enrichment: unchanged += 1 outcome = "unchanged" else: unchanged += 1 outcome = "no-repo" processed[wiki_slug] = { "result": outcome, "at": _now_iso(), "fields": list(enrichment.keys()) if enrichment else [], "source_slug": source_slug, } if report_progress: fields = ",".join(enrichment.keys()) if enrichment else "none" print( f" [{attempted}] [{outcome}] {wiki_slug} " f"(source={source_slug}, fields={fields})", flush=True, ) if attempted % flush_every == 0: save_checkpoint(wiki_path, checkpoint) # Polite pacing — only between LIVE fetches, not cache hits. # We can't cheaply tell from here whether fetch_text hit cache, # so use a short default (0.5s) that's fine either way. if sleep_seconds > 0: time.sleep(sleep_seconds) save_checkpoint(wiki_path, checkpoint) if report_progress: tail = " (interrupted)" if graceful and graceful.requested else "" print( f"\nEnrich summary{tail}: attempted={attempted}, enriched={enriched}, " f"unchanged/no-repo={unchanged}, failed={failed}, skipped_resume={skipped}", flush=True, ) return checkpoint # ── CLI ────────────────────────────────────────────────────────────────────── def _print_status(wiki_path: Path, source: str) -> None: cp = load_checkpoint(wiki_path, source) print(f"source: {cp['source']}") print(f"started_at: {cp['started_at']}") print(f"updated_at: {cp['updated_at']}") print(f"total_seen: {cp['total_seen']}") print(f"processed: {len(cp['processed'])}") print(f"failures: {len(cp['failures'])}") # Breakdown by outcome. outcomes: dict[str, int] = {} for entry in cp["processed"].values(): result = str(entry.get("result", "?")) outcomes[result] = outcomes.get(result, 0) + 1 if outcomes: print("\nProcessed breakdown:") for k in sorted(outcomes): print(f" {k}: {outcomes[k]}") if cp["failures"]: print("\nLast 5 failures:") for slug, info in list(cp["failures"].items())[-5:]: print(f" {slug}: {info['error'][:120]}") def _build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="ctx-mcp-enrich", description=( "Enrich MCP entity frontmatter from per-source detail pages. " "Currently supports --source pulsemcp (github_url + stars)." ), ) parser.add_argument("--source", required=True, help="Source name (e.g. pulsemcp)") group = parser.add_mutually_exclusive_group() group.add_argument("--slug", help="Enrich one slug only (dry-run candidate)") group.add_argument( "--status", action="store_true", help="Print checkpoint summary and exit", ) group.add_argument( "--reset", action="store_true", help="Delete the checkpoint for --source before starting", ) parser.add_argument("--limit", type=int, default=None, help="Cap the number of entities attempted this run") parser.add_argument("--refresh", action="store_true", help="Bypass the raw detail-page cache") parser.add_argument("--dry-run", action="store_true", help="Fetch but do not write any frontmatter") parser.add_argument( "--skip-failures", action="store_true", help="Do not retry slugs already present in checkpoint.failures", ) parser.add_argument( "--flush-every", type=int, default=DEFAULT_FLUSH_EVERY, help=f"Checkpoint flush cadence (default {DEFAULT_FLUSH_EVERY})", ) parser.add_argument( "--sleep", type=float, default=DEFAULT_SLEEP_SECONDS, help=f"Seconds between fetches (default {DEFAULT_SLEEP_SECONDS})", ) parser.add_argument("--wiki", default=str(cfg.wiki_dir), help="Wiki root") parser.add_argument("--quiet", action="store_true", help="Suppress per-entity progress lines") return parser def _force_utf8_stdio() -> None: """Mirror of mcp_fetch/mcp_ingest helper — Windows cp1252 crashes on CJK/emoji in pulsemcp descriptions, so force UTF-8 at entry.""" for stream in (sys.stdout, sys.stderr): reconfigure = getattr(stream, "reconfigure", None) if reconfigure is None: continue try: reconfigure(encoding="utf-8", errors="replace") except (OSError, ValueError): pass def main() -> None: _force_utf8_stdio() parser = _build_parser() args = parser.parse_args() wiki_path = Path(os.path.expanduser(args.wiki)) if args.status: _print_status(wiki_path, args.source) sys.exit(0) if args.reset: try: _checkpoint_path(wiki_path, args.source).unlink() print(f"Reset checkpoint for {args.source!r}.", file=sys.stderr) except FileNotFoundError: pass checkpoint = load_checkpoint(wiki_path, args.source) if args.slug: # Single-slug path: bypass the discovery loop, build a one-file list. root = wiki_path / _MCP_ENTITY_SUBDIR # Shard lookup mirrors McpRecord.entity_relpath. shard = args.slug[0] if args.slug and args.slug[0].isalpha() else "0-9" entity_paths = [root / shard / f"{args.slug}.md"] if _read_entity_text(wiki_path, entity_paths[0]) is None: print( f"Error: no entity at {entity_paths[0]} — has it been ingested?", file=sys.stderr, ) sys.exit(1) else: entity_paths = list(_iter_entities(wiki_path)) if not entity_paths: print("No MCP entities found — run ctx-mcp-ingest first.", file=sys.stderr) sys.exit(1) graceful = _GracefulExit() graceful.install() try: enrich_entities( entity_paths, source_name=args.source, wiki_path=wiki_path, checkpoint=checkpoint, refresh=args.refresh, dry_run=args.dry_run, limit=args.limit, skip_failures=args.skip_failures, flush_every=args.flush_every, sleep_seconds=args.sleep, graceful=graceful, report_progress=not args.quiet, ) finally: graceful.uninstall() sys.exit(1 if checkpoint["failures"] else 0) if __name__ == "__main__": main()