333 lines
9.3 KiB
Python
333 lines
9.3 KiB
Python
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 build_skill_detail_payload
|
|
from personal_mcp.catalog.server import build_skills_index_payload
|
|
from personal_mcp.catalog.server import get_pattern_by_id_payload
|
|
from personal_mcp.catalog.server import get_prompt_by_id_payload
|
|
from personal_mcp.catalog.server import search_patterns_payload
|
|
from personal_mcp.catalog.server import search_prompts_payload
|
|
from personal_mcp.registry.load import load_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.registry.read import read_skill_document
|
|
from personal_mcp.registry.read import read_skill_reference
|
|
|
|
DOCS_ROOT = os.getenv("PERSONAL_MCP_DOCS_ROOT", "../../docs")
|
|
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")
|
|
REGISTRY: DocsRegistry = load_docs_registry(
|
|
package_anchor="personal_mcp",
|
|
docs_root=DOCS_ROOT,
|
|
)
|
|
|
|
mcp = FastMCP("personal-mcp", on_duplicate="error")
|
|
|
|
|
|
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() -> 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 {
|
|
"readOnlyHint": True,
|
|
"idempotentHint": True,
|
|
}
|
|
|
|
|
|
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 _make_prompt_handler(content: str):
|
|
def prompt_handler(**kwargs: Any) -> str:
|
|
return _render_prompt_markdown(content, kwargs)
|
|
|
|
return prompt_handler
|
|
|
|
|
|
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
|
|
default = Parameter.empty if arg.required else arg.default
|
|
params.append(
|
|
Parameter(
|
|
arg_name,
|
|
kind=Parameter.KEYWORD_ONLY,
|
|
default=default,
|
|
annotation=arg_type,
|
|
)
|
|
)
|
|
|
|
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),
|
|
)
|
|
|
|
|
|
@mcp.resource(
|
|
"resource://catalog/skills_index",
|
|
mime_type="application/json",
|
|
tags={"catalog"},
|
|
annotations=_ro_annotations(),
|
|
)
|
|
def skills_index() -> dict[str, Any]:
|
|
return build_skills_index_payload(REGISTRY)
|
|
|
|
|
|
@mcp.resource(
|
|
"resource://catalog/skills_index{?q,tag,capability,cursor,limit}",
|
|
mime_type="application/json",
|
|
tags={"catalog"},
|
|
annotations=_ro_annotations(),
|
|
)
|
|
def skills_index_query(
|
|
q: str | None = None,
|
|
tag: str | None = None,
|
|
capability: str | None = None,
|
|
cursor: str | None = None,
|
|
limit: int | None = None,
|
|
) -> dict[str, Any]:
|
|
return build_skills_index_payload(
|
|
REGISTRY,
|
|
query=q,
|
|
tag=tag,
|
|
capability=capability,
|
|
cursor=cursor,
|
|
limit=limit,
|
|
)
|
|
|
|
|
|
@mcp.resource(
|
|
"resource://catalog/skills/{skill_id}",
|
|
mime_type="application/json",
|
|
tags={"catalog"},
|
|
annotations=_ro_annotations(),
|
|
)
|
|
def skill_detail(skill_id: str) -> dict[str, Any]:
|
|
return build_skill_detail_payload(REGISTRY, skill_id)
|
|
|
|
|
|
@mcp.resource(
|
|
"resource://skills/{skill_id}/document",
|
|
mime_type="text/markdown",
|
|
tags={"skill-doc"},
|
|
annotations=_ro_annotations(),
|
|
)
|
|
def skill_document(skill_id: str) -> dict[str, str]:
|
|
return read_skill_document(REGISTRY, skill_id)
|
|
|
|
|
|
@mcp.resource(
|
|
"resource://skills/{skill_id}/references/{ref_id}",
|
|
mime_type="text/markdown",
|
|
tags={"reference"},
|
|
annotations=_ro_annotations(),
|
|
)
|
|
def skill_reference(skill_id: str, ref_id: str) -> dict[str, str]:
|
|
return read_skill_reference(REGISTRY, skill_id=skill_id, ref_id=ref_id)
|
|
|
|
|
|
@mcp.resource(
|
|
"resource://docs/{path*}",
|
|
mime_type="text/markdown",
|
|
tags={"docs"},
|
|
annotations=_ro_annotations(),
|
|
)
|
|
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 = "",
|
|
tags: list[str] | None = None,
|
|
skip: int = 0,
|
|
limit: int = 20,
|
|
) -> dict[str, Any]:
|
|
"""Search normalized pattern metadata with optional tags and pagination."""
|
|
return search_patterns_payload(
|
|
REGISTRY,
|
|
query=query,
|
|
tags=tags,
|
|
skip=skip,
|
|
limit=limit,
|
|
)
|
|
|
|
|
|
@mcp.tool
|
|
def get_pattern_by_id(id: str) -> dict[str, Any]:
|
|
"""Return one normalized pattern by stable id."""
|
|
return get_pattern_by_id_payload(REGISTRY, id)
|
|
|
|
|
|
@mcp.tool
|
|
def get_skill_document_by_id(skill_id: str) -> dict[str, Any]:
|
|
"""Return the canonical skill document payload for a stable skill id."""
|
|
if skill_id not in REGISTRY.skills_by_id:
|
|
return {"found": False, "id": skill_id}
|
|
|
|
return {
|
|
"found": True,
|
|
"document": read_skill_document(REGISTRY, skill_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)
|
|
|
|
|
|
_install_tool_fallback_transforms()
|
|
_register_prompt_objects()
|