MCP SQL Server

"""Execute SELECT query tool.""" from typing import Any, Optional from fastmcp import Context from .base import MCPTool class ExecuteSelectService(MCPTool): """Service to execute SELECT queries safely.""" @property def name(self) -> str: return "execute_select_query" @property def description(self) -> str: return "Execute a SELECT query to verify/test queries. Only SELECT queries are allowed - no ALTER, INSERT, CREATE, UPDATE, DELETE, etc." async def execute( self, ctx: Context, query: str, database: Optional[str] = None, server_name: Optional[str] = None, limit: int = 100, user: Optional[str] = None, password: Optional[str] = None, driver: Optional[str] = None, port: Optional[int] = None ) -> dict[str, Any]: """Execute a SELECT query safely. Args: ctx: FastMCP context query: SQL SELECT query to execute database: Database name server_name: Server hostname/address limit: Maximum number of rows to return (default 100, max 1000) user: Database username password: Database password driver: ODBC driver name port: Server port Returns: dict: Query results with columns, rows, and metadata, or error dictionary """ creds = self.creds_manager.get_from_context( ctx, user, password, server_name, database, driver, port ) if not creds.database: return {"error": "Database name is required"} if not query or not query.strip(): return {"error": "Query cannot be empty"} if not creds.is_valid(): return {"error": "Missing credentials"} return self.executor.execute_select_query(creds, query, limit)