diff --git a/qiskit_ibm_runtime/api/clients/runtime.py b/qiskit_ibm_runtime/api/clients/runtime.py
index ddbd41771..82b1d6814 100644
--- a/qiskit_ibm_runtime/api/clients/runtime.py
+++ b/qiskit_ibm_runtime/api/clients/runtime.py
@@ -230,6 +230,23 @@ def job_metadata(self, job_id: str) -> Dict[str, Any]:
         """
         return self._api.program_job(job_id).metadata()
 
+    def create_session(
+        self,
+        backend: Optional[str] = None,
+        instance: Optional[str] = None,
+        max_time: Optional[int] = None,
+        channel: Optional[str] = None,
+        mode: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Create a session.
+
+        Args:
+            mode: Execution mode.
+        """
+        return self._api.runtime_session(session_id=None).create(
+            backend, instance, max_time, channel, mode
+        )
+
     def cancel_session(self, session_id: str) -> None:
         """Close all jobs in the runtime session.
 
diff --git a/qiskit_ibm_runtime/api/rest/runtime.py b/qiskit_ibm_runtime/api/rest/runtime.py
index 354fc4e07..bde211df2 100644
--- a/qiskit_ibm_runtime/api/rest/runtime.py
+++ b/qiskit_ibm_runtime/api/rest/runtime.py
@@ -49,7 +49,7 @@ def program_job(self, job_id: str) -> "ProgramJob":
         """
         return ProgramJob(self.session, job_id)
 
-    def runtime_session(self, session_id: str) -> "RuntimeSession":
+    def runtime_session(self, session_id: str = None) -> "RuntimeSession":
         """Return an adapter for the session.
 
         Args:
diff --git a/qiskit_ibm_runtime/api/rest/runtime_session.py b/qiskit_ibm_runtime/api/rest/runtime_session.py
index fb4b3b944..87f66c076 100644
--- a/qiskit_ibm_runtime/api/rest/runtime_session.py
+++ b/qiskit_ibm_runtime/api/rest/runtime_session.py
@@ -12,7 +12,7 @@
 
 """Runtime Session REST adapter."""
 
-from typing import Dict, Any
+from typing import Dict, Any, Optional
 from .base import RestAdapterBase
 from ..session import RetrySession
 from ..exceptions import RequestsApiError
@@ -35,7 +35,34 @@ def __init__(self, session: RetrySession, session_id: str, url_prefix: str = "")
             session_id: Job ID of the first job in a runtime session.
             url_prefix: Prefix to use in the URL.
         """
-        super().__init__(session, "{}/sessions/{}".format(url_prefix, session_id))
+        if not session_id:
+            super().__init__(session, "{}/sessions".format(url_prefix))
+        else:
+            super().__init__(session, "{}/sessions/{}".format(url_prefix, session_id))
+
+    def create(
+        self,
+        backend: Optional[str] = None,
+        instance: Optional[str] = None,
+        max_time: Optional[int] = None,
+        channel: Optional[str] = None,
+        mode: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Create a session"""
+        url = self.get_url("self")
+        payload = {}
+        if mode:
+            payload["mode"] = mode
+        if backend:
+            payload["backend"] = backend
+        if instance:
+            payload["instance"] = instance
+        if max_time:
+            if channel == "ibm_quantum":
+                payload["max_session_ttl"] = max_time  # type: ignore[assignment]
+            else:
+                payload["max_ttl"] = max_time  # type: ignore[assignment]
+        return self.session.post(url, json=payload).json()
 
     def cancel(self) -> None:
         """Cancel all jobs in the session."""
diff --git a/qiskit_ibm_runtime/batch.py b/qiskit_ibm_runtime/batch.py
index 6947ddc81..18663ee09 100644
--- a/qiskit_ibm_runtime/batch.py
+++ b/qiskit_ibm_runtime/batch.py
@@ -12,10 +12,27 @@
 
 """Qiskit Runtime batch mode."""
 
+from typing import Optional, Union
+from qiskit_ibm_runtime import QiskitRuntimeService
+from .ibm_backend import IBMBackend
+
 from .session import Session
 
 
 class Batch(Session):
     """Class for creating a batch mode in Qiskit Runtime."""
 
-    pass
+    def __init__(
+        self,
+        service: Optional[QiskitRuntimeService] = None,
+        backend: Optional[Union[str, IBMBackend]] = None,
+        max_time: Optional[Union[int, str]] = None,
+    ):
+        super().__init__(service=service, backend=backend, max_time=max_time)
+
+    def _create_session(self) -> str:
+        """Create a session."""
+        session = self._service._api_client.create_session(
+            self._backend, self._instance, self._max_time, self._service.channel, "batch"
+        )
+        return session.get("id")
diff --git a/qiskit_ibm_runtime/runtime_job.py b/qiskit_ibm_runtime/runtime_job.py
index cac6221d9..21e9307a4 100644
--- a/qiskit_ibm_runtime/runtime_job.py
+++ b/qiskit_ibm_runtime/runtime_job.py
@@ -657,7 +657,7 @@ def session_id(self) -> str:
         """Session ID.
 
         Returns:
-            Job ID of the first job in a runtime session.
+            Session ID. None if the backend is a simulator.
         """
         if not self._session_id:
             response = self._api_client.job_get(job_id=self.job_id())
diff --git a/qiskit_ibm_runtime/session.py b/qiskit_ibm_runtime/session.py
index cb9da574d..1c2520059 100644
--- a/qiskit_ibm_runtime/session.py
+++ b/qiskit_ibm_runtime/session.py
@@ -15,7 +15,6 @@
 from typing import Dict, Optional, Type, Union, Callable, Any
 from types import TracebackType
 from functools import wraps
-from threading import Lock
 
 from qiskit_ibm_runtime import QiskitRuntimeService
 from .runtime_job import RuntimeJob
@@ -110,7 +109,6 @@ def __init__(
                     if QiskitRuntimeService.global_service is None
                     else QiskitRuntimeService.global_service
                 )
-
         else:
             self._service = service
 
@@ -118,13 +116,7 @@ def __init__(
             raise ValueError('"backend" is required for ``ibm_quantum`` channel.')
 
         self._instance = None
-        if isinstance(backend, IBMBackend):
-            self._instance = backend._instance
-            backend = backend.name
-        self._backend = backend
 
-        self._setup_lock = Lock()
-        self._session_id: Optional[str] = None
         self._active = True
         self._max_time = (
             max_time
@@ -132,6 +124,28 @@ def __init__(
             else hms_to_seconds(max_time, "Invalid max_time value: ")
         )
 
+        if isinstance(backend, IBMBackend):
+            self._instance = backend._instance
+            sim_backend = backend.configuration().simulator
+            backend = backend.name
+        else:
+            backend_obj = self._service.backend(backend)
+            self._instance = backend_obj._instance
+            sim_backend = backend_obj.configuration().simulator
+        self._backend = backend
+
+        if not sim_backend:
+            self._session_id = self._create_session()
+        else:
+            self._session_id = None
+
+    def _create_session(self) -> str:
+        """Create a session."""
+        session = self._service._api_client.create_session(
+            self._backend, self._instance, self._max_time, self._service.channel
+        )
+        return session.get("id")
+
     @_active_session
     def run(
         self,
@@ -162,29 +176,15 @@ def run(
 
         options["backend"] = self._backend
 
-        if not self._session_id:
-            # Make sure only one thread can send the session starter job.
-            self._setup_lock.acquire()
-            # TODO: What happens if session max time != first job max time?
-            # Use session max time if this is first job.
-            options["session_time"] = self._max_time
-
-        try:
-            job = self._service.run(
-                program_id=program_id,
-                options=options,
-                inputs=inputs,
-                session_id=self._session_id,
-                start_session=self._session_id is None,
-                callback=callback,
-                result_decoder=result_decoder,
-            )
-
-            if self._session_id is None:
-                self._session_id = job.job_id()
-        finally:
-            if self._setup_lock.locked():
-                self._setup_lock.release()
+        job = self._service.run(
+            program_id=program_id,
+            options=options,
+            inputs=inputs,
+            session_id=self._session_id,
+            start_session=False,
+            callback=callback,
+            result_decoder=result_decoder,
+        )
 
         if self._backend is None:
             self._backend = job.backend().name
@@ -278,11 +278,11 @@ def details(self) -> Optional[Dict[str, Any]]:
         return None
 
     @property
-    def session_id(self) -> str:
+    def session_id(self) -> Optional[str]:
         """Return the session ID.
 
         Returns:
-            Session ID. None until a job is submitted.
+            Session ID. None if the backend is a simulator.
         """
         return self._session_id
 
diff --git a/qiskit_ibm_runtime/utils/json.py b/qiskit_ibm_runtime/utils/json.py
index 6b1468bea..f86db4261 100644
--- a/qiskit_ibm_runtime/utils/json.py
+++ b/qiskit_ibm_runtime/utils/json.py
@@ -215,7 +215,7 @@ def default(self, obj: Any) -> Any:  # pylint: disable=arguments-differ
         if hasattr(obj, "to_json"):
             return {"__type__": "to_json", "__value__": obj.to_json()}
         if isinstance(obj, QuantumCircuit):
-            kwargs: dict[str, object] = {"use_symengine": bool(optionals.HAS_SYMENGINE)}
+            kwargs: Dict[str, object] = {"use_symengine": bool(optionals.HAS_SYMENGINE)}
             if _TERRA_VERSION[0] >= 1:
                 # NOTE: This can be updated only after the server side has
                 # updated to a newer qiskit version.
diff --git a/releasenotes/notes/session-modes-5c22b68620f8d690.yaml b/releasenotes/notes/session-modes-5c22b68620f8d690.yaml
new file mode 100644
index 000000000..b9e887fe6
--- /dev/null
+++ b/releasenotes/notes/session-modes-5c22b68620f8d690.yaml
@@ -0,0 +1,10 @@
+---
+features:
+  - |
+    Sessions will now be started with a new ``/sessions`` endpoint that allows for different 
+    execution modes. Batch mode is now supported through :class:`~qiskit_ibm_runtime.Batch`, and 
+    :class:`~qiskit_ibm_runtime.Session` will work the same as way as before. 
+    Please see https://docs.quantum.ibm.com/run/sessions for more information.
+
+    Note that ``Session`` and ``Batch`` created from ``qiskit-ibm-runtime`` prior to this release will no longer be 
+    supported after March 31, 2024. Please update your ``qiskit-ibm-runtime`` version as soon as possible before this date. 
diff --git a/test/integration/test_session.py b/test/integration/test_session.py
index 15e4a92f5..674537a2e 100644
--- a/test/integration/test_session.py
+++ b/test/integration/test_session.py
@@ -131,22 +131,20 @@ def test_backend_run_with_session(self):
         )
 
     def test_backend_and_primitive_in_session(self):
-        """Test Sampler.run and backend.run in the same session."""
+        """Test using simulator does not start a session."""
         backend = self.service.get_backend("ibmq_qasm_simulator")
         with Session(backend=backend) as session:
             sampler = Sampler(session=session)
             job1 = sampler.run(circuits=bell())
             with warnings.catch_warnings(record=True):
                 job2 = backend.run(circuits=bell())
-            self.assertEqual(job1.session_id, job1.job_id())
+            self.assertIsNone(job1.session_id)
             self.assertIsNone(job2.session_id)
         with backend.open_session() as session:
             with warnings.catch_warnings(record=True):
                 sampler = Sampler(backend=backend)
             job1 = backend.run(bell())
             job2 = sampler.run(circuits=bell())
-            session_id = session.session_id
-            self.assertEqual(session_id, job1.job_id())
             self.assertIsNone(job2.session_id)
 
     def test_session_cancel(self):
diff --git a/test/unit/mock/fake_runtime_client.py b/test/unit/mock/fake_runtime_client.py
index 8c32fa985..4cf3e101c 100644
--- a/test/unit/mock/fake_runtime_client.py
+++ b/test/unit/mock/fake_runtime_client.py
@@ -285,6 +285,22 @@ def set_job_classes(self, classes):
             classes = [classes]
         self._job_classes = classes
 
+    # pylint: disable=unused-argument
+    def create_session(
+        self,
+        backend: Optional[str] = None,
+        instance: Optional[str] = None,
+        max_time: Optional[int] = None,
+        channel: Optional[str] = None,
+        mode: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Create a session."""
+        return {"id": uuid.uuid4().hex}
+
+    def close_session(self, session_id: str) -> None:
+        """Close a session."""
+        pass
+
     def is_qctrl_enabled(self):
         """Return whether or not channel_strategy q-ctrl is enabled."""
         return False
diff --git a/test/unit/test_batch.py b/test/unit/test_batch.py
index 3b54b1094..d5fb053a2 100644
--- a/test/unit/test_batch.py
+++ b/test/unit/test_batch.py
@@ -12,25 +12,48 @@
 
 """Tests for Batch class."""
 
-from unittest.mock import patch
+from unittest.mock import MagicMock
 
 from qiskit_ibm_runtime import Batch
+from qiskit_ibm_runtime.ibm_backend import IBMBackend
 from qiskit_ibm_runtime.utils.default_session import _DEFAULT_SESSION
 
 from ..ibm_test_case import IBMTestCase
 
 
 class TestBatch(IBMTestCase):
-    """Class for testing the Session class."""
+    """Class for testing the Batch class."""
 
     def tearDown(self) -> None:
         super().tearDown()
         _DEFAULT_SESSION.set(None)
 
-    @patch("qiskit_ibm_runtime.session.QiskitRuntimeService", autospec=True)
-    def test_default_batch(self, mock_service):
-        """Test using default batch mode."""
-        mock_service.global_service = None
-        batch = Batch(backend="ibm_gotham")
-        self.assertIsNotNone(batch.service)
-        mock_service.assert_called_once()
+    def test_passing_ibm_backend(self):
+        """Test passing in IBMBackend instance."""
+        backend = MagicMock(spec=IBMBackend)
+        backend._instance = None
+        backend.name = "ibm_gotham"
+        session = Batch(service=MagicMock(), backend=backend)
+        self.assertEqual(session.backend(), "ibm_gotham")
+
+    def test_using_ibm_backend_service(self):
+        """Test using service from an IBMBackend instance."""
+        backend = MagicMock(spec=IBMBackend)
+        backend._instance = None
+        backend.name = "ibm_gotham"
+        session = Batch(backend=backend)
+        self.assertEqual(session.service, backend.service)
+
+    def test_run_after_close(self):
+        """Test running after session is closed."""
+        session = Batch(service=MagicMock(), backend="ibm_gotham")
+        session.cancel()
+        with self.assertRaises(RuntimeError):
+            session.run(program_id="program_id", inputs={})
+
+    def test_context_manager(self):
+        """Test session as a context manager."""
+        with Batch(service=MagicMock(), backend="ibm_gotham") as session:
+            session.run(program_id="foo", inputs={})
+            session.cancel()
+        self.assertFalse(session._active)
diff --git a/test/unit/test_runtime_ws.py b/test/unit/test_runtime_ws.py
index 3dc232a1b..0b9dab26f 100644
--- a/test/unit/test_runtime_ws.py
+++ b/test/unit/test_runtime_ws.py
@@ -91,6 +91,7 @@ def _patched_run(callback, *args, **kwargs):  # pylint: disable=unused-argument
         service = MagicMock(spec=QiskitRuntimeService)
         service.run = _patched_run
         service._channel_strategy = None
+        service._api_client = MagicMock()
 
         circ = bell()
         obs = SparsePauliOp.from_list([("IZ", 1)])
diff --git a/test/unit/test_session.py b/test/unit/test_session.py
index 5e228acbb..b2b84fbd2 100644
--- a/test/unit/test_session.py
+++ b/test/unit/test_session.py
@@ -12,11 +12,7 @@
 
 """Tests for Session classession."""
 
-import sys
-import time
-from concurrent.futures import ThreadPoolExecutor, wait
-
-from unittest.mock import MagicMock, Mock, patch
+from unittest.mock import MagicMock, patch
 
 from qiskit_ibm_runtime.fake_provider import FakeManila
 from qiskit_ibm_runtime import Session
@@ -33,7 +29,7 @@ def tearDown(self) -> None:
         super().tearDown()
         _DEFAULT_SESSION.set(None)
 
-    @patch("qiskit_ibm_runtime.session.QiskitRuntimeService", autospec=True)
+    @patch("qiskit_ibm_runtime.session.QiskitRuntimeService")
     def test_default_service(self, mock_service):
         """Test using default service."""
         mock_service.global_service = None
@@ -107,58 +103,19 @@ def test_run(self):
         decoder = MagicMock()
         max_time = 42
         session = Session(service=service, backend=backend, max_time=max_time)
-        session_ids = [None, job.job_id()]
-        start_sessions = [True, False]
-
-        for idx in range(2):
-            session.run(
-                program_id=program_id,
-                inputs=inputs,
-                options=options,
-                result_decoder=decoder,
-            )
-            _, kwargs = service.run.call_args
-            self.assertEqual(kwargs["program_id"], program_id)
-            self.assertDictEqual(kwargs["options"], {"backend": backend, **options})
-            self.assertTrue({"session_time": 42}.items() <= kwargs["options"].items())
-            self.assertDictEqual(kwargs["inputs"], inputs)
-            self.assertEqual(kwargs["session_id"], session_ids[idx])
-            self.assertEqual(kwargs["start_session"], start_sessions[idx])
-            self.assertEqual(kwargs["result_decoder"], decoder)
-            self.assertEqual(session.session_id, job.job_id())
-            self.assertEqual(session.backend(), backend)
-
-    def test_run_is_thread_safe(self):
-        """Test the session sends a session starter job once, and only once."""
-        service = MagicMock()
-        api = MagicMock()
-        service._api_client = api
-
-        def _wait_a_bit(*args, **kwargs):
-            # pylint: disable=unused-argument
-            switchinterval = sys.getswitchinterval()
-            time.sleep(switchinterval * 2)
-            return MagicMock()
-
-        service.run = Mock(side_effect=_wait_a_bit)
 
-        session = Session(service=service, backend="ibm_gotham")
-        with ThreadPoolExecutor(max_workers=2) as executor:
-            results = list(map(lambda _: executor.submit(session.run, "", {}), range(5)))
-            wait(results)
-
-        calls = service.run.call_args_list
-        session_starters = list(filter(lambda c: c.kwargs["start_session"] is True, calls))
-        self.assertEqual(len(session_starters), 1)
-
-    def test_close_without_run(self):
-        """Test closing without run."""
-        service = MagicMock()
-        api = MagicMock()
-        service._api_client = api
-        session = Session(service=service, backend="ibm_gotham")
-        session.close()
-        api.close_session.assert_not_called()
+        session.run(
+            program_id=program_id,
+            inputs=inputs,
+            options=options,
+            result_decoder=decoder,
+        )
+        _, kwargs = service.run.call_args
+        self.assertEqual(kwargs["program_id"], program_id)
+        self.assertDictEqual(kwargs["options"], {"backend": backend, **options})
+        self.assertDictEqual(kwargs["inputs"], inputs)
+        self.assertEqual(kwargs["result_decoder"], decoder)
+        self.assertEqual(session.backend(), backend)
 
     def test_context_manager(self):
         """Test session as a context manager."""
@@ -182,14 +139,14 @@ def test_default_backend(self):
     def test_global_service(self):
         """Test that global service is used in Session"""
         _ = FakeRuntimeService(channel="ibm_quantum", token="abc")
-        session = Session(backend="ibmq_qasm_simulator")
+        session = Session(backend="common_backend")
         self.assertTrue(isinstance(session._service, FakeRuntimeService))
         self.assertEqual(session._service._account.token, "abc")
         _ = FakeRuntimeService(channel="ibm_quantum", token="xyz")
-        session = Session(backend="ibmq_qasm_simulator")
+        session = Session(backend="common_backend")
         self.assertEqual(session._service._account.token, "xyz")
         with Session(
-            service=FakeRuntimeService(channel="ibm_quantum", token="uvw"), backend="ibm_gotham"
+            service=FakeRuntimeService(channel="ibm_quantum", token="uvw"), backend="common_backend"
         ) as session:
             self.assertEqual(session._service._account.token, "uvw")