diff --git a/remotespark/sparkkernelbase.py b/remotespark/sparkkernelbase.py
index 3abecfd..bd4c70f 100644
--- a/remotespark/sparkkernelbase.py
+++ b/remotespark/sparkkernelbase.py
@@ -9,6 +9,11 @@ from remotespark.utils.utils import get_connection_string
 
 
 class SparkKernelBase(IPythonKernel):
+    run_command = "run"
+    config_command = "config"
+    sql_command = "sql"
+    hive_command = "hive"
+
     def __init__(self, implementation, implementation_version, language, language_version, language_info,
                  kernel_conf_name, session_language, client_name, **kwargs):
         # Required by Jupyter - Override
@@ -32,7 +37,7 @@ class SparkKernelBase(IPythonKernel):
         # Disable warnings for test env in HDI
         requests.packages.urllib3.disable_warnings()
 
-        if "testing" not in kwargs.keys():
+        if not kwargs.get("testing", False):
             (username, password, url) = self._get_configuration()
             self.connection_string = get_connection_string(url, username, password)
             self._load_magics_extension()
@@ -41,29 +46,40 @@ class SparkKernelBase(IPythonKernel):
         if self._fatal_error is not None:
             self._abort_with_fatal_error(self._fatal_error)
 
-        if not self.session_started:
-            self._start_session()
-
-        # Modify code by prepending spark magic text
-        if code.lower().startswith("%sql\n") or code.lower().startswith("%sql "):
-            code = "%%spark -c sql\n{}".format(code[5:])
-        elif code.lower().startswith("%%sql\n") or code.lower().startswith("%%sql "):
-            code = "%%spark -c sql\n{}".format(code[6:])
-        elif code.lower().startswith("%hive\n") or code.lower().startswith("%hive "):
-            code = "%%spark -c hive\n{}".format(code[6:])
-        elif code.lower().startswith("%%hive\n") or code.lower().startswith("%%hive "):
-            code = "%%spark -c hive\n{}".format(code[7:])
+        subcommand, flags, code_to_run = self._parse_user_command(code)
+
+        if subcommand == self.run_command:
+            code_to_run = "%%spark\n{}".format(code_to_run)
+            return self._run_starting_session(code_to_run, silent, store_history, user_expressions, allow_stdin)
+        elif subcommand == self.sql_command:
+            code_to_run = "%%spark -c sql\n{}".format(code_to_run)
+            return self._run_starting_session(code_to_run, silent, store_history, user_expressions, allow_stdin)
+        elif subcommand == self.hive_command:
+            code_to_run = "%%spark -c hive\n{}".format(code_to_run)
+            return self._run_starting_session(code_to_run, silent, store_history, user_expressions, allow_stdin)
+        elif subcommand == self.config_command:
+            restart_session = False
+
+            if self.session_started:
+                if "f" not in flags:
+                    raise KeyError("A session has already been started. In order to modify the Spark configuration, "
+                                   "please provide the '-f' flag at the beginning of the config magic:\n\te.g. `%config"
+                                   " -f {}`\n\nNote that this will kill the current session and will create a new one "
+                                   "with the configuration provided. All previously run commands in the session will be"
+                                   " lost.")
+                else:
+                    restart_session = True
+
+            code_to_run = "%%spark config {}".format(code_to_run)
+
+            return self._run_restarting_session(code_to_run, silent, store_history, user_expressions, allow_stdin,
+                                                restart_session)
         else:
-            code = "%%spark\n{}".format(code)
-
-        return self._execute_cell(code, silent, store_history, user_expressions, allow_stdin)
+            raise KeyError("Magic '{}' not supported.".format(subcommand))
 
     def do_shutdown(self, restart):
         # Cleanup
-        if self.session_started:
-            code = "%spark cleanup"
-            self._execute_cell_for_user(code, True, False)
-            self.session_started = False
+        self._delete_session()
 
         return self._do_shutdown_ipykernel(restart)
 
@@ -83,6 +99,27 @@ class SparkKernelBase(IPythonKernel):
                                log_if_error="Failed to create a Livy session.")
             self.logger.debug("Added session.")
 
+    def _delete_session(self):
+        if self.session_started:
+            code = "%spark cleanup"
+            self._execute_cell_for_user(code, True, False)
+            self.session_started = False
+
+    def _run_starting_session(self, code, silent, store_history, user_expressions, allow_stdin):
+        self._start_session()
+        return self._execute_cell(code, silent, store_history, user_expressions, allow_stdin)
+
+    def _run_restarting_session(self, code, silent, store_history, user_expressions, allow_stdin, restart):
+        if restart:
+            self._delete_session()
+
+        res = self._execute_cell(code, silent, store_history, user_expressions, allow_stdin)
+
+        if restart:
+            self._start_session()
+
+        return res
+
     def _get_configuration(self):
         try:
             credentials = getattr(conf, 'kernel_' + self.kernel_conf_name + '_credentials')()
@@ -95,6 +132,35 @@ class SparkKernelBase(IPythonKernel):
                 self.kernel_conf_name)
             self._abort_with_fatal_error(message)
 
+    def _parse_user_command(self, code):
+        # Normalize 2 signs to 1
+        if code.startswith("%%"):
+            code = code[1:]
+
+        # When no magic, return run command
+        if not code.startswith("%"):
+            code = "%{} {}".format(self.run_command, code)
+
+        # Remove percentage sign
+        code = code[1:]
+
+        split_code = code.split(None, 1)
+        subcommand = split_code[0].lower()
+        flags = []
+        rest = split_code[1]
+
+        # Get all flags
+        flag_split = rest.split(None, 1)
+        while len(flag_split) >= 2 and flag_split[0].startswith("-"):
+            flags.append(flag_split[0][1:].lower())
+            rest = flag_split[1]
+            flag_split = rest.split(None, 1)
+
+        # flags to lower
+        flags = [i.lower() for i in flags]
+
+        return subcommand, flags, rest
+
     def _execute_cell(self, code, silent, store_history=True, user_expressions=None, allow_stdin=False,
                       shutdown_if_error=False, log_if_error=None):
         reply_content = self._execute_cell_for_user(code, silent, store_history, user_expressions, allow_stdin)

Expose session configs through wrapper kernel
diff --git a/tests/test_sparkkernelbase.py b/tests/test_sparkkernelbase.py
index a3c3ce1..d5f105c 100644
--- a/tests/test_sparkkernelbase.py
+++ b/tests/test_sparkkernelbase.py
@@ -1,5 +1,5 @@
 from mock import MagicMock, call
-from nose.tools import with_setup
+from nose.tools import with_setup, raises
 
 from remotespark.sparkkernelbase import SparkKernelBase
 import remotespark.utils.configuration as conf
@@ -53,7 +53,7 @@ def test_get_config():
     pwd = "p"
     url = "url"
 
-    config = { "kernel_python_credentials": {user_ev: usr, pass_ev: pwd, url_ev: url} }
+    config = {"kernel_python_credentials": {user_ev: usr, pass_ev: pwd, url_ev: url}}
     conf.override(config)
 
     u, p, r = kernel._get_configuration()
@@ -93,7 +93,55 @@ def test_start_session():
 
     assert kernel.session_started
     assert call("%spark add TestKernel python {} skip".format(conn_str), True, False, None, False) \
-           in execute_cell_mock.mock_calls
+        in execute_cell_mock.mock_calls
+
+
+@with_setup(_setup(), _teardown())
+def test_delete_session():
+    kernel.session_started = True
+
+    kernel._delete_session()
+
+    assert not kernel.session_started
+    assert call("%spark cleanup", True, False) in execute_cell_mock.mock_calls
+
+
+@with_setup(_setup, _teardown)
+def test_set_config():
+    def _check(prepend, session_started=False, key_error_expected=False):
+        # Set up
+        properties = """{"extra": 2}"""
+        code = prepend + properties
+        kernel.session_started = session_started
+        execute_cell_mock.reset_mock()
+
+        # Call method
+        try:
+            kernel.do_execute(code, False)
+        except KeyError:
+            if not key_error_expected:
+                assert False
+
+            # When exception is expected, nothing to check
+            return
+
+        assert session_started == kernel.session_started
+        assert call("%%spark config {}".format(properties), False, True, None, False) \
+            in execute_cell_mock.mock_calls
+
+        if session_started and not key_error_expected:
+            # This means -f must be present, so check that a restart happened
+            assert call("%spark cleanup", True, False) in execute_cell_mock.mock_calls
+            assert call("%spark add TestKernel python {} skip".format(conn_str), True, False, None, False) \
+                in execute_cell_mock.mock_calls
+
+    _check("%config ")
+    _check("%config\n")
+    _check("%%config ")
+    _check("%%config\n")
+    _check("%config -f ")
+    _check("%config ", True, True)
+    _check("%config -f ", True, False)
 
 
 @with_setup(_setup, _teardown)
@@ -111,6 +159,17 @@ def test_do_execute_initializes_magics_if_not_run():
     assert call("%%spark\n{}".format(code), False, True, None, False) in execute_cell_mock.mock_calls
 
 
+@with_setup(_setup, _teardown)
+@raises(KeyError)
+def test_magic_not_supported():
+    # Set up
+    code = "%alex some spark code"
+    kernel.session_started = True
+
+    # Call method
+    kernel.do_execute(code, False)
+
+
 @with_setup(_setup, _teardown)
 def test_call_spark():
     # Set up

pip install -e .
parse_log_pytest
requirements.txt
pytest
apt-get update apt-get install -y gcc
3.7
requirements.txt
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
anyio==3.7.1
argon2-cffi==23.1.0
argon2-cffi-bindings==21.2.0
attrs==24.2.0
beautifulsoup4==4.13.3
bleach==6.0.0
certifi @ file:///croot/certifi_1671487769961/work/certifi
cffi==1.15.1
charset-normalizer==3.4.1
comm==0.1.4
decorator==5.1.1
defusedxml==0.7.1
entrypoints==0.4
exceptiongroup==1.2.2
fastjsonschema==2.21.1
idna==3.10
importlib-metadata==6.7.0
importlib-resources==5.12.0
iniconfig==2.0.0
ipykernel==4.1.1
ipython==4.0.0
ipython-genutils==0.2.0
ipywidgets==7.8.5
Jinja2==3.1.6
jsonschema==4.17.3
jupyter-server==1.24.0
jupyter_client==7.4.9
jupyter_core==4.12.0
jupyterlab-pygments==0.2.2
jupyterlab_widgets==1.1.11
MarkupSafe==2.1.5
mistune==3.0.2
mock==5.2.0
nbclassic==1.2.0
nbclient==0.7.4
nbconvert==7.6.0
nbformat==5.8.0
nest-asyncio==1.6.0
nose==1.3.7
notebook==6.5.7
notebook_shim==0.2.4
numpy==1.21.6
packaging==24.0
pandas==1.3.5
pandocfilters==1.5.1
pexpect==4.9.0
pickleshare==0.7.5
pkgutil_resolve_name==1.3.10
plotly==5.18.0
pluggy==1.2.0
prometheus-client==0.17.1
ptyprocess==0.7.0
pycparser==2.21
Pygments==2.17.2
pyrsistent==0.19.3
pytest==7.4.4
python-dateutil==2.9.0.post0
pytz==2025.2
pyzmq==26.2.1
-e git+https://github.com/jupyter-incubator/sparkmagic.git@5d7c9a29da1f4a3a12fc9cd821807b474625afc1#egg=remotespark
requests==2.31.0
Send2Trash==1.8.3
simplegeneric==0.8.1
six==1.17.0
sniffio==1.3.1
soupsieve==2.4.1
tenacity==8.2.3
terminado==0.17.1
tinycss2==1.2.1
tomli==2.0.1
tornado==6.2
traitlets==5.9.0
typing_extensions==4.7.1
urllib3==2.0.7
webencodings==0.5.1
websocket-client==1.6.1
widgetsnbextension==3.6.10
zipp==3.15.0

name: sparkmagic
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2022.12.7=py37h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.4.4=h6a678d5_1
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - pip=22.3.1=py37h06a4308_0
  - python=3.7.16=h7a1cb2a_0
  - readline=8.2=h5eee18b_0
  - setuptools=65.6.3=py37h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - wheel=0.38.4=py37h06a4308_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - anyio==3.7.1
      - argon2-cffi==23.1.0
      - argon2-cffi-bindings==21.2.0
      - attrs==24.2.0
      - beautifulsoup4==4.13.3
      - bleach==6.0.0
      - cffi==1.15.1
      - charset-normalizer==3.4.1
      - comm==0.1.4
      - decorator==5.1.1
      - defusedxml==0.7.1
      - entrypoints==0.4
      - exceptiongroup==1.2.2
      - fastjsonschema==2.21.1
      - idna==3.10
      - importlib-metadata==6.7.0
      - importlib-resources==5.12.0
      - iniconfig==2.0.0
      - ipykernel==4.1.1
      - ipython==4.0.0
      - ipython-genutils==0.2.0
      - ipywidgets==7.8.5
      - jinja2==3.1.6
      - jsonschema==4.17.3
      - jupyter-client==7.4.9
      - jupyter-core==4.12.0
      - jupyter-server==1.24.0
      - jupyterlab-pygments==0.2.2
      - jupyterlab-widgets==1.1.11
      - markupsafe==2.1.5
      - mistune==3.0.2
      - mock==5.2.0
      - nbclassic==1.2.0
      - nbclient==0.7.4
      - nbconvert==7.6.0
      - nbformat==5.8.0
      - nest-asyncio==1.6.0
      - nose==1.3.7
      - notebook==6.5.7
      - notebook-shim==0.2.4
      - numpy==1.21.6
      - packaging==24.0
      - pandas==1.3.5
      - pandocfilters==1.5.1
      - pexpect==4.9.0
      - pickleshare==0.7.5
      - pkgutil-resolve-name==1.3.10
      - plotly==5.18.0
      - pluggy==1.2.0
      - prometheus-client==0.17.1
      - ptyprocess==0.7.0
      - pycparser==2.21
      - pygments==2.17.2
      - pyrsistent==0.19.3
      - pytest==7.4.4
      - python-dateutil==2.9.0.post0
      - pytz==2025.2
      - pyzmq==26.2.1
      - requests==2.31.0
      - send2trash==1.8.3
      - simplegeneric==0.8.1
      - six==1.17.0
      - sniffio==1.3.1
      - soupsieve==2.4.1
      - tenacity==8.2.3
      - terminado==0.17.1
      - tinycss2==1.2.1
      - tomli==2.0.1
      - tornado==6.2
      - traitlets==5.9.0
      - typing-extensions==4.7.1
      - urllib3==2.0.7
      - webencodings==0.5.1
      - websocket-client==1.6.1
      - widgetsnbextension==3.6.10
      - zipp==3.15.0
prefix: /opt/conda/envs/sparkmagic

tests/test_sparkkernelbase.py::test_delete_session tests/test_sparkkernelbase.py::test_set_config tests/test_sparkkernelbase.py::test_magic_not_supported
tests/test_sparkkernelbase.py::test_get_config tests/test_sparkkernelbase.py::test_get_config_not_set tests/test_sparkkernelbase.py::test_initialize_magics tests/test_sparkkernelbase.py::test_start_session tests/test_sparkkernelbase.py::test_do_execute_initializes_magics_if_not_run tests/test_sparkkernelbase.py::test_call_spark tests/test_sparkkernelbase.py::test_execute_throws_if_fatal_error_happened tests/test_sparkkernelbase.py::test_execute_throws_if_fatal_error_happens_for_execution tests/test_sparkkernelbase.py::test_call_spark_sql_new_line tests/test_sparkkernelbase.py::test_call_spark_hive_new_line tests/test_sparkkernelbase.py::test_shutdown_cleans_up
remotespark/sparkkernelbase.py
remotespark/sparkkernelbase.py