This commit is contained in:
John Lancaster
2026-06-21 17:35:01 -05:00
parent c189677717
commit 36347ff4a5
6 changed files with 297 additions and 151 deletions
+8 -50
View File
@@ -4,57 +4,15 @@ from typing import Any
from personal_mcp.registry.contracts import DocsRegistry from personal_mcp.registry.contracts import DocsRegistry
from personal_mcp.registry.models.registry import PromptRecord 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 SkillRecord
from personal_mcp.registry.models.registry import SkillSummaryPayload
DEFAULT_LIMIT = 20 DEFAULT_LIMIT = 20
MAX_LIMIT = 100 MAX_LIMIT = 100
def _pattern_payload(skill: SkillRecord) -> dict[str, Any]:
return {
"id": skill.skill_id,
"name": skill.name,
"version": skill.version,
"description": skill.description,
"tags": list(skill.tags),
"depends_on": list(skill.depends_on),
"capabilities": list(skill.capabilities),
"resources": list(skill.capabilities),
}
def _summary_payload(skill: SkillRecord) -> dict[str, Any]:
return {
"id": skill.skill_id,
"name": skill.name,
"description": skill.description,
"tags": list(skill.tags),
"capabilities": list(skill.capabilities),
"version": skill.version,
"document_uri": skill.document_uri,
"detail_uri": f"resource://catalog/skills/{skill.skill_id}",
"resources": {
"document": skill.document_uri,
"references": [
f"resource://skills/{skill.skill_id}/references/{ref_id}" for ref_id in sorted(skill.references)
],
},
}
def _prompt_summary_payload(prompt: PromptRecord) -> dict[str, Any]:
return {
"id": prompt.prompt_id,
"name": prompt.name,
"description": prompt.description,
"tags": list(prompt.tags),
"capabilities": list(prompt.capabilities),
"version": prompt.version,
"document_uri": prompt.document_uri,
"detail_uri": f"resource://catalog/prompts/{prompt.prompt_id}",
}
def _skill_matches( def _skill_matches(
skill: SkillRecord, skill: SkillRecord,
*, *,
@@ -130,7 +88,7 @@ def build_skills_index_payload(
next_cursor = start + normalized_limit next_cursor = start + normalized_limit
return { return {
"skills": [_summary_payload(skill) for skill in page], "skills": [SkillSummaryPayload.from_record(skill).model_dump() for skill in page],
"total": len(matches), "total": len(matches),
"cursor": str(start), "cursor": str(start),
"limit": normalized_limit, "limit": normalized_limit,
@@ -187,7 +145,7 @@ def build_prompts_index_payload(
next_cursor = start + normalized_limit next_cursor = start + normalized_limit
return { return {
"prompts": [_prompt_summary_payload(prompt) for prompt in page], "prompts": [PromptSummaryPayload.from_record(prompt).model_dump() for prompt in page],
"total": len(matches), "total": len(matches),
"cursor": str(start), "cursor": str(start),
"limit": normalized_limit, "limit": normalized_limit,
@@ -240,7 +198,7 @@ def search_patterns_payload(
page = matches[normalized_skip : normalized_skip + normalized_limit] page = matches[normalized_skip : normalized_skip + normalized_limit]
return { return {
"patterns": [_pattern_payload(skill) for skill in page], "patterns": [SkillPatternPayload.from_record(skill).model_dump() for skill in page],
"total": len(matches), "total": len(matches),
"skip": normalized_skip, "skip": normalized_skip,
"limit": normalized_limit, "limit": normalized_limit,
@@ -250,7 +208,7 @@ def search_patterns_payload(
def get_pattern_by_id_payload(registry: DocsRegistry, skill_id: str) -> dict[str, Any]: def get_pattern_by_id_payload(registry: DocsRegistry, skill_id: str) -> dict[str, Any]:
if skill_id not in registry.skills_by_id: if skill_id not in registry.skills_by_id:
return {"found": False, "id": skill_id} return {"found": False, "id": skill_id}
return {"found": True, "pattern": _pattern_payload(registry.skills_by_id[skill_id])} return {"found": True, "pattern": SkillPatternPayload.from_record(registry.skills_by_id[skill_id]).model_dump()}
def search_prompts_payload( def search_prompts_payload(
@@ -277,7 +235,7 @@ def search_prompts_payload(
page = matches[normalized_skip : normalized_skip + normalized_limit] page = matches[normalized_skip : normalized_skip + normalized_limit]
return { return {
"prompts": [_prompt_summary_payload(prompt) for prompt in page], "prompts": [PromptSummaryPayload.from_record(prompt).model_dump() for prompt in page],
"total": len(matches), "total": len(matches),
"skip": normalized_skip, "skip": normalized_skip,
"limit": normalized_limit, "limit": normalized_limit,
+56
View File
@@ -1,11 +1,20 @@
from collections.abc import Iterable from collections.abc import Iterable
from collections.abc import Mapping
from dataclasses import dataclass from dataclasses import dataclass
from fnmatch import fnmatch from fnmatch import fnmatch
from importlib.resources.abc import Traversable from importlib.resources.abc import Traversable
from itertools import groupby from itertools import groupby
from itertools import starmap from itertools import starmap
from pathlib import PurePosixPath
from typing import Self from typing import Self
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 personal_mcp.skills.document_loader import _reference_id_from_filename
from personal_mcp.skills.document_loader import _title_from_reference_filename
from .document import MarkdownDocument from .document import MarkdownDocument
@@ -49,3 +58,50 @@ def group_skill_paths(docs: Iterable[MarkdownDocument]) -> dict[str, set[Markdow
) )
grouped = groupby(s, key=lambda doc: doc.skill_slug) grouped = groupby(s, key=lambda doc: doc.skill_slug)
return {k: set(g) for k, g in grouped if k} return {k: set(g) for k, g in grouped if k}
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").joinpath(reference_doc.relpath.name).as_posix(),
title=_title_from_reference_filename(reference_doc.relpath.name),
)
return discovered
def build_stored_skill(
*,
bundle: SkillFilesBundle,
docs_by_relpath: Mapping[PurePosixPath, 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").joinpath(bundle.slug).joinpath(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,
}
)
+23 -83
View File
@@ -4,129 +4,67 @@ import importlib
from collections import defaultdict from collections import defaultdict
from pathlib import Path from pathlib import Path
from pathlib import PurePosixPath from pathlib import PurePosixPath
from typing import Any
import yaml
from personal_mcp.registry.ingest.document import MarkdownDocument from personal_mcp.registry.ingest.document import MarkdownDocument
from personal_mcp.registry.ingest.prompt import PromptFilesBundle from personal_mcp.registry.ingest.prompt import PromptFilesBundle
from personal_mcp.registry.ingest.skill import SkillFilesBundle from personal_mcp.registry.ingest.skill import SkillFilesBundle
from personal_mcp.registry.models.common import ReferenceEntry from personal_mcp.registry.ingest.skill import build_stored_skill
from personal_mcp.registry.models.common import _normalize_docs_path from personal_mcp.registry.models.common import _normalize_docs_path
from personal_mcp.registry.models.prompt import StoredPrompt
from personal_mcp.registry.models.registry import DocsRegistry from personal_mcp.registry.models.registry import DocsRegistry
from personal_mcp.registry.models.registry import PromptRecord from personal_mcp.registry.models.registry import PromptRecord
from personal_mcp.registry.models.registry import PromptSummaryRecord from personal_mcp.registry.models.registry import PromptSummaryRecord
from personal_mcp.registry.models.registry import ReferenceRecord from personal_mcp.registry.models.registry import ReferenceRecord
from personal_mcp.registry.models.registry import SkillRecord from personal_mcp.registry.models.registry import SkillRecord
from personal_mcp.registry.models.registry import SkillSummaryRecord from personal_mcp.registry.models.registry import SkillSummaryRecord
from personal_mcp.skills.document_loader import _reference_id_from_filename
from personal_mcp.skills.document_loader import _title_from_reference_filename
from personal_mcp.skills.document_loader import _validate_prompt_frontmatter
from personal_mcp.skills.document_loader import _validate_skill_frontmatter
def _parse_frontmatter(raw_frontmatter: str | None, *, path: PurePosixPath) -> dict[str, Any]:
if raw_frontmatter is None:
raise ValueError(f"missing YAML frontmatter: {path.as_posix()}")
parsed = yaml.safe_load(raw_frontmatter)
if not isinstance(parsed, dict):
raise TypeError(f"frontmatter must parse to an object: {path.as_posix()}")
return parsed
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").joinpath(reference_doc.relpath.name).as_posix(),
title=_title_from_reference_filename(reference_doc.relpath.name),
)
return discovered
def _build_skill_record( def _build_skill_record(
*, bundle: SkillFilesBundle, docs_by_relpath: dict[PurePosixPath, MarkdownDocument] *, bundle: SkillFilesBundle, docs_by_relpath: dict[PurePosixPath, MarkdownDocument]
) -> SkillRecord: ) -> SkillRecord:
frontmatter_raw = _parse_frontmatter(bundle.skill.frontmatter, path=bundle.skill.relpath) stored = build_stored_skill(bundle=bundle, docs_by_relpath=docs_by_relpath)
frontmatter = _validate_skill_frontmatter(frontmatter_raw, skill_dir_name=bundle.slug) metadata = stored.frontmatter.x_personal_mcp
metadata = frontmatter.x_personal_mcp
merged_entries = _discover_reference_entries(bundle)
merged_entries.update(dict(metadata.references))
references: dict[str, ReferenceRecord] = {} references: dict[str, ReferenceRecord] = {}
for ref_id, entry in sorted(merged_entries.items()): for ref_id, ref in sorted(stored.references.items()):
ref_relpath = PurePosixPath("skills").joinpath(bundle.slug).joinpath(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] = ReferenceRecord( references[ref_id] = ReferenceRecord(
ref_id=ref_id, ref_id=ref_id,
uri=f"resource://skills/{metadata.id}/references/{ref_id}", uri=f"resource://skills/{metadata.id}/references/{ref_id}",
relpath=ref_relpath.as_posix(), relpath=ref.relpath.as_posix(),
mime_type=entry.mime_type, mime_type=ref.entry.mime_type,
title=entry.title, title=ref.entry.title,
content=ref_doc.content, content=ref.content,
) )
return SkillRecord( return SkillRecord(
skill_id=metadata.id, skill_id=metadata.id,
name=frontmatter.name, name=stored.frontmatter.name,
description=frontmatter.description, description=stored.frontmatter.description,
version=metadata.version, version=metadata.version,
tags=tuple(metadata.tags), tags=tuple(metadata.tags),
capabilities=tuple(metadata.capabilities), capabilities=tuple(metadata.capabilities),
depends_on=tuple(metadata.depends_on), depends_on=tuple(metadata.depends_on),
document_uri=f"resource://skills/{metadata.id}/document", document_uri=f"resource://skills/{metadata.id}/document",
document_relpath=bundle.skill.relpath.as_posix(), document_relpath=stored.relpath.as_posix(),
document_content=bundle.skill.content, document_content=stored.content,
references=references, references=references,
) )
def _build_prompt_record(*, bundle: PromptFilesBundle) -> PromptRecord: def _build_prompt_record(*, bundle: PromptFilesBundle) -> PromptRecord:
frontmatter_raw = _parse_frontmatter(bundle.prompt.frontmatter, path=bundle.prompt.relpath) stored = StoredPrompt.from_bundle(bundle)
frontmatter = _validate_prompt_frontmatter(frontmatter_raw, prompt_dir_name=bundle.slug) metadata = stored.frontmatter.x_personal_mcp
metadata = frontmatter.x_personal_mcp
return PromptRecord( return PromptRecord(
prompt_id=metadata.id, prompt_id=metadata.id,
name=frontmatter.name, name=stored.frontmatter.name,
description=frontmatter.description, description=stored.frontmatter.description,
version=metadata.version, version=metadata.version,
tags=tuple(metadata.tags), tags=tuple(metadata.tags),
capabilities=tuple(metadata.capabilities), capabilities=tuple(metadata.capabilities),
arguments=dict(metadata.arguments), arguments=dict(metadata.arguments),
document_uri=f"resource://prompts/{metadata.id}/document", document_uri=f"resource://prompts/{metadata.id}/document",
document_relpath=bundle.prompt.relpath.as_posix(), document_relpath=stored.relpath.as_posix(),
document_content=bundle.prompt.content, document_content=stored.content,
)
def _to_summary(skill: SkillRecord) -> SkillSummaryRecord:
return SkillSummaryRecord(
skill_id=skill.skill_id,
name=skill.name,
description=skill.description,
tags=skill.tags,
capabilities=skill.capabilities,
document_uri=skill.document_uri,
version=skill.version,
)
def _to_prompt_summary(prompt: PromptRecord) -> PromptSummaryRecord:
return PromptSummaryRecord(
prompt_id=prompt.prompt_id,
name=prompt.name,
description=prompt.description,
tags=prompt.tags,
capabilities=prompt.capabilities,
document_uri=prompt.document_uri,
version=prompt.version,
) )
@@ -215,7 +153,9 @@ def load_docs_registry(*, package_anchor: str, docs_root: str = "docs") -> DocsR
return DocsRegistry( return DocsRegistry(
skills_by_id=skills_by_id, skills_by_id=skills_by_id,
skills_in_load_order=skills_in_order_tuple, skills_in_load_order=skills_in_order_tuple,
skills_summary_in_load_order=tuple(_to_summary(skills_by_id[skill_id]) for skill_id in 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_by_path=docs_markdown_by_path,
docs_markdown_path_index=tuple(sorted(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), tag_to_skill_ids=_build_tag_index_skills(skills_in_order_tuple, skills_by_id),
@@ -223,7 +163,7 @@ def load_docs_registry(*, package_anchor: str, docs_root: str = "docs") -> DocsR
prompts_by_id=prompts_by_id, prompts_by_id=prompts_by_id,
prompts_in_load_order=prompts_in_order_tuple, prompts_in_load_order=prompts_in_order_tuple,
prompts_summary_in_load_order=tuple( prompts_summary_in_load_order=tuple(
_to_prompt_summary(prompts_by_id[prompt_id]) for prompt_id in prompts_in_order_tuple PromptSummaryRecord.from_record(prompts_by_id[prompt_id]) for prompt_id in prompts_in_order_tuple
), ),
tag_to_prompt_ids=_build_tag_index_prompts(prompts_in_order_tuple, prompts_by_id), tag_to_prompt_ids=_build_tag_index_prompts(prompts_in_order_tuple, prompts_by_id),
) )
+54 -9
View File
@@ -1,18 +1,23 @@
import re import re
from collections.abc import Mapping from collections.abc import Mapping
from pathlib import PurePosixPath from pathlib import PurePosixPath
from typing import TYPE_CHECKING
from typing import Any from typing import Any
from typing import Literal from typing import Literal
import yaml import yaml
from pydantic import Field from pydantic import Field
from pydantic import field_validator from pydantic import field_validator
from pydantic import model_validator
from .common import SEMVER_RE from .common import SEMVER_RE
from .common import SKILL_ID_RE from .common import SKILL_ID_RE
from .common import StrictFrozenModel from .common import StrictFrozenModel
from .common import frozen_mapping from .common import frozen_mapping
if TYPE_CHECKING:
from personal_mcp.registry.ingest.prompt import PromptFilesBundle
type PromptArgumentType = Literal[ type PromptArgumentType = Literal[
"string", "string",
"number", "number",
@@ -101,6 +106,18 @@ class PromptFrontmatter(StrictFrozenModel):
raise ValueError("name must not contain reserved words anthropic or claude") raise ValueError("name must not contain reserved words anthropic or claude")
return value return value
@classmethod
def from_raw_yaml(cls, raw: str | None) -> "PromptFrontmatter":
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 StoredPrompt(StrictFrozenModel): class StoredPrompt(StrictFrozenModel):
"""Normalized prompt document content with path and frontmatter for storage in the registry.""" """Normalized prompt document content with path and frontmatter for storage in the registry."""
@@ -108,15 +125,43 @@ class StoredPrompt(StrictFrozenModel):
prompt_id: str prompt_id: str
relpath: PurePosixPath relpath: PurePosixPath
content: str content: str
frontmatter: PromptFrontmatter | None = None frontmatter: PromptFrontmatter
@field_validator("frontmatter", mode="before") @field_validator("frontmatter", mode="before")
@classmethod @classmethod
def parse_frontmatter_yaml(cls, value: str | None): def parse_frontmatter_yaml(cls, value: PromptFrontmatter | str | None) -> PromptFrontmatter:
if value is None: if isinstance(value, PromptFrontmatter):
return None return value
try: return PromptFrontmatter.from_raw_yaml(value)
data = yaml.safe_load(value)
except yaml.YAMLError as e: @model_validator(mode="after")
raise ValueError(f"invalid YAML in frontmatter: {e}") from e def validate_contract(self) -> "StoredPrompt":
return PromptFrontmatter.model_validate(data) parts = self.relpath.parts
if len(parts) < 3 or parts[0] != "prompts":
raise ValueError("prompt relpath must be under prompts/<slug>/")
prompt_dir_name = parts[1]
if self.frontmatter.name != prompt_dir_name:
raise ValueError("frontmatter name must exactly match prompt 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://prompts/{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.prompt_id != self.frontmatter.x_personal_mcp.id:
raise ValueError("prompt_id must exactly match x-personal-mcp.id")
return self
@classmethod
def from_bundle(cls, bundle: "PromptFilesBundle") -> "StoredPrompt":
frontmatter = PromptFrontmatter.from_raw_yaml(bundle.prompt.frontmatter)
return cls.model_validate(
{
"prompt_id": frontmatter.x_personal_mcp.id,
"relpath": bundle.prompt.relpath,
"content": bundle.prompt.content,
"frontmatter": frontmatter,
}
)
@@ -51,6 +51,18 @@ class SkillSummaryRecord(StrictFrozenModel):
document_uri: str document_uri: str
version: 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): class PromptRecord(StrictFrozenModel):
"""Registry record containing a fully resolved prompt document.""" """Registry record containing a fully resolved prompt document."""
@@ -83,6 +95,103 @@ class PromptSummaryRecord(StrictFrozenModel):
document_uri: str document_uri: str
version: str version: str
@classmethod
def from_record(cls, record: PromptRecord) -> "PromptSummaryRecord":
return cls(
prompt_id=record.prompt_id,
name=record.name,
description=record.description,
tags=record.tags,
capabilities=record.capabilities,
document_uri=record.document_uri,
version=record.version,
)
class SkillPatternPayload(StrictFrozenModel):
"""Catalog payload model for skill pattern search results."""
id: str
name: str
version: str
description: str
tags: list[str]
depends_on: 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),
depends_on=list(record.depends_on),
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."""
id: str
name: str
description: str
tags: list[str]
capabilities: list[str]
version: str
document_uri: str
detail_uri: str
@classmethod
def from_record(cls, record: PromptRecord) -> "PromptSummaryPayload":
return cls(
id=record.prompt_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/prompts/{record.prompt_id}",
)
class DocsRegistry(StrictFrozenModel): class DocsRegistry(StrictFrozenModel):
"""In-memory index of loaded skills, prompts, and docs content.""" """In-memory index of loaded skills, prompts, and docs content."""
+47 -9
View File
@@ -4,6 +4,7 @@ from pathlib import PurePosixPath
import yaml import yaml
from pydantic import Field from pydantic import Field
from pydantic import field_validator from pydantic import field_validator
from pydantic import model_validator
from .common import SEMVER_RE from .common import SEMVER_RE
from .common import SKILL_ID_RE from .common import SKILL_ID_RE
@@ -107,6 +108,18 @@ class SkillFrontmatter(StrictFrozenModel):
raise ValueError("name must not contain reserved words anthropic or claude") raise ValueError("name must not contain reserved words anthropic or claude")
return value 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): class StoredSkillReference(StrictFrozenModel):
"""Structured representation of a skill reference markdown document.""" """Structured representation of a skill reference markdown document."""
@@ -124,15 +137,40 @@ class StoredSkill(StrictFrozenModel):
relpath: PurePosixPath relpath: PurePosixPath
content: str content: str
frontmatter: SkillFrontmatter frontmatter: SkillFrontmatter
references: dict[str, StoredSkillReference] = Field(default_factory=dict) references: Mapping[str, StoredSkillReference] = Field(default_factory=frozen_mapping)
@field_validator("frontmatter", mode="before") @field_validator("frontmatter", mode="before")
@classmethod @classmethod
def parse_frontmatter_yaml(cls, value: str | None): def parse_frontmatter_yaml(cls, value: SkillFrontmatter | str | None) -> SkillFrontmatter:
if value is None: if isinstance(value, SkillFrontmatter):
return None return value
try: return SkillFrontmatter.from_raw_yaml(value)
data = yaml.safe_load(value)
except yaml.YAMLError as e: @field_validator("references", mode="before")
raise ValueError(f"invalid YAML in frontmatter: {e}") from e @classmethod
return SkillFrontmatter.model_validate(data) 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