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

migration

Browse Source
This commit is contained in:
John Lancaster
2026-08-07 20:07:21 -05:00
parent 2a2700b78c
commit 5b6d5aaec4
70 changed files with 1182 additions and 3633 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/personal_mcp/catalog/__init__.py
-11
View File
@@ -1,11 +0,0 @@
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 get_prompt_by_id_payload
from personal_mcp.catalog.server import search_prompts_payload
__all__ = [
"build_prompt_detail_payload",
"build_prompts_index_payload",
"get_prompt_by_id_payload",
"search_prompts_payload",
]
src/personal_mcp/catalog/server.py
-125
View File
@@ -1,125 +0,0 @@
from __future__ import annotations
from typing import Any
from personal_mcp.registry.models.registry import DocsRegistry
from personal_mcp.registry.models.registry import PromptRecord
from personal_mcp.registry.models.registry import PromptSummaryPayload
DEFAULT_LIMIT = 20
MAX_LIMIT = 100
def _prompt_matches(
prompt: PromptRecord,
*,
query: str | None,
tag: str | None,
) -> bool:
if query:
lowered = query.strip().lower()
if lowered:
haystack = " ".join(
[
prompt.prompt_id,
prompt.name,
prompt.description,
" ".join(prompt.tags),
" ".join(sorted(prompt.arguments)),
]
).lower()
terms = [term for term in lowered.replace("-", " ").split() if term]
if any(term not in haystack for term in terms):
return False
return not (tag and tag not in prompt.tags)
def build_prompts_index_payload(
registry: DocsRegistry,
*,
query: str | None = None,
tag: str | None = None,
cursor: str | None = None,
limit: int | None = None,
) -> dict[str, Any]:
normalized_limit = DEFAULT_LIMIT if limit is None else max(1, min(limit, MAX_LIMIT))
try:
start = 0 if cursor is None else max(0, int(cursor))
except ValueError as exc:
raise ValueError("cursor must be an integer string") from exc
ordered = [registry.prompts_by_id[prompt_id] for prompt_id in registry.prompts_in_load_order]
matches = [prompt for prompt in ordered if _prompt_matches(prompt, query=query, tag=tag)]
page = matches[start : start + normalized_limit]
next_cursor = start + normalized_limit
return {
"prompts": [PromptSummaryPayload.from_record(prompt).model_dump() for prompt in page],
"total": len(matches),
"cursor": str(start),
"limit": normalized_limit,
"next_cursor": str(next_cursor) if next_cursor < len(matches) else None,
}
def build_prompt_detail_payload(registry: DocsRegistry, prompt_id: str) -> dict[str, Any]:
if prompt_id not in registry.prompts_by_id:
raise KeyError(prompt_id)
prompt = registry.prompts_by_id[prompt_id]
return {
"id": prompt.prompt_id,
"name": prompt.name,
"description": prompt.description,
"version": prompt.version,
"tags": list(prompt.tags),
"capabilities": list(prompt.capabilities),
"resources": {
"document": prompt.document_uri,
},
"arguments": {
arg_name: arg.model_dump(exclude_none=True) for arg_name, arg in sorted(prompt.arguments.items())
},
}
def search_prompts_payload(
registry: DocsRegistry,
*,
query: str = "",
tags: list[str] | None = None,
skip: int = 0,
limit: int = DEFAULT_LIMIT,
) -> dict[str, Any]:
normalized_skip = max(skip, 0)
normalized_limit = max(1, min(limit, MAX_LIMIT))
requested_tags = [tag.strip() for tag in (tags or []) if tag and tag.strip()]
matches: list[PromptRecord] = []
for prompt_id in registry.prompts_in_load_order:
prompt = registry.prompts_by_id[prompt_id]
if not _prompt_matches(prompt, query=query, tag=None):
continue
if requested_tags and any(tag not in prompt.tags for tag in requested_tags):
continue
matches.append(prompt)
page = matches[normalized_skip : normalized_skip + normalized_limit]
return {
"prompts": [PromptSummaryPayload.from_record(prompt).model_dump() for prompt in page],
"total": len(matches),
"skip": normalized_skip,
"limit": normalized_limit,
}
def get_prompt_by_id_payload(registry: DocsRegistry, prompt_id: str) -> dict[str, Any]:
if prompt_id not in registry.prompts_by_id:
return {"found": False, "id": prompt_id}
return {
"found": True,
"prompt": build_prompt_detail_payload(registry, prompt_id),
}
src/personal_mcp/config.py
-2
View File
@@ -1,6 +1,5 @@
from functools import cache
from pathlib import Path
from typing import Literal
from pydantic import BaseModel
from pydantic import DirectoryPath
@@ -29,7 +28,6 @@ class Settings(BaseSettings):
debug: bool = False
log_level: str = "info"
mounts: Mounts = Field(default_factory=Mounts)
mcp_transport: Literal["http", "sse"] = "http"
site_dir: DirectoryPath = Field(default=_REPO_ROOT / "site")
src/personal_mcp/mcp.py
+2 -172
View File
@@ -1,66 +1,13 @@
from __future__ import annotations
import os
import re
from inspect import Parameter
from inspect import Signature
from typing import Any
from typing import cast
from fastmcp import FastMCP
from fastmcp.server.transforms import ResourcesAsTools
from fastmcp.server.transforms.search import BM25SearchTransform
from fastmcp.server.transforms.search import RegexSearchTransform
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 get_prompt_by_id_payload
from personal_mcp.catalog.server import search_prompts_payload
from personal_mcp.prompts import create_prompts_provider
from personal_mcp.registry.load import get_docs_registry
from personal_mcp.registry.models.registry import DocsRegistry
from personal_mcp.registry.read import read_docs_markdown_path
from personal_mcp.registry.read import read_prompt_document
from personal_mcp.skills import create_skills_provider
TOOL_SEARCH_MODE = os.getenv("PERSONAL_MCP_TOOL_SEARCH", "none").strip().lower()
TOOL_SEARCH_MAX_RESULTS = os.getenv("PERSONAL_MCP_TOOL_SEARCH_MAX_RESULTS", "5")
def _parse_positive_int(value: str, *, env_name: str) -> int:
try:
parsed = int(value)
except ValueError as exc:
raise ValueError(f"{env_name} must be an integer") from exc
if parsed <= 0:
raise ValueError(f"{env_name} must be greater than zero")
return parsed
def _install_tool_fallback_transforms(mcp: FastMCP) -> None:
# Expose list_resources/read_resource for tool-only clients.
mcp.add_transform(ResourcesAsTools(mcp))
if TOOL_SEARCH_MODE in {"", "none"}:
return
max_results = _parse_positive_int(
TOOL_SEARCH_MAX_RESULTS,
env_name="PERSONAL_MCP_TOOL_SEARCH_MAX_RESULTS",
)
kwargs: dict[str, Any] = {
"max_results": max_results,
"always_visible": ["list_resources", "read_resource"],
}
if TOOL_SEARCH_MODE == "regex":
mcp.add_transform(RegexSearchTransform(**kwargs))
return
if TOOL_SEARCH_MODE == "bm25":
mcp.add_transform(BM25SearchTransform(**kwargs))
return
raise ValueError("PERSONAL_MCP_TOOL_SEARCH must be one of: none, regex, bm25")
def _ro_annotations() -> dict[str, bool]:
return {
@@ -69,54 +16,6 @@ def _ro_annotations() -> dict[str, bool]:
}
def _render_prompt_markdown(content: str, arguments: dict[str, Any]) -> str:
rendered = content
for key, value in arguments.items():
rendered = rendered.replace(f"{{{{{key}}}}}", str(value))
return rendered
def _make_prompt_handler(content: str):
def prompt_handler(**kwargs: Any) -> str:
return _render_prompt_markdown(content, kwargs)
return prompt_handler
def _register_prompt_objects(mcp: FastMCP, registry: DocsRegistry) -> None:
for prompt_id in registry.prompts_in_load_order:
prompt = registry.prompts_by_id[prompt_id]
annotations: dict[str, Any] = {}
params: list[Parameter] = []
for arg_name, arg in sorted(prompt.arguments.items()):
annotations[arg_name] = str
default = Parameter.empty if arg.required else None
params.append(
Parameter(
arg_name,
kind=Parameter.KEYWORD_ONLY,
default=default,
annotation=str,
)
)
signature = Signature(parameters=params, return_annotation=str)
prompt_handler = _make_prompt_handler(prompt.document_content)
prompt_handler.__name__ = re.sub(r"[^a-zA-Z0-9_]", "_", prompt_id)
prompt_handler.__doc__ = prompt.description
prompt_handler.__annotations__ = annotations
cast(Any, prompt_handler).__signature__ = signature
mcp.prompt(
prompt_handler,
name=prompt_id,
description=prompt.description,
tags=set(prompt.tags),
)
def _register_components(mcp: FastMCP, registry: DocsRegistry) -> None:
@mcp.resource(
"resource://docs/{path*}",
@@ -127,80 +26,11 @@ def _register_components(mcp: FastMCP, registry: DocsRegistry) -> None:
def docs_markdown(path: str) -> dict[str, str]:
return read_docs_markdown_path(registry, path)
@mcp.resource(
"resource://catalog/prompts_index",
mime_type="application/json",
tags={"catalog"},
annotations=_ro_annotations(),
)
def prompts_index() -> dict[str, Any]:
return build_prompts_index_payload(registry)
@mcp.resource(
"resource://catalog/prompts_index{?q,tag,cursor,limit}",
mime_type="application/json",
tags={"catalog"},
annotations=_ro_annotations(),
)
def prompts_index_query(
q: str | None = None,
tag: str | None = None,
cursor: str | None = None,
limit: int | None = None,
) -> dict[str, Any]:
return build_prompts_index_payload(
registry,
query=q,
tag=tag,
cursor=cursor,
limit=limit,
)
@mcp.resource(
"resource://catalog/prompts/{prompt_id}",
mime_type="application/json",
tags={"catalog"},
annotations=_ro_annotations(),
)
def prompt_detail(prompt_id: str) -> dict[str, Any]:
return build_prompt_detail_payload(registry, prompt_id)
@mcp.resource(
"resource://prompts/{prompt_id}/document",
mime_type="text/markdown",
tags={"prompt-doc"},
annotations=_ro_annotations(),
)
def prompt_document(prompt_id: str) -> dict[str, str]:
return read_prompt_document(registry, prompt_id)
@mcp.tool
def search_prompts(
query: str = "",
tags: list[str] | None = None,
skip: int = 0,
limit: int = 20,
) -> dict[str, Any]:
"""Search prompt metadata with optional tags and pagination."""
return search_prompts_payload(
registry,
query=query,
tags=tags,
skip=skip,
limit=limit,
)
@mcp.tool
def get_prompt_by_id(prompt_id: str) -> dict[str, Any]:
"""Return one prompt by stable id."""
return get_prompt_by_id_payload(registry, prompt_id)
def create_mcp() -> FastMCP:
registry = get_docs_registry()
mcp = FastMCP("personal-mcp", on_duplicate="error")
_register_components(mcp, registry)
_register_prompt_objects(mcp, registry)
mcp.add_provider(create_prompts_provider())
mcp.add_provider(create_skills_provider())
_install_tool_fallback_transforms(mcp)
return mcp
src/personal_mcp/prompts/__init__.py
+3
View File
@@ -0,0 +1,3 @@
from .provider import create_prompts_provider
__all__ = ["create_prompts_provider"]
src/personal_mcp/prompts/components/__init__.py
View File
src/personal_mcp/prompts/components/authoring.py
+45
View File
@@ -0,0 +1,45 @@
from typing import Annotated
from typing import Literal
from fastmcp.prompts import prompt
from pydantic import Field
from personal_mcp.prompts.content import render_prompt
@prompt(
name="authoring",
version="1.0.0",
description=(
"Provide a practical checklist and baseline template for authoring docs-first MCP modules and "
"repository-specific Copilot instruction shims."
),
tags={"authoring", "mcp", "fastmcp", "copilot", "prompts", "scaffolding"},
)
def authoring(
artifact_type: Annotated[
Literal["skill", "prompt", "shim"],
Field(description="Artifact type to create."),
],
artifact_id: Annotated[
str,
Field(description="Lowercase kebab-case id for the module or shim."),
],
goal: Annotated[
str,
Field(description="One-sentence capability statement."),
],
scope_glob: Annotated[
str | None,
Field(description="Optional applyTo glob for shim outputs."),
] = None,
) -> str:
return render_prompt(
"authoring",
{
"artifact_type": artifact_type,
"artifact_id": artifact_id,
"goal": goal,
"scope_glob": scope_glob,
},
)
src/personal_mcp/prompts/components/greenfield_architecture.py
+45
View File
@@ -0,0 +1,45 @@
from typing import Annotated
from typing import Literal
from fastmcp.prompts import prompt
from pydantic import Field
from personal_mcp.prompts.content import render_prompt
@prompt(
name="greenfield-architecture",
version="1.0.0",
description=(
"Research established patterns and design a high-level architecture for a "
"new app or library with explicit tradeoffs and test strategy."
),
tags={"architecture", "planning", "greenfield", "design", "testing", "prompts"},
)
def greenfield_architecture(
scope_type: Annotated[
Literal["app", "library"],
Field(description="Scope type to design."),
],
intent_document: Annotated[
str | None,
Field(description="Optional full document describing goals and context."),
] = None,
problem_domain: Annotated[
str | None,
Field(description="Problem domain and business goal."),
] = None,
constraints: Annotated[
str | None,
Field(description="Runtime, deployment, and non-functional constraints."),
] = None,
) -> str:
return render_prompt(
"greenfield-architecture",
{
"scope_type": scope_type,
"intent_document": intent_document,
"problem_domain": problem_domain,
"constraints": constraints,
},
)
src/personal_mcp/prompts/components/jsfiddle_page_layout.py
+31
View File
@@ -0,0 +1,31 @@
from typing import Annotated
from fastmcp.prompts import prompt
from pydantic import Field
from personal_mcp.prompts.content import render_prompt
@prompt(
name="jsfiddle-page-layout",
version="1.1.0",
description=(
"Create a responsive sample page layout for a user-supplied domain and return paste-ready HTML and CSS "
"for JSFiddle."
),
tags={"frontend", "html", "css", "jsfiddle", "layout", "prototyping", "prompts"},
)
def jsfiddle_page_layout(
domain: Annotated[
str,
Field(description="Product, service, organization, or subject represented by the page."),
],
layout_brief: Annotated[
str | None,
Field(description="Optional page type, sections, priorities, or visual constraints."),
] = None,
) -> str:
return render_prompt(
"jsfiddle-page-layout",
{"domain": domain, "layout_brief": layout_brief},
)
src/personal_mcp/prompts/components/mcp_consumer_repo_shim.py
+44
View File
@@ -0,0 +1,44 @@
from typing import Annotated
from fastmcp.prompts import prompt
from pydantic import Field
from personal_mcp.prompts.content import render_prompt
@prompt(
name="mcp-consumer-repo-shim",
version="1.0.0",
description=(
"Create one repository-specific thin shim instruction file that binds a file scope to a user-selected "
"Personal MCP skill resource."
),
tags={"copilot", "mcp", "instructions", "shims", "prompts"},
)
def mcp_consumer_repo_shim(
apply_to_glob: Annotated[
str,
Field(description="File glob scope for the shim applyTo field."),
],
primary_skill_resource: Annotated[
str,
Field(description="Primary native skill:// resource URI."),
],
shim_title: Annotated[
str | None,
Field(description="Optional human-readable instruction shim name."),
] = None,
companion_docs_page: Annotated[
str | None,
Field(description="Optional relative companion documentation link."),
] = None,
) -> str:
return render_prompt(
"mcp-consumer-repo-shim",
{
"apply_to_glob": apply_to_glob,
"primary_skill_resource": primary_skill_resource,
"shim_title": shim_title,
"companion_docs_page": companion_docs_page,
},
)
src/personal_mcp/prompts/components/nicegui_component_extraction.py
+44
View File
@@ -0,0 +1,44 @@
from typing import Annotated
from fastmcp.prompts import prompt
from pydantic import Field
from personal_mcp.prompts.content import render_prompt
@prompt(
name="nicegui-component-extraction",
version="1.0.0",
description=(
"Extract a user-selected component from a JSFiddle page layout and implement it as a reusable NiceGUI "
"render function."
),
tags={"nicegui", "components", "frontend", "refactoring", "jsfiddle", "prompts"},
)
def nicegui_component_extraction(
component: Annotated[
str,
Field(description="Visible label, semantic role, or selector identifying the component."),
],
source_layout: Annotated[
str | None,
Field(description="Optional source HTML and CSS."),
] = None,
target_location: Annotated[
str | None,
Field(description="Optional target NiceGUI page, module, or package."),
] = None,
behavior_requirements: Annotated[
str | None,
Field(description="Optional interactions, state, callbacks, or variations."),
] = None,
) -> str:
return render_prompt(
"nicegui-component-extraction",
{
"component": component,
"source_layout": source_layout,
"target_location": target_location,
"behavior_requirements": behavior_requirements,
},
)
src/personal_mcp/prompts/components/pytest_fill_scaffold.py
+45
View File
@@ -0,0 +1,45 @@
from typing import Annotated
from typing import Literal
from fastmcp.prompts import prompt
from pydantic import Field
from personal_mcp.prompts.content import render_prompt
@prompt(
name="pytest-fill-scaffold",
version="1.0.0",
description=(
"Fill scaffolded pytest methods with assertions, fixtures, and minimal test data while preserving reviewed "
"structure."
),
tags={"pytest", "testing", "scaffolding", "prompts"},
)
def pytest_fill_scaffold(
target_files: Annotated[
str,
Field(description="Target test file paths under tests/."),
],
stack: Annotated[
Literal["pure-python", "fastapi", "sqlalchemy-sync", "sqlalchemy-async", "mixed"],
Field(description="Runtime stack type for fixture and marker choices."),
],
strategy: Annotated[
str | None,
Field(description="Optional minimal or comprehensive implementation preference."),
] = None,
marker_lane: Annotated[
str | None,
Field(description="Optional pytest marker lane."),
] = None,
) -> str:
return render_prompt(
"pytest-fill-scaffold",
{
"target_files": target_files,
"stack": stack,
"strategy": strategy,
"marker_lane": marker_lane,
},
)
src/personal_mcp/prompts/components/pytest_scaffold.py
+42
View File
@@ -0,0 +1,42 @@
from typing import Annotated
from typing import Literal
from fastmcp.prompts import prompt
from pydantic import Field
from personal_mcp.prompts.content import render_prompt
@prompt(
name="pytest-scaffold",
version="1.0.0",
description="Plan and optionally scaffold pytest file and class structure for selected Python modules.",
tags={"pytest", "testing", "scaffolding", "prompts"},
)
def pytest_scaffold(
target_modules: Annotated[
str,
Field(description="Target module paths under src/."),
],
mode: Annotated[
Literal["plan-only", "scaffold"],
Field(description="Whether to plan only or create scaffold files."),
],
path_strategy: Annotated[
str | None,
Field(description="Optional src-to-tests path mapping preference."),
] = None,
naming_style: Annotated[
str | None,
Field(description="Optional concise test naming preference."),
] = None,
) -> str:
return render_prompt(
"pytest-scaffold",
{
"target_modules": target_modules,
"mode": mode,
"path_strategy": path_strategy,
"naming_style": naming_style,
},
)
src/personal_mcp/prompts/content.py
+30
View File
@@ -0,0 +1,30 @@
import re
from importlib.resources import files
from typing import Any
_PROMPT_ID_RE = re.compile(r"^[a-z][a-z0-9-]*$")
_FRONTMATTER_RE = re.compile(r"\A---\r?\n.*?\r?\n---\r?\n?", re.DOTALL)
_PLACEHOLDER_RE = re.compile(r"\{\{([A-Za-z_][A-Za-z0-9_]*)\}\}")
def render_prompt(prompt_id: str, arguments: dict[str, Any]) -> str:
if not _PROMPT_ID_RE.fullmatch(prompt_id):
raise ValueError("prompt_id must be lowercase kebab-case")
resource = files("personal_mcp").joinpath("docs", "prompts", prompt_id, "PROMPT.md")
if not resource.is_file():
raise FileNotFoundError(f"prompt document does not exist: {prompt_id}")
content = _FRONTMATTER_RE.sub("", resource.read_text(encoding="utf-8"), count=1)
placeholders = set(_PLACEHOLDER_RE.findall(content))
argument_names = set(arguments)
if placeholders != argument_names:
missing = sorted(argument_names - placeholders)
unknown = sorted(placeholders - argument_names)
raise ValueError(f"prompt placeholders do not match arguments; missing={missing}, unknown={unknown}")
rendered = content
for name, value in arguments.items():
replacement = "Not provided" if value is None else str(value)
rendered = rendered.replace(f"{{{{{name}}}}}", replacement)
return rendered
src/personal_mcp/prompts/provider.py
+10
View File
@@ -0,0 +1,10 @@
from pathlib import Path
from fastmcp.server.providers import FileSystemProvider
def create_prompts_provider() -> FileSystemProvider:
components_root = Path(__file__).parent / "components"
if not components_root.is_dir():
raise FileNotFoundError(f"prompt components root does not exist or is not a directory: {components_root}")
return FileSystemProvider(root=components_root, reload=False)
src/personal_mcp/registry/contracts.py
+1 -7
View File
@@ -1,9 +1,3 @@
from .models.registry import DocsRegistry
from .models.registry import PromptRecord
from .models.registry import PromptSummaryRecord
__all__ = [
"DocsRegistry",
"PromptRecord",
"PromptSummaryRecord",
]
__all__ = ["DocsRegistry"]
src/personal_mcp/registry/ingest/document.py
+2 -30
View File
@@ -1,7 +1,5 @@
from collections.abc import Generator
from collections.abc import Iterator
from dataclasses import dataclass
from dataclasses import field
from importlib.resources.abc import Traversable
from itertools import starmap
from pathlib import PurePosixPath
@@ -17,10 +15,8 @@ class MarkdownDocument:
relpath: DocsPath
"""The relative path of the document within the package resources."""
content: str = field(repr=False)
content: str
"""The raw markdown content of the document."""
frontmatter: str | None = field(repr=False, default=None)
"""The raw YAML frontmatter of the document, if present."""
def __post_init__(self) -> None:
object.__setattr__(self, "relpath", parse_docs_path(self.relpath))
@@ -34,15 +30,7 @@ class MarkdownDocument:
@classmethod
def from_resource(cls, relpath: DocsPath, resource: Traversable) -> Self:
"""Load a markdown document from a package resource."""
raw = resource.read_text(encoding="utf-8")
frontmatter = get_raw_frontmatter(raw)
return cls(relpath=relpath, content=raw, frontmatter=frontmatter)
@property
def prompt_slug(self) -> str | None:
parts = self.relpath.parts
if parts[0] == "prompts" and len(parts) >= 3:
return parts[1]
return cls(relpath=relpath, content=resource.read_text(encoding="utf-8"))
def walk_resources(
@@ -59,19 +47,3 @@ def walk_resources(
yield from walk_resources(child, suffix=suffix, prefix=relpath)
elif child.is_file() and child.name.lower().endswith(suffix):
yield relpath, child
def get_raw_frontmatter(raw: str) -> str | None:
delimiter = iter(get_frontmatter_delim_idx(raw, delimiter="---"))
try:
start = next(delimiter) + 1
end = next(delimiter)
except StopIteration:
return None
return "\n".join(raw.splitlines()[start:end])
def get_frontmatter_delim_idx(raw: str, *, delimiter: str = "---") -> Generator[int]:
for i, line in enumerate(raw.splitlines()):
if line.strip().startswith(delimiter):
yield i
src/personal_mcp/registry/ingest/prompt.py
-48
View File
@@ -1,48 +0,0 @@
from collections.abc import Iterable
from dataclasses import dataclass
from importlib.resources.abc import Traversable
from itertools import starmap
from typing import Self
from .document import MarkdownDocument
@dataclass(frozen=True, slots=True)
class PromptFilesBundle:
"""Represents a prompt and all of its associated markdown files."""
slug: str
prompt: MarkdownDocument
other: tuple[MarkdownDocument, ...]
@classmethod
def from_root(cls, root: Traversable) -> list[Self]:
# Should only be used for testing
return list(cls.from_docs(MarkdownDocument.from_root(root).values()))
@classmethod
def from_docs(cls, docs: Iterable[MarkdownDocument]) -> tuple[Self, ...]:
return tuple(starmap(cls.from_paths, group_prompt_paths(docs).items()))
@classmethod
def from_paths(cls, slug: str, paths: set[MarkdownDocument]) -> Self:
prompt = next(iter(p for p in paths if p.relpath.name == "PROMPT.md"))
sorted_paths = tuple(sorted(paths, key=lambda p: p.relpath))
other = tuple(p for p in sorted_paths if p != prompt)
return cls(
slug=slug,
prompt=prompt,
other=other,
)
def group_prompt_paths(docs: Iterable[MarkdownDocument]) -> dict[str, set[MarkdownDocument]]:
"""Group prompts from a list of markdown documents by their prompt slug."""
grouped: dict[str, set[MarkdownDocument]] = {}
for doc in sorted(
filter(lambda d: d.prompt_slug is not None, docs),
key=lambda d: d.relpath,
):
if doc.prompt_slug:
grouped.setdefault(doc.prompt_slug, set()).add(doc)
return grouped
src/personal_mcp/registry/load.py
+1 -54
View File
@@ -1,44 +1,10 @@
from __future__ import annotations
from collections import defaultdict
from functools import cache
from importlib.resources import files
from personal_mcp.registry.ingest.document import MarkdownDocument
from personal_mcp.registry.ingest.prompt import PromptFilesBundle
from personal_mcp.registry.models.prompt import StoredPrompt
from personal_mcp.registry.models.registry import DocsRegistry
from personal_mcp.registry.models.registry import PromptRecord
from personal_mcp.registry.models.registry import PromptSummaryRecord
def _build_prompt_record(*, bundle: PromptFilesBundle) -> PromptRecord:
stored = StoredPrompt.from_bundle(bundle)
metadata = stored.frontmatter.x_personal_mcp
return PromptRecord(
prompt_id=metadata.id,
name=stored.frontmatter.name,
description=stored.frontmatter.description,
version=metadata.version,
tags=tuple(metadata.tags),
capabilities=tuple(metadata.capabilities),
arguments=dict(metadata.arguments),
document_uri=f"resource://prompts/{metadata.id}/document",
document_relpath=stored.relpath,
document_content=stored.content,
)
def _build_tag_index_prompts(
prompts_in_order: tuple[str, ...],
prompts_by_id: dict[str, PromptRecord],
) -> dict[str, tuple[str, ...]]:
tag_index: defaultdict[str, list[str]] = defaultdict(list)
for prompt_id in prompts_in_order:
for tag in prompts_by_id[prompt_id].tags:
tag_index[tag].append(prompt_id)
return {tag: tuple(ids) for tag, ids in sorted(tag_index.items())}
@cache
@@ -48,28 +14,9 @@ def get_docs_registry() -> DocsRegistry:
raise FileNotFoundError(f"docs root does not exist or is not a directory: {root}")
docs = MarkdownDocument.from_root(root)
docs_markdown_by_path = {relpath: doc.content for relpath, doc in docs.items()}
prompt_bundles = PromptFilesBundle.from_docs(docs.values())
prompts_by_id: dict[str, PromptRecord] = {}
prompts_in_load_order: list[str] = []
for bundle in prompt_bundles:
record = _build_prompt_record(bundle=bundle)
if record.prompt_id in prompts_by_id:
raise ValueError(f"duplicate prompt_id detected: {record.prompt_id}")
prompts_by_id[record.prompt_id] = record
prompts_in_load_order.append(record.prompt_id)
prompts_in_order_tuple = tuple(prompts_in_load_order)
docs_markdown_by_path = {relpath: doc.content for relpath, doc in docs.items() if relpath.parts[0] != "skills"}
return DocsRegistry(
docs_markdown_by_path=docs_markdown_by_path,
docs_markdown_path_index=tuple(sorted(docs_markdown_by_path)),
prompts_by_id=prompts_by_id,
prompts_in_load_order=prompts_in_order_tuple,
prompts_summary_in_load_order=tuple(
PromptSummaryRecord.from_record(prompts_by_id[prompt_id]) for prompt_id in prompts_in_order_tuple
),
tag_to_prompt_ids=_build_tag_index_prompts(prompts_in_order_tuple, prompts_by_id),
)
src/personal_mcp/registry/models/common.py
-7
View File
@@ -1,18 +1,13 @@
import re
from collections.abc import Mapping
from pathlib import PurePosixPath
from types import MappingProxyType
from typing import Annotated
from typing import ClassVar
from typing import Final
from pydantic import BaseModel
from pydantic import BeforeValidator
from pydantic import ConfigDict
SKILL_ID_RE: Final[re.Pattern[str]] = re.compile(r"^[a-z][a-z0-9-]*$")
SEMVER_RE: Final[re.Pattern[str]] = re.compile(r"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)(?:[-+][0-9A-Za-z.-]+)?$")
class StrictFrozenModel(BaseModel):
"""Immutable base model with strict field validation rules."""
@@ -20,8 +15,6 @@ class StrictFrozenModel(BaseModel):
model_config: ClassVar[ConfigDict] = ConfigDict(
extra="forbid",
frozen=True,
validate_by_alias=True,
validate_by_name=True,
str_strip_whitespace=True,
)
src/personal_mcp/registry/models/prompt.py
-147
View File
@@ -1,147 +0,0 @@
import re
from collections.abc import Mapping
from typing import TYPE_CHECKING
import yaml
from pydantic import Field
from pydantic import field_validator
from pydantic import model_validator
from .common import SEMVER_RE
from .common import SKILL_ID_RE
from .common import DocsPath
from .common import StrictFrozenModel
from .common import frozen_mapping
if TYPE_CHECKING:
from personal_mcp.registry.ingest.prompt import PromptFilesBundle
class PromptArgumentEntry(StrictFrozenModel):
"""Schema for a single prompt argument definition."""
title: str | None = None
description: str | None = None
required: bool = False
class PromptMetadata(StrictFrozenModel):
"""Canonical metadata describing a prompt contract and arguments."""
id: str
version: str
tags: tuple[str, ...] = ()
capabilities: tuple[str, ...] = Field(min_length=1)
arguments: Mapping[str, PromptArgumentEntry] = Field(default_factory=frozen_mapping)
@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: tuple[str, ...]) -> tuple[str, ...]:
for tag in value:
if not SKILL_ID_RE.fullmatch(tag):
raise ValueError(f"invalid tag: {tag}")
return value
@field_validator("arguments", mode="before")
@classmethod
def freeze_arguments(cls, value: Mapping[str, PromptArgumentEntry] | None) -> Mapping[str, PromptArgumentEntry]:
return frozen_mapping(value)
@field_validator("arguments")
@classmethod
def validate_argument_names(cls, value: Mapping[str, PromptArgumentEntry]) -> Mapping[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(StrictFrozenModel):
"""Parsed PROMPT frontmatter including personal-mcp metadata."""
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
@classmethod
def from_raw_yaml(cls, raw: str | None) -> "PromptFrontmatter":
if raw is None:
raise ValueError("missing YAML frontmatter")
try:
data = yaml.safe_load(raw)
except yaml.YAMLError as e:
raise ValueError(f"invalid YAML in frontmatter: {e}") from e
if not isinstance(data, dict):
raise TypeError("frontmatter must parse to an object")
return cls.model_validate(data)
class StoredPrompt(StrictFrozenModel):
"""Normalized prompt document content with path and frontmatter for storage in the registry."""
prompt_id: str
relpath: DocsPath
content: str
frontmatter: PromptFrontmatter
@field_validator("frontmatter", mode="before")
@classmethod
def parse_frontmatter_yaml(cls, value: PromptFrontmatter | str | None) -> PromptFrontmatter:
if isinstance(value, PromptFrontmatter):
return value
return PromptFrontmatter.from_raw_yaml(value)
@model_validator(mode="after")
def validate_contract(self) -> "StoredPrompt":
parts = self.relpath.parts
if len(parts) < 3 or parts[0] != "prompts":
raise ValueError("prompt relpath must be under prompts/<slug>/")
prompt_dir_name = parts[1]
if self.frontmatter.name != prompt_dir_name:
raise ValueError("frontmatter name must exactly match prompt directory name")
if self.frontmatter.x_personal_mcp.id != self.frontmatter.name:
raise ValueError("x-personal-mcp.id must exactly match name")
expected_capability = f"resource://prompts/{self.frontmatter.name}/document"
if expected_capability not in self.frontmatter.x_personal_mcp.capabilities:
raise ValueError(f"capabilities must include {expected_capability}")
if self.prompt_id != self.frontmatter.x_personal_mcp.id:
raise ValueError("prompt_id must exactly match x-personal-mcp.id")
return self
@classmethod
def from_bundle(cls, bundle: "PromptFilesBundle") -> "StoredPrompt":
frontmatter = PromptFrontmatter.from_raw_yaml(bundle.prompt.frontmatter)
return cls.model_validate(
{
"prompt_id": frontmatter.x_personal_mcp.id,
"relpath": bundle.prompt.relpath,
"content": bundle.prompt.content,
"frontmatter": frontmatter,
}
)
src/personal_mcp/registry/models/registry.py
+2 -86
View File
@@ -6,105 +6,21 @@ from pydantic import field_validator
from .common import DocsPath
from .common import StrictFrozenModel
from .common import frozen_mapping
from .prompt import PromptArgumentEntry
def _empty_docs_mapping() -> Mapping[DocsPath, str]:
return frozen_mapping()
class PromptRecord(StrictFrozenModel):
"""Registry record containing a fully resolved prompt document."""
prompt_id: str
name: str
description: str
version: str
tags: tuple[str, ...]
capabilities: tuple[str, ...]
arguments: Mapping[str, PromptArgumentEntry] = Field(default_factory=frozen_mapping)
document_uri: str
document_relpath: DocsPath
document_content: str
@field_validator("arguments", mode="before")
@classmethod
def freeze_arguments(cls, value: Mapping[str, PromptArgumentEntry] | None) -> Mapping[str, PromptArgumentEntry]:
return frozen_mapping(value)
class PromptSummaryRecord(StrictFrozenModel):
"""Compact prompt summary exposed by catalog listing APIs."""
prompt_id: str
name: str
description: str
tags: tuple[str, ...]
capabilities: tuple[str, ...]
document_uri: str
version: str
@classmethod
def from_record(cls, record: PromptRecord) -> "PromptSummaryRecord":
return cls(
prompt_id=record.prompt_id,
name=record.name,
description=record.description,
tags=record.tags,
capabilities=record.capabilities,
document_uri=record.document_uri,
version=record.version,
)
class PromptSummaryPayload(StrictFrozenModel):
"""Catalog payload model for prompt index summaries."""
id: str
name: str
description: str
tags: list[str]
capabilities: list[str]
version: str
document_uri: str
detail_uri: str
@classmethod
def from_record(cls, record: PromptRecord) -> "PromptSummaryPayload":
return cls(
id=record.prompt_id,
name=record.name,
description=record.description,
tags=list(record.tags),
capabilities=list(record.capabilities),
version=record.version,
document_uri=record.document_uri,
detail_uri=f"resource://catalog/prompts/{record.prompt_id}",
)
class DocsRegistry(StrictFrozenModel):
"""In-memory index of loaded prompts and documentation content."""
"""In-memory index of documentation content."""
docs_markdown_by_path: Mapping[DocsPath, str] = Field(default_factory=_empty_docs_mapping)
"""Maps each documentation path to its loaded Markdown content."""
docs_markdown_path_index: tuple[DocsPath, ...]
"""Lists documentation paths in deterministic index order."""
prompts_by_id: Mapping[str, PromptRecord] = Field(default_factory=frozen_mapping)
"""Maps each prompt identifier to its fully resolved registry record."""
prompts_in_load_order: tuple[str, ...] = ()
"""Preserves prompt identifiers in deterministic source loading order."""
prompts_summary_in_load_order: tuple[PromptSummaryRecord, ...] = ()
"""Stores compact prompt summaries in the same deterministic loading order."""
tag_to_prompt_ids: Mapping[str, tuple[str, ...]] = Field(default_factory=frozen_mapping)
"""Indexes prompt identifiers by tag for catalog filtering and search."""
@field_validator(
"docs_markdown_by_path",
"prompts_by_id",
"tag_to_prompt_ids",
mode="before",
)
@field_validator("docs_markdown_by_path", mode="before")
@classmethod
def freeze_mappings(cls, value: Mapping[str, object] | None) -> Mapping[str, object]:
return frozen_mapping(value)
src/personal_mcp/registry/read.py
+2 -13
View File
@@ -4,6 +4,8 @@ from .models.registry import DocsRegistry
def read_docs_markdown_path(registry: DocsRegistry, path: str) -> dict[str, str]:
docs_path = parse_docs_path(path)
if docs_path.parts[0] == "skills":
raise KeyError(f"unknown docs path: {docs_path.as_posix()}")
if docs_path not in registry.docs_markdown_by_path:
raise KeyError(f"unknown docs path: {docs_path.as_posix()}")
return {
@@ -12,16 +14,3 @@ def read_docs_markdown_path(registry: DocsRegistry, path: str) -> dict[str, str]
"source_path": f"docs/{docs_path.as_posix()}",
"content": registry.docs_markdown_by_path[docs_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.as_posix()}",
"content": prompt.document_content,
}
src/personal_mcp/web/app.py
+1 -1
View File
@@ -13,7 +13,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
path=runtime_settings.mounts.mcp,
json_response=True,
stateless_http=True,
transport=runtime_settings.mcp_transport,
transport="http",
)
app = FastAPI(
debug=runtime_settings.debug,