"""
MCP tools for response synthesis operations
"""
import logging
from typing import Dict, Any, List, Optional
from services.synthesis_service import SynthesisService
from shared.models import SemanticSearchResult, QuestionClassification
logger = logging.getLogger(__name__)
class SynthesisTools:
"""
MCP tool implementations for response synthesis operations
"""
def __init__(self, synthesis_service: SynthesisService):
self.synthesis_service = synthesis_service
async def synthesize_response(
self,
question: str,
sql_results: Optional[List[Dict]] = None,
semantic_results: Optional[List[Dict]] = None,
schema_info: Optional[Dict] = None,
classification: Optional[Dict] = None
) -> Dict[str, Any]:
"""
MCP tool: Synthesize comprehensive response from multiple data sources
Args:
question: Original question
sql_results: Results from SQL queries
semantic_results: Results from semantic search
schema_info: Database schema information
classification: Question classification
Returns:
Synthesized response
"""
try:
logger.info(f"MCP Tool - Synthesize Response: {question}")
# Convert semantic results to objects if provided
semantic_objs = []
if semantic_results:
for result in semantic_results:
semantic_objs.append(SemanticSearchResult(
content=result.get('content', ''),
score=result.get('score', 0.0),
metadata=result.get('metadata', {})
))
# Convert classification if provided
classification_obj = None
if classification:
# This would need proper QuestionClassification object creation
# For now, pass as-is and let the service handle it
pass
response = self.synthesis_service.synthesize_response(
question=question,
sql_results=sql_results or [],
semantic_results=semantic_objs,
schema_info=schema_info,
classification=classification_obj
)
return {
'success': True,
'question': question,
'response': response,
'components_used': {
'sql_results': bool(sql_results),
'semantic_results': bool(semantic_results),
'schema_info': bool(schema_info),
'classification': bool(classification)
}
}
except Exception as e:
logger.error(f"Synthesize response tool failed: {e}")
return {
'success': False,
'error': str(e),
'response': f"I encountered an error while processing your question: {str(e)}"
}
async def clean_markdown(self, text: str) -> Dict[str, Any]:
"""
MCP tool: Clean and format markdown text
Args:
text: Text to clean
Returns:
Cleaned text
"""
try:
logger.info("MCP Tool - Clean Markdown")
cleaned_text = self.synthesis_service.clean_markdown(text)
return {
'success': True,
'original_text': text,
'cleaned_text': cleaned_text
}
except Exception as e:
logger.error(f"Clean markdown tool failed: {e}")
return {
'success': False,
'error': str(e),
'cleaned_text': text # Return original on error
}
async def format_sql_results(self, sql_results: List[Dict]) -> Dict[str, Any]:
"""
MCP tool: Format SQL results for display
Args:
sql_results: SQL query results
Returns:
Formatted results
"""
try:
logger.info("MCP Tool - Format SQL Results")
formatted = self.synthesis_service._format_sql_results(sql_results)
return {
'success': True,
'formatted_results': formatted,
'result_count': len(sql_results)
}
except Exception as e:
logger.error(f"Format SQL results tool failed: {e}")
return {
'success': False,
'error': str(e),
'formatted_results': f"Found {len(sql_results)} results (formatting error)"
}
async def format_semantic_results(self, semantic_results: List[Dict]) -> Dict[str, Any]:
"""
MCP tool: Format semantic search results for display
Args:
semantic_results: Semantic search results
Returns:
Formatted results
"""
try:
logger.info("MCP Tool - Format Semantic Results")
# Convert to SemanticSearchResult objects
semantic_objs = []
for result in semantic_results:
semantic_objs.append(SemanticSearchResult(
content=result.get('content', ''),
score=result.get('score', 0.0),
metadata=result.get('metadata', {})
))
formatted = self.synthesis_service._format_semantic_results(semantic_objs)
return {
'success': True,
'formatted_results': formatted,
'result_count': len(semantic_results)
}
except Exception as e:
logger.error(f"Format semantic results tool failed: {e}")
return {
'success': False,
'error': str(e),
'formatted_results': f"Found {len(semantic_results)} related documents (formatting error)"
}
