From dab539489a352004332ffc9e2e74677cba3c7e31 Mon Sep 17 00:00:00 2001 From: John Lancaster <32917998+jsl12@users.noreply.github.com> Date: Sun, 21 Jun 2026 09:13:30 -0500 Subject: [PATCH] started manual refactor --- ruff.toml | 61 ++ src/personal_mcp/catalog/__init__.py | 24 +- src/personal_mcp/catalog/server.py | 14 +- src/personal_mcp/mcp.py | 43 +- src/personal_mcp/registry/__init__.py | 0 src/personal_mcp/registry/contracts.py | 80 ++ src/personal_mcp/registry/file.py | 51 ++ src/personal_mcp/registry/issues.py | 22 + src/personal_mcp/registry/load.py | 416 +++++++++ src/personal_mcp/registry/models/__init__.py | 0 src/personal_mcp/registry/models/common.py | 29 + src/personal_mcp/registry/models/prompt.py | 99 +++ src/personal_mcp/registry/models/skill.py | 90 ++ src/personal_mcp/registry/prompts.py | 0 src/personal_mcp/registry/read.py | 61 ++ src/personal_mcp/registry/skills.py | 43 + src/personal_mcp/skills/document_loader.py | 822 +----------------- src/personal_mcp/skills/parse.py | 0 src/personal_mcp/web/app.py | 3 +- src/personal_mcp/web/config.py | 3 +- src/personal_mcp/web/docs_mount.py | 4 +- .../catalog/test_prompt_loader_and_catalog.py | 131 --- .../integration/catalog/test_tool_fallback.py | 94 -- .../skills/test_document_loader_references.py | 144 --- 24 files changed, 999 insertions(+), 1235 deletions(-) create mode 100644 ruff.toml create mode 100644 src/personal_mcp/registry/__init__.py create mode 100644 src/personal_mcp/registry/contracts.py create mode 100644 src/personal_mcp/registry/file.py create mode 100644 src/personal_mcp/registry/issues.py create mode 100644 src/personal_mcp/registry/load.py create mode 100644 src/personal_mcp/registry/models/__init__.py create mode 100644 src/personal_mcp/registry/models/common.py create mode 100644 src/personal_mcp/registry/models/prompt.py create mode 100644 src/personal_mcp/registry/models/skill.py create mode 100644 src/personal_mcp/registry/prompts.py create mode 100644 src/personal_mcp/registry/read.py create mode 100644 src/personal_mcp/registry/skills.py create mode 100644 src/personal_mcp/skills/parse.py delete mode 100644 tests/integration/catalog/test_prompt_loader_and_catalog.py delete mode 100644 tests/integration/catalog/test_tool_fallback.py delete mode 100644 tests/unit/skills/test_document_loader_references.py diff --git a/ruff.toml b/ruff.toml new file mode 100644 index 0000000..25429e7 --- /dev/null +++ b/ruff.toml @@ -0,0 +1,61 @@ +line-length = 120 +indent-width = 4 +target-version = "py313" + +exclude = [ + ".venv", + ".devenv", + ".git", + ".vscode", + "build", + "site", + "__pycache__", +] + +[lint] +preview = true + +extend-select = [ + "ARG", # https://docs.astral.sh/ruff/rules/#flake8-unused-arguments-arg + "B", # https://docs.astral.sh/ruff/rules/#flake8-bugbear-b + "C4", # https://docs.astral.sh/ruff/rules/#flake8-comprehensions-c4 + "DOC102", # https://docs.astral.sh/ruff/rules/docstring-extraneous-parameter/ + "DOC202", # https://docs.astral.sh/ruff/rules/docstring-extraneous-returns/ + "DOC403", # https://docs.astral.sh/ruff/rules/docstring-extraneous-yields/ + "DOC502", # https://docs.astral.sh/ruff/rules/docstring-extraneous-exception/ + "E", "W", # https://docs.astral.sh/ruff/rules/#pycodestyle-e-w + "F", # https://docs.astral.sh/ruff/rules/#pyflakes-f + "FURB", # https://docs.astral.sh/ruff/rules/#refurb-furb + "I", # https://docs.astral.sh/ruff/rules/#isort-i + "N", # https://docs.astral.sh/ruff/rules/#pep8-naming-n + "PD", # https://docs.astral.sh/ruff/rules/#pandas-vet-pd + "PTH", # https://docs.astral.sh/ruff/rules/#flake8-use-pathlib-pth + "UP", # https://docs.astral.sh/ruff/rules/#pyupgrade-up + "SIM", # https://docs.astral.sh/ruff/rules/#flake8-simplify-sim + "PLR0202", # https://docs.astral.sh/ruff/rules/no-classmethod-decorator/ + "PLR0203", # https://docs.astral.sh/ruff/rules/no-staticmethod-decorator/ + "PLR0206", # https://docs.astral.sh/ruff/rules/property-with-parameters/ + "PLR0915", # https://docs.astral.sh/ruff/rules/too-many-statements/ + "PLR1702", # https://docs.astral.sh/ruff/rules/too-many-nested-blocks/ + "TRY002", +] +extend-fixable = ["ALL"] +ignore = [ + "UP046", + "UP047", +] + +[lint.extend-per-file-ignores] +"*.ipynb" = [ + "F401", # unused imports + "F841", # unused local variable +] + +[lint.isort] +force-single-line = true + +[format] +quote-style = "double" +indent-style = "space" +skip-magic-trailing-comma = false +line-ending = "auto" diff --git a/src/personal_mcp/catalog/__init__.py b/src/personal_mcp/catalog/__init__.py index bce0ab7..563171d 100644 --- a/src/personal_mcp/catalog/__init__.py +++ b/src/personal_mcp/catalog/__init__.py @@ -1,21 +1,19 @@ -from personal_mcp.catalog.server import ( - 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, -) +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_skill_detail_payload", "build_prompt_detail_payload", "build_prompts_index_payload", + "build_skill_detail_payload", "build_skills_index_payload", - "get_prompt_by_id_payload", "get_pattern_by_id_payload", - "search_prompts_payload", + "get_prompt_by_id_payload", "search_patterns_payload", + "search_prompts_payload", ] diff --git a/src/personal_mcp/catalog/server.py b/src/personal_mcp/catalog/server.py index 5972bca..64f6066 100644 --- a/src/personal_mcp/catalog/server.py +++ b/src/personal_mcp/catalog/server.py @@ -2,7 +2,9 @@ from __future__ import annotations from typing import Any -from personal_mcp.skills.document_loader import DocsRegistry, PromptRecord, SkillRecord +from personal_mcp.skills.document_loader import DocsRegistry +from personal_mcp.skills.document_loader import PromptRecord +from personal_mcp.skills.document_loader import SkillRecord DEFAULT_LIMIT = 20 MAX_LIMIT = 100 @@ -79,10 +81,7 @@ def _skill_matches( if tag and tag not in skill.tags: return False - if capability and capability not in skill.capabilities: - return False - - return True + return not (capability and capability not in skill.capabilities) def _prompt_matches( @@ -107,10 +106,7 @@ def _prompt_matches( if any(term not in haystack for term in terms): return False - if tag and tag not in prompt.tags: - return False - - return True + return not (tag and tag not in prompt.tags) def build_skills_index_payload( diff --git a/src/personal_mcp/mcp.py b/src/personal_mcp/mcp.py index d871e4c..db4aa2b 100644 --- a/src/personal_mcp/mcp.py +++ b/src/personal_mcp/mcp.py @@ -2,31 +2,29 @@ from __future__ import annotations import os import re -from inspect import Parameter, Signature +from inspect import Parameter +from inspect import Signature from typing import Any from fastmcp import FastMCP from fastmcp.server.transforms import ResourcesAsTools -from fastmcp.server.transforms.search import BM25SearchTransform, RegexSearchTransform +from fastmcp.server.transforms.search import BM25SearchTransform +from fastmcp.server.transforms.search import RegexSearchTransform -from personal_mcp.catalog.server import ( - 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, -) -from personal_mcp.skills.document_loader import ( - DocsRegistry, - load_docs_registry, - read_docs_markdown_path, - read_prompt_document, - read_skill_document, - read_skill_reference, -) +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.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 DOCS_ROOT = os.getenv("PERSONAL_MCP_DOCS_ROOT", "../../docs") TOOL_SEARCH_MODE = os.getenv("PERSONAL_MCP_TOOL_SEARCH", "none").strip().lower() @@ -116,10 +114,7 @@ def _register_prompt_objects() -> None: for arg_name, arg in sorted(prompt.arguments.items()): arg_type = _python_type(arg.type) annotations[arg_name] = arg_type - if arg.required: - default = Parameter.empty - else: - default = arg.default + default = Parameter.empty if arg.required else arg.default params.append( Parameter( arg_name, diff --git a/src/personal_mcp/registry/__init__.py b/src/personal_mcp/registry/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/personal_mcp/registry/contracts.py b/src/personal_mcp/registry/contracts.py new file mode 100644 index 0000000..139714b --- /dev/null +++ b/src/personal_mcp/registry/contracts.py @@ -0,0 +1,80 @@ +from dataclasses import dataclass +from dataclasses import field + +from .models.prompt import PromptArgumentEntry + + +@dataclass(frozen=True) +class ReferenceRecord: + ref_id: str + uri: str + relpath: str + mime_type: str + title: str | None + content: str + + +@dataclass(frozen=True) +class SkillRecord: + skill_id: str + name: str + description: str + version: str + tags: tuple[str, ...] + capabilities: tuple[str, ...] + depends_on: tuple[str, ...] + document_uri: str + document_relpath: str + document_content: str + references: dict[str, ReferenceRecord] + + +@dataclass(frozen=True) +class SkillSummaryRecord: + skill_id: str + name: str + description: str + tags: tuple[str, ...] + capabilities: tuple[str, ...] + document_uri: str + version: str + + +@dataclass(frozen=True) +class PromptRecord: + prompt_id: str + name: str + description: str + version: str + tags: tuple[str, ...] + capabilities: tuple[str, ...] + arguments: dict[str, PromptArgumentEntry] + document_uri: str + document_relpath: str + document_content: str + + +@dataclass(frozen=True) +class PromptSummaryRecord: + prompt_id: str + name: str + description: str + tags: tuple[str, ...] + capabilities: tuple[str, ...] + document_uri: str + version: str + + +@dataclass(frozen=True) +class DocsRegistry: + skills_by_id: dict[str, SkillRecord] + skills_in_load_order: tuple[str, ...] + skills_summary_in_load_order: tuple[SkillSummaryRecord, ...] + docs_markdown_by_path: dict[str, str] + docs_markdown_path_index: tuple[str, ...] + tag_to_skill_ids: dict[str, tuple[str, ...]] + capability_to_skill_ids: dict[str, tuple[str, ...]] + prompts_by_id: dict[str, PromptRecord] = field(default_factory=dict) + prompts_in_load_order: tuple[str, ...] = () + prompts_summary_in_load_order: tuple[PromptSummaryRecord, ...] = () + tag_to_prompt_ids: dict[str, tuple[str, ...]] = field(default_factory=dict) diff --git a/src/personal_mcp/registry/file.py b/src/personal_mcp/registry/file.py new file mode 100644 index 0000000..8809091 --- /dev/null +++ b/src/personal_mcp/registry/file.py @@ -0,0 +1,51 @@ +from collections.abc import Generator +from importlib.abc import Traversable +from pathlib import PurePosixPath + +import yaml + +from .models.skill import SkillFrontmatter + + +def walk_resources( + node: Traversable, + *, + suffix: str = ".md", + prefix: PurePosixPath | None = None, +) -> Generator[tuple[str, Traversable]]: + """Recursively yield all resources in node, with their full path.""" + prefix = prefix if prefix is not None else PurePosixPath() + for child in sorted(node.iterdir(), key=lambda item: item.name): + relpath = prefix.joinpath(child.name) + if child.is_dir(): + yield from walk_resources(child, suffix=suffix, prefix=relpath) + continue + if not child.is_file() or not child.name.lower().endswith(suffix): + continue + yield relpath.as_posix(), child + + +def get_markdown_content(resource: Traversable) -> dict[str, str]: + """Read the content of a markdown resource as text.""" + return {relpath: doc_file.read_text(encoding="utf-8") for relpath, doc_file in walk_resources(resource)} + + +def get_idx(raw): + for i, line in enumerate(raw.splitlines()): + if line.strip().startswith("---"): + yield i + + +def get_raw_frontmatter(raw: str) -> str: + delimiter = iter(get_idx(raw)) + start = next(delimiter) + 1 + end = next(delimiter) + return "\n".join(raw.splitlines()[start:end]) + + +def gen_valid_frontmatter(content: dict[str, str]) -> Generator[SkillFrontmatter]: + for relpath, raw in content.items(): + if relpath.endswith("SKILL.md"): + fm = get_raw_frontmatter(raw) + validated = SkillFrontmatter.model_validate(yaml.safe_load(fm)) + yield validated diff --git a/src/personal_mcp/registry/issues.py b/src/personal_mcp/registry/issues.py new file mode 100644 index 0000000..7eb52bf --- /dev/null +++ b/src/personal_mcp/registry/issues.py @@ -0,0 +1,22 @@ +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) diff --git a/src/personal_mcp/registry/load.py b/src/personal_mcp/registry/load.py new file mode 100644 index 0000000..8077dac --- /dev/null +++ b/src/personal_mcp/registry/load.py @@ -0,0 +1,416 @@ +from importlib.resources import files +from pathlib import PurePosixPath + +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 + + +def load_docs_registry( + *, + 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", + ) + ] + ) + + skills_by_id: dict[str, SkillRecord] = {} + summaries: list[SkillSummaryRecord] = [] + prompts_by_id: dict[str, PromptRecord] = {} + prompt_summaries: list[PromptSummaryRecord] = [] + + for skill_dir in sorted(skills_root.iterdir(), key=lambda item: item.name): + if not skill_dir.is_dir(): + 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.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, + ) + 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, + ) + ) + + prompts_root = docs_dir.joinpath("prompts") + 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") + + 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, + ) + ) + + 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, + ) + ) + + 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()): + 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", + ) + ) + seen_uris.add(uri) + + 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) + + return DocsRegistry( + skills_by_id=skills_by_id, + skills_in_load_order=skill_ids, + skills_summary_in_load_order=ordered_summaries, + 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())}, + 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())}, + ) diff --git a/src/personal_mcp/registry/models/__init__.py b/src/personal_mcp/registry/models/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/personal_mcp/registry/models/common.py b/src/personal_mcp/registry/models/common.py new file mode 100644 index 0000000..0d8ec48 --- /dev/null +++ b/src/personal_mcp/registry/models/common.py @@ -0,0 +1,29 @@ +import re +from pathlib import PurePosixPath + +from pydantic import BaseModel +from pydantic import ConfigDict +from pydantic import field_validator + +SKILL_ID_RE = re.compile(r"^[a-z][a-z0-9-]*$") +SEMVER_RE = re.compile(r"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)(?:[-+][0-9A-Za-z.-]+)?$") + + +class ReferenceEntry(BaseModel): + model_config = ConfigDict(extra="ignore", str_strip_whitespace=True) + + path: str + mime_type: str = "text/markdown" + title: str | None = None + + @field_validator("path") + @classmethod + def validate_reference_path(cls, value: str) -> str: + path = PurePosixPath(value) + if path.is_absolute() or ".." in path.parts: + raise ValueError("reference path must be a relative in-skill path") + if not str(path).startswith("references/"): + raise ValueError("reference path must stay under references/") + if path.suffix.lower() != ".md": + raise ValueError("reference path must target a markdown file") + return path.as_posix() diff --git a/src/personal_mcp/registry/models/prompt.py b/src/personal_mcp/registry/models/prompt.py new file mode 100644 index 0000000..13f0c79 --- /dev/null +++ b/src/personal_mcp/registry/models/prompt.py @@ -0,0 +1,99 @@ +import re +from typing import Any + +from pydantic import BaseModel +from pydantic import ConfigDict +from pydantic import Field +from pydantic import field_validator + +from .common import SEMVER_RE +from .common import SKILL_ID_RE + + +class PromptArgumentEntry(BaseModel): + model_config = ConfigDict(extra="ignore", str_strip_whitespace=True) + + type: str = Field(min_length=1) + description: str | None = None + required: bool = False + default: Any | None = None + enum: list[str] | None = None + + @field_validator("type") + @classmethod + def validate_type(cls, value: str) -> str: + allowed_types = { + "string", + "number", + "integer", + "boolean", + "array", + "object", + } + if value not in allowed_types: + raise ValueError(f"unsupported prompt argument type: {value}") + return value + + @field_validator("enum") + @classmethod + def validate_enum(cls, value: list[str] | None) -> list[str] | None: + if value is not None and not value: + raise ValueError("enum must contain at least one value when provided") + return value + + +class PromptMetadata(BaseModel): + model_config = ConfigDict(extra="ignore", str_strip_whitespace=True) + + id: str + version: str + tags: list[str] = Field(default_factory=list) + capabilities: list[str] = Field(min_length=1) + arguments: dict[str, PromptArgumentEntry] = Field(default_factory=dict) + + @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: list[str]) -> list[str]: + for tag in value: + if not SKILL_ID_RE.fullmatch(tag): + raise ValueError(f"invalid tag: {tag}") + return value + + @field_validator("arguments") + @classmethod + def validate_argument_names(cls, value: dict[str, PromptArgumentEntry]) -> dict[str, PromptArgumentEntry]: + for name in value: + if not re.fullmatch(r"^[A-Za-z_][A-Za-z0-9_]*$", name): + raise ValueError(f"invalid prompt argument name: {name}") + return value + + +class PromptFrontmatter(BaseModel): + model_config = ConfigDict(extra="ignore", str_strip_whitespace=True) + + name: str = Field(min_length=1, max_length=64) + description: str = Field(min_length=1, max_length=1024) + x_personal_mcp: PromptMetadata = 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 diff --git a/src/personal_mcp/registry/models/skill.py b/src/personal_mcp/registry/models/skill.py new file mode 100644 index 0000000..dd4357b --- /dev/null +++ b/src/personal_mcp/registry/models/skill.py @@ -0,0 +1,90 @@ +from pydantic import BaseModel +from pydantic import ConfigDict +from pydantic import Field +from pydantic import field_validator + +from .common import SEMVER_RE +from .common import SKILL_ID_RE +from .common import ReferenceEntry + + +class PersonalMcpMetadata(BaseModel): + model_config = ConfigDict(extra="ignore", str_strip_whitespace=True) + + id: str + version: str + tags: list[str] = Field(default_factory=list) + capabilities: list[str] = Field(min_length=1) + depends_on: list[str] = Field(default_factory=list) + references: dict[str, ReferenceEntry] = Field(default_factory=dict) + + @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: list[str]) -> list[str]: + for tag in value: + if not SKILL_ID_RE.fullmatch(tag): + raise ValueError(f"invalid tag: {tag}") + return value + + @field_validator("depends_on") + @classmethod + def validate_depends_on(cls, value: list[str]) -> list[str]: + for dep in value: + if not SKILL_ID_RE.fullmatch(dep): + raise ValueError(f"invalid depends_on skill id: {dep}") + return value + + @field_validator("references") + @classmethod + def validate_reference_ids(cls, value: dict[str, ReferenceEntry]) -> dict[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(BaseModel): + model_config = ConfigDict(extra="ignore", str_strip_whitespace=True) + + name: str = Field(min_length=1, max_length=64) + description: str = Field(min_length=1, max_length=1024) + when_to_use: str | None = None + allowed_tools: str | list[str] | None = Field(default=None, alias="allowed-tools") + disallowed_tools: str | list[str] | None = Field( + default=None, + alias="disallowed-tools", + ) + disable_model_invocation: bool | None = Field( + default=None, + alias="disable-model-invocation", + ) + user_invocable: bool | None = Field(default=None, alias="user-invocable") + argument_hint: str | None = Field(default=None, alias="argument-hint") + arguments: str | list[str] | None = None + license: str | None = None + compatibility: str | None = None + metadata: dict[str, str] | None = None + x_personal_mcp: PersonalMcpMetadata = 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 diff --git a/src/personal_mcp/registry/prompts.py b/src/personal_mcp/registry/prompts.py new file mode 100644 index 0000000..e69de29 diff --git a/src/personal_mcp/registry/read.py b/src/personal_mcp/registry/read.py new file mode 100644 index 0000000..3d91bf0 --- /dev/null +++ b/src/personal_mcp/registry/read.py @@ -0,0 +1,61 @@ +from .contracts 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}", + "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}", + "content": reference.content, + } + + +def read_docs_markdown_path(registry: DocsRegistry, path: str) -> dict[str, str]: + normalized_path = _normalize_docs_path(path) + if normalized_path not in registry.docs_markdown_by_path: + raise KeyError(f"unknown docs path: {normalized_path}") + return { + "uri": f"resource://docs/{normalized_path}", + "format": "markdown", + "source_path": f"docs/{normalized_path}", + "content": registry.docs_markdown_by_path[normalized_path], + } + + +def read_prompt_document(registry: DocsRegistry, prompt_id: str) -> dict[str, str]: + if prompt_id not in registry.prompts_by_id: + raise KeyError(f"unknown prompt_id: {prompt_id}") + prompt = registry.prompts_by_id[prompt_id] + return { + "id": prompt.prompt_id, + "uri": prompt.document_uri, + "format": "markdown", + "source_path": f"docs/{prompt.document_relpath}", + "content": prompt.document_content, + } diff --git a/src/personal_mcp/registry/skills.py b/src/personal_mcp/registry/skills.py new file mode 100644 index 0000000..bc45a06 --- /dev/null +++ b/src/personal_mcp/registry/skills.py @@ -0,0 +1,43 @@ +from collections.abc import Generator +from dataclasses import dataclass +from fnmatch import fnmatch +from functools import partial +from itertools import groupby + + +def get_skill_filemap(content: dict[str, str]) -> dict[str, list[str]]: + """Get the map of skill slugs to their associated files.""" + + def get_skill_name(relpath: str) -> str | None: + if relpath.startswith("skills/"): + return relpath.split("/")[1] + + grouped = groupby(content.keys(), key=get_skill_name) + return {k: list(v) for k, v in grouped if k is not None} + + +@dataclass(frozen=True, slots=True) +class SkillBundle: + slug: str + skill: str + references: list[str] + other: list[str] + + +def gen_skill_bundles(content: dict[str, str]) -> Generator[SkillBundle]: + """Generate the skill bundles from the map of raw markdown content.""" + matcher = partial(fnmatch, pat="skills/*/references/*.md") + for slug, paths in get_skill_filemap(content).items(): + skill = next(iter(p for p in paths if fnmatch(p, "skills/*/SKILL.md"))) + groups = {k: list(v) for k, v in groupby(paths, key=matcher)} + yield SkillBundle( + slug=slug, + skill=skill, + references=list(groups.get(True, [])), + other=[f for f in groups.get(False, []) if f != skill], + ) + + +def get_all_skill_bundles(content: dict[str, str]) -> list[SkillBundle]: + """Get all skill bundles from the map of raw markdown content.""" + return list(gen_skill_bundles(content)) diff --git a/src/personal_mcp/skills/document_loader.py b/src/personal_mcp/skills/document_loader.py index fca0efb..13ca6b2 100644 --- a/src/personal_mcp/skills/document_loader.py +++ b/src/personal_mcp/skills/document_loader.py @@ -1,318 +1,14 @@ from __future__ import annotations import re -from dataclasses import dataclass, field -from importlib.resources import files from importlib.resources.abc import Traversable from pathlib import PurePosixPath from typing import Any import yaml -from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator -SKILL_ID_RE = re.compile(r"^[a-z][a-z0-9-]*$") -SEMVER_RE = re.compile( - r"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)(?:[-+][0-9A-Za-z.-]+)?$" -) - - -@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} " - f"(skill={issue.skill_id or 'unknown'}, path={issue.path})" - ) - for issue in errors - ] - ) - super().__init__(summary) - - -class ReferenceEntry(BaseModel): - model_config = ConfigDict(extra="ignore", str_strip_whitespace=True) - - path: str - mime_type: str = "text/markdown" - title: str | None = None - - @field_validator("path") - @classmethod - def validate_reference_path(cls, value: str) -> str: - path = PurePosixPath(value) - if path.is_absolute() or ".." in path.parts: - raise ValueError("reference path must be a relative in-skill path") - if not str(path).startswith("references/"): - raise ValueError("reference path must stay under references/") - if path.suffix.lower() != ".md": - raise ValueError("reference path must target a markdown file") - return path.as_posix() - - -class PersonalMcpMetadata(BaseModel): - model_config = ConfigDict(extra="ignore", str_strip_whitespace=True) - - id: str - version: str - tags: list[str] = Field(default_factory=list) - capabilities: list[str] = Field(min_length=1) - depends_on: list[str] = Field(default_factory=list) - references: dict[str, ReferenceEntry] = Field(default_factory=dict) - - @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: list[str]) -> list[str]: - for tag in value: - if not SKILL_ID_RE.fullmatch(tag): - raise ValueError(f"invalid tag: {tag}") - return value - - @field_validator("depends_on") - @classmethod - def validate_depends_on(cls, value: list[str]) -> list[str]: - for dep in value: - if not SKILL_ID_RE.fullmatch(dep): - raise ValueError(f"invalid depends_on skill id: {dep}") - return value - - @field_validator("references") - @classmethod - def validate_reference_ids( - cls, value: dict[str, ReferenceEntry] - ) -> dict[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 PromptArgumentEntry(BaseModel): - model_config = ConfigDict(extra="ignore", str_strip_whitespace=True) - - type: str = Field(min_length=1) - description: str | None = None - required: bool = False - default: Any | None = None - enum: list[str] | None = None - - @field_validator("type") - @classmethod - def validate_type(cls, value: str) -> str: - allowed_types = { - "string", - "number", - "integer", - "boolean", - "array", - "object", - } - if value not in allowed_types: - raise ValueError(f"unsupported prompt argument type: {value}") - return value - - @field_validator("enum") - @classmethod - def validate_enum(cls, value: list[str] | None) -> list[str] | None: - if value is not None and not value: - raise ValueError("enum must contain at least one value when provided") - return value - - -class PromptMetadata(BaseModel): - model_config = ConfigDict(extra="ignore", str_strip_whitespace=True) - - id: str - version: str - tags: list[str] = Field(default_factory=list) - capabilities: list[str] = Field(min_length=1) - arguments: dict[str, PromptArgumentEntry] = Field(default_factory=dict) - - @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: list[str]) -> list[str]: - for tag in value: - if not SKILL_ID_RE.fullmatch(tag): - raise ValueError(f"invalid tag: {tag}") - return value - - @field_validator("arguments") - @classmethod - def validate_argument_names( - cls, value: dict[str, PromptArgumentEntry] - ) -> dict[str, PromptArgumentEntry]: - for name in value: - if not re.fullmatch(r"^[A-Za-z_][A-Za-z0-9_]*$", name): - raise ValueError(f"invalid prompt argument name: {name}") - return value - - -class SkillFrontmatter(BaseModel): - model_config = ConfigDict(extra="ignore", str_strip_whitespace=True) - - name: str = Field(min_length=1, max_length=64) - description: str = Field(min_length=1, max_length=1024) - when_to_use: str | None = None - allowed_tools: str | list[str] | None = Field(default=None, alias="allowed-tools") - disallowed_tools: str | list[str] | None = Field( - default=None, - alias="disallowed-tools", - ) - disable_model_invocation: bool | None = Field( - default=None, - alias="disable-model-invocation", - ) - user_invocable: bool | None = Field(default=None, alias="user-invocable") - argument_hint: str | None = Field(default=None, alias="argument-hint") - arguments: str | list[str] | None = None - license: str | None = None - compatibility: str | None = None - metadata: dict[str, str] | None = None - x_personal_mcp: PersonalMcpMetadata = 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 - - -class PromptFrontmatter(BaseModel): - model_config = ConfigDict(extra="ignore", str_strip_whitespace=True) - - name: str = Field(min_length=1, max_length=64) - description: str = Field(min_length=1, max_length=1024) - x_personal_mcp: PromptMetadata = 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 - - -@dataclass(frozen=True) -class ReferenceRecord: - ref_id: str - uri: str - relpath: str - mime_type: str - title: str | None - content: str - - -@dataclass(frozen=True) -class SkillRecord: - skill_id: str - name: str - description: str - version: str - tags: tuple[str, ...] - capabilities: tuple[str, ...] - depends_on: tuple[str, ...] - document_uri: str - document_relpath: str - document_content: str - references: dict[str, ReferenceRecord] - - -@dataclass(frozen=True) -class SkillSummaryRecord: - skill_id: str - name: str - description: str - tags: tuple[str, ...] - capabilities: tuple[str, ...] - document_uri: str - version: str - - -@dataclass(frozen=True) -class PromptRecord: - prompt_id: str - name: str - description: str - version: str - tags: tuple[str, ...] - capabilities: tuple[str, ...] - arguments: dict[str, PromptArgumentEntry] - document_uri: str - document_relpath: str - document_content: str - - -@dataclass(frozen=True) -class PromptSummaryRecord: - prompt_id: str - name: str - description: str - tags: tuple[str, ...] - capabilities: tuple[str, ...] - document_uri: str - version: str - - -@dataclass(frozen=True) -class DocsRegistry: - skills_by_id: dict[str, SkillRecord] - skills_in_load_order: tuple[str, ...] - skills_summary_in_load_order: tuple[SkillSummaryRecord, ...] - docs_markdown_by_path: dict[str, str] - docs_markdown_path_index: tuple[str, ...] - tag_to_skill_ids: dict[str, tuple[str, ...]] - capability_to_skill_ids: dict[str, tuple[str, ...]] - prompts_by_id: dict[str, PromptRecord] = field(default_factory=dict) - prompts_in_load_order: tuple[str, ...] = () - prompts_summary_in_load_order: tuple[PromptSummaryRecord, ...] = () - tag_to_prompt_ids: dict[str, tuple[str, ...]] = field(default_factory=dict) +from ..registry.models.common import ReferenceEntry +from ..registry.models.skill import SkillFrontmatter def _parse_frontmatter(markdown: str, *, path: str) -> tuple[dict[str, Any], str]: @@ -343,7 +39,7 @@ def _parse_frontmatter(markdown: str, *, path: str) -> tuple[dict[str, Any], str def _walk_markdown( node: Traversable, *, - prefix: PurePosixPath = PurePosixPath(""), + prefix: PurePosixPath = PurePosixPath(), ) -> list[tuple[str, Traversable]]: results: list[tuple[str, Traversable]] = [] for child in sorted(node.iterdir(), key=lambda item: item.name): @@ -357,9 +53,7 @@ def _walk_markdown( return results -def _validate_skill_frontmatter( - raw: dict[str, Any], *, skill_dir_name: str -) -> SkillFrontmatter: +def _validate_skill_frontmatter(raw: dict[str, Any], *, skill_dir_name: str) -> SkillFrontmatter: model = SkillFrontmatter.model_validate(raw) if model.name != skill_dir_name: raise ValueError("frontmatter name must exactly match skill directory name") @@ -371,9 +65,7 @@ def _validate_skill_frontmatter( return model -def _validate_prompt_frontmatter( - raw: dict[str, Any], *, prompt_dir_name: str -) -> PromptFrontmatter: +def _validate_prompt_frontmatter(raw: dict[str, Any], *, prompt_dir_name: str) -> PromptFrontmatter: model = PromptFrontmatter.model_validate(raw) if model.name != prompt_dir_name: raise ValueError("frontmatter name must exactly match prompt directory name") @@ -437,507 +129,3 @@ def _discover_top_level_references( title=_title_from_reference_filename(child.name), ) return discovered - - -def _ensure_no_cycles(skills_by_id: dict[str, SkillRecord]) -> list[tuple[str, str]]: - visiting: set[str] = set() - visited: set[str] = set() - cycles: list[tuple[str, str]] = [] - - def walk(skill_id: str, stack: list[str]) -> None: - if skill_id in visited: - return - if skill_id in visiting: - cycle_from = stack[stack.index(skill_id) :] - cycles.append((skill_id, " -> ".join(cycle_from + [skill_id]))) - return - - visiting.add(skill_id) - stack.append(skill_id) - for dep in skills_by_id[skill_id].depends_on: - if dep in skills_by_id: - walk(dep, stack) - stack.pop() - visiting.remove(skill_id) - visited.add(skill_id) - - for skill_id in sorted(skills_by_id): - walk(skill_id, []) - return cycles - - -def load_docs_registry( - *, - 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", - ) - ] - ) - - skills_by_id: dict[str, SkillRecord] = {} - summaries: list[SkillSummaryRecord] = [] - prompts_by_id: dict[str, PromptRecord] = {} - prompt_summaries: list[PromptSummaryRecord] = [] - - for skill_dir in sorted(skills_root.iterdir(), key=lambda item: item.name): - if not skill_dir.is_dir(): - 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.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, - ) - 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, - ) - ) - - prompts_root = docs_dir.joinpath("prompts") - 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") - - 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, - ) - ) - - 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, - ) - ) - - 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()): - 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", - ) - ) - seen_uris.add(uri) - - 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) - - return DocsRegistry( - skills_by_id=skills_by_id, - skills_in_load_order=skill_ids, - skills_summary_in_load_order=ordered_summaries, - 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()) - }, - 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()) - }, - ) - - -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}", - "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}", - "content": reference.content, - } - - -def read_docs_markdown_path(registry: DocsRegistry, path: str) -> dict[str, str]: - normalized_path = _normalize_docs_path(path) - if normalized_path not in registry.docs_markdown_by_path: - raise KeyError(f"unknown docs path: {normalized_path}") - return { - "uri": f"resource://docs/{normalized_path}", - "format": "markdown", - "source_path": f"docs/{normalized_path}", - "content": registry.docs_markdown_by_path[normalized_path], - } - - -def read_prompt_document(registry: DocsRegistry, prompt_id: str) -> dict[str, str]: - if prompt_id not in registry.prompts_by_id: - raise KeyError(f"unknown prompt_id: {prompt_id}") - prompt = registry.prompts_by_id[prompt_id] - return { - "id": prompt.prompt_id, - "uri": prompt.document_uri, - "format": "markdown", - "source_path": f"docs/{prompt.document_relpath}", - "content": prompt.document_content, - } diff --git a/src/personal_mcp/skills/parse.py b/src/personal_mcp/skills/parse.py new file mode 100644 index 0000000..e69de29 diff --git a/src/personal_mcp/web/app.py b/src/personal_mcp/web/app.py index d179642..2f4353c 100644 --- a/src/personal_mcp/web/app.py +++ b/src/personal_mcp/web/app.py @@ -1,7 +1,8 @@ from fastapi import FastAPI from personal_mcp.mcp import mcp -from personal_mcp.web.config import Settings, get_settings +from personal_mcp.web.config import Settings +from personal_mcp.web.config import get_settings from personal_mcp.web.docs_mount import mount_docs_static from personal_mcp.web.health import router as health_router diff --git a/src/personal_mcp/web/config.py b/src/personal_mcp/web/config.py index e24bb63..b14e646 100644 --- a/src/personal_mcp/web/config.py +++ b/src/personal_mcp/web/config.py @@ -1,7 +1,8 @@ from pathlib import Path from pydantic import Field -from pydantic_settings import BaseSettings, SettingsConfigDict +from pydantic_settings import BaseSettings +from pydantic_settings import SettingsConfigDict _REPO_ROOT = Path(__file__).resolve().parents[3] diff --git a/src/personal_mcp/web/docs_mount.py b/src/personal_mcp/web/docs_mount.py index f35d516..9a7eb67 100644 --- a/src/personal_mcp/web/docs_mount.py +++ b/src/personal_mcp/web/docs_mount.py @@ -1,6 +1,8 @@ from pathlib import Path -from fastapi import FastAPI, Response, status +from fastapi import FastAPI +from fastapi import Response +from fastapi import status from fastapi.responses import RedirectResponse from fastapi.staticfiles import StaticFiles diff --git a/tests/integration/catalog/test_prompt_loader_and_catalog.py b/tests/integration/catalog/test_prompt_loader_and_catalog.py deleted file mode 100644 index 8fcb5ae..0000000 --- a/tests/integration/catalog/test_prompt_loader_and_catalog.py +++ /dev/null @@ -1,131 +0,0 @@ -from __future__ import annotations - -from pathlib import Path - -import pytest - -from personal_mcp.catalog.server import ( - build_prompt_detail_payload, - build_prompts_index_payload, -) -from personal_mcp.skills import document_loader - - -def _write_skill(root: Path, *, skill_id: str) -> None: - skill_dir = root / "docs" / "skills" / skill_id - skill_dir.mkdir(parents=True, exist_ok=True) - skill_doc = "\n".join( - [ - "---", - f"name: {skill_id}", - "description: Example skill", - "x-personal-mcp:", - f" id: {skill_id}", - " version: 1.0.0", - " tags: [example]", - " capabilities:", - f" - resource://skills/{skill_id}/document", - "---", - "", - f"# {skill_id}", - "", - ] - ) - (skill_dir / "SKILL.md").write_text(skill_doc, encoding="utf-8") - - -def _write_frontmatter_prompt(root: Path, *, prompt_id: str) -> None: - prompt_dir = root / "docs" / "prompts" / prompt_id - prompt_dir.mkdir(parents=True, exist_ok=True) - prompt_doc = "\n".join( - [ - "---", - f"name: {prompt_id}", - "description: Scaffold initial tests", - "x-personal-mcp:", - f" id: {prompt_id}", - " version: 1.0.0", - " tags: [pytest, testing]", - " capabilities:", - f" - resource://prompts/{prompt_id}/document", - " arguments:", - " target_scope:", - " type: string", - " description: The test scope", - " required: true", - "---", - "", - "Create tests for {{target_scope}}.", - "", - ] - ) - (prompt_dir / "PROMPT.md").write_text(prompt_doc, encoding="utf-8") - - -def _write_legacy_prompt(root: Path) -> None: - legacy = root / "docs" / "prompts" / "testing" - legacy.mkdir(parents=True, exist_ok=True) - (legacy / "inital_test_structure.md").write_text( - "# Prompt For Creating Initial Test Structure\n", - encoding="utf-8", - ) - - -def _load_registry(monkeypatch: pytest.MonkeyPatch, root: Path): - monkeypatch.setattr(document_loader, "files", lambda _: root) - return document_loader.load_docs_registry(package_anchor="unused") - - -def test_loads_frontmatter_prompt_and_arguments(monkeypatch, tmp_path: Path) -> None: - _write_skill(tmp_path, skill_id="demo-skill") - _write_frontmatter_prompt(tmp_path, prompt_id="initial-test-structure") - - registry = _load_registry(monkeypatch, tmp_path) - - assert "initial-test-structure" in registry.prompts_by_id - prompt = registry.prompts_by_id["initial-test-structure"] - assert prompt.arguments["target_scope"].type == "string" - assert prompt.arguments["target_scope"].required is True - - -def test_loads_legacy_prompt_markdown(monkeypatch, tmp_path: Path) -> None: - _write_skill(tmp_path, skill_id="demo-skill") - _write_legacy_prompt(tmp_path) - - registry = _load_registry(monkeypatch, tmp_path) - - assert "inital-test-structure" in registry.prompts_by_id - prompt = registry.prompts_by_id["inital-test-structure"] - assert prompt.document_relpath == "prompts/testing/inital_test_structure.md" - assert "Legacy prompt loaded" in prompt.description - - -def test_rejects_prompt_skill_id_collision(monkeypatch, tmp_path: Path) -> None: - _write_skill(tmp_path, skill_id="shared-id") - _write_frontmatter_prompt(tmp_path, prompt_id="shared-id") - - with pytest.raises(document_loader.DocsRegistryValidationError) as exc: - _load_registry(monkeypatch, tmp_path) - - codes = [issue.code for issue in exc.value.errors] - assert "prompt_skill_id_collision" in codes - - -def test_prompt_catalog_index_and_detail(monkeypatch, tmp_path: Path) -> None: - _write_skill(tmp_path, skill_id="demo-skill") - _write_frontmatter_prompt(tmp_path, prompt_id="initial-test-structure") - _write_frontmatter_prompt(tmp_path, prompt_id="route-tests") - - registry = _load_registry(monkeypatch, tmp_path) - - index = build_prompts_index_payload( - registry, - query="initial-test-structure", - limit=10, - ) - assert index["total"] == 1 - assert index["prompts"][0]["id"] == "initial-test-structure" - - detail = build_prompt_detail_payload(registry, "initial-test-structure") - assert "target_scope" in detail["arguments"] - assert detail["arguments"]["target_scope"]["required"] is True diff --git a/tests/integration/catalog/test_tool_fallback.py b/tests/integration/catalog/test_tool_fallback.py deleted file mode 100644 index 2cd8bf3..0000000 --- a/tests/integration/catalog/test_tool_fallback.py +++ /dev/null @@ -1,94 +0,0 @@ -from __future__ import annotations - -import asyncio - -import pytest - -from personal_mcp import mcp as mcp_module - - -def _listed_tool_names() -> set[str]: - tools = asyncio.run(mcp_module.mcp.list_tools()) - return {tool.name for tool in tools} - - -@pytest.fixture -def listed_tool_names() -> set[str]: - return _listed_tool_names() - - -def test_canonical_step6_tools_are_listed(listed_tool_names: set[str]) -> None: - assert { - "list_resources", - "read_resource", - "search_patterns", - "get_pattern_by_id", - "get_skill_document_by_id", - "search_prompts", - "get_prompt_by_id", - }.issubset(listed_tool_names) - - -def test_catalog_compatibility_alias_tools_are_listed( - listed_tool_names: set[str], -) -> None: - assert { - "catalog_search_patterns", - "catalog_get_pattern_by_id", - "catalog_get_skill_document_by_id", - "catalog_search_prompts", - "catalog_get_prompt_by_id", - }.issubset(listed_tool_names) - - -@pytest.mark.parametrize( - "skill_id", - [mcp_module.REGISTRY.skills_in_load_order[0], "missing-step6-skill"], -) -def test_skill_document_alias_matches_canonical_payload(skill_id: str) -> None: - canonical = mcp_module.get_skill_document_by_id(skill_id) - alias = mcp_module.catalog_get_skill_document_by_id(skill_id) - - assert alias == canonical - - -def test_search_and_pattern_aliases_match_canonical_payloads() -> None: - search_canonical = mcp_module.search_patterns( - query="mcp", - tags=[], - skip=0, - limit=5, - ) - search_alias = mcp_module.catalog_search_patterns( - query="mcp", - tags=[], - skip=0, - limit=5, - ) - assert search_alias == search_canonical - - sample_skill_id = mcp_module.REGISTRY.skills_in_load_order[0] - pattern_canonical = mcp_module.get_pattern_by_id(sample_skill_id) - pattern_alias = mcp_module.catalog_get_pattern_by_id(sample_skill_id) - assert pattern_alias == pattern_canonical - - -def test_prompt_aliases_match_canonical_payloads() -> None: - search_canonical = mcp_module.search_prompts( - query="test", - tags=[], - skip=0, - limit=5, - ) - search_alias = mcp_module.catalog_search_prompts( - query="test", - tags=[], - skip=0, - limit=5, - ) - assert search_alias == search_canonical - - sample_prompt_id = mcp_module.REGISTRY.prompts_in_load_order[0] - prompt_canonical = mcp_module.get_prompt_by_id(sample_prompt_id) - prompt_alias = mcp_module.catalog_get_prompt_by_id(sample_prompt_id) - assert prompt_alias == prompt_canonical diff --git a/tests/unit/skills/test_document_loader_references.py b/tests/unit/skills/test_document_loader_references.py deleted file mode 100644 index 9e92496..0000000 --- a/tests/unit/skills/test_document_loader_references.py +++ /dev/null @@ -1,144 +0,0 @@ -from __future__ import annotations - -from pathlib import Path -from textwrap import dedent - -import pytest - -from personal_mcp.skills import document_loader - - -def _write_skill( - root: Path, - *, - skill_id: str, - references_block: str = "", -) -> Path: - skill_dir = root / "docs" / "skills" / skill_id - skill_dir.mkdir(parents=True, exist_ok=True) - frontmatter_lines = [ - "---", - f"name: {skill_id}", - "description: Example skill", - "x-personal-mcp:", - f" id: {skill_id}", - " version: 1.0.0", - " tags: [example]", - " capabilities:", - f" - resource://skills/{skill_id}/document", - ] - if references_block: - frontmatter_lines.extend(f" {line}" for line in references_block.splitlines()) - frontmatter_lines.append("---") - - skill_doc = "\n".join(frontmatter_lines) + f"\n\n# {skill_id}\n" - (skill_dir / "SKILL.md").write_text(skill_doc, encoding="utf-8") - return skill_dir - - -def _load_registry(monkeypatch: pytest.MonkeyPatch, root: Path): - monkeypatch.setattr(document_loader, "files", lambda _: root) - return document_loader.load_docs_registry(package_anchor="unused") - - -def test_auto_discovers_top_level_references(monkeypatch, tmp_path: Path) -> None: - skill_dir = _write_skill(tmp_path, skill_id="demo-skill") - refs = skill_dir / "references" - refs.mkdir() - (refs / "index.md").write_text("# Source Map\n", encoding="utf-8") - (refs / "feature-catalog.md").write_text("# Feature Catalog\n", encoding="utf-8") - - registry = _load_registry(monkeypatch, tmp_path) - skill = registry.skills_by_id["demo-skill"] - - assert set(skill.references) == {"index", "feature-catalog"} - assert skill.references["index"].title == "Index" - assert skill.references["feature-catalog"].title == "Feature Catalog" - - -def test_explicit_reference_overrides_discovered(monkeypatch, tmp_path: Path) -> None: - references_block = dedent( - """\ - references: - index: - path: references/index.md - mime_type: text/plain - title: Explicit Title - """ - ).rstrip() - skill_dir = _write_skill( - tmp_path, - skill_id="override-skill", - references_block=references_block, - ) - refs = skill_dir / "references" - refs.mkdir() - (refs / "index.md").write_text("# Source Map\n", encoding="utf-8") - - registry = _load_registry(monkeypatch, tmp_path) - record = registry.skills_by_id["override-skill"].references["index"] - - assert record.mime_type == "text/plain" - assert record.title == "Explicit Title" - - -def test_nested_references_are_not_auto_discovered(monkeypatch, tmp_path: Path) -> None: - references_block = dedent( - """\ - references: - architecture-overview: - path: references/nested/architecture-overview.md - title: Architecture Overview - """ - ).rstrip() - skill_dir = _write_skill( - tmp_path, - skill_id="nested-skill", - references_block=references_block, - ) - refs = skill_dir / "references" - (refs / "nested").mkdir(parents=True) - (refs / "nested" / "architecture-overview.md").write_text( - "# Architecture\n", - encoding="utf-8", - ) - - registry = _load_registry(monkeypatch, tmp_path) - skill = registry.skills_by_id["nested-skill"] - - assert set(skill.references) == {"architecture-overview"} - - -@pytest.mark.parametrize( - ("skill_id", "files", "expected_refs"), - [ - ( - "non-markdown-skill", - {"guide.txt": "plain text", "guide.md": "# Guide\n"}, - {"guide"}, - ), - ( - "normalized-id-skill", - {"implicit_io.md": "# Implicit IO\n"}, - {"implicit-io"}, - ), - ], -) -def test_auto_discovery_file_filtering_and_ref_id_normalization( - monkeypatch, - tmp_path: Path, - skill_id: str, - files: dict[str, str], - expected_refs: set[str], -) -> None: - skill_dir = _write_skill(tmp_path, skill_id=skill_id) - refs = skill_dir / "references" - refs.mkdir() - - for filename, content in files.items(): - (refs / filename).write_text(content, encoding="utf-8") - - registry = _load_registry(monkeypatch, tmp_path) - skill = registry.skills_by_id[skill_id] - - assert set(skill.references) == expected_refs