Dynamic Per-User Tool Generation MCP Server

""" Exact Match Service Handles exact (case-insensitive) matching for entity names. This is the first stage of resolution before fuzzy matching. """ import logging from typing import Optional, Dict, Any from slot_resolution.core.models import ResolvedEntity, ResolutionMethod logger = logging.getLogger(__name__) class ExactMatchService: """ Service for exact case-insensitive matching. Checks if the normalized input exactly matches a known entity name. This is faster than fuzzy matching and should be tried first. """ def __init__(self): """Initialize the exact match service.""" # In a production system, this could maintain an in-memory # cache of entity names for fast exact matching self._entity_cache: Dict[str, Dict[str, Any]] = {} logger.info("ExactMatchService initialized") async def find_exact_match( self, entity_type: str, normalized_query: str, tenant_id: str, filters: Optional[Dict[str, Any]] = None ) -> Optional[ResolvedEntity]: """ Find exact match for normalized query. Args: entity_type: Type of entity to search normalized_query: Normalized search query tenant_id: Tenant identifier filters: Optional contextual filters Returns: ResolvedEntity if exact match found, None otherwise """ # TODO: Implement exact match logic # This could query a cache or database for exact name matches # For now, we return None to fall through to fuzzy matching logger.debug( f"Exact match check: entity_type={entity_type}, " f"query='{normalized_query}', tenant_id={tenant_id}" ) # Placeholder: In production, this would check a cache or database return None def cache_entity( self, entity_type: str, entity_id: str, canonical_name: str, normalized_name: str, tenant_id: str ): """ Cache an entity for fast exact matching. Args: entity_type: Type of entity entity_id: Entity ID canonical_name: Canonical name normalized_name: Normalized name for matching tenant_id: Tenant identifier """ cache_key = f"{tenant_id}:{entity_type}:{normalized_name}" self._entity_cache[cache_key] = { "id": entity_id, "canonical_name": canonical_name, "entity_type": entity_type } logger.debug(f"Cached entity: {cache_key}") def clear_cache(self, tenant_id: Optional[str] = None): """ Clear entity cache. Args: tenant_id: If specified, clear only for this tenant """ if tenant_id: # Clear only for specific tenant keys_to_remove = [ k for k in self._entity_cache.keys() if k.startswith(f"{tenant_id}:") ] for key in keys_to_remove: del self._entity_cache[key] logger.info(f"Cleared cache for tenant {tenant_id}") else: # Clear all self._entity_cache.clear() logger.info("Cleared all entity cache")