started manual refactor
This commit is contained in:
John Lancaster
@@ -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)
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -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-id>/SKILL.md",
|
||||
)
|
||||
)
|
||||
continue
|
||||
|
||||
skill_markdown = skill_doc_file.read_text(encoding="utf-8")
|
||||
try:
|
||||
raw_frontmatter, _ = _parse_frontmatter(
|
||||
skill_markdown,
|
||||
path=skill_doc_relpath,
|
||||
)
|
||||
frontmatter = _validate_skill_frontmatter(
|
||||
raw_frontmatter,
|
||||
skill_dir_name=skill_dir_name,
|
||||
)
|
||||
except (ValueError, ValidationError) as exc:
|
||||
issues.append(
|
||||
RegistryIssue(
|
||||
code="invalid_frontmatter",
|
||||
message=str(exc),
|
||||
skill_id=skill_dir_name,
|
||||
path=skill_doc_relpath,
|
||||
hint="fix SKILL.md YAML frontmatter to match the contract",
|
||||
)
|
||||
)
|
||||
continue
|
||||
|
||||
effective_reference_entries = _discover_top_level_references(skill_dir=skill_dir)
|
||||
effective_reference_entries.update(frontmatter.x_personal_mcp.references)
|
||||
|
||||
references: dict[str, ReferenceRecord] = {}
|
||||
for ref_id, ref_entry in effective_reference_entries.items():
|
||||
ref_relpath = skill_rel_root.joinpath(ref_entry.path).as_posix()
|
||||
if ref_relpath not in docs_markdown_by_path:
|
||||
issues.append(
|
||||
RegistryIssue(
|
||||
code="missing_reference",
|
||||
message=f"reference target is missing for ref_id '{ref_id}'",
|
||||
skill_id=frontmatter.name,
|
||||
path=ref_relpath,
|
||||
hint="fix x-personal-mcp.references path or add the referenced markdown file",
|
||||
)
|
||||
)
|
||||
continue
|
||||
|
||||
references[ref_id] = ReferenceRecord(
|
||||
ref_id=ref_id,
|
||||
uri=f"resource://skills/{frontmatter.name}/references/{ref_id}",
|
||||
relpath=ref_relpath,
|
||||
mime_type=ref_entry.mime_type,
|
||||
title=ref_entry.title,
|
||||
content=docs_markdown_by_path[ref_relpath],
|
||||
)
|
||||
|
||||
skill_id = frontmatter.name
|
||||
if skill_id in skills_by_id:
|
||||
issues.append(
|
||||
RegistryIssue(
|
||||
code="duplicate_skill_id",
|
||||
message="duplicate skill id discovered",
|
||||
skill_id=skill_id,
|
||||
path=skill_doc_relpath,
|
||||
hint="ensure each skill directory has a unique id",
|
||||
)
|
||||
)
|
||||
continue
|
||||
|
||||
record = SkillRecord(
|
||||
skill_id=skill_id,
|
||||
name=frontmatter.name,
|
||||
description=frontmatter.description,
|
||||
version=frontmatter.x_personal_mcp.version,
|
||||
tags=tuple(frontmatter.x_personal_mcp.tags),
|
||||
capabilities=tuple(frontmatter.x_personal_mcp.capabilities),
|
||||
depends_on=tuple(frontmatter.x_personal_mcp.depends_on),
|
||||
document_uri=f"resource://skills/{skill_id}/document",
|
||||
document_relpath=skill_doc_relpath,
|
||||
document_content=skill_markdown,
|
||||
references=references,
|
||||
)
|
||||
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())},
|
||||
)
|
||||
@@ -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()
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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,
|
||||
}
|
||||
@@ -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))
|
||||
Reference in New Issue
Block a user