From caa4a5079ac1a17541ae2064ffbac935d4fa7f63 Mon Sep 17 00:00:00 2001 From: John Lancaster <32917998+jsl12@users.noreply.github.com> Date: Sun, 21 Jun 2026 10:55:44 -0500 Subject: [PATCH] WIP loading --- src/personal_mcp/registry/file.py | 31 ++++++++----- src/personal_mcp/registry/skills.py | 70 +++++++++++++++++------------ 2 files changed, 63 insertions(+), 38 deletions(-) diff --git a/src/personal_mcp/registry/file.py b/src/personal_mcp/registry/file.py index 0b763fb..3a6783e 100644 --- a/src/personal_mcp/registry/file.py +++ b/src/personal_mcp/registry/file.py @@ -1,5 +1,6 @@ from collections.abc import Generator from dataclasses import dataclass +from dataclasses import field from importlib.abc import Traversable from itertools import starmap from pathlib import PurePosixPath @@ -7,20 +8,21 @@ from typing import Self @dataclass(frozen=True, slots=True) -class LoadedMarkdownDocument: +class MarkdownDocument: """Represents a loaded markdown document with its content and frontmatter.""" relpath: PurePosixPath """The relative path of the document within the package resources.""" - content: str + content: str = field(repr=False) """The raw markdown content of the document.""" - frontmatter: str + frontmatter: str | None = field(repr=False, default=None) """The raw YAML frontmatter of the document, if present.""" @classmethod - def from_root(cls, root: Traversable) -> set[Self]: + def from_root(cls, root: Traversable): """Recursively load all markdown documents from the root resource.""" - return set(starmap(cls.from_resource, walk_resources(root))) + mapped = starmap(cls.from_resource, walk_resources(root)) + return {d.relpath: d for d in mapped} @classmethod def from_resource(cls, relpath: PurePosixPath, resource: Traversable) -> Self: @@ -29,13 +31,19 @@ class LoadedMarkdownDocument: frontmatter = get_raw_frontmatter(raw) return cls(relpath=relpath, content=raw, frontmatter=frontmatter) + @property + def skill_slug(self) -> str | None: + parts = self.relpath.parts + if parts[0] == "skills" and len(parts) >= 3: + return parts[1] + def walk_resources( node: Traversable, *, suffix: str = ".md", prefix: PurePosixPath | None = None, -) -> Generator[tuple[str, Traversable]]: +) -> Generator[tuple[PurePosixPath, Traversable]]: """Recursively yield all resources in node, with their full path.""" prefix = prefix if prefix is not None else PurePosixPath() for child in sorted(node.iterdir(), key=lambda item: item.name): @@ -45,13 +53,16 @@ def walk_resources( continue if not child.is_file() or not child.name.lower().endswith(suffix): continue - yield relpath.as_posix(), child + yield relpath, child -def get_raw_frontmatter(raw: str) -> str: +def get_raw_frontmatter(raw: str) -> str | None: delimiter = iter(get_frontmatter_delim_idx(raw, delimiter="---")) - start = next(delimiter) + 1 - end = next(delimiter) + try: + start = next(delimiter) + 1 + end = next(delimiter) + except StopIteration: + return None return "\n".join(raw.splitlines()[start:end]) diff --git a/src/personal_mcp/registry/skills.py b/src/personal_mcp/registry/skills.py index bc45a06..0e5db8a 100644 --- a/src/personal_mcp/registry/skills.py +++ b/src/personal_mcp/registry/skills.py @@ -1,43 +1,57 @@ -from collections.abc import Generator +from collections.abc import Iterable from dataclasses import dataclass from fnmatch import fnmatch -from functools import partial +from importlib.resources.abc import Traversable from itertools import groupby +from itertools import starmap +from typing import Self - -def get_skill_filemap(content: dict[str, str]) -> dict[str, list[str]]: - """Get the map of skill slugs to their associated files.""" - - def get_skill_name(relpath: str) -> str | None: - if relpath.startswith("skills/"): - return relpath.split("/")[1] - - grouped = groupby(content.keys(), key=get_skill_name) - return {k: list(v) for k, v in grouped if k is not None} +from .file import MarkdownDocument @dataclass(frozen=True, slots=True) -class SkillBundle: +class SkillFilesBundle: + """Represents a skill and all of its associated markdown files.""" + slug: str - skill: str - references: list[str] - other: list[str] + skill: MarkdownDocument + references: tuple[MarkdownDocument, ...] + other: tuple[MarkdownDocument, ...] + @classmethod + def from_root(cls, root: Traversable) -> list[Self]: + # Should only be used for testing + return list(cls.from_docs(MarkdownDocument.from_root(root).values())) -def gen_skill_bundles(content: dict[str, str]) -> Generator[SkillBundle]: - """Generate the skill bundles from the map of raw markdown content.""" - matcher = partial(fnmatch, pat="skills/*/references/*.md") - for slug, paths in get_skill_filemap(content).items(): - skill = next(iter(p for p in paths if fnmatch(p, "skills/*/SKILL.md"))) - groups = {k: list(v) for k, v in groupby(paths, key=matcher)} - yield SkillBundle( + @classmethod + def from_docs(cls, docs: Iterable[MarkdownDocument]) -> tuple[Self, ...]: + return tuple(starmap(cls.from_paths, group_skill_paths(docs).items())) + + @classmethod + def from_paths(cls, slug: str, paths: set[MarkdownDocument]) -> Self: + skill = next(iter(p for p in paths if p.relpath.name == "SKILL.md")) + groups = { + k: tuple(v) + for k, v in groupby( + paths, + key=lambda p: fnmatch(p.relpath.as_posix(), f"skills/{slug}/references/*.md"), + ) + } + references = groups.get(True, ()) + other = tuple(p for p in groups.get(False, ()) if p != skill) + return cls( slug=slug, skill=skill, - references=list(groups.get(True, [])), - other=[f for f in groups.get(False, []) if f != skill], + references=references, + other=other, ) -def get_all_skill_bundles(content: dict[str, str]) -> list[SkillBundle]: - """Get all skill bundles from the map of raw markdown content.""" - return list(gen_skill_bundles(content)) +def group_skill_paths(docs: Iterable[MarkdownDocument]) -> dict[str, set[MarkdownDocument]]: + """Group skills from a list of markdown documents by their skill slug.""" + s = sorted( + filter(lambda d: d.skill_slug is not None, docs), + key=lambda d: (d.skill_slug or "", d.relpath.stem), + ) + grouped = groupby(s, key=lambda doc: doc.skill_slug) + return {k: set(g) for k, g in grouped if k}