"""
MCP tools for semantic search operations
"""
import logging
from typing import Dict, Any
from services.semantic_service import SemanticService
logger = logging.getLogger(__name__)
class SemanticTools:
"""
MCP tool implementations for semantic search operations
"""
def __init__(self, semantic_service: SemanticService):
self.semantic_service = semantic_service
async def semantic_search(
self,
query: str,
limit: int = 10,
table_filter: str = None,
fk_filter: str = None
) -> Dict[str, Any]:
"""
MCP tool: Perform semantic search
Args:
query: Search query
limit: Maximum number of results
table_filter: Optional table filter
fk_filter: Optional foreign key filter
Returns:
Semantic search results
"""
try:
logger.info(f"MCP Tool - Semantic Search: {query}")
filters = {}
if table_filter:
filters['table_filter'] = table_filter
if fk_filter:
filters['fk_filter'] = fk_filter
results = self.semantic_service.search(
query=query,
limit=limit,
filters=filters if filters else None
)
return {
'success': True,
'query': query,
'results': [
{
'content': result.content,
'score': result.score,
'metadata': result.metadata
}
for result in results
],
'count': len(results)
}
except Exception as e:
logger.error(f"Semantic search tool failed: {e}")
return {
'success': False,
'error': str(e),
'results': [],
'count': 0
}
async def get_search_stats(self) -> Dict[str, Any]:
"""
MCP tool: Get semantic search system statistics
Returns:
Search system statistics
"""
try:
logger.info("MCP Tool - Get Search Stats")
stats = self.semantic_service.get_search_stats()
return {
'success': True,
'stats': stats
}
except Exception as e:
logger.error(f"Get search stats tool failed: {e}")
return {
'success': False,
'error': str(e),
'stats': {}
}
async def extract_search_terms(self, query: str) -> Dict[str, Any]:
"""
MCP tool: Extract search terms from a query
Args:
query: Query to extract terms from
Returns:
Extracted search terms
"""
try:
logger.info(f"MCP Tool - Extract Search Terms: {query}")
terms = self.semantic_service.extract_search_terms(query)
return {
'success': True,
'query': query,
'terms': terms,
'count': len(terms)
}
except Exception as e:
logger.error(f"Extract search terms tool failed: {e}")
return {
'success': False,
'error': str(e),
'terms': [],
'count': 0
}
