ty checking

This commit is contained in:
John Lancaster
2026-06-21 15:29:57 -05:00
parent c5b7733528
commit 7fec3a4337
12 changed files with 372 additions and 239 deletions
+10 -25
View File
@@ -2,9 +2,9 @@ from __future__ import annotations
from typing import Any
from personal_mcp.skills.document_loader import DocsRegistry
from personal_mcp.skills.document_loader import PromptRecord
from personal_mcp.skills.document_loader import SkillRecord
from personal_mcp.registry.contracts import DocsRegistry
from personal_mcp.registry.models.registry import PromptRecord
from personal_mcp.registry.models.registry import SkillRecord
DEFAULT_LIMIT = 20
MAX_LIMIT = 100
@@ -36,8 +36,7 @@ def _summary_payload(skill: SkillRecord) -> dict[str, Any]:
"resources": {
"document": skill.document_uri,
"references": [
f"resource://skills/{skill.skill_id}/references/{ref_id}"
for ref_id in sorted(skill.references)
f"resource://skills/{skill.skill_id}/references/{ref_id}" for ref_id in sorted(skill.references)
],
},
}
@@ -124,14 +123,8 @@ def build_skills_index_payload(
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)
]
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
@@ -187,13 +180,8 @@ def build_prompts_index_payload(
except ValueError as exc:
raise ValueError("cursor must be an integer string") from exc
ordered = [
registry.prompts_by_id[prompt_id]
for prompt_id in registry.prompts_in_load_order
]
matches = [
prompt for prompt in ordered if _prompt_matches(prompt, query=query, tag=tag)
]
ordered = [registry.prompts_by_id[prompt_id] for prompt_id in registry.prompts_in_load_order]
matches = [prompt for prompt in ordered if _prompt_matches(prompt, query=query, tag=tag)]
page = matches[start : start + normalized_limit]
next_cursor = start + normalized_limit
@@ -207,9 +195,7 @@ def build_prompts_index_payload(
}
def build_prompt_detail_payload(
registry: DocsRegistry, prompt_id: str
) -> dict[str, Any]:
def build_prompt_detail_payload(registry: DocsRegistry, prompt_id: str) -> dict[str, Any]:
if prompt_id not in registry.prompts_by_id:
raise KeyError(prompt_id)
@@ -225,8 +211,7 @@ def build_prompt_detail_payload(
"document": prompt.document_uri,
},
"arguments": {
arg_name: arg.model_dump(exclude_none=True)
for arg_name, arg in sorted(prompt.arguments.items())
arg_name: arg.model_dump(exclude_none=True) for arg_name, arg in sorted(prompt.arguments.items())
},
}
+17 -14
View File
@@ -5,6 +5,7 @@ import re
from inspect import Parameter
from inspect import Signature
from typing import Any
from typing import cast
from fastmcp import FastMCP
from fastmcp.server.transforms import ResourcesAsTools
@@ -19,12 +20,12 @@ 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.skills.document_loader import DocsRegistry
from personal_mcp.skills.document_loader import load_docs_registry
from personal_mcp.skills.document_loader import read_docs_markdown_path
from personal_mcp.skills.document_loader import read_prompt_document
from personal_mcp.skills.document_loader import read_skill_document
from personal_mcp.skills.document_loader import read_skill_reference
from personal_mcp.registry.contracts import DocsRegistry
from personal_mcp.registry.load import load_docs_registry
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
DOCS_ROOT = os.getenv("PERSONAL_MCP_DOCS_ROOT", "../../docs")
TOOL_SEARCH_MODE = os.getenv("PERSONAL_MCP_TOOL_SEARCH", "none").strip().lower()
@@ -70,9 +71,7 @@ def _install_tool_fallback_transforms() -> None:
mcp.add_transform(BM25SearchTransform(**kwargs))
return
raise ValueError(
"PERSONAL_MCP_TOOL_SEARCH must be one of: none, regex, bm25"
)
raise ValueError("PERSONAL_MCP_TOOL_SEARCH must be one of: none, regex, bm25")
def _ro_annotations() -> dict[str, bool]:
@@ -105,6 +104,13 @@ def _python_type(prompt_arg_type: str) -> type[Any]:
return str
def _make_prompt_handler(content: str):
def prompt_handler(**kwargs: Any) -> str:
return _render_prompt_markdown(content, kwargs)
return prompt_handler
def _register_prompt_objects() -> None:
for prompt_id in REGISTRY.prompts_in_load_order:
prompt = REGISTRY.prompts_by_id[prompt_id]
@@ -126,15 +132,12 @@ def _register_prompt_objects() -> None:
signature = Signature(parameters=params, return_annotation=str)
prompt_content = prompt.document_content
def prompt_handler(**kwargs: Any) -> str:
return _render_prompt_markdown(prompt_content, kwargs)
prompt_handler = _make_prompt_handler(prompt.document_content)
prompt_handler.__name__ = re.sub(r"[^a-zA-Z0-9_]", "_", prompt_id)
prompt_handler.__doc__ = prompt.description
prompt_handler.__annotations__ = annotations
prompt_handler.__signature__ = signature # type: ignore[attr-defined]
cast(Any, prompt_handler).__signature__ = signature
mcp.prompt(
prompt_handler,
name=prompt_id,
+3 -9
View File
@@ -30,15 +30,9 @@ class SkillFilesBundle:
@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)
sorted_paths = tuple(sorted(paths, key=lambda p: p.relpath.as_posix()))
references = tuple(p for p in sorted_paths if fnmatch(p.relpath.as_posix(), f"skills/{slug}/references/*.md"))
other = tuple(p for p in sorted_paths if p not in references and p != skill)
return cls(
slug=slug,
skill=skill,
+279 -187
View File
@@ -1,6 +1,14 @@
from importlib.resources import files
from pathlib import PurePosixPath
from pydantic import ValidationError
from ..skills.document_loader import _discover_top_level_references
from ..skills.document_loader import _parse_frontmatter
from ..skills.document_loader import _reference_id_from_filename
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
@@ -11,49 +19,47 @@ from .issues import DocsRegistryValidationError
from .issues import RegistryIssue
def load_docs_registry(
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 _load_skills(
*,
package_anchor: str,
docs_root: str = "docs",
) -> DocsRegistry:
docs_dir = files(package_anchor).joinpath(docs_root)
issues: list[RegistryIssue] = []
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: dict[str, str] = {}
for relpath, doc_file in _walk_markdown(docs_dir):
docs_markdown_by_path[relpath] = doc_file.read_text(encoding="utf-8")
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",
)
]
)
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] = []
prompts_by_id: dict[str, PromptRecord] = {}
prompt_summaries: list[PromptSummaryRecord] = []
issues: list[RegistryIssue] = []
for skill_dir in sorted(skills_root.iterdir(), key=lambda item: item.name):
if not skill_dir.is_dir():
@@ -164,153 +170,173 @@ def load_docs_registry(
)
)
prompts_root = docs_dir.joinpath("prompts")
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()
if prompts_root.is_dir():
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")
for prompt_dir in sorted(prompts_root.iterdir(), key=lambda item: item.name):
if not prompt_dir.is_dir():
continue
if not prompt_doc_file.is_file():
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")
discovered_prompt_docs.add(prompt_doc_relpath)
if not prompt_doc_file.is_file():
continue
prompt_markdown = prompt_doc_file.read_text(encoding="utf-8")
try:
raw_frontmatter, _ = _parse_frontmatter(
prompt_markdown,
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,
)
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,
hint="fix PROMPT.md YAML frontmatter to match the contract",
)
)
continue
for relpath, prompt_file in _walk_markdown(
prompts_root,
prefix=PurePosixPath("prompts"),
):
if relpath in discovered_prompt_docs or relpath.endswith("/PROMPT.md"):
continue
prompt_markdown = prompt_file.read_text(encoding="utf-8")
prompt_id = _reference_id_from_filename(PurePosixPath(relpath).name)
if prompt_id is None:
continue
if prompt_id in prompts_by_id:
issues.append(
RegistryIssue(
code="duplicate_prompt_id",
message="duplicate prompt id discovered",
skill_id=prompt_id,
path=relpath,
hint="ensure each prompt id is unique",
)
)
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=relpath,
hint="use a unique prompt id that does not match a skill id",
)
)
continue
prompt_record = PromptRecord(
prompt_id=prompt_id,
name=prompt_id,
description=f"Legacy prompt loaded from docs/{relpath}",
version="1.0.0",
tags=("prompt", "legacy"),
capabilities=(f"resource://prompts/{prompt_id}/document",),
arguments={},
document_uri=f"resource://prompts/{prompt_id}/document",
document_relpath=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,
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,
)
)
for relpath, prompt_file in _walk_markdown(
prompts_root,
prefix=PurePosixPath("prompts"),
):
if relpath in discovered_prompt_docs or relpath.endswith("/PROMPT.md"):
continue
prompt_markdown = prompt_file.read_text(encoding="utf-8")
prompt_id = _reference_id_from_filename(PurePosixPath(relpath).name)
if prompt_id is None:
continue
if prompt_id in prompts_by_id:
issues.append(
RegistryIssue(
code="duplicate_prompt_id",
message="duplicate prompt id discovered",
skill_id=prompt_id,
path=relpath,
hint="ensure each prompt id is unique",
)
)
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=relpath,
hint="use a unique prompt id that does not match a skill id",
)
)
continue
prompt_record = PromptRecord(
prompt_id=prompt_id,
name=prompt_id,
description=f"Legacy prompt loaded from docs/{relpath}",
version="1.0.0",
tags=("prompt", "legacy"),
capabilities=(f"resource://prompts/{prompt_id}/document",),
arguments={},
document_uri=f"resource://prompts/{prompt_id}/document",
document_relpath=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:
@@ -363,18 +389,84 @@ def load_docs_registry(
seen_uris.add(uri)
for prompt_id, record in sorted(prompts_by_id.items()):
for uri in [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",
)
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)
)
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,
)
prompts_by_id: dict[str, PromptRecord] = {}
prompt_summaries: list[PromptSummaryRecord] = []
prompt_issues: list[RegistryIssue] = []
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,
)
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)
+1
View File
@@ -1,3 +1,4 @@
from ..skills.document_loader import _normalize_docs_path
from .contracts import DocsRegistry
+5 -2
View File
@@ -7,7 +7,9 @@ from typing import Any
import yaml
from ..registry.models.common import SKILL_ID_RE
from ..registry.models.common import ReferenceEntry
from ..registry.models.prompt import PromptFrontmatter
from ..registry.models.skill import SkillFrontmatter
@@ -32,15 +34,16 @@ def _parse_frontmatter(markdown: str, *, path: str) -> tuple[dict[str, Any], str
body = "\n".join(lines[end_index + 1 :])
parsed = yaml.safe_load(raw_yaml)
if not isinstance(parsed, dict):
raise ValueError(f"frontmatter must parse to an object: {path}")
raise TypeError(f"frontmatter must parse to an object: {path}")
return parsed, body
def _walk_markdown(
node: Traversable,
*,
prefix: PurePosixPath = PurePosixPath(),
prefix: PurePosixPath | None = None,
) -> list[tuple[str, Traversable]]:
prefix = PurePosixPath() if prefix is None else prefix
results: list[tuple[str, Traversable]] = []
for child in sorted(node.iterdir(), key=lambda item: item.name):
relpath = prefix.joinpath(child.name)