Jason Thistlethwaite
81 lines
4.0 KiB
Python
81 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from typing import Any, Dict, Optional
|
|
|
|
from .models import SearchQuery, search_response
|
|
|
|
|
|
class SemanticMCP:
|
|
def __init__(self, search_service: Any, backfill_service: Optional[Any], store: Optional[Any] = None, refresh_service: Optional[Any] = None) -> None:
|
|
self.search_service = search_service
|
|
self.backfill_service = backfill_service
|
|
self.store = store
|
|
self.refresh_service = refresh_service
|
|
|
|
def tools(self) -> Dict[str, Dict[str, str]]:
|
|
return {
|
|
"semantic_search": {"description": "Search the semantic index and return cited snippets."},
|
|
"semantic_get_document": {"description": "Fetch one indexed document by stable id."},
|
|
"semantic_list_projects": {"description": "List indexed project identifiers and document counts."},
|
|
"semantic_backfill_redmine_sample": {"description": "Rebuild the Redmine sample collection."},
|
|
"semantic_refresh_redmine": {"description": "Refresh recent Redmine issues without re-embedding unchanged documents."},
|
|
}
|
|
|
|
def call_tool(self, name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
|
|
if name == "semantic_search":
|
|
query = SearchQuery(
|
|
text=arguments.get("query") or arguments.get("text") or "",
|
|
source=arguments.get("source"),
|
|
project_id=arguments.get("project_id"),
|
|
project_identifier=arguments.get("project_identifier"),
|
|
doc_type=arguments.get("doc_type"),
|
|
issue_id=arguments.get("issue_id"),
|
|
contact_id=arguments.get("contact_id"),
|
|
contact_email=arguments.get("contact_email"),
|
|
date_from=arguments.get("date_from"),
|
|
date_to=arguments.get("date_to"),
|
|
limit=int(arguments.get("limit", 10)),
|
|
include_snippets=bool(arguments.get("include_snippets", True)),
|
|
)
|
|
results = self.search_service.search(query)
|
|
return search_response(query, results)
|
|
if name == "semantic_get_document":
|
|
return self.search_service.get_document(arguments["id"]) or {"error": "not_found", "id": arguments["id"]}
|
|
if name == "semantic_list_projects":
|
|
if self.store is None:
|
|
return {"error": "project_listing_unavailable"}
|
|
return {"projects": self.store.list_projects(source=arguments.get("source", "redmine"))}
|
|
if name == "semantic_backfill_redmine_sample":
|
|
if self.backfill_service is None:
|
|
return {"error": "backfill_unavailable"}
|
|
return self.backfill_service.backfill_redmine_sample(limit=int(arguments.get("limit", 500)))
|
|
if name == "semantic_refresh_redmine":
|
|
if self.refresh_service is None:
|
|
return {"error": "refresh_unavailable"}
|
|
project_limits = arguments.get("project_limits")
|
|
if not project_limits:
|
|
project = arguments.get("project_identifier")
|
|
if not project:
|
|
return {"error": "project_required"}
|
|
project_limits = {project: int(arguments.get("limit", 500))}
|
|
return self.refresh_service.refresh_redmine_project_limits(
|
|
{str(project): int(limit) for project, limit in project_limits.items()},
|
|
dry_run=bool(arguments.get("dry_run", False)),
|
|
force_rebuild=bool(arguments.get("force_rebuild", False)),
|
|
overlap_minutes=int(arguments.get("overlap_minutes", 15)),
|
|
)
|
|
raise ValueError(f"unknown tool: {name}")
|
|
|
|
|
|
def serve_stdio(mcp: SemanticMCP) -> None:
|
|
for line in sys.stdin:
|
|
request = json.loads(line)
|
|
try:
|
|
result = mcp.call_tool(request["name"], request.get("arguments") or {})
|
|
response = {"id": request.get("id"), "result": result}
|
|
except Exception as exc:
|
|
response = {"id": request.get("id"), "error": str(exc)}
|
|
print(json.dumps(response), flush=True)
|