SQL Server MCP

Instructions

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

Implementation Reference

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

  Core handler function for the 'execute_query' MCP tool. Includes registration via @mcp.tool() decorator, input schema via type hints and docstring, validation logic, query execution, and result formatting.

  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/query.py:15-32 (schema)

  Input/output schema defined by function signature type hints and comprehensive docstring describing parameters and return structure.

  """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
  """

• src/mssql_mcp/tools/query.py:14-14 (registration)

  Tool registration using the @mcp.tool() decorator from FastMCP.

  def execute_query(query: str, max_rows: int | None = None) -> dict[str, Any]: