"""
MCP tools for smart search operations
Provides high-level intelligent search functionality
"""
import logging
from typing import Dict, Any
from services.smart_search_service import SmartSearchService
logger = logging.getLogger(__name__)
class SmartSearchTools:
"""
MCP tool implementations for smart search operations
High-level orchestrated search functionality
"""
def __init__(self, smart_search_service: SmartSearchService):
self.smart_search_service = smart_search_service
async def smart_search(
self,
question: str,
include_schema: bool = True,
max_sql_queries: int = 3,
max_semantic_results: int = 10
) -> Dict[str, Any]:
"""
MCP tool: Intelligent search that automatically determines the best strategy
Args:
question: Natural language question
include_schema: Whether to include schema info in response
max_sql_queries: Maximum SQL queries to execute
max_semantic_results: Maximum semantic search results
Returns:
Comprehensive search response
"""
try:
logger.info(f"MCP Tool - Smart Search: {question}")
result = await self.smart_search_service.search(
question=question,
include_schema=include_schema,
max_sql_queries=max_sql_queries,
max_semantic_results=max_semantic_results
)
return result
except Exception as e:
logger.error(f"Smart search tool failed: {e}")
return {
'success': False,
'error': str(e),
'response': f"I encountered an error while searching: {str(e)}"
}
async def get_search_capabilities(self) -> Dict[str, Any]:
"""
MCP tool: Get information about search system capabilities
Returns:
Search system capabilities and statistics
"""
try:
logger.info("MCP Tool - Get Search Capabilities")
capabilities = await self.smart_search_service.get_search_capabilities()
return {
'success': True,
'capabilities': capabilities
}
except Exception as e:
logger.error(f"Get search capabilities tool failed: {e}")
return {
'success': False,
'error': str(e),
'capabilities': {}
}
async def suggest_questions(self, topic: str = None) -> Dict[str, Any]:
"""
MCP tool: Suggest example questions based on available data
Args:
topic: Optional topic to focus suggestions on
Returns:
List of suggested questions
"""
try:
logger.info(f"MCP Tool - Suggest Questions: {topic or 'general'}")
suggestions = await self.smart_search_service.suggest_questions(topic)
return {
'success': True,
'topic': topic,
'suggestions': suggestions,
'count': len(suggestions)
}
except Exception as e:
logger.error(f"Suggest questions tool failed: {e}")
return {
'success': False,
'error': str(e),
'suggestions': []
}
