"""
MCP tools for SQL operations
"""
import logging
from typing import Dict, Any
from services.sql_service import SQLService
logger = logging.getLogger(__name__)
class SQLTools:
"""
MCP tool implementations for SQL operations
"""
def __init__(self, sql_service: SQLService):
self.sql_service = sql_service
async def execute_sql(self, sql: str, limit: bool = True) -> Dict[str, Any]:
"""
MCP tool: Execute a SQL query safely
Args:
sql: SQL query to execute
limit: Whether to apply row limit for safety
Returns:
Query execution results
"""
try:
logger.info(f"MCP Tool - Execute SQL: {sql[:100]}...")
result = self.sql_service.execute_safe(sql, limit=limit)
return {
'success': result.success,
'data': result.data,
'rows_affected': result.rows_affected,
'execution_time': result.execution_time,
'query': result.query,
'error': result.error
}
except Exception as e:
logger.error(f"Execute SQL tool failed: {e}")
return {
'success': False,
'error': str(e),
'data': [],
'rows_affected': 0
}
async def validate_sql(self, sql: str) -> Dict[str, Any]:
"""
MCP tool: Validate SQL without executing
Args:
sql: SQL query to validate
Returns:
Validation results
"""
try:
logger.info(f"MCP Tool - Validate SQL: {sql[:100]}...")
error = self.sql_service.validate_sql(sql)
return {
'valid': error is None,
'error': error
}
except Exception as e:
logger.error(f"Validate SQL tool failed: {e}")
return {
'valid': False,
'error': str(e)
}
async def explain_sql(self, sql: str) -> Dict[str, Any]:
"""
MCP tool: Get execution plan for SQL query
Args:
sql: SQL query to explain
Returns:
Query execution plan
"""
try:
logger.info(f"MCP Tool - Explain SQL: {sql[:100]}...")
return self.sql_service.explain_sql(sql)
except Exception as e:
logger.error(f"Explain SQL tool failed: {e}")
return {
'success': False,
'error': str(e)
}
