126 lines
4.0 KiB
Python
126 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from personal_mcp.registry.models.registry import DocsRegistry
|
|
from personal_mcp.registry.models.registry import PromptRecord
|
|
from personal_mcp.registry.models.registry import PromptSummaryPayload
|
|
|
|
DEFAULT_LIMIT = 20
|
|
MAX_LIMIT = 100
|
|
|
|
|
|
def _prompt_matches(
|
|
prompt: PromptRecord,
|
|
*,
|
|
query: str | None,
|
|
tag: str | None,
|
|
) -> bool:
|
|
if query:
|
|
lowered = query.strip().lower()
|
|
if lowered:
|
|
haystack = " ".join(
|
|
[
|
|
prompt.prompt_id,
|
|
prompt.name,
|
|
prompt.description,
|
|
" ".join(prompt.tags),
|
|
" ".join(sorted(prompt.arguments)),
|
|
]
|
|
).lower()
|
|
terms = [term for term in lowered.replace("-", " ").split() if term]
|
|
if any(term not in haystack for term in terms):
|
|
return False
|
|
|
|
return not (tag and tag not in prompt.tags)
|
|
|
|
|
|
def build_prompts_index_payload(
|
|
registry: DocsRegistry,
|
|
*,
|
|
query: str | None = None,
|
|
tag: 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.prompts_by_id[prompt_id] for prompt_id in registry.prompts_in_load_order]
|
|
matches = [prompt for prompt in ordered if _prompt_matches(prompt, query=query, tag=tag)]
|
|
|
|
page = matches[start : start + normalized_limit]
|
|
next_cursor = start + normalized_limit
|
|
|
|
return {
|
|
"prompts": [PromptSummaryPayload.from_record(prompt).model_dump() for prompt 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_prompt_detail_payload(registry: DocsRegistry, prompt_id: str) -> dict[str, Any]:
|
|
if prompt_id not in registry.prompts_by_id:
|
|
raise KeyError(prompt_id)
|
|
|
|
prompt = registry.prompts_by_id[prompt_id]
|
|
return {
|
|
"id": prompt.prompt_id,
|
|
"name": prompt.name,
|
|
"description": prompt.description,
|
|
"version": prompt.version,
|
|
"tags": list(prompt.tags),
|
|
"capabilities": list(prompt.capabilities),
|
|
"resources": {
|
|
"document": prompt.document_uri,
|
|
},
|
|
"arguments": {
|
|
arg_name: arg.model_dump(exclude_none=True) for arg_name, arg in sorted(prompt.arguments.items())
|
|
},
|
|
}
|
|
|
|
|
|
def search_prompts_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[PromptRecord] = []
|
|
for prompt_id in registry.prompts_in_load_order:
|
|
prompt = registry.prompts_by_id[prompt_id]
|
|
if not _prompt_matches(prompt, query=query, tag=None):
|
|
continue
|
|
if requested_tags and any(tag not in prompt.tags for tag in requested_tags):
|
|
continue
|
|
matches.append(prompt)
|
|
|
|
page = matches[normalized_skip : normalized_skip + normalized_limit]
|
|
return {
|
|
"prompts": [PromptSummaryPayload.from_record(prompt).model_dump() for prompt in page],
|
|
"total": len(matches),
|
|
"skip": normalized_skip,
|
|
"limit": normalized_limit,
|
|
}
|
|
|
|
|
|
def get_prompt_by_id_payload(registry: DocsRegistry, prompt_id: str) -> dict[str, Any]:
|
|
if prompt_id not in registry.prompts_by_id:
|
|
return {"found": False, "id": prompt_id}
|
|
return {
|
|
"found": True,
|
|
"prompt": build_prompt_detail_payload(registry, prompt_id),
|
|
}
|