prompts/src/personal_mcp/catalog/server.py

from __future__ import annotations

from typing import Any

from personal_mcp.skills.document_loader import DocsRegistry, SkillRecord

DEFAULT_LIMIT = 20
MAX_LIMIT = 100


def _pattern_payload(skill: SkillRecord) -> dict[str, Any]:
    return {
        "id": skill.skill_id,
        "name": skill.name,
        "version": skill.version,
        "description": skill.description,
        "tags": list(skill.tags),
        "depends_on": list(skill.depends_on),
        "capabilities": list(skill.capabilities),
        "resources": list(skill.capabilities),
    }


def _summary_payload(skill: SkillRecord) -> dict[str, Any]:
    return {
        "id": skill.skill_id,
        "name": skill.name,
        "description": skill.description,
        "tags": list(skill.tags),
        "capabilities": list(skill.capabilities),
        "version": skill.version,
        "document_uri": skill.document_uri,
        "detail_uri": f"resource://catalog/skills/{skill.skill_id}",
        "resources": {
            "document": skill.document_uri,
            "references": [
                f"resource://skills/{skill.skill_id}/references/{ref_id}"
                for ref_id in sorted(skill.references)
            ],
        },
    }


def _skill_matches(
    skill: SkillRecord,
    *,
    query: str | None,
    tag: str | None,
    capability: str | None,
) -> bool:
    if query:
        lowered = query.strip().lower()
        if lowered:
            haystack = " ".join(
                [
                    skill.skill_id,
                    skill.name,
                    skill.description,
                    " ".join(skill.tags),
                ]
            ).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 skill.tags:
        return False

    if capability and capability not in skill.capabilities:
        return False

    return True


def build_skills_index_payload(
    registry: DocsRegistry,
    *,
    query: str | None = None,
    tag: str | None = None,
    capability: 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.skills_by_id[skill_id] for skill_id in registry.skills_in_load_order]
    matches = [
        skill
        for skill in ordered
        if _skill_matches(skill, query=query, tag=tag, capability=capability)
    ]

    page = matches[start : start + normalized_limit]
    next_cursor = start + normalized_limit

    return {
        "skills": [_summary_payload(skill) for skill 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_skill_detail_payload(registry: DocsRegistry, skill_id: str) -> dict[str, Any]:
    if skill_id not in registry.skills_by_id:
        raise KeyError(skill_id)

    skill = registry.skills_by_id[skill_id]
    return {
        "id": skill.skill_id,
        "name": skill.name,
        "description": skill.description,
        "version": skill.version,
        "tags": list(skill.tags),
        "depends_on": list(skill.depends_on),
        "capabilities": list(skill.capabilities),
        "resources": {
            "document": skill.document_uri,
            "references": {
                ref_id: {
                    "uri": ref.uri,
                    "mime_type": ref.mime_type,
                    "title": ref.title,
                    "path": ref.relpath,
                }
                for ref_id, ref in sorted(skill.references.items())
            },
        },
    }


def search_patterns_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[SkillRecord] = []
    for skill_id in registry.skills_in_load_order:
        skill = registry.skills_by_id[skill_id]
        if not _skill_matches(skill, query=query, tag=None, capability=None):
            continue
        if requested_tags and any(tag not in skill.tags for tag in requested_tags):
            continue
        matches.append(skill)

    page = matches[normalized_skip : normalized_skip + normalized_limit]
    return {
        "patterns": [_pattern_payload(skill) for skill in page],
        "total": len(matches),
        "skip": normalized_skip,
        "limit": normalized_limit,
    }


def get_pattern_by_id_payload(registry: DocsRegistry, skill_id: str) -> dict[str, Any]:
    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])}