SQL Server MCP

Instructions

Implementation Reference

• src/mssql_mcp/tools/query.py:13-82 (handler)

  The core handler function for the 'execute_query' MCP tool. Validates input query as read-only SELECT, applies row limits, executes via connection manager, and returns structured results including columns, rows, and metadata.
  @mcp.tool() def execute_query(query: str, max_rows: int | None = None) -> dict[str, Any]: """Execute a read-only SQL query and return results. Only SELECT statements are allowed. The query will have a row limit applied automatically if not specified. Args: query: SQL SELECT statement to execute max_rows: Maximum rows to return (overrides default, capped by MSSQL_MAX_ROWS) Returns: Dictionary with: - query: The original query - executed_query: The query that was actually executed (may include TOP) - columns: List of column names - rows: List of row dictionaries - row_count: Number of rows returned - max_rows: The effective row limit applied """ try: manager = get_connection_manager() config = manager.config # Create validator validator = SQLValidator( blocked_commands=config.blocked_commands, read_only=True, # execute_query is always read-only allowed_schemas=config.allowed_schemas if config.allowed_schemas else None, ) # Validate query is SELECT-only if not validator.is_select_only(query): return { "error": "Only SELECT queries are allowed. Use other tools for data modification.", "query": query, } # Check blocked commands is_valid, error = validator.validate(query) if not is_valid: return {"error": error, "query": query} # Determine effective row limit effective_max_rows = min(max_rows or config.max_rows, config.max_rows) # Inject row limit executed_query = validator.inject_row_limit(query, effective_max_rows) # Execute query rows = manager.execute_query(executed_query) # Extract column names from first row or return empty columns: list[str] = [] if rows: columns = list(rows[0].keys()) return { "query": query, "executed_query": executed_query, "columns": columns, "rows": rows, "row_count": len(rows), "max_rows": effective_max_rows, } except Exception as e: logger.error(f"Error executing query: {e}") return {"error": str(e), "query": query}
• src/mssql_mcp/tools/__init__.py:1-14 (registration)

  Registers the execute_query tool by importing the query module, which triggers the @mcp.tool() decorator to register it with the FastMCP instance.
  """MCP tools for SQL Server operations.""" # Import all tool modules to register their tools with the FastMCP instance from . import crud, databases, export, query, stored_procs, tables __all__ = [ "crud", "databases", "export", "query", "stored_procs", "tables", ]
• src/mssql_mcp/tools/query.py:14-32 (schema)

  Type hints and docstring define the input schema (query: str, max_rows: Optional[int]) and output schema (dict with query, executed_query, columns, rows, row_count, max_rows). Used by FastMCP for tool schema generation.
  def execute_query(query: str, max_rows: int | None = None) -> dict[str, Any]: """Execute a read-only SQL query and return results. Only SELECT statements are allowed. The query will have a row limit applied automatically if not specified. Args: query: SQL SELECT statement to execute max_rows: Maximum rows to return (overrides default, capped by MSSQL_MAX_ROWS) Returns: Dictionary with: - query: The original query - executed_query: The query that was actually executed (may include TOP) - columns: List of column names - rows: List of row dictionaries - row_count: Number of rows returned - max_rows: The effective row limit applied """