SSO MCP Server

"""Process service for retrieving and listing development processes. Provides business logic for process operations. """ from __future__ import annotations from typing import TYPE_CHECKING, Any from sso_mcp_server import get_logger from sso_mcp_server.processes.discovery import discover_processes from sso_mcp_server.processes.parser import normalize_name, parse_process_file if TYPE_CHECKING: from pathlib import Path _logger = get_logger("process_service") class ProcessService: """Service for managing development process operations. Provides methods to get, list, and search processes from the configured directory. """ def __init__(self, process_dir: Path) -> None: """Initialize the process service. Args: process_dir: Directory containing process markdown files. """ self._process_dir = process_dir _logger.debug("process_service_initialized", directory=str(process_dir)) def get_process(self, name: str) -> dict[str, Any] | None: """Get a process by name. Performs case-insensitive matching against process names (from frontmatter) and filenames (without extension). Args: name: Name of the process to retrieve. Returns: Dictionary with name, description, content, and path. Returns None if process not found. """ _logger.debug("getting_process", name=name) name_normalized = normalize_name(name) name_lower = name.lower() for file_path in discover_processes(self._process_dir): parsed = parse_process_file(file_path) if parsed is None: continue # Match by normalized name from frontmatter (case-insensitive) if parsed.get("normalized_name") == name_normalized: _logger.info("process_found", name=parsed["name"], path=str(file_path)) return parsed # Match by name from frontmatter (case-insensitive) if parsed["name"].lower() == name_lower: _logger.info("process_found_by_name", name=parsed["name"], path=str(file_path)) return parsed # Match by filename (without extension, case-insensitive) if file_path.stem.lower() == name_lower: _logger.info( "process_found_by_filename", name=parsed["name"], path=str(file_path), ) return parsed # Match by normalized filename if normalize_name(file_path.stem) == name_normalized: _logger.info( "process_found_by_normalized_filename", name=parsed["name"], path=str(file_path), ) return parsed _logger.debug("process_not_found", name=name) return None def list_processes(self) -> list[dict[str, Any]]: """List all available processes. Returns metadata (name, description) for all processes. Does NOT include full content. Returns: List of dictionaries with name and description for each process. """ _logger.debug("listing_processes") processes = [] for file_path in discover_processes(self._process_dir): parsed = parse_process_file(file_path) if parsed is None: continue # Return metadata only, not content processes.append( { "name": parsed["name"], "description": parsed.get("description", ""), } ) _logger.info("processes_listed", count=len(processes)) return processes def get_available_names(self) -> list[str]: """Get list of available process names. Useful for error messages when a process is not found (FR-015). Returns: List of process names. """ processes = self.list_processes() return [p["name"] for p in processes] def search_processes(self, query: str) -> list[dict[str, Any]]: """Search processes by keyword. Searches across process name, description, and content. Results are ordered by relevance. Args: query: Search keyword or phrase. Returns: List of search results with name, description, relevance_score, and snippet. """ from sso_mcp_server.processes.search import SearchEngine _logger.debug("searching_processes", query=query) engine = SearchEngine(self) results = engine.search(query) _logger.info("search_completed", query=query, result_count=len(results)) return results