MCP SQL Server

"""Query specific columns tool.""" from typing import Any, Optional from fastmcp import Context from .base import MCPTool class QueryColumnsService(MCPTool): """Service to query specific columns from a table.""" @property def name(self) -> str: return "query_table_with_columns" @property def description(self) -> str: return "Query specific columns from a table" async def execute( self, ctx: Context, table: str, columns: list[str], database: Optional[str] = None, server_name: Optional[str] = None, limit: int = 10, user: Optional[str] = None, password: Optional[str] = None, driver: Optional[str] = None, port: Optional[int] = None ) -> list[dict[str, Any]]: """Query specific columns. Args: ctx: FastMCP context table: Table name to query columns: List of column names to retrieve database: Database name server_name: Server hostname/address limit: Maximum number of rows (default 10, max 100) user: Database username password: Database password driver: ODBC driver name port: Server port Returns: list[dict]: Query results or error dictionary """ creds = self.creds_manager.get_from_context( ctx, user, password, server_name, database, driver, port ) if not creds.database or not table: return [{"error": "Database and table names are required"}] if not creds.is_valid(): return [{"error": "Missing credentials"}] return self.executor.query_columns(creds, table, columns, limit)