SQL Server MCP

Instructions

Implementation Reference

• src/mssql_mcp/tools/crud.py:43-131 (handler)

  The main execution logic for the 'read_rows' tool. Decorated with @mcp.tool(), it handles reading rows from MSSQL tables using ID, IDs, filter, or all rows with limits and column selection. Uses helpers for PK columns and table parsing.
  def read_rows( table: str, id: Any | None = None, ids: list[Any] | None = None, filter: str | None = None, columns: list[str] | None = None, max_rows: int | None = None, ) -> dict[str, Any]: """Read rows from a table by primary key or filter. Provide one of: id (single row), ids (multiple rows), or filter (WHERE clause). Args: table: Table name (can include schema: 'dbo.Users' or 'Users') id: Single primary key value (for composite keys, use filter) ids: List of primary key values filter: WHERE clause without 'WHERE' keyword (e.g., "status = 'active'") columns: List of columns to return (default: all columns) max_rows: Maximum rows to return Returns: Dictionary with: - table: Full table name - rows: List of row dictionaries - count: Number of rows returned """ try: manager = get_connection_manager() config = manager.config schema, table_name = parse_table_name(table) # Determine columns col_list = ", ".join(columns) if columns else "*" # Determine effective row limit effective_max_rows = min(max_rows or config.max_rows, config.max_rows) # Build query params: list[Any] = [] if id is not None: # Single row by primary key pk_cols = _get_primary_key_columns(schema, table_name) if not pk_cols: return {"error": f"No primary key found for table {schema}.{table_name}"} query = ( f"SELECT TOP {effective_max_rows} {col_list} " f"FROM [{schema}].[{table_name}] WHERE [{pk_cols[0]}] = %s" ) params = [id] elif ids is not None: # Multiple rows by primary keys pk_cols = _get_primary_key_columns(schema, table_name) if not pk_cols: return {"error": f"No primary key found for table {schema}.{table_name}"} placeholders = ", ".join(["%s"] * len(ids)) query = ( f"SELECT TOP {effective_max_rows} {col_list} " f"FROM [{schema}].[{table_name}] WHERE [{pk_cols[0]}] IN ({placeholders})" ) params = list(ids) elif filter is not None: # Custom filter # Note: filter params not supported - use execute_query for complex cases query = ( f"SELECT TOP {effective_max_rows} {col_list} " f"FROM [{schema}].[{table_name}] WHERE {filter}" ) else: # All rows (with limit) query = ( f"SELECT TOP {effective_max_rows} {col_list} " f"FROM [{schema}].[{table_name}]" ) rows = manager.execute_query(query, tuple(params) if params else None) return { "table": f"{schema}.{table_name}", "rows": rows, "count": len(rows), } except Exception as e: logger.error(f"Error reading rows from {table}: {e}") return {"error": str(e)}
• src/mssql_mcp/tools/crud.py:14-39 (helper)

  Supporting function to retrieve primary key column names from INFORMATION_SCHEMA, used by read_rows for ID-based queries.
  def _get_primary_key_columns(schema: str, table: str) -> list[str]: """Get primary key column(s) for a table. Args: schema: Schema name table: Table name Returns: List of primary key column names """ manager = get_connection_manager() query = """ SELECT c.COLUMN_NAME FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS tc JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE c ON tc.CONSTRAINT_NAME = c.CONSTRAINT_NAME AND tc.TABLE_SCHEMA = c.TABLE_SCHEMA AND tc.TABLE_NAME = c.TABLE_NAME WHERE tc.TABLE_SCHEMA = %s AND tc.TABLE_NAME = %s AND tc.CONSTRAINT_TYPE = 'PRIMARY KEY' ORDER BY c.ORDINAL_POSITION """ rows = manager.execute_query(query, (schema, table)) return [row["COLUMN_NAME"] for row in rows]
• src/mssql_mcp/server.py:203-203 (registration)

  Import statement that loads the crud module, triggering registration of all @mcp.tool() decorated functions including read_rows.
  from .tools import crud, databases, export, query, stored_procs, tables # noqa: E402, F401
• src/mssql_mcp/tools/__init__.py:4-4 (registration)

  Import in tools __init__.py that ensures crud tools are registered when the tools package is imported.
  from . import crud, databases, export, query, stored_procs, tables