sten 6 implementation
This commit is contained in:
@@ -4,6 +4,8 @@ import os
|
||||
from typing import Any
|
||||
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.transforms import ResourcesAsTools
|
||||
from fastmcp.server.transforms.search import BM25SearchTransform, RegexSearchTransform
|
||||
|
||||
from personal_mcp.catalog.server import (
|
||||
build_skill_detail_payload,
|
||||
@@ -20,6 +22,8 @@ from personal_mcp.skills.document_loader import (
|
||||
)
|
||||
|
||||
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,
|
||||
@@ -28,6 +32,44 @@ REGISTRY: DocsRegistry = load_docs_registry(
|
||||
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,
|
||||
@@ -141,3 +183,6 @@ def get_skill_document_by_id(skill_id: str) -> dict[str, Any]:
|
||||
"found": True,
|
||||
"document": read_skill_document(REGISTRY, skill_id),
|
||||
}
|
||||
|
||||
|
||||
_install_tool_fallback_transforms()
|
||||
|
||||
Reference in New Issue
Block a user