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

mcp updates

Browse Source
This commit is contained in:
John Lancaster
2026-08-27 18:48:21 -05:00
parent e999437b93
commit f3bbbfc25f
8 changed files with 43 additions and 180 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/personal_mcp/mcp.py
+6 -147
View File
@@ -1,15 +1,10 @@
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 mcp_types import CompletionArgument
from mcp_types import CompletionContext
from mcp_types import Icon
from mcp_types import PromptReference
from personal_mcp.prompts import create_prompts_provider
from personal_mcp.prompts.models import MarkdownPrompt
@@ -21,9 +16,8 @@ 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, 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.
Use prompts for parameterized workflows. For task-specific guidance, browse native skill resources,
select one skill://<name>/SKILL.md resource, and read its manifest only when supporting detail is needed.
"""
_SERVER_ICON = Icon(
src=(
@@ -47,51 +41,6 @@ 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*}",
@@ -106,95 +55,6 @@ def _register_components(mcp: FastMCP, registry: DocsRegistry) -> None:
return read_docs_markdown_path(registry, path)
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",
description=(
"List available MCP resources and URI templates. Use before read_resource to discover skill guidance."
),
annotations=_ro_annotations(),
)
async def list_resources() -> dict[str, list[dict[str, str | None]]]:
resources = await mcp.list_resources()
templates = await mcp.list_resource_templates()
return {
"resources": [
{
"uri": str(resource.uri),
"name": resource.name,
"description": resource.description,
"mime_type": resource.mime_type,
}
for resource in resources
],
"templates": [
{
"uri_template": template.uri_template,
"name": template.name,
"description": template.description,
"mime_type": template.mime_type,
}
for template in templates
],
}
@mcp.tool(
name="read_resource",
title="Read Resource",
description="Read a resource URI returned by list_resources, including skill files, manifests, and references.",
annotations=_ro_annotations(),
)
async def read_resource(uri: str) -> dict[str, object]:
result = await mcp.read_resource(uri)
return result.model_dump(mode="json", exclude_none=True)
def _register_prompt_completions(mcp: FastMCP, provider: MarkdownPromptsProvider) -> None:
@mcp.completion
async def complete_prompt_argument(
@@ -230,6 +90,5 @@ def create_mcp() -> FastMCP:
prompts_provider = create_prompts_provider()
mcp.add_provider(prompts_provider)
mcp.add_provider(create_skills_provider())
_register_resource_tools(mcp)
_register_prompt_completions(mcp, prompts_provider)
return mcp