"""
MCP tools for search operations
"""
import logging
from typing import Dict, Any, Optional
from services.smart_search import SmartSearch
from shared.models import SearchResponse
logger = logging.getLogger(__name__)
class SearchTools:
"""
MCP tool implementations for search operations
Thin layer that delegates to business logic services
"""
def __init__(self, smart_search: SmartSearch):
self.smart_search = smart_search
async def answer_question(
self,
question: str,
filters: Optional[Dict[str, Any]] = None,
limit: int = 10,
include_sql: bool = True,
include_semantic: bool = True
) -> Dict[str, Any]:
"""
MCP tool: Answer a natural language question about the database
Args:
question: Natural language question
filters: Optional filters to apply
limit: Maximum number of results
include_sql: Whether to include SQL-based search
include_semantic: Whether to include semantic search
Returns:
Structured response with answer and metadata
"""
try:
logger.info(f"MCP Tool - Answer Question: {question}")
response = self.smart_search.answer(
question=question,
filters=filters,
limit=limit,
include_sql=include_sql,
include_semantic=include_semantic
)
return self._format_search_response(response)
except Exception as e:
logger.error(f"Answer question tool failed: {e}")
return {
'success': False,
'error': str(e),
'answer': f"I encountered an error while processing your question: {str(e)}"
}
async def semantic_search(
self,
query: str,
limit: int = 10,
filters: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""
MCP tool: Perform semantic search on documents
Args:
query: Search query
limit: Maximum number of results
filters: Optional filters to apply
Returns:
Search results with metadata
"""
try:
logger.info(f"MCP Tool - Semantic Search: {query}")
results = self.smart_search.semantic_service.search(
query=query,
limit=limit,
filters=filters
)
return {
'success': True,
'results': [
{
'content': result.content,
'score': result.score,
'source': result.source,
'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_capabilities(self) -> Dict[str, Any]:
"""
MCP tool: Get information about available search capabilities
Returns:
Capabilities information
"""
try:
logger.info("MCP Tool - Get Search Capabilities")
return self.smart_search.get_capabilities()
except Exception as e:
logger.error(f"Get capabilities tool failed: {e}")
return {
'error': str(e),
'vector_search_available': False,
'tables_available': 0
}
def _format_search_response(self, response: SearchResponse) -> Dict[str, Any]:
"""Format SearchResponse for MCP tool output"""
return {
'success': response.success,
'answer': response.answer_markdown,
'strategy': response.strategy.value,
'sources_used': response.sources_used,
'processing_time': response.processing_time,
'sql_queries': [
{
'sql': query.sql,
'description': query.description,
'confidence': query.confidence
}
for query in response.sql_queries
],
'semantic_results': [
{
'content': result.content[:200] + "..." if len(result.content) > 200 else result.content,
'score': result.score,
'source': result.source
}
for result in response.semantic_results
],
'error': response.error
}
