SQL Server MCP

Instructions

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

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