started prompt mechanics

src/personal_mcp/catalog/__init__.py

@@ -1,13 +1,21 @@
 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,
 )
 
 __all__ = [
     "build_skill_detail_payload",
+    "build_prompt_detail_payload",
+    "build_prompts_index_payload",
     "build_skills_index_payload",
+    "get_prompt_by_id_payload",
     "get_pattern_by_id_payload",
+    "search_prompts_payload",
     "search_patterns_payload",
 ]

src/personal_mcp/catalog/server.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from typing import Any
 
-from personal_mcp.skills.document_loader import DocsRegistry, SkillRecord
+from personal_mcp.skills.document_loader import DocsRegistry, PromptRecord, SkillRecord
 
 DEFAULT_LIMIT = 20
 MAX_LIMIT = 100
@@ -41,6 +41,19 @@ def _summary_payload(skill: SkillRecord) -> dict[str, Any]:
     }
 
 
+def _prompt_summary_payload(prompt: PromptRecord) -> dict[str, Any]:
+    return {
+        "id": prompt.prompt_id,
+        "name": prompt.name,
+        "description": prompt.description,
+        "tags": list(prompt.tags),
+        "capabilities": list(prompt.capabilities),
+        "version": prompt.version,
+        "document_uri": prompt.document_uri,
+        "detail_uri": f"resource://catalog/prompts/{prompt.prompt_id}",
+    }
+
+
 def _skill_matches(
     skill: SkillRecord,
     *,
@@ -72,6 +85,34 @@ def _skill_matches(
     return True
 
 
+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
+
+    if tag and tag not in prompt.tags:
+        return False
+
+    return True
+
+
 def build_skills_index_payload(
     registry: DocsRegistry,
     *,
@@ -136,6 +177,64 @@ def build_skill_detail_payload(registry: DocsRegistry, skill_id: str) -> dict[st
     }
 
 
+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": [_prompt_summary_payload(prompt) 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_patterns_payload(
     registry: DocsRegistry,
     *,
@@ -171,3 +270,43 @@ def get_pattern_by_id_payload(registry: DocsRegistry, skill_id: str) -> dict[str
     if skill_id not in registry.skills_by_id:
         return {"found": False, "id": skill_id}
     return {"found": True, "pattern": _pattern_payload(registry.skills_by_id[skill_id])}
+
+
+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": [_prompt_summary_payload(prompt) 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/mcp.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import os
+import re
+from inspect import Parameter, Signature
 from typing import Any
 
 from fastmcp import FastMCP
@@ -8,15 +10,20 @@ from fastmcp.server.transforms import ResourcesAsTools
 from fastmcp.server.transforms.search import BM25SearchTransform, 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,
 )
@@ -77,6 +84,70 @@ 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 _python_type(prompt_arg_type: str) -> type[Any]:
+    if prompt_arg_type == "string":
+        return str
+    if prompt_arg_type == "number":
+        return float
+    if prompt_arg_type == "integer":
+        return int
+    if prompt_arg_type == "boolean":
+        return bool
+    if prompt_arg_type == "array":
+        return list
+    if prompt_arg_type == "object":
+        return dict
+    return str
+
+
+def _register_prompt_objects() -> 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()):
+            arg_type = _python_type(arg.type)
+            annotations[arg_name] = arg_type
+            if arg.required:
+                default = Parameter.empty
+            else:
+                default = arg.default
+            params.append(
+                Parameter(
+                    arg_name,
+                    kind=Parameter.KEYWORD_ONLY,
+                    default=default,
+                    annotation=arg_type,
+                )
+            )
+
+        signature = Signature(parameters=params, return_annotation=str)
+
+        prompt_content = prompt.document_content
+
+        def prompt_handler(**kwargs: Any) -> str:
+            return _render_prompt_markdown(prompt_content, kwargs)
+
+        prompt_handler.__name__ = re.sub(r"[^a-zA-Z0-9_]", "_", prompt_id)
+        prompt_handler.__doc__ = prompt.description
+        prompt_handler.__annotations__ = annotations
+        prompt_handler.__signature__ = signature  # type: ignore[attr-defined]
+        mcp.prompt(
+            prompt_handler,
+            name=prompt_id,
+            description=prompt.description,
+            tags=set(prompt.tags),
+        )
+
+
 @mcp.resource(
     "resource://catalog/skills_index",
     mime_type="application/json",
@@ -150,6 +221,57 @@ 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_patterns(
     query: str = "",
@@ -185,6 +307,29 @@ def get_skill_document_by_id(skill_id: str) -> dict[str, Any]:
     }
 
 
+@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)
+
+
 @mcp.tool
 def catalog_search_patterns(
     query: str = "",
@@ -213,4 +358,27 @@ def catalog_get_skill_document_by_id(skill_id: str) -> dict[str, Any]:
     return get_skill_document_by_id(skill_id)
 
 
+@mcp.tool
+def catalog_search_prompts(
+    query: str = "",
+    tags: list[str] | None = None,
+    skip: int = 0,
+    limit: int = 20,
+) -> dict[str, Any]:
+    """Compatibility alias for clients expecting catalog_* tool naming."""
+    return search_prompts(
+        query=query,
+        tags=tags,
+        skip=skip,
+        limit=limit,
+    )
+
+
+@mcp.tool
+def catalog_get_prompt_by_id(prompt_id: str) -> dict[str, Any]:
+    """Compatibility alias for clients expecting catalog_* tool naming."""
+    return get_prompt_by_id(prompt_id)
+
+
 _install_tool_fallback_transforms()
+_register_prompt_objects()

src/personal_mcp/skills/document_loader.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import re
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from importlib.resources import files
 from importlib.resources.abc import Traversable
 from pathlib import PurePosixPath
@@ -111,6 +111,80 @@ class PersonalMcpMetadata(BaseModel):
         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)
 
@@ -146,6 +220,25 @@ class SkillFrontmatter(BaseModel):
         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
@@ -182,6 +275,31 @@ class SkillSummaryRecord:
     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]
@@ -191,6 +309,10 @@ class DocsRegistry:
     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)
 
 
 def _parse_frontmatter(markdown: str, *, path: str) -> tuple[dict[str, Any], str]:
@@ -249,6 +371,20 @@ def _validate_skill_frontmatter(
     return model
 
 
+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")
+    if model.x_personal_mcp.id != model.name:
+        raise ValueError("x-personal-mcp.id must exactly match name")
+    expected_capability = f"resource://prompts/{model.name}/document"
+    if expected_capability not in model.x_personal_mcp.capabilities:
+        raise ValueError(f"capabilities must include {expected_capability}")
+    return model
+
+
 def _normalize_docs_path(path: str) -> str:
     normalized = PurePosixPath(path)
     if normalized.is_absolute() or ".." in normalized.parts:
@@ -371,6 +507,8 @@ def load_docs_registry(
 
     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():
@@ -481,6 +619,154 @@ def load_docs_registry(
             )
         )
 
+    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:
@@ -531,15 +817,37 @@ def load_docs_registry(
                 )
             seen_uris.add(uri)
 
+    for prompt_id, record in sorted(prompts_by_id.items()):
+        for uri in [record.document_uri]:
+            if uri in seen_uris:
+                issues.append(
+                    RegistryIssue(
+                        code="duplicate_uri",
+                        message=f"duplicate resource URI generated: {uri}",
+                        skill_id=prompt_id,
+                        path=record.document_relpath,
+                        hint="ensure unique prompt ids",
+                    )
+                )
+            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:
@@ -547,6 +855,11 @@ def load_docs_registry(
         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,
@@ -560,6 +873,13 @@ def load_docs_registry(
             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())
+        },
     )
 
 
@@ -608,3 +928,16 @@ def read_docs_markdown_path(registry: DocsRegistry, path: str) -> dict[str, str]
         "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,
+    }