john/prompts
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

changed to skill provider

Browse Source
This commit is contained in:
John Lancaster
2026-08-07 19:09:54 -05:00
parent c6817a074e
commit 44edffb8b7
48 changed files with 552 additions and 2946 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/personal_mcp/catalog/__init__.py
-8
View File
@@ -1,19 +1,11 @@
from personal_mcp.catalog.server import build_prompt_detail_payload
from personal_mcp.catalog.server import build_prompts_index_payload
from personal_mcp.catalog.server import build_skill_detail_payload
from personal_mcp.catalog.server import build_skills_index_payload
from personal_mcp.catalog.server import get_pattern_by_id_payload
from personal_mcp.catalog.server import get_prompt_by_id_payload
from personal_mcp.catalog.server import search_patterns_payload
from personal_mcp.catalog.server import search_prompts_payload
__all__ = [
"build_prompt_detail_payload",
"build_prompts_index_payload",
"build_skill_detail_payload",
"build_skills_index_payload",
"get_pattern_by_id_payload",
"get_prompt_by_id_payload",
"search_patterns_payload",
"search_prompts_payload",
]
src/personal_mcp/catalog/server.py
-125
View File
@@ -5,42 +5,11 @@ from typing import Any
from personal_mcp.registry.models.registry import DocsRegistry
from personal_mcp.registry.models.registry import PromptRecord
from personal_mcp.registry.models.registry import PromptSummaryPayload
from personal_mcp.registry.models.registry import SkillPatternPayload
from personal_mcp.registry.models.registry import SkillRecord
from personal_mcp.registry.models.registry import SkillSummaryPayload
DEFAULT_LIMIT = 20
MAX_LIMIT = 100
def _skill_matches(
skill: SkillRecord,
*,
query: str | None,
tag: str | None,
capability: str | None,
) -> bool:
if query:
lowered = query.strip().lower()
if lowered:
haystack = " ".join(
[
skill.skill_id,
skill.name,
skill.description,
" ".join(skill.tags),
]
).lower()
terms = [term for term in lowered.replace("-", " ").split() if term]
if any(term not in haystack for term in terms):
return False
if tag and tag not in skill.tags:
return False
return not (capability and capability not in skill.capabilities)
def _prompt_matches(
prompt: PromptRecord,
*,
@@ -66,63 +35,6 @@ def _prompt_matches(
return not (tag and tag not in prompt.tags)
def build_skills_index_payload(
registry: DocsRegistry,
*,
query: str | None = None,
tag: str | None = None,
capability: str | None = None,
cursor: str | None = None,
limit: int | None = None,
) -> dict[str, Any]:
normalized_limit = DEFAULT_LIMIT if limit is None else max(1, min(limit, MAX_LIMIT))
try:
start = 0 if cursor is None else max(0, int(cursor))
except ValueError as exc:
raise ValueError("cursor must be an integer string") from exc
ordered = [registry.skills_by_id[skill_id] for skill_id in registry.skills_in_load_order]
matches = [skill for skill in ordered if _skill_matches(skill, query=query, tag=tag, capability=capability)]
page = matches[start : start + normalized_limit]
next_cursor = start + normalized_limit
return {
"skills": [SkillSummaryPayload.from_record(skill).model_dump() for skill in page],
"total": len(matches),
"cursor": str(start),
"limit": normalized_limit,
"next_cursor": str(next_cursor) if next_cursor < len(matches) else None,
}
def build_skill_detail_payload(registry: DocsRegistry, skill_id: str) -> dict[str, Any]:
if skill_id not in registry.skills_by_id:
raise KeyError(skill_id)
skill = registry.skills_by_id[skill_id]
return {
"id": skill.skill_id,
"name": skill.name,
"description": skill.description,
"version": skill.version,
"tags": list(skill.tags),
"capabilities": list(skill.capabilities),
"resources": {
"document": skill.document_uri,
"references": {
ref_id: {
"uri": ref.uri,
"mime_type": ref.mime_type,
"title": ref.title,
"path": ref.relpath.as_posix(),
}
for ref_id, ref in sorted(skill.references.items())
},
},
}
def build_prompts_index_payload(
registry: DocsRegistry,
*,
@@ -173,43 +85,6 @@ def build_prompt_detail_payload(registry: DocsRegistry, prompt_id: str) -> dict[
}
def search_patterns_payload(
registry: DocsRegistry,
*,
query: str = "",
tags: list[str] | None = None,
skip: int = 0,
limit: int = DEFAULT_LIMIT,
) -> dict[str, Any]:
normalized_skip = max(skip, 0)
normalized_limit = max(1, min(limit, MAX_LIMIT))
requested_tags = [tag.strip() for tag in (tags or []) if tag and tag.strip()]
matches: list[SkillRecord] = []
for skill_id in registry.skills_in_load_order:
skill = registry.skills_by_id[skill_id]
if not _skill_matches(skill, query=query, tag=None, capability=None):
continue
if requested_tags and any(tag not in skill.tags for tag in requested_tags):
continue
matches.append(skill)
page = matches[normalized_skip : normalized_skip + normalized_limit]
return {
"patterns": [SkillPatternPayload.from_record(skill).model_dump() for skill in page],
"total": len(matches),
"skip": normalized_skip,
"limit": normalized_limit,
}
def get_pattern_by_id_payload(registry: DocsRegistry, skill_id: str) -> dict[str, Any]:
if skill_id not in registry.skills_by_id:
return {"found": False, "id": skill_id}
return {"found": True, "pattern": SkillPatternPayload.from_record(registry.skills_by_id[skill_id]).model_dump()}
def search_prompts_payload(
registry: DocsRegistry,
*,
src/personal_mcp/mcp.py
+2 -96
View File
@@ -14,18 +14,13 @@ from fastmcp.server.transforms.search import RegexSearchTransform
from personal_mcp.catalog.server import build_prompt_detail_payload
from personal_mcp.catalog.server import build_prompts_index_payload
from personal_mcp.catalog.server import build_skill_detail_payload
from personal_mcp.catalog.server import build_skills_index_payload
from personal_mcp.catalog.server import get_pattern_by_id_payload
from personal_mcp.catalog.server import get_prompt_by_id_payload
from personal_mcp.catalog.server import search_patterns_payload
from personal_mcp.catalog.server import search_prompts_payload
from personal_mcp.registry.load import get_docs_registry
from personal_mcp.registry.models.registry import DocsRegistry
from personal_mcp.registry.read import read_docs_markdown_path
from personal_mcp.registry.read import read_prompt_document
from personal_mcp.registry.read import read_skill_document
from personal_mcp.registry.read import read_skill_reference
from personal_mcp.skills import create_skills_provider
TOOL_SEARCH_MODE = os.getenv("PERSONAL_MCP_TOOL_SEARCH", "none").strip().lower()
TOOL_SEARCH_MAX_RESULTS = os.getenv("PERSONAL_MCP_TOOL_SEARCH_MAX_RESULTS", "5")
@@ -123,64 +118,6 @@ def _register_prompt_objects(mcp: FastMCP, registry: DocsRegistry) -> None:
def _register_components(mcp: FastMCP, registry: DocsRegistry) -> None:
@mcp.resource(
"resource://catalog/skills_index",
mime_type="application/json",
tags={"catalog"},
annotations=_ro_annotations(),
)
def skills_index() -> dict[str, Any]:
return build_skills_index_payload(registry)
@mcp.resource(
"resource://catalog/skills_index{?q,tag,capability,cursor,limit}",
mime_type="application/json",
tags={"catalog"},
annotations=_ro_annotations(),
)
def skills_index_query(
q: str | None = None,
tag: str | None = None,
capability: str | None = None,
cursor: str | None = None,
limit: int | None = None,
) -> dict[str, Any]:
return build_skills_index_payload(
registry,
query=q,
tag=tag,
capability=capability,
cursor=cursor,
limit=limit,
)
@mcp.resource(
"resource://catalog/skills/{skill_id}",
mime_type="application/json",
tags={"catalog"},
annotations=_ro_annotations(),
)
def skill_detail(skill_id: str) -> dict[str, Any]:
return build_skill_detail_payload(registry, skill_id)
@mcp.resource(
"resource://skills/{skill_id}/document",
mime_type="text/markdown",
tags={"skill-doc"},
annotations=_ro_annotations(),
)
def skill_document(skill_id: str) -> dict[str, str]:
return read_skill_document(registry, skill_id)
@mcp.resource(
"resource://skills/{skill_id}/references/{ref_id}",
mime_type="text/markdown",
tags={"reference"},
annotations=_ro_annotations(),
)
def skill_reference(skill_id: str, ref_id: str) -> dict[str, str]:
return read_skill_reference(registry, skill_id=skill_id, ref_id=ref_id)
@mcp.resource(
"resource://docs/{path*}",
mime_type="text/markdown",
@@ -237,38 +174,6 @@ def _register_components(mcp: FastMCP, registry: DocsRegistry) -> None:
def prompt_document(prompt_id: str) -> dict[str, str]:
return read_prompt_document(registry, prompt_id)
@mcp.tool
def search_patterns(
query: str = "",
tags: list[str] | None = None,
skip: int = 0,
limit: int = 20,
) -> dict[str, Any]:
"""Search normalized pattern metadata with optional tags and pagination."""
return search_patterns_payload(
registry,
query=query,
tags=tags,
skip=skip,
limit=limit,
)
@mcp.tool
def get_pattern_by_id(id: str) -> dict[str, Any]:
"""Return one normalized pattern by stable id."""
return get_pattern_by_id_payload(registry, id)
@mcp.tool
def get_skill_document_by_id(skill_id: str) -> dict[str, Any]:
"""Return the canonical skill document payload for a stable skill id."""
if skill_id not in registry.skills_by_id:
return {"found": False, "id": skill_id}
return {
"found": True,
"document": read_skill_document(registry, skill_id),
}
@mcp.tool
def search_prompts(
query: str = "",
@@ -296,5 +201,6 @@ def create_mcp() -> FastMCP:
mcp = FastMCP("personal-mcp", on_duplicate="error")
_register_components(mcp, registry)
_register_prompt_objects(mcp, registry)
mcp.add_provider(create_skills_provider())
_install_tool_fallback_transforms(mcp)
return mcp
src/personal_mcp/registry/contracts.py
-6
View File
@@ -1,15 +1,9 @@
from .models.registry import DocsRegistry
from .models.registry import PromptRecord
from .models.registry import PromptSummaryRecord
from .models.registry import ReferenceRecord
from .models.registry import SkillRecord
from .models.registry import SkillSummaryRecord
__all__ = [
"DocsRegistry",
"PromptRecord",
"PromptSummaryRecord",
"ReferenceRecord",
"SkillRecord",
"SkillSummaryRecord",
]
src/personal_mcp/registry/ingest/document.py
-6
View File
@@ -38,12 +38,6 @@ class MarkdownDocument:
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]
@property
def prompt_slug(self) -> str | None:
parts = self.relpath.parts
src/personal_mcp/registry/ingest/skill.py
-125
View File
@@ -1,125 +0,0 @@
import re
from collections.abc import Iterable
from collections.abc import Mapping
from dataclasses import dataclass
from importlib.resources.abc import Traversable
from itertools import groupby
from itertools import starmap
from pathlib import PurePosixPath
from typing import Self
from personal_mcp.registry.models.common import SKILL_ID_RE
from personal_mcp.registry.models.common import DocsPath
from personal_mcp.registry.models.common import ReferenceEntry
from personal_mcp.registry.models.skill import SkillFrontmatter
from personal_mcp.registry.models.skill import StoredSkill
from personal_mcp.registry.models.skill import StoredSkillReference
from .document import MarkdownDocument
@dataclass(frozen=True, slots=True)
class SkillFilesBundle:
"""Represents a skill and all of its associated markdown files."""
slug: 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()))
@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"))
sorted_paths = tuple(sorted(paths, key=lambda p: p.relpath))
references_dir = PurePosixPath("skills", slug, "references")
references = tuple(p for p in sorted_paths if p.relpath.parent == references_dir)
other = tuple(p for p in sorted_paths if p not in references and p != skill)
return cls(
slug=slug,
skill=skill,
references=references,
other=other,
)
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}
def _title_from_reference_filename(filename: str) -> str:
stem = PurePosixPath(filename).stem
normalized = stem.replace("-", " ").replace("_", " ").split()
if not normalized:
return stem
return " ".join(token.capitalize() for token in normalized)
def _reference_id_from_filename(filename: str) -> str | None:
stem = PurePosixPath(filename).stem.strip().lower().replace("_", "-")
normalized = re.sub(r"[^a-z0-9-]+", "-", stem)
normalized = re.sub(r"-+", "-", normalized).strip("-")
if not normalized or not SKILL_ID_RE.fullmatch(normalized):
return None
return normalized
def _discover_reference_entries(bundle: SkillFilesBundle) -> dict[str, ReferenceEntry]:
discovered: dict[str, ReferenceEntry] = {}
for reference_doc in bundle.references:
ref_id = _reference_id_from_filename(reference_doc.relpath.name)
if ref_id is None:
continue
discovered[ref_id] = ReferenceEntry(
path=PurePosixPath("references", reference_doc.relpath.name),
title=_title_from_reference_filename(reference_doc.relpath.name),
)
return discovered
def build_stored_skill(
*,
bundle: SkillFilesBundle,
docs_by_relpath: Mapping[DocsPath, MarkdownDocument],
) -> StoredSkill:
frontmatter = SkillFrontmatter.from_raw_yaml(bundle.skill.frontmatter)
metadata = frontmatter.x_personal_mcp
merged_entries = _discover_reference_entries(bundle)
merged_entries.update(dict(metadata.references))
references: dict[str, StoredSkillReference] = {}
for ref_id, entry in sorted(merged_entries.items()):
ref_relpath = PurePosixPath("skills", bundle.slug, entry.path)
if ref_relpath not in docs_by_relpath:
raise KeyError(f"reference document not found for '{metadata.id}:{ref_id}' at {ref_relpath.as_posix()}")
ref_doc = docs_by_relpath[ref_relpath]
references[ref_id] = StoredSkillReference(
ref_id=ref_id,
relpath=ref_relpath,
content=ref_doc.content,
entry=entry,
)
return StoredSkill.model_validate(
{
"skill_id": metadata.id,
"relpath": bundle.skill.relpath,
"content": bundle.skill.content,
"frontmatter": frontmatter,
"references": references,
}
)
src/personal_mcp/registry/load.py
-76
View File
@@ -6,44 +6,10 @@ from importlib.resources import files
from personal_mcp.registry.ingest.document import MarkdownDocument
from personal_mcp.registry.ingest.prompt import PromptFilesBundle
from personal_mcp.registry.ingest.skill import SkillFilesBundle
from personal_mcp.registry.ingest.skill import build_stored_skill
from personal_mcp.registry.models.common import DocsPath
from personal_mcp.registry.models.prompt import StoredPrompt
from personal_mcp.registry.models.registry import DocsRegistry
from personal_mcp.registry.models.registry import PromptRecord
from personal_mcp.registry.models.registry import PromptSummaryRecord
from personal_mcp.registry.models.registry import ReferenceRecord
from personal_mcp.registry.models.registry import SkillRecord
from personal_mcp.registry.models.registry import SkillSummaryRecord
def _build_skill_record(*, bundle: SkillFilesBundle, docs_by_relpath: dict[DocsPath, MarkdownDocument]) -> SkillRecord:
stored = build_stored_skill(bundle=bundle, docs_by_relpath=docs_by_relpath)
metadata = stored.frontmatter.x_personal_mcp
references: dict[str, ReferenceRecord] = {}
for ref_id, ref in sorted(stored.references.items()):
references[ref_id] = ReferenceRecord(
ref_id=ref_id,
uri=f"resource://skills/{metadata.id}/references/{ref_id}",
relpath=ref.relpath,
mime_type=ref.entry.mime_type,
title=ref.entry.title,
content=ref.content,
)
return SkillRecord(
skill_id=metadata.id,
name=stored.frontmatter.name,
description=stored.frontmatter.description,
version=metadata.version,
tags=tuple(metadata.tags),
capabilities=tuple(metadata.capabilities),
document_uri=f"resource://skills/{metadata.id}/document",
document_relpath=stored.relpath,
document_content=stored.content,
references=references,
)
def _build_prompt_record(*, bundle: PromptFilesBundle) -> PromptRecord:
@@ -64,26 +30,6 @@ def _build_prompt_record(*, bundle: PromptFilesBundle) -> PromptRecord:
)
def _build_tag_index_skills(
skills_in_order: tuple[str, ...], skills_by_id: dict[str, SkillRecord]
) -> dict[str, tuple[str, ...]]:
tag_index: defaultdict[str, list[str]] = defaultdict(list)
for skill_id in skills_in_order:
for tag in skills_by_id[skill_id].tags:
tag_index[tag].append(skill_id)
return {tag: tuple(ids) for tag, ids in sorted(tag_index.items())}
def _build_capability_index(
skills_in_order: tuple[str, ...], skills_by_id: dict[str, SkillRecord]
) -> dict[str, tuple[str, ...]]:
capability_index: defaultdict[str, list[str]] = defaultdict(list)
for skill_id in skills_in_order:
for capability in skills_by_id[skill_id].capabilities:
capability_index[capability].append(skill_id)
return {capability: tuple(ids) for capability, ids in sorted(capability_index.items())}
def _build_tag_index_prompts(
prompts_in_order: tuple[str, ...],
prompts_by_id: dict[str, PromptRecord],
@@ -104,44 +50,22 @@ def get_docs_registry() -> DocsRegistry:
docs_markdown_by_path = {relpath: doc.content for relpath, doc in docs.items()}
skill_bundles = SkillFilesBundle.from_docs(docs.values())
prompt_bundles = PromptFilesBundle.from_docs(docs.values())
docs_by_relpath = {doc.relpath: doc for doc in docs.values()}
skills_by_id: dict[str, SkillRecord] = {}
skills_in_load_order: list[str] = []
for bundle in skill_bundles:
record = _build_skill_record(bundle=bundle, docs_by_relpath=docs_by_relpath)
if record.skill_id in skills_by_id:
raise ValueError(f"duplicate skill_id detected: {record.skill_id}")
skills_by_id[record.skill_id] = record
skills_in_load_order.append(record.skill_id)
prompts_by_id: dict[str, PromptRecord] = {}
prompts_in_load_order: list[str] = []
for bundle in prompt_bundles:
record = _build_prompt_record(bundle=bundle)
if record.prompt_id in prompts_by_id:
raise ValueError(f"duplicate prompt_id detected: {record.prompt_id}")
if record.prompt_id in skills_by_id:
raise ValueError(f"prompt_id collides with existing skill_id: {record.prompt_id}")
prompts_by_id[record.prompt_id] = record
prompts_in_load_order.append(record.prompt_id)
skills_in_order_tuple = tuple(skills_in_load_order)
prompts_in_order_tuple = tuple(prompts_in_load_order)
return DocsRegistry(
skills_by_id=skills_by_id,
skills_in_load_order=skills_in_order_tuple,
skills_summary_in_load_order=tuple(
SkillSummaryRecord.from_record(skills_by_id[skill_id]) for skill_id in skills_in_order_tuple
),
docs_markdown_by_path=docs_markdown_by_path,
docs_markdown_path_index=tuple(sorted(docs_markdown_by_path)),
tag_to_skill_ids=_build_tag_index_skills(skills_in_order_tuple, skills_by_id),
capability_to_skill_ids=_build_capability_index(skills_in_order_tuple, skills_by_id),
prompts_by_id=prompts_by_id,
prompts_in_load_order=prompts_in_order_tuple,
prompts_summary_in_load_order=tuple(
src/personal_mcp/registry/models/common.py
-16
View File
@@ -45,20 +45,4 @@ def parse_docs_path(value: str | PurePosixPath) -> PurePosixPath:
return path
def parse_reference_path(value: str | PurePosixPath) -> PurePosixPath:
path = parse_docs_path(value)
if len(path.parts) < 2 or path.parts[0] != "references":
raise ValueError("reference path must stay under references/")
return path
type DocsPath = Annotated[PurePosixPath, BeforeValidator(parse_docs_path)]
type ReferencePath = Annotated[PurePosixPath, BeforeValidator(parse_reference_path)]
class ReferenceEntry(StrictFrozenModel):
"""Reference metadata for a markdown file within a skill."""
path: ReferencePath
mime_type: str = "text/markdown"
title: str | None = None
src/personal_mcp/registry/models/registry.py
+1 -126
View File
@@ -13,61 +13,6 @@ def _empty_docs_mapping() -> Mapping[DocsPath, str]:
return frozen_mapping()
class ReferenceRecord(StrictFrozenModel):
"""Registry record for a resolved skill reference document."""
ref_id: str
uri: str
relpath: DocsPath
mime_type: str
title: str | None
content: str
class SkillRecord(StrictFrozenModel):
"""Registry record containing a fully resolved skill and references."""
skill_id: str
name: str
description: str
version: str
tags: tuple[str, ...]
capabilities: tuple[str, ...]
document_uri: str
document_relpath: DocsPath
document_content: str
references: Mapping[str, ReferenceRecord] = Field(default_factory=frozen_mapping)
@field_validator("references", mode="before")
@classmethod
def freeze_references(cls, value: Mapping[str, ReferenceRecord] | None) -> Mapping[str, ReferenceRecord]:
return frozen_mapping(value)
class SkillSummaryRecord(StrictFrozenModel):
"""Compact skill summary exposed by catalog listing APIs."""
skill_id: str
name: str
description: str
tags: tuple[str, ...]
capabilities: tuple[str, ...]
document_uri: str
version: str
@classmethod
def from_record(cls, record: SkillRecord) -> "SkillSummaryRecord":
return cls(
skill_id=record.skill_id,
name=record.name,
description=record.description,
tags=record.tags,
capabilities=record.capabilities,
document_uri=record.document_uri,
version=record.version,
)
class PromptRecord(StrictFrozenModel):
"""Registry record containing a fully resolved prompt document."""
@@ -112,63 +57,6 @@ class PromptSummaryRecord(StrictFrozenModel):
)
class SkillPatternPayload(StrictFrozenModel):
"""Catalog payload model for skill pattern search results."""
id: str
name: str
version: str
description: str
tags: list[str]
capabilities: list[str]
resources: list[str]
@classmethod
def from_record(cls, record: SkillRecord) -> "SkillPatternPayload":
return cls(
id=record.skill_id,
name=record.name,
version=record.version,
description=record.description,
tags=list(record.tags),
capabilities=list(record.capabilities),
resources=list(record.capabilities),
)
class SkillSummaryPayload(StrictFrozenModel):
"""Catalog payload model for skill index summaries."""
id: str
name: str
description: str
tags: list[str]
capabilities: list[str]
version: str
document_uri: str
detail_uri: str
resources: dict[str, str | list[str]]
@classmethod
def from_record(cls, record: SkillRecord) -> "SkillSummaryPayload":
return cls(
id=record.skill_id,
name=record.name,
description=record.description,
tags=list(record.tags),
capabilities=list(record.capabilities),
version=record.version,
document_uri=record.document_uri,
detail_uri=f"resource://catalog/skills/{record.skill_id}",
resources={
"document": record.document_uri,
"references": [
f"resource://skills/{record.skill_id}/references/{ref_id}" for ref_id in sorted(record.references)
],
},
)
class PromptSummaryPayload(StrictFrozenModel):
"""Catalog payload model for prompt index summaries."""
@@ -196,22 +84,12 @@ class PromptSummaryPayload(StrictFrozenModel):
class DocsRegistry(StrictFrozenModel):
"""In-memory index of loaded skills, prompts, and docs content."""
"""In-memory index of loaded prompts and documentation content."""
skills_by_id: Mapping[str, SkillRecord] = Field(default_factory=frozen_mapping)
"""Maps each skill identifier to its fully resolved registry record."""
skills_in_load_order: tuple[str, ...]
"""Preserves skill identifiers in deterministic source loading order."""
skills_summary_in_load_order: tuple[SkillSummaryRecord, ...]
"""Stores compact skill summaries in the same deterministic loading order."""
docs_markdown_by_path: Mapping[DocsPath, str] = Field(default_factory=_empty_docs_mapping)
"""Maps each documentation path to its loaded Markdown content."""
docs_markdown_path_index: tuple[DocsPath, ...]
"""Lists documentation paths in deterministic index order."""
tag_to_skill_ids: Mapping[str, tuple[str, ...]] = Field(default_factory=frozen_mapping)
"""Indexes skill identifiers by tag for catalog filtering and search."""
capability_to_skill_ids: Mapping[str, tuple[str, ...]] = Field(default_factory=frozen_mapping)
"""Indexes skill identifiers by the capabilities they provide."""
prompts_by_id: Mapping[str, PromptRecord] = Field(default_factory=frozen_mapping)
"""Maps each prompt identifier to its fully resolved registry record."""
prompts_in_load_order: tuple[str, ...] = ()
@@ -222,10 +100,7 @@ class DocsRegistry(StrictFrozenModel):
"""Indexes prompt identifiers by tag for catalog filtering and search."""
@field_validator(
"skills_by_id",
"docs_markdown_by_path",
"tag_to_skill_ids",
"capability_to_skill_ids",
"prompts_by_id",
"tag_to_prompt_ids",
mode="before",
src/personal_mcp/registry/models/skill.py
-142
View File
@@ -1,142 +0,0 @@
from collections.abc import Mapping
import yaml
from pydantic import Field
from pydantic import field_validator
from pydantic import model_validator
from .common import SEMVER_RE
from .common import SKILL_ID_RE
from .common import DocsPath
from .common import ReferenceEntry
from .common import StrictFrozenModel
from .common import frozen_mapping
class SkillMetadata(StrictFrozenModel):
"""Canonical metadata describing a skill."""
id: str
version: str
tags: tuple[str, ...] = ()
capabilities: tuple[str, ...] = Field(min_length=1)
references: Mapping[str, ReferenceEntry] = Field(default_factory=frozen_mapping)
@field_validator("id")
@classmethod
def validate_id(cls, value: str) -> str:
if not SKILL_ID_RE.fullmatch(value):
raise ValueError("id must be lowercase kebab-case and start with a letter")
return value
@field_validator("version")
@classmethod
def validate_version(cls, value: str) -> str:
if not SEMVER_RE.fullmatch(value):
raise ValueError("version must be semver")
return value
@field_validator("tags")
@classmethod
def validate_tags(cls, value: tuple[str, ...]) -> tuple[str, ...]:
for tag in value:
if not SKILL_ID_RE.fullmatch(tag):
raise ValueError(f"invalid tag: {tag}")
return value
@field_validator("references", mode="before")
@classmethod
def freeze_references(cls, value: Mapping[str, ReferenceEntry] | None) -> Mapping[str, ReferenceEntry]:
return frozen_mapping(value)
@field_validator("references")
@classmethod
def validate_reference_ids(cls, value: Mapping[str, ReferenceEntry]) -> Mapping[str, ReferenceEntry]:
for ref_id in value:
if not SKILL_ID_RE.fullmatch(ref_id):
raise ValueError(f"invalid reference id: {ref_id}")
return value
class SkillFrontmatter(StrictFrozenModel):
"""Parsed SKILL frontmatter including standard and personal-mcp fields."""
name: str = Field(min_length=1, max_length=64)
description: str = Field(min_length=1, max_length=1024)
x_personal_mcp: SkillMetadata = Field(alias="x-personal-mcp")
@field_validator("name")
@classmethod
def validate_name(cls, value: str) -> str:
if not SKILL_ID_RE.fullmatch(value):
raise ValueError("name must be lowercase kebab-case and start with a letter")
if "anthropic" in value or "claude" in value:
raise ValueError("name must not contain reserved words anthropic or claude")
return value
@classmethod
def from_raw_yaml(cls, raw: str | None) -> "SkillFrontmatter":
if raw is None:
raise ValueError("missing YAML frontmatter")
try:
data = yaml.safe_load(raw)
except yaml.YAMLError as e:
raise ValueError(f"invalid YAML in frontmatter: {e}") from e
if not isinstance(data, dict):
raise TypeError("frontmatter must parse to an object")
return cls.model_validate(data)
class StoredSkillReference(StrictFrozenModel):
"""Structured representation of a skill reference markdown document."""
ref_id: str
relpath: DocsPath
content: str
entry: ReferenceEntry
class StoredSkill(StrictFrozenModel):
"""Structured representation of a skill markdown document."""
skill_id: str
relpath: DocsPath
content: str
frontmatter: SkillFrontmatter
references: Mapping[str, StoredSkillReference] = Field(default_factory=frozen_mapping)
@field_validator("frontmatter", mode="before")
@classmethod
def parse_frontmatter_yaml(cls, value: SkillFrontmatter | str | None) -> SkillFrontmatter:
if isinstance(value, SkillFrontmatter):
return value
return SkillFrontmatter.from_raw_yaml(value)
@field_validator("references", mode="before")
@classmethod
def freeze_references(cls, value: Mapping[str, StoredSkillReference] | None) -> Mapping[str, StoredSkillReference]:
return frozen_mapping(value)
@model_validator(mode="after")
def validate_contract(self) -> "StoredSkill":
parts = self.relpath.parts
if len(parts) < 3 or parts[0] != "skills":
raise ValueError("skill relpath must be under skills/<slug>/")
skill_dir_name = parts[1]
if self.frontmatter.name != skill_dir_name:
raise ValueError("frontmatter name must exactly match skill directory name")
if self.frontmatter.x_personal_mcp.id != self.frontmatter.name:
raise ValueError("x-personal-mcp.id must exactly match name")
expected_capability = f"resource://skills/{self.frontmatter.name}/document"
if expected_capability not in self.frontmatter.x_personal_mcp.capabilities:
raise ValueError(f"capabilities must include {expected_capability}")
if self.skill_id != self.frontmatter.x_personal_mcp.id:
raise ValueError("skill_id must exactly match x-personal-mcp.id")
for ref_id, ref in self.references.items():
if ref.ref_id != ref_id:
raise ValueError(f"reference key must match ref_id: {ref_id}")
return self
src/personal_mcp/registry/read.py
-35
View File
@@ -2,41 +2,6 @@ from .models.common import parse_docs_path
from .models.registry import DocsRegistry
def read_skill_document(registry: DocsRegistry, skill_id: str) -> dict[str, str]:
if skill_id not in registry.skills_by_id:
raise KeyError(f"unknown skill_id: {skill_id}")
skill = registry.skills_by_id[skill_id]
return {
"id": skill.skill_id,
"uri": skill.document_uri,
"format": "markdown",
"source_path": f"docs/{skill.document_relpath.as_posix()}",
"content": skill.document_content,
}
def read_skill_reference(
registry: DocsRegistry,
*,
skill_id: str,
ref_id: str,
) -> dict[str, str]:
if skill_id not in registry.skills_by_id:
raise KeyError(f"unknown skill_id: {skill_id}")
skill = registry.skills_by_id[skill_id]
if ref_id not in skill.references:
raise KeyError(f"unknown ref_id '{ref_id}' for skill '{skill_id}'")
reference = skill.references[ref_id]
return {
"id": ref_id,
"skill_id": skill_id,
"uri": reference.uri,
"format": "markdown",
"source_path": f"docs/{reference.relpath.as_posix()}",
"content": reference.content,
}
def read_docs_markdown_path(registry: DocsRegistry, path: str) -> dict[str, str]:
docs_path = parse_docs_path(path)
if docs_path not in registry.docs_markdown_by_path:
src/personal_mcp/skills/__init__.py
+5 -1
View File
@@ -1 +1,5 @@
"""Docs registry and markdown loading utilities for personal MCP skills."""
"""FastMCP provider composition for personal MCP skills."""
from .provider import create_skills_provider
__all__ = ["create_skills_provider"]
src/personal_mcp/skills/parse.py
View File
src/personal_mcp/skills/provider.py
+20
View File
@@ -0,0 +1,20 @@
from pathlib import Path
from fastmcp.server.providers.skills import SkillsDirectoryProvider
def create_skills_provider() -> SkillsDirectoryProvider:
"""Create the provider for skills packaged with personal-mcp."""
skills_root = Path(__file__).resolve().parents[1] / "docs" / "skills"
if not skills_root.is_dir():
raise FileNotFoundError(f"packaged skills directory does not exist: {skills_root}")
has_skills = any(skill_dir.is_dir() and (skill_dir / "SKILL.md").is_file() for skill_dir in skills_root.iterdir())
if not has_skills:
raise ValueError(f"packaged skills directory contains no skills: {skills_root}")
return SkillsDirectoryProvider(
roots=skills_root,
reload=False,
supporting_files="template",
)