search_skills

src/personal_mcp/mcp.py

@@ -1,10 +1,15 @@
 from __future__ import annotations
 
+import re
+from typing import Annotated
+
 from fastmcp import FastMCP
 from mcp.types import CompletionArgument
 from mcp.types import CompletionContext
 from mcp.types import Icon
 from mcp.types import PromptReference
+from pydantic import BaseModel
+from pydantic import Field
 
 from personal_mcp.prompts import create_prompts_provider
 from personal_mcp.prompts.models import MarkdownPrompt
@@ -16,9 +21,9 @@ from personal_mcp.skills import create_skills_provider
 
 _SERVER_INSTRUCTIONS = """Personal development guidance exposed as native MCP resources and prompts.
 
-Use prompts for parameterized workflows. For task-specific guidance, list resources, select one
-skill://<name>/SKILL.md by description, and read its manifest only when supporting detail is needed.
-Read-only compatibility tools expose the same resource catalog to clients without native resource access.
+Use prompts for parameterized workflows. For task-specific guidance, search skills, select one
+skill://<name>/SKILL.md result, and read its manifest only when supporting detail is needed. Read-only
+compatibility tools expose the same resource catalog to clients without native resource access.
 """
 _SERVER_ICON = Icon(
     src=(
@@ -42,6 +47,51 @@ def _ro_annotations() -> dict[str, bool]:
     }
 
 
+class SkillSearchResult(BaseModel):
+    """Metadata needed to select a native skill resource."""
+
+    name: str
+    description: str
+    uri: str
+
+
+class SkillSearchResponse(BaseModel):
+    """Bounded skill matches for an agent search query."""
+
+    results: list[SkillSearchResult]
+
+
+def _skill_search_score(name: str, description: str, query: str) -> int:
+    normalized_name = name.casefold().replace("-", " ")
+    normalized_description = description.casefold()
+    normalized_query = " ".join(query.casefold().split())
+    terms = tuple(dict.fromkeys(re.findall(r"[a-z0-9]+", normalized_query)))
+    if not terms:
+        return 0
+
+    name_terms = set(normalized_name.split())
+    description_terms = set(re.findall(r"[a-z0-9]+", normalized_description))
+    score = 100 if normalized_query == normalized_name else 0
+    matched_terms = 0
+    for term in terms:
+        term_score = 0
+        if term in name_terms:
+            term_score = 20
+        elif term in normalized_name:
+            term_score = 10
+        elif term in description_terms:
+            term_score = 4
+        elif term in normalized_description:
+            term_score = 1
+        if term_score:
+            matched_terms += 1
+            score += term_score
+
+    if matched_terms == len(terms):
+        score += 10
+    return score
+
+
 def _register_components(mcp: FastMCP, registry: DocsRegistry) -> None:
     @mcp.resource(
         "resource://docs/{path*}",
@@ -57,6 +107,51 @@ def _register_components(mcp: FastMCP, registry: DocsRegistry) -> None:
 
 
 def _register_resource_tools(mcp: FastMCP) -> None:
+    @mcp.tool(
+        name="search_skills",
+        title="Search Skills",
+        description=(
+            "Find task-specific skill guidance by capability, technology, problem, or workflow. "
+            "Returns skill metadata and canonical URIs only; use read_resource to load a selected result."
+        ),
+        tags={"search", "skills"},
+        annotations=_ro_annotations(),
+    )
+    async def search_skills(
+        query: Annotated[
+            str,
+            Field(
+                min_length=2,
+                max_length=300,
+                description="Capability, technology, problem, or workflow to find.",
+            ),
+        ],
+        limit: Annotated[
+            int,
+            Field(ge=1, le=10, description="Maximum number of matches to return."),
+        ] = 5,
+    ) -> SkillSearchResponse:
+        resources = await mcp.list_resources()
+        matches: list[tuple[int, SkillSearchResult]] = []
+        for resource in resources:
+            uri = str(resource.uri)
+            if not uri.startswith("skill://") or not uri.endswith("/SKILL.md"):
+                continue
+
+            name = uri.removeprefix("skill://").removesuffix("/SKILL.md")
+            description = resource.description or ""
+            score = _skill_search_score(name, description, query)
+            if score:
+                matches.append(
+                    (
+                        score,
+                        SkillSearchResult(name=name, description=description, uri=uri),
+                    )
+                )
+
+        matches.sort(key=lambda match: (-match[0], match[1].name))
+        return SkillSearchResponse(results=[match[1] for match in matches[:limit]])
+
     @mcp.tool(
         name="list_resources",
         title="List Resources",