"""MCP tool for searching processes by keyword.
Implements the search_processes tool per mcp-tools.json contract.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from sso_mcp_server import get_logger
if TYPE_CHECKING:
from sso_mcp_server.processes.service import ProcessService
_logger = get_logger("search_processes_tool")
def search_processes_impl(query: str, service: ProcessService) -> dict[str, Any]:
"""Implementation of search_processes tool.
Searches across all development processes for a keyword.
Returns matching processes ranked by relevance.
Args:
query: Search keyword or phrase (case-insensitive).
service: ProcessService instance.
Returns:
Dictionary with search results, query, count, and total_processes.
"""
_logger.info("search_processes_request", query=query)
# Get total process count
all_processes = service.list_processes()
total_processes = len(all_processes)
# Perform search
results = service.search_processes(query)
result: dict[str, Any] = {
"results": results,
"query": query,
"count": len(results),
"total_processes": total_processes,
}
if len(results) == 0:
result["message"] = f"No processes matched the search query '{query}'."
_logger.info(
"search_processes_completed",
query=query,
result_count=len(results),
total_processes=total_processes,
)
return result