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

better data models

Browse Source
This commit is contained in:
John Lancaster
2026-06-21 16:53:43 -05:00
parent 34923b51d7
commit 37fa9b6c6f
8 changed files with 261 additions and 525 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/personal_mcp/registry/contracts.py
+15
View File
@@ -0,0 +1,15 @@
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/issues.py
-22
View File
@@ -1,22 +0,0 @@
from dataclasses import dataclass
@dataclass(frozen=True)
class RegistryIssue:
code: str
message: str
skill_id: str | None
path: str
hint: str
class DocsRegistryValidationError(Exception):
def __init__(self, errors: list[RegistryIssue]) -> None:
self.errors = errors
summary = "\n".join(
[
(f"{issue.code}: {issue.message} (skill={issue.skill_id or 'unknown'}, path={issue.path})")
for issue in errors
]
)
super().__init__(summary)
src/personal_mcp/registry/load.py
+206 -425
View File
@@ -1,448 +1,229 @@
from importlib.resources import files
from __future__ import annotations
import importlib
from collections import defaultdict
from pathlib import Path
from pathlib import PurePosixPath
from typing import Any
from pydantic import ValidationError
import yaml
from ..skills.document_loader import _discover_top_level_references
from ..skills.document_loader import _parse_frontmatter
from ..skills.document_loader import _validate_prompt_frontmatter
from ..skills.document_loader import _validate_skill_frontmatter
from ..skills.document_loader import _walk_markdown
from .contracts import DocsRegistry
from .contracts import PromptRecord
from .contracts import PromptSummaryRecord
from .contracts import ReferenceRecord
from .contracts import SkillRecord
from .contracts import SkillSummaryRecord
from .issues import DocsRegistryValidationError
from .issues import RegistryIssue
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.models.common import ReferenceEntry
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
from personal_mcp.skills.document_loader import _normalize_docs_path
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 _ensure_no_cycles(
skills_by_id: dict[str, SkillRecord],
) -> list[tuple[str, str]]:
state: dict[str, int] = {}
stack: list[str] = []
cycles: set[tuple[str, str]] = set()
def visit(skill_id: str) -> None:
state[skill_id] = 1
stack.append(skill_id)
for dependency in skills_by_id[skill_id].depends_on:
if dependency not in skills_by_id:
continue
dep_state = state.get(dependency, 0)
if dep_state == 0:
visit(dependency)
continue
if dep_state == 1:
cycle_start = stack[stack.index(dependency)]
cycle_path = stack[stack.index(dependency) :] + [dependency]
cycles.add((cycle_start, " -> ".join(cycle_path)))
stack.pop()
state[skill_id] = 2
for skill_id in sorted(skills_by_id):
if state.get(skill_id, 0) == 0:
visit(skill_id)
return sorted(cycles)
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 _load_skills(
*,
docs_markdown_by_path: dict[str, str],
skills_root,
) -> tuple[dict[str, SkillRecord], list[SkillSummaryRecord], list[RegistryIssue]]:
skills_by_id: dict[str, SkillRecord] = {}
summaries: list[SkillSummaryRecord] = []
issues: list[RegistryIssue] = []
for skill_dir in sorted(skills_root.iterdir(), key=lambda item: item.name):
if not skill_dir.is_dir():
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
skill_dir_name = skill_dir.name
skill_rel_root = PurePosixPath("skills").joinpath(skill_dir_name)
skill_doc_relpath = skill_rel_root.joinpath("SKILL.md").as_posix()
skill_doc_file = skill_dir.joinpath("SKILL.md")
if not skill_doc_file.is_file():
issues.append(
RegistryIssue(
code="missing_skill_document",
message="missing required SKILL.md",
skill_id=skill_dir_name,
path=skill_doc_relpath,
hint="add docs/skills/<skill-id>/SKILL.md",
)
)
continue
skill_markdown = skill_doc_file.read_text(encoding="utf-8")
try:
raw_frontmatter, _ = _parse_frontmatter(
skill_markdown,
path=skill_doc_relpath,
)
frontmatter = _validate_skill_frontmatter(
raw_frontmatter,
skill_dir_name=skill_dir_name,
)
except (ValueError, ValidationError) as exc:
issues.append(
RegistryIssue(
code="invalid_frontmatter",
message=str(exc),
skill_id=skill_dir_name,
path=skill_doc_relpath,
hint="fix SKILL.md YAML frontmatter to match the contract",
)
)
continue
effective_reference_entries = _discover_top_level_references(skill_dir=skill_dir)
effective_reference_entries.update(frontmatter.x_personal_mcp.references)
references: dict[str, ReferenceRecord] = {}
for ref_id, ref_entry in effective_reference_entries.items():
ref_relpath = skill_rel_root.joinpath(ref_entry.path).as_posix()
if ref_relpath not in docs_markdown_by_path:
issues.append(
RegistryIssue(
code="missing_reference",
message=f"reference target is missing for ref_id '{ref_id}'",
skill_id=frontmatter.name,
path=ref_relpath,
hint="fix x-personal-mcp.references path or add the referenced markdown file",
)
)
continue
references[ref_id] = ReferenceRecord(
ref_id=ref_id,
uri=f"resource://skills/{frontmatter.name}/references/{ref_id}",
relpath=ref_relpath,
mime_type=ref_entry.mime_type,
title=ref_entry.title,
content=docs_markdown_by_path[ref_relpath],
)
skill_id = frontmatter.name
if skill_id in skills_by_id:
issues.append(
RegistryIssue(
code="duplicate_skill_id",
message="duplicate skill id discovered",
skill_id=skill_id,
path=skill_doc_relpath,
hint="ensure each skill directory has a unique id",
)
)
continue
record = SkillRecord(
skill_id=skill_id,
name=frontmatter.name,
description=frontmatter.description,
version=frontmatter.x_personal_mcp.version,
tags=tuple(frontmatter.x_personal_mcp.tags),
capabilities=tuple(frontmatter.x_personal_mcp.capabilities),
depends_on=tuple(frontmatter.x_personal_mcp.depends_on),
document_uri=f"resource://skills/{skill_id}/document",
document_relpath=skill_doc_relpath,
document_content=skill_markdown,
references=references,
discovered[ref_id] = ReferenceEntry(
path=PurePosixPath("references").joinpath(reference_doc.relpath.name).as_posix(),
title=_title_from_reference_filename(reference_doc.relpath.name),
)
skills_by_id[skill_id] = record
summaries.append(
SkillSummaryRecord(
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,
)
return discovered
def _build_skill_record(
*, bundle: SkillFilesBundle, docs_by_relpath: dict[PurePosixPath, MarkdownDocument]
) -> SkillRecord:
frontmatter_raw = _parse_frontmatter(bundle.skill.frontmatter, path=bundle.skill.relpath)
frontmatter = _validate_skill_frontmatter(frontmatter_raw, skill_dir_name=bundle.slug)
metadata = frontmatter.x_personal_mcp
merged_entries = _discover_reference_entries(bundle)
merged_entries.update(dict(metadata.references))
references: dict[str, ReferenceRecord] = {}
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] = ReferenceRecord(
ref_id=ref_id,
uri=f"resource://skills/{metadata.id}/references/{ref_id}",
relpath=ref_relpath.as_posix(),
mime_type=entry.mime_type,
title=entry.title,
content=ref_doc.content,
)
return skills_by_id, summaries, issues
def _load_prompts(
*,
prompts_root,
skills_by_id: dict[str, SkillRecord],
) -> tuple[dict[str, PromptRecord], list[PromptSummaryRecord], list[RegistryIssue]]:
prompts_by_id: dict[str, PromptRecord] = {}
prompt_summaries: list[PromptSummaryRecord] = []
issues: list[RegistryIssue] = []
discovered_prompt_docs: set[str] = set()
for prompt_dir in sorted(prompts_root.iterdir(), key=lambda item: item.name):
if not prompt_dir.is_dir():
continue
prompt_dir_name = prompt_dir.name
prompt_rel_root = PurePosixPath("prompts").joinpath(prompt_dir_name)
prompt_doc_relpath = prompt_rel_root.joinpath("PROMPT.md").as_posix()
prompt_doc_file = prompt_dir.joinpath("PROMPT.md")
if not prompt_doc_file.is_file():
continue
discovered_prompt_docs.add(prompt_doc_relpath)
prompt_markdown = prompt_doc_file.read_text(encoding="utf-8")
try:
raw_frontmatter, _ = _parse_frontmatter(
prompt_markdown,
path=prompt_doc_relpath,
)
frontmatter = _validate_prompt_frontmatter(
raw_frontmatter,
prompt_dir_name=prompt_dir_name,
)
except (ValueError, ValidationError) as exc:
issues.append(
RegistryIssue(
code="invalid_prompt_frontmatter",
message=str(exc),
skill_id=prompt_dir_name,
path=prompt_doc_relpath,
hint="fix PROMPT.md YAML frontmatter to match the contract",
)
)
continue
prompt_id = frontmatter.name
if prompt_id in prompts_by_id:
issues.append(
RegistryIssue(
code="duplicate_prompt_id",
message="duplicate prompt id discovered",
skill_id=prompt_id,
path=prompt_doc_relpath,
hint="ensure each prompt directory has a unique id",
)
)
continue
if prompt_id in skills_by_id:
issues.append(
RegistryIssue(
code="prompt_skill_id_collision",
message="prompt id collides with an existing skill id",
skill_id=prompt_id,
path=prompt_doc_relpath,
hint="use a unique prompt id that does not match a skill id",
)
)
continue
prompt_record = PromptRecord(
prompt_id=prompt_id,
name=frontmatter.name,
description=frontmatter.description,
version=frontmatter.x_personal_mcp.version,
tags=tuple(frontmatter.x_personal_mcp.tags),
capabilities=tuple(frontmatter.x_personal_mcp.capabilities),
arguments=frontmatter.x_personal_mcp.arguments,
document_uri=f"resource://prompts/{prompt_id}/document",
document_relpath=prompt_doc_relpath,
document_content=prompt_markdown,
)
prompts_by_id[prompt_id] = prompt_record
prompt_summaries.append(
PromptSummaryRecord(
prompt_id=prompt_record.prompt_id,
name=prompt_record.name,
description=prompt_record.description,
tags=prompt_record.tags,
capabilities=prompt_record.capabilities,
document_uri=prompt_record.document_uri,
version=prompt_record.version,
)
)
return prompts_by_id, prompt_summaries, issues
def _collect_relationship_and_uri_issues(
*,
skills_by_id: dict[str, SkillRecord],
prompts_by_id: dict[str, PromptRecord],
) -> list[RegistryIssue]:
issues: list[RegistryIssue] = []
for skill_id, record in sorted(skills_by_id.items()):
for dependency in record.depends_on:
if dependency == skill_id:
issues.append(
RegistryIssue(
code="self_dependency",
message="skill must not depend on itself",
skill_id=skill_id,
path=record.document_relpath,
hint="remove the skill id from depends_on",
)
)
elif dependency not in skills_by_id:
issues.append(
RegistryIssue(
code="missing_dependency",
message=f"depends_on target '{dependency}' does not exist",
skill_id=skill_id,
path=record.document_relpath,
hint="add the missing skill or remove it from depends_on",
)
)
for cycle_start, cycle in _ensure_no_cycles(skills_by_id):
issues.append(
RegistryIssue(
code="dependency_cycle",
message=f"depends_on cycle detected: {cycle}",
skill_id=cycle_start,
path=skills_by_id[cycle_start].document_relpath,
hint="remove at least one dependency edge in the cycle",
)
)
seen_uris: set[str] = set()
for skill_id, record in sorted(skills_by_id.items()):
uris = [record.document_uri] + [ref.uri for ref in record.references.values()]
for uri in uris:
if uri in seen_uris:
issues.append(
RegistryIssue(
code="duplicate_uri",
message=f"duplicate resource URI generated: {uri}",
skill_id=skill_id,
path=record.document_relpath,
hint="ensure unique skill ids and reference ids",
)
)
seen_uris.add(uri)
for prompt_id, record in sorted(prompts_by_id.items()):
uri = record.document_uri
if uri in seen_uris:
issues.append(
RegistryIssue(
code="duplicate_uri",
message=f"duplicate resource URI generated: {uri}",
skill_id=prompt_id,
path=record.document_relpath,
hint="ensure unique prompt ids",
)
)
seen_uris.add(uri)
return issues
def load_docs_registry(
*,
package_anchor: str,
docs_root: str = "docs",
) -> DocsRegistry:
docs_dir = files(package_anchor).joinpath(docs_root)
if not docs_dir.is_dir():
raise DocsRegistryValidationError(
[
RegistryIssue(
code="missing_docs_root",
message="docs root directory does not exist",
skill_id=None,
path=docs_root,
hint="configure docs_root to a valid packaged docs path",
)
]
)
docs_markdown_by_path = {
relpath: doc_file.read_text(encoding="utf-8") for relpath, doc_file in _walk_markdown(docs_dir)
}
skills_root = docs_dir.joinpath("skills")
if not skills_root.is_dir():
raise DocsRegistryValidationError(
[
RegistryIssue(
code="missing_skills_root",
message="skills directory does not exist under docs root",
skill_id=None,
path=f"{docs_root}/skills",
hint="ensure docs/skills is included in packaged docs",
)
]
)
skills_by_id, summaries, skill_issues = _load_skills(
docs_markdown_by_path=docs_markdown_by_path,
skills_root=skills_root,
return SkillRecord(
skill_id=metadata.id,
name=frontmatter.name,
description=frontmatter.description,
version=metadata.version,
tags=tuple(metadata.tags),
capabilities=tuple(metadata.capabilities),
depends_on=tuple(metadata.depends_on),
document_uri=f"resource://skills/{metadata.id}/document",
document_relpath=bundle.skill.relpath.as_posix(),
document_content=bundle.skill.content,
references=references,
)
def _build_prompt_record(*, bundle: PromptFilesBundle) -> PromptRecord:
frontmatter_raw = _parse_frontmatter(bundle.prompt.frontmatter, path=bundle.prompt.relpath)
frontmatter = _validate_prompt_frontmatter(frontmatter_raw, prompt_dir_name=bundle.slug)
metadata = frontmatter.x_personal_mcp
return PromptRecord(
prompt_id=metadata.id,
name=frontmatter.name,
description=frontmatter.description,
version=metadata.version,
tags=tuple(metadata.tags),
capabilities=tuple(metadata.capabilities),
arguments=dict(metadata.arguments),
document_uri=f"resource://prompts/{metadata.id}/document",
document_relpath=bundle.prompt.relpath.as_posix(),
document_content=bundle.prompt.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,
)
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],
) -> dict[str, tuple[str, ...]]:
tag_index: defaultdict[str, list[str]] = defaultdict(list)
for prompt_id in prompts_in_order:
for tag in prompts_by_id[prompt_id].tags:
tag_index[tag].append(prompt_id)
return {tag: tuple(ids) for tag, ids in sorted(tag_index.items())}
def _resolve_docs_root(*, package_anchor: str, docs_root: str) -> Path:
package = importlib.import_module(package_anchor)
package_file = getattr(package, "__file__", None)
if package_file is None:
raise ValueError(f"package anchor '{package_anchor}' has no file location")
resolved = Path(package_file).resolve().parent.joinpath(docs_root).resolve()
if not resolved.exists() or not resolved.is_dir():
raise FileNotFoundError(f"docs root does not exist or is not a directory: {resolved}")
return resolved
def load_docs_registry(*, package_anchor: str, docs_root: str = "docs") -> DocsRegistry:
docs_path = _resolve_docs_root(package_anchor=package_anchor, docs_root=docs_root)
docs = MarkdownDocument.from_root(docs_path)
docs_markdown_by_path = {_normalize_docs_path(relpath.as_posix()): 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] = {}
prompt_summaries: list[PromptSummaryRecord] = []
prompt_issues: list[RegistryIssue] = []
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)
prompts_root = docs_dir.joinpath("prompts")
if prompts_root.is_dir():
prompts_by_id, prompt_summaries, prompt_issues = _load_prompts(
prompts_root=prompts_root,
skills_by_id=skills_by_id,
)
for skill_id, skill in skills_by_id.items():
for dependency in skill.depends_on:
if dependency not in skills_by_id:
raise ValueError(f"skill '{skill_id}' depends_on unknown skill '{dependency}'")
issues = [
*skill_issues,
*prompt_issues,
*_collect_relationship_and_uri_issues(
skills_by_id=skills_by_id,
prompts_by_id=prompts_by_id,
),
]
if issues:
raise DocsRegistryValidationError(issues)
skill_ids = tuple(sorted(skills_by_id))
summary_by_id = {summary.skill_id: summary for summary in summaries}
ordered_summaries = tuple(summary_by_id[skill_id] for skill_id in skill_ids)
prompt_ids = tuple(sorted(prompts_by_id))
prompt_summary_by_id = {summary.prompt_id: summary for summary in prompt_summaries}
ordered_prompt_summaries = tuple(prompt_summary_by_id[prompt_id] for prompt_id in prompt_ids)
tag_index: dict[str, list[str]] = {}
capability_index: dict[str, list[str]] = {}
prompt_tag_index: dict[str, list[str]] = {}
for skill_id in skill_ids:
record = skills_by_id[skill_id]
for tag in record.tags:
tag_index.setdefault(tag, []).append(skill_id)
for capability in record.capabilities:
capability_index.setdefault(capability, []).append(skill_id)
for prompt_id in prompt_ids:
record = prompts_by_id[prompt_id]
for tag in record.tags:
prompt_tag_index.setdefault(tag, []).append(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=skill_ids,
skills_summary_in_load_order=ordered_summaries,
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),
docs_markdown_by_path=docs_markdown_by_path,
docs_markdown_path_index=tuple(sorted(docs_markdown_by_path)),
tag_to_skill_ids={key: tuple(sorted(values)) for key, values in sorted(tag_index.items())},
capability_to_skill_ids={key: tuple(sorted(values)) for key, values in sorted(capability_index.items())},
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=prompt_ids,
prompts_summary_in_load_order=ordered_prompt_summaries,
tag_to_prompt_ids={key: tuple(sorted(values)) for key, values in sorted(prompt_tag_index.items())},
prompts_in_load_order=prompts_in_order_tuple,
prompts_summary_in_load_order=tuple(
_to_prompt_summary(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),
)
src/personal_mcp/registry/models/document.py
-56
View File
@@ -1,56 +0,0 @@
from pathlib import PurePosixPath
from typing import Self
from .common import StrictFrozenModel
class MarkdownDocumentModel(StrictFrozenModel):
"""Normalized markdown document content with path and frontmatter."""
relpath: str
content: str
frontmatter: str | None = None
skill_slug: str | None = None
class SkillFilesBundleModel(StrictFrozenModel):
"""Grouped skill documents partitioned into skill, references, and other files."""
slug: str
skill: MarkdownDocumentModel
references: tuple[MarkdownDocumentModel, ...] = ()
other: tuple[MarkdownDocumentModel, ...] = ()
@classmethod
def from_parts(
cls,
*,
slug: str,
skill: MarkdownDocumentModel,
references: tuple[MarkdownDocumentModel, ...] = (),
other: tuple[MarkdownDocumentModel, ...] = (),
) -> Self:
# Ensure deterministic ordering even when the source collection is unordered.
references_sorted = tuple(sorted(references, key=lambda doc: doc.relpath))
other_sorted = tuple(sorted(other, key=lambda doc: doc.relpath))
return cls(
slug=slug,
skill=skill,
references=references_sorted,
other=other_sorted,
)
def to_markdown_document_model(
*,
relpath: PurePosixPath,
content: str,
frontmatter: str | None,
skill_slug: str | None,
) -> MarkdownDocumentModel:
return MarkdownDocumentModel(
relpath=relpath.as_posix(),
content=content,
frontmatter=frontmatter,
skill_slug=skill_slug,
)
src/personal_mcp/registry/models/prompt.py
+17 -4
View File
@@ -1,8 +1,10 @@
import re
from collections.abc import Mapping
from pathlib import PurePosixPath
from typing import Any
from typing import Literal
import yaml
from pydantic import Field
from pydantic import field_validator
@@ -100,10 +102,21 @@ class PromptFrontmatter(StrictFrozenModel):
return value
class PromptDocumentModel(StrictFrozenModel):
"""Structured representation of a prompt markdown document."""
class StoredPrompt(StrictFrozenModel):
"""Normalized prompt document content with path and frontmatter for storage in the registry."""
prompt_id: str
relpath: str
relpath: PurePosixPath
content: str
frontmatter: PromptFrontmatter
frontmatter: PromptFrontmatter | None = None
@field_validator("frontmatter", mode="before")
@classmethod
def parse_frontmatter_yaml(cls, value: str | None):
if value is None:
return None
try:
data = yaml.safe_load(value)
except yaml.YAMLError as e:
raise ValueError(f"invalid YAML in frontmatter: {e}") from e
return PromptFrontmatter.model_validate(data)
src/personal_mcp/registry/models/skill.py
+23 -10
View File
@@ -1,6 +1,7 @@
from collections.abc import Mapping
from pathlib import PurePosixPath
import yaml
from pydantic import Field
from pydantic import field_validator
@@ -107,19 +108,31 @@ class SkillFrontmatter(StrictFrozenModel):
return value
class SkillDocumentModel(StrictFrozenModel):
"""Structured representation of a skill markdown document."""
skill_id: str
relpath: PurePosixPath
content: str
frontmatter: SkillFrontmatter
class SkillReferenceDocumentModel(StrictFrozenModel):
class StoredSkillReference(StrictFrozenModel):
"""Structured representation of a skill reference markdown document."""
ref_id: str
relpath: PurePosixPath
content: str
entry: ReferenceEntry
class StoredSkill(StrictFrozenModel):
"""Structured representation of a skill markdown document."""
skill_id: str
relpath: PurePosixPath
content: str
frontmatter: SkillFrontmatter
references: dict[str, StoredSkillReference] = Field(default_factory=dict)
@field_validator("frontmatter", mode="before")
@classmethod
def parse_frontmatter_yaml(cls, value: str | None):
if value is None:
return None
try:
data = yaml.safe_load(value)
except yaml.YAMLError as e:
raise ValueError(f"invalid YAML in frontmatter: {e}") from e
return SkillFrontmatter.model_validate(data)
src/personal_mcp/registry/prompts.py
View File