Repository: Qiskit/qiskit-ibm-runtime
Pull request #453 (2022-08-01 22:15:09)

Flexible session
<!--
⚠️ The pull request title should be short detailed and understandable for all.
⚠️ Also, please add a release note file using reno if the change needs to be
  documented in the release notes.
⚠️ If your pull request fixes an open issue, please link to the issue.

✅ I have added the tests to cover my changes.
✅ I have updated the documentation accordingly.
✅ I have read the CONTRIBUTING document.
-->

### Summary

Prototype of a new `Session` that allows multiple calls to the same or different primitives. Had to create a new class to avoid circular import. Does not do data caching.


### Details and comments
Fixes #



coveralls: ## Pull Request Test Coverage Report for [Build 2778346655](https://coveralls.io/builds/51331118)

* **-40** of **184**   **(78.26%)**  changed or added relevant lines in **11** files are covered.
* **1** unchanged line in **1** file lost coverage.
* Overall coverage decreased (**-1.8%**) to **60.6%**

---

|  Changes Missing Coverage | Covered Lines | Changed/Added Lines | % |
| :-----|--------------|--------|---: |
| [qiskit_ibm_runtime/runtime_program.py](https://coveralls.io/builds/51331118/source?filename=qiskit_ibm_runtime%2Fruntime_program.py#L443) | 1 | 2 | 50.0%
| [qiskit_ibm_runtime/utils/deprecation.py](https://coveralls.io/builds/51331118/source?filename=qiskit_ibm_runtime%2Futils%2Fdeprecation.py#L37) | 14 | 16 | 87.5%
| [qiskit_ibm_runtime/runtime_options.py](https://coveralls.io/builds/51331118/source?filename=qiskit_ibm_runtime%2Fruntime_options.py#L57) | 18 | 24 | 75.0%
| [qiskit_ibm_runtime/session.py](https://coveralls.io/builds/51331118/source?filename=qiskit_ibm_runtime%2Fsession.py#L35) | 39 | 51 | 76.47%
| [qiskit_ibm_runtime/sampler.py](https://coveralls.io/builds/51331118/source?filename=qiskit_ibm_runtime%2Fsampler.py#L169) | 49 | 68 | 72.06%
<!-- | **Total:** | **144** | **184** | **78.26%** | -->

|  Files with Coverage Reduction | New Missed Lines | % |
| :-----|--------------|--: |
| [qiskit_ibm_runtime/sampler.py](https://coveralls.io/builds/51331118/source?filename=qiskit_ibm_runtime%2Fsampler.py#L290) | 1 | 72.09% |
<!-- | **Total:** | **1** |  | -->

|  Totals | [![Coverage Status](https://coveralls.io/builds/51331118/badge)](https://coveralls.io/builds/51331118) |
| :-- | --: |
| Change from base [Build 2767598743](https://coveralls.io/builds/51288372): |  -1.8% |
| Covered Lines: | 2867 |
| Relevant Lines: | 4731 |

---
##### 💛  - [Coveralls](https://coveralls.io)

renier: @tmittal947 could you take a look at the latest comments here?
review-notebook-app[bot]: Check out this pull request on&nbsp; <a href="https://app.reviewnb.com/Qiskit/qiskit-ibm-runtime/pull/453"><img align="absmiddle"  alt="ReviewNB" height="28" class="BotMessageButtonImage" src="https://raw.githubusercontent.com/ReviewNB/support/master/images/button_reviewnb.png"/></a> 

 See visual diffs & provide feedback on Jupyter Notebooks. 

---

 <i>Powered by <a href='https://www.reviewnb.com/?utm_source=gh'>ReviewNB</a></i>

Patch:
diff --git a/docs/tutorials/how-to-getting-started-with-estimator.ipynb b/docs/tutorials/how-to-getting-started-with-estimator.ipynb
index 4cdda204..0c0eaccf 100644
--- a/docs/tutorials/how-to-getting-started-with-estimator.ipynb
+++ b/docs/tutorials/how-to-getting-started-with-estimator.ipynb
@@ -28,22 +28,25 @@
     "\n",
     "With Qiskit Runtime primitives, we introduce the concept of a session or a factory that allows you to define a job as a collection of iterative calls to the quantum computer. When you start a session, it caches the data you send so it doesn't have to be transmitted to the Quantum Datacenter on each iteration.\n",
     "\n",
+    "### Create an Estimator instance\n",
     "\n",
-    "### Specify program inputs\n",
+    "The `Estimator` class takes in a `session` object that represents the session to use, and an `options` variable to control the execution environment.\n",
     "\n",
-    "The Estimator takes in the following arguments:\n",
+    "`options` can be either a dictionary or a `qiskit_ibm_runtime.Options` class instance. Initializing it as an `Options` class allows you to do auto-complete. \n",
     "\n",
-    "- **circuits**: a list of (parameterized) circuits that you want to investigate.\n",
-    "- **observables**: a list of observables to measure the expectation values.\n",
-    "- **parameters**: a list of parameters for the parameterized circuits. It should be omitted if the circuits provided are not parameterized.\n",
-    "- **skip_transpilation**: circuit transpilation is skipped if set to `True`. Default value is `False`.\n",
-    "- **service**: the `QiskitRuntimeService` instance to run the program on. If not specified, the default saved account for `QiskitRuntimeService` is initialized.\n",
-    "- **options**: Runtime options dictionary that control the execution environment.\n",
-    "  - **backend**: The backend to run on. This option is required if you are running on [IBM Quantum](https://quantum-computing.ibm.com/). However, if you are running on [IBM Cloud](https://cloud.ibm.com/quantum), you can choose not to specify the backend, in which case the least busy backend is used.\n",
-    "  \n",
-    "You can find more details in [the Estimator API reference](https://qiskit.org/documentation/partners/qiskit_ibm_runtime/stubs/qiskit_ibm_runtime.Estimator.html).\n",
+    "Some of the settings you can specify using `options`:\n",
+    "\n",
+    "* **backend**: Optional string name of backend. If not specified a backend will be selected\n",
+    "    automatically (IBM Cloud only).\n",
+    "* **log_level**: logging level to set in the execution environment. The valid\n",
+    "        log levels are: `DEBUG`, `INFO`, `WARNING`, `ERROR`, and `CRITICAL`.\n",
+    "        The default level is `WARNING`.\n",
+    "* **optimization_level**: How much optimization to perform on the circuits. The default is 1.\n",
+    "* **resilience_level**: How much resilience to build against errors. Higher levels generate more accurate results, at the expense of longer processing times.\n",
     "\n",
-    "Example:"
+    "You can find more details in [the Options API reference](https://qiskit.org/documentation/partners/qiskit_ibm_runtime/stubs/qiskit_ibm_runtime.Options.html).\n",
+    "\n",
+    "With these arguments, you can create a `Estimator` instance with the desired options. Example:"
    ]
   },
   {
@@ -53,18 +56,15 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "from qiskit_ibm_runtime import QiskitRuntimeService, Estimator\n",
+    "from qiskit_ibm_runtime import QiskitRuntimeService, Session, Estimator, Options\n",
     "from qiskit.circuit.library import RealAmplitudes\n",
     "from qiskit.quantum_info import SparsePauliOp\n",
     "\n",
     "service = QiskitRuntimeService()\n",
+    "options = Options(backend=\"ibmq_qasm_simulator\", optimization_level=1)\n",
     "\n",
-    "psi1 = RealAmplitudes(num_qubits=2, reps=2)\n",
-    "psi2 = RealAmplitudes(num_qubits=2, reps=3)\n",
-    "\n",
-    "H1 = SparsePauliOp.from_list([(\"II\", 1), (\"IZ\", 2), (\"XI\", 3)])\n",
-    "H2 = SparsePauliOp.from_list([(\"IZ\", 1)])\n",
-    "H3 = SparsePauliOp.from_list([(\"ZI\", 1), (\"ZZ\", 1)])"
+    "with Session(service) as session:\n",
+    "    estimator = Estimator(session=session, options=options)"
    ]
   },
   {
@@ -72,27 +72,37 @@
    "id": "c8803a0e",
    "metadata": {},
    "source": [
-    "## Write to & read from a session\n",
+    "## Invoke the Sampler primitive\n",
+    "\n",
+    "You can invoke the `Estimator` primitive program using the `run()` method of the `Estimator` instance you just created. The method returns a `IBMRuntimeJob` instance which you can use to query for things like job ID and status. The `result()` method of the job will return the result. \n",
+    "\n",
+    "You can invoke the `run()` method multiple times with the different inputs within a session.\n",
+    "\n",
+    "### Specify program inputs\n",
     "\n",
-    "Running a job and returning the results are done by writing to and reading from the session. After the results are returned, the session is automatically closed.\n",
+    "The `Estimator.run()` method takes in the following arguments:\n",
     "\n",
+    "- **circuits**: A list of (parameterized) circuits that you want to investigate.\n",
+    "- **observables**: A list of observables to measure the expectation values.\n",
+    "- **parameters**: A list of parameters for the parameterized circuits. It should be omitted if the circuits provided are not parameterized.\n",
+    "- **parameter_values**: an optional list of concrete parameters to be bound.\n",
+    "- **kwargs**: Additional options to overwrite the default values. \n",
+    "\n",
+    "You can find more details in [the Estimator API reference](https://qiskit.org/documentation/partners/qiskit_ibm_runtime/stubs/qiskit_ibm_runtime.Estimator.html).\n",
     "\n",
     "### Run the job & print results\n",
     "\n",
     "Run the job, specifying your previously defined inputs and options.  Use `circuits`, `observables`, and `parameter_values` to use a specific parameter and observable with the specified circuit.\n",
     "\n",
-    "For example, this line `psi1_H23_result = estimator(circuits=[0, 0], observables=[1, 2], parameter_values=[theta1]*2)` specifies the following:\n",
+    "For example, this line `psi1_H23_result = estimator.run(circuits=[psi1, psi1], observables=[H2, H3], parameter_values=[theta1]*2).result()` specifies the following:\n",
     "\n",
     "- Return the value for using observable `H2` and parameter `theta1` with the circuit `psi1`.\n",
-    "- Return the value for using observable `H3` and parameter `theta1` with the circuit `psi1`.\n",
-    "\n",
-    "You can also pass circuits and observables as objects.\n",
-    "For example, `psi1_H23_result = estimator(circuits=[ps1, ps1], observables=[H2, H3], parameter_values=[theta1]*2)`"
+    "- Return the value for using observable `H3` and parameter `theta1` with the circuit `psi1`.\n"
    ]
   },
   {
    "cell_type": "code",
-   "execution_count": 2,
+   "execution_count": null,
    "id": "786b8037",
    "metadata": {},
    "outputs": [
@@ -100,49 +110,41 @@
      "name": "stdout",
      "output_type": "stream",
      "text": [
-      "EstimatorResult(values=array([1.515625]), metadata=[{'variance': 9.6634521484375, 'shots': 1024}])\n",
-      "EstimatorResult(values=array([-0.52929688,  0.08984375]), metadata=[{'variance': 0.7198448181152344, 'shots': 1024}, {'variance': 1.9958419799804688, 'shots': 1024}])\n",
-      "EstimatorResult(values=array([0.16992188]), metadata=[{'variance': 0.9711265563964844, 'shots': 1024}])\n",
-      "EstimatorResult(values=array([1.58203125, 1.15234375]), metadata=[{'variance': 8.664047241210938, 'shots': 1024}, {'variance': 12.394515991210938, 'shots': 1024}])\n",
-      "EstimatorResult(values=array([ 1.5625    ,  0.17578125, -1.0859375 ]), metadata=[{'variance': 8.88671875, 'shots': 1024}, {'variance': 0.9691009521484375, 'shots': 1024}, {'variance': 1.24774169921875, 'shots': 1024}])\n"
+      "EstimatorResult(values=array([1.59570312]), metadata=[{'variance': 8.661632537841797, 'shots': 1024}])\n",
+      "EstimatorResult(values=array([-0.5390625 ,  0.07421875]), metadata=[{'variance': 0.70941162109375, 'shots': 1024}, {'variance': 1.9894332885742188, 'shots': 1024}])\n",
+      "EstimatorResult(values=array([0.17773438]), metadata=[{'variance': 0.9684104919433594, 'shots': 1024}])\n"
      ]
     }
    ],
    "source": [
-    "with Estimator(\n",
-    "    circuits=[psi1, psi2],\n",
-    "    observables=[H1, H2, H3],\n",
-    "    service=service, \n",
-    "    options={ \"backend\": \"ibmq_qasm_simulator\" }\n",
-    ") as estimator:\n",
+    "with Session(service) as session:\n",
+    "    estimator = Estimator(session=session, options=options)\n",
+    "\n",
     "    theta1 = [0, 1, 1, 2, 3, 5]\n",
     "    theta2 = [0, 1, 1, 2, 3, 5, 8, 13]\n",
     "    theta3 = [1, 2, 3, 4, 5, 6]\n",
     "\n",
     "    # calculate [ <psi1(theta1)|H1|psi1(theta1)> ]\n",
-    "    # pass indices of circuits and observables\n",
-    "    psi1_H1_result = estimator(circuits=[0], observables=[0], parameter_values=[theta1])\n",
-    "    print(psi1_H1_result)\n",
+    "    psi1_H1 = estimator.run(circuits=[psi1], observables=[H1], parameter_values=[theta1])\n",
+    "    print(psi1_H1.result())\n",
     "\n",
     "    # calculate [ <psi1(theta1)|H2|psi1(theta1)>, <psi1(theta1)|H3|psi1(theta1)> ]\n",
-    "    # alternatively you can also pass circuits and observables as objects\n",
-    "    psi1_H23_result = estimator(circuits=[psi1, psi1], observables=[H2, H3], parameter_values=[theta1]*2)\n",
-    "    print(psi1_H23_result)\n",
+    "    psi1_H23 = estimator.run(circuits=[psi1, psi1], observables=[H2, H3], parameter_values=[theta1]*2)\n",
+    "    print(psi1_H23.result())\n",
     "\n",
     "    # calculate [ <psi2(theta2)|H2|psi2(theta2)> ]\n",
-    "    # Note that you don't need to specify the labels \"circuits\", \"observables\", or \"parameter_values\", as long as they are specified in that order.\n",
-    "    psi2_H2_result = estimator([1], [1], [theta2])\n",
-    "    print(psi2_H2_result)\n",
+    "    psi2_H2 = estimator.run(circuits=[psi2], observables=[H2], parameter_values=[theta2])\n",
+    "    print(psi2_H2.result())\n",
     "\n",
     "    # calculate [ <psi1(theta1)|H1|psi1(theta1)>, <psi1(theta3)|H1|psi1(theta3)> ]\n",
-    "    psi1_H1_result2 = estimator([0, 0], [0, 0], [theta1, theta3])\n",
-    "    print(psi1_H1_result2)\n",
+    "    psi1_H1_job = estimator.run(circuits=[psi1, psi1], observables=[H1, H1], parameter_values=[theta1, theta3])\n",
+    "    print(psi1_H1_job.result())\n",
     "\n",
     "    # calculate [ <psi1(theta1)|H1|psi1(theta1)>,\n",
     "    #             <psi2(theta2)|H2|psi2(theta2)>,\n",
     "    #             <psi1(theta3)|H3|psi1(theta3)> ]\n",
-    "    psi12_H23_result = estimator([0, 1, 0], [0, 1, 2], [theta1, theta2, theta3])\n",
-    "    print(psi12_H23_result)"
+    "    psi12_H23 = estimator.run(circuits=[psi1, psi2, psi1], observables=[H1, H2, H3], parameter_values=[theta1, theta2, theta3])\n",
+    "    print(psi12_H23.result())"
    ]
   },
   {
@@ -150,8 +152,16 @@
    "id": "4baad90e",
    "metadata": {},
    "source": [
-    "The results align with the parameter - circuit - observable tuples specified previously.  For example, the first result: `EstimatorResult(values=array([1.55273438]), metadata=[{'variance': 8.897655487060547, 'shots': 1024}])` is the output of the parameter labeled `theta1` and observable `H1` being sent to the first circuit."
+    "The results align with the parameter - circuit - observable tuples specified previously.  For example, the first result: `EstimatorResult(values=array([1.53710938]), metadata=[{'variance': 9.249675750732422, 'shots': 1024}])` is the output of the parameter labeled `theta1` and observable `H1` being sent to the first circuit."
    ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "a0a4eec1",
+   "metadata": {},
+   "outputs": [],
+   "source": []
   }
  ],
  "metadata": {
@@ -170,7 +180,7 @@
    "name": "python",
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
-   "version": "3.8.8"
+   "version": "3.9.12"
   }
  },
  "nbformat": 4,
diff --git a/docs/tutorials/how-to-getting-started-with-sampler.ipynb b/docs/tutorials/how-to-getting-started-with-sampler.ipynb
index 20cebd9e..036d9e5d 100644
--- a/docs/tutorials/how-to-getting-started-with-sampler.ipynb
+++ b/docs/tutorials/how-to-getting-started-with-sampler.ipynb
@@ -26,20 +26,26 @@
     "With Qiskit Runtime primitives, we introduce the concept of a session or a factory that allows you to define a job as a collection of iterative calls to the quantum computer. When you start a session, it caches the data you send so it doesn't have to be transmitted to the Quantum Datacenter on each iteration.\n",
     "\n",
     "\n",
-    "### Specify program inputs\n",
+    "### Create a Sampler instance\n",
     "\n",
-    "The Sampler takes in the following arguments:\n",
+    "The `Sampler` class takes in a `session` object that represents the session to use, and an `options` variable to control the execution environment.\n",
     "\n",
-    "- **circuits**: a list of (parameterized) circuits that you want to investigate.\n",
-    "- **parameters**: a list of parameters for the parameterized circuits. It should be omitted if the circuits provided are not parameterized.\n",
-    "- **skip_transpilation**: circuit transpilation is skipped if set to `True`. Default value is `False`.\n",
-    "- **service**: the `QiskitRuntimeService` instance to run the program on. If not specified, the default saved account for `QiskitRuntimeService` is initialized.\n",
-    "- **options**: Runtime options dictionary that control the execution environment.\n",
-    "  - **backend**: The backend to run on. This option is required if you are running on [IBM Quantum](https://quantum-computing.ibm.com/). However, if you are running on [IBM Cloud](https://cloud.ibm.com/quantum), you can choose not to specify the backend, in which case the least busy backend is used.\n",
-    "  \n",
-    "You can find more details in [the Sampler API reference](https://qiskit.org/documentation/partners/qiskit_ibm_runtime/stubs/qiskit_ibm_runtime.Sampler.html).\n",
+    "`options` can be either a dictionary or a `qiskit_ibm_runtime.Options` class instance. Initializing it as an `Options` class allows you to do auto-complete. \n",
     "\n",
-    "Example:"
+    "Some of the settings you can specify using `options`:\n",
+    "\n",
+    "* **backend**: Optional string name of backend. If not specified a backend will be selected\n",
+    "    automatically (IBM Cloud only).\n",
+    "* **log_level**: logging level to set in the execution environment. The valid\n",
+    "        log levels are: `DEBUG`, `INFO`, `WARNING`, `ERROR`, and `CRITICAL`.\n",
+    "        The default level is `WARNING`.\n",
+    "* **optimization_level**: How much optimization to perform on the circuits. The default is 1.\n",
+    "* **resilience_level**: How much resilience to build against errors. Higher levels generate more accurate results, at the expense of longer processing times.\n",
+    "\n",
+    "You can find more details in [the Options API reference](https://qiskit.org/documentation/partners/qiskit_ibm_runtime/stubs/qiskit_ibm_runtime.Options.html).\n",
+    "\n",
+    "\n",
+    "With these arguments, you can create a `Sampler` instance with the desired options. Example:"
    ]
   },
   {
@@ -47,17 +53,69 @@
    "execution_count": 1,
    "id": "ff4e5c44",
    "metadata": {},
-   "outputs": [],
+   "outputs": [
+    {
+     "ename": "ConnectionError",
+     "evalue": "HTTPSConnectionPool(host='iam.auth.quantum-computing.ibm.com', port=443): Max retries exceeded with url: /identity/token (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x1259bc6a0>: Failed to establish a new connection: [Errno 8] nodename nor servname provided, or not known'))",
+     "output_type": "error",
+     "traceback": [
+      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
+      "\u001b[0;31mgaierror\u001b[0m                                  Traceback (most recent call last)",
+      "\u001b[0;32m~/.pyenv/versions/3.8.1/envs/qiskit-ibm-runtime/lib/python3.8/site-packages/urllib3/connection.py\u001b[0m in \u001b[0;36m_new_conn\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    173\u001b[0m         \u001b[0;32mtry\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 174\u001b[0;31m             conn = connection.create_connection(\n\u001b[0m\u001b[1;32m    175\u001b[0m                 \u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_dns_host\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mport\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtimeout\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m**\u001b[0m\u001b[0mextra_kw\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;32m~/.pyenv/versions/3.8.1/envs/qiskit-ibm-runtime/lib/python3.8/site-packages/urllib3/util/connection.py\u001b[0m in \u001b[0;36mcreate_connection\u001b[0;34m(address, timeout, source_address, socket_options)\u001b[0m\n\u001b[1;32m     72\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 73\u001b[0;31m     \u001b[0;32mfor\u001b[0m \u001b[0mres\u001b[0m \u001b[0;32min\u001b[0m \u001b[0msocket\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mgetaddrinfo\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mhost\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mport\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mfamily\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0msocket\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mSOCK_STREAM\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m     74\u001b[0m         \u001b[0maf\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0msocktype\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mproto\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mcanonname\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0msa\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mres\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;32m~/.pyenv/versions/3.8.1/lib/python3.8/socket.py\u001b[0m in \u001b[0;36mgetaddrinfo\u001b[0;34m(host, port, family, type, proto, flags)\u001b[0m\n\u001b[1;32m    917\u001b[0m     \u001b[0maddrlist\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;34m[\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 918\u001b[0;31m     \u001b[0;32mfor\u001b[0m \u001b[0mres\u001b[0m \u001b[0;32min\u001b[0m \u001b[0m_socket\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mgetaddrinfo\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mhost\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mport\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mfamily\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mtype\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mproto\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mflags\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    919\u001b[0m         \u001b[0maf\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0msocktype\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mproto\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mcanonname\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0msa\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mres\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;31mgaierror\u001b[0m: [Errno 8] nodename nor servname provided, or not known",
+      "\nDuring handling of the above exception, another exception occurred:\n",
+      "\u001b[0;31mNewConnectionError\u001b[0m                        Traceback (most recent call last)",
+      "\u001b[0;32m~/.pyenv/versions/3.8.1/envs/qiskit-ibm-runtime/lib/python3.8/site-packages/urllib3/connectionpool.py\u001b[0m in \u001b[0;36murlopen\u001b[0;34m(self, method, url, body, headers, retries, redirect, assert_same_host, timeout, pool_timeout, release_conn, chunked, body_pos, **response_kw)\u001b[0m\n\u001b[1;32m    698\u001b[0m             \u001b[0;31m# Make the request on the httplib connection object.\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 699\u001b[0;31m             httplib_response = self._make_request(\n\u001b[0m\u001b[1;32m    700\u001b[0m                 \u001b[0mconn\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;32m~/.pyenv/versions/3.8.1/envs/qiskit-ibm-runtime/lib/python3.8/site-packages/urllib3/connectionpool.py\u001b[0m in \u001b[0;36m_make_request\u001b[0;34m(self, conn, method, url, timeout, chunked, **httplib_request_kw)\u001b[0m\n\u001b[1;32m    381\u001b[0m         \u001b[0;32mtry\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 382\u001b[0;31m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_validate_conn\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mconn\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    383\u001b[0m         \u001b[0;32mexcept\u001b[0m \u001b[0;34m(\u001b[0m\u001b[0mSocketTimeout\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mBaseSSLError\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0me\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;32m~/.pyenv/versions/3.8.1/envs/qiskit-ibm-runtime/lib/python3.8/site-packages/urllib3/connectionpool.py\u001b[0m in \u001b[0;36m_validate_conn\u001b[0;34m(self, conn)\u001b[0m\n\u001b[1;32m   1009\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0;32mnot\u001b[0m \u001b[0mgetattr\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mconn\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m\"sock\"\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;32mNone\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m  \u001b[0;31m# AppEngine might not have  `.sock`\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1010\u001b[0;31m             \u001b[0mconn\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mconnect\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   1011\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;32m~/.pyenv/versions/3.8.1/envs/qiskit-ibm-runtime/lib/python3.8/site-packages/urllib3/connection.py\u001b[0m in \u001b[0;36mconnect\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    357\u001b[0m         \u001b[0;31m# Add certificate verification\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 358\u001b[0;31m         \u001b[0mconn\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_new_conn\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    359\u001b[0m         \u001b[0mhostname\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mhost\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;32m~/.pyenv/versions/3.8.1/envs/qiskit-ibm-runtime/lib/python3.8/site-packages/urllib3/connection.py\u001b[0m in \u001b[0;36m_new_conn\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    185\u001b[0m         \u001b[0;32mexcept\u001b[0m \u001b[0mSocketError\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0me\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 186\u001b[0;31m             raise NewConnectionError(\n\u001b[0m\u001b[1;32m    187\u001b[0m                 \u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m\"Failed to establish a new connection: %s\"\u001b[0m \u001b[0;34m%\u001b[0m \u001b[0me\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;31mNewConnectionError\u001b[0m: <urllib3.connection.HTTPSConnection object at 0x1259bc6a0>: Failed to establish a new connection: [Errno 8] nodename nor servname provided, or not known",
+      "\nDuring handling of the above exception, another exception occurred:\n",
+      "\u001b[0;31mMaxRetryError\u001b[0m                             Traceback (most recent call last)",
+      "\u001b[0;32m~/.pyenv/versions/3.8.1/envs/qiskit-ibm-runtime/lib/python3.8/site-packages/requests/adapters.py\u001b[0m in \u001b[0;36msend\u001b[0;34m(self, request, stream, timeout, verify, cert, proxies)\u001b[0m\n\u001b[1;32m    439\u001b[0m             \u001b[0;32mif\u001b[0m \u001b[0;32mnot\u001b[0m \u001b[0mchunked\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 440\u001b[0;31m                 resp = conn.urlopen(\n\u001b[0m\u001b[1;32m    441\u001b[0m                     \u001b[0mmethod\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mrequest\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mmethod\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;32m~/.pyenv/versions/3.8.1/envs/qiskit-ibm-runtime/lib/python3.8/site-packages/urllib3/connectionpool.py\u001b[0m in \u001b[0;36murlopen\u001b[0;34m(self, method, url, body, headers, retries, redirect, assert_same_host, timeout, pool_timeout, release_conn, chunked, body_pos, **response_kw)\u001b[0m\n\u001b[1;32m    754\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 755\u001b[0;31m             retries = retries.increment(\n\u001b[0m\u001b[1;32m    756\u001b[0m                 \u001b[0mmethod\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0murl\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0merror\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0me\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0m_pool\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0m_stacktrace\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0msys\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mexc_info\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0;36m2\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;32m~/.pyenv/versions/3.8.1/envs/qiskit-ibm-runtime/lib/python3.8/site-packages/urllib3/util/retry.py\u001b[0m in \u001b[0;36mincrement\u001b[0;34m(self, method, url, response, error, _pool, _stacktrace)\u001b[0m\n\u001b[1;32m    573\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mnew_retry\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mis_exhausted\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 574\u001b[0;31m             \u001b[0;32mraise\u001b[0m \u001b[0mMaxRetryError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0m_pool\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0murl\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0merror\u001b[0m \u001b[0;32mor\u001b[0m \u001b[0mResponseError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mcause\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    575\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;31mMaxRetryError\u001b[0m: HTTPSConnectionPool(host='iam.auth.quantum-computing.ibm.com', port=443): Max retries exceeded with url: /identity/token (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x1259bc6a0>: Failed to establish a new connection: [Errno 8] nodename nor servname provided, or not known'))",
+      "\nDuring handling of the above exception, another exception occurred:\n",
+      "\u001b[0;31mConnectionError\u001b[0m                           Traceback (most recent call last)",
+      "\u001b[0;32m/var/folders/30/grx88hyn4jbb3wmkl8930v2r0000gn/T/ipykernel_50090/3774291390.py\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m      2\u001b[0m \u001b[0;32mfrom\u001b[0m \u001b[0mqiskit\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mQuantumCircuit\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      3\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 4\u001b[0;31m \u001b[0mservice\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mQiskitRuntimeService\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m      5\u001b[0m \u001b[0moptions\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mOptions\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mbackend\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;34m\"ibmq_qasm_simulator\"\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0moptimization_level\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;36m1\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      6\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;32m~/Documents/Q/Github/qiskit-ibm-runtime/qiskit_ibm_runtime/qiskit_runtime_service.py\u001b[0m in \u001b[0;36m__init__\u001b[0;34m(self, channel, auth, token, url, name, instance, proxies, verify)\u001b[0m\n\u001b[1;32m    168\u001b[0m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_auth_warning\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    169\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 170\u001b[0;31m         self._account = self._discover_account(\n\u001b[0m\u001b[1;32m    171\u001b[0m             \u001b[0mtoken\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mtoken\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    172\u001b[0m             \u001b[0murl\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0murl\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;32m~/Documents/Q/Github/qiskit-ibm-runtime/qiskit_ibm_runtime/qiskit_runtime_service.py\u001b[0m in \u001b[0;36m_discover_account\u001b[0;34m(self, token, url, instance, channel, auth, name, proxies, verify)\u001b[0m\n\u001b[1;32m    287\u001b[0m         \u001b[0;31m# resolve CRN if needed\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    288\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0maccount\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mchannel\u001b[0m \u001b[0;34m==\u001b[0m \u001b[0;34m\"ibm_cloud\"\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 289\u001b[0;31m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_resolve_crn\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0maccount\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    290\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    291\u001b[0m         \u001b[0;31m# ensure account is valid, fail early if not\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;32m~/Documents/Q/Github/qiskit-ibm-runtime/qiskit_ibm_runtime/qiskit_runtime_service.py\u001b[0m in \u001b[0;36m_resolve_crn\u001b[0;34m(self, account)\u001b[0m\n\u001b[1;32m    319\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    320\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0m_resolve_crn\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0maccount\u001b[0m\u001b[0;34m:\u001b[0m \u001b[0mAccount\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;34m->\u001b[0m \u001b[0;32mNone\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 321\u001b[0;31m         \u001b[0maccount\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mresolve_crn\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    322\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    323\u001b[0m     def _authenticate_ibm_quantum_account(\n",
+      "\u001b[0;32m~/Documents/Q/Github/qiskit-ibm-runtime/qiskit_ibm_runtime/accounts/account.py\u001b[0m in \u001b[0;36mresolve_crn\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m     98\u001b[0m         \"\"\"\n\u001b[1;32m     99\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mchannel\u001b[0m \u001b[0;34m==\u001b[0m \u001b[0;34m\"ibm_cloud\"\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 100\u001b[0;31m             crn = resolve_crn(\n\u001b[0m\u001b[1;32m    101\u001b[0m                 \u001b[0mchannel\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mchannel\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    102\u001b[0m                 \u001b[0murl\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0murl\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;32m~/Documents/Q/Github/qiskit-ibm-runtime/qiskit_ibm_runtime/utils/utils.py\u001b[0m in \u001b[0;36mresolve_crn\u001b[0;34m(channel, url, instance, token)\u001b[0m\n\u001b[1;32m    102\u001b[0m             \u001b[0mclient\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mset_service_url\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mget_resource_controller_api_url\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0murl\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    103\u001b[0m             \u001b[0mclient\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mset_http_client\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0msession\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 104\u001b[0;31m             \u001b[0mlist_response\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mclient\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mlist_resource_instances\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mname\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0minstance\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    105\u001b[0m             \u001b[0mresult\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mlist_response\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mget_result\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    106\u001b[0m             \u001b[0mrow_count\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mresult\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0;34m\"rows_count\"\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;32m~/.pyenv/versions/3.8.1/envs/qiskit-ibm-runtime/lib/python3.8/site-packages/ibm_platform_services/resource_controller_v2.py\u001b[0m in \u001b[0;36mlist_resource_instances\u001b[0;34m(self, guid, name, resource_group_id, resource_id, resource_plan_id, type, sub_type, limit, start, state, updated_from, updated_to, **kwargs)\u001b[0m\n\u001b[1;32m    159\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    160\u001b[0m         \u001b[0murl\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;34m'/v2/resource_instances'\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 161\u001b[0;31m         request = self.prepare_request(method='GET',\n\u001b[0m\u001b[1;32m    162\u001b[0m                                        \u001b[0murl\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0murl\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    163\u001b[0m                                        \u001b[0mheaders\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mheaders\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;32m~/.pyenv/versions/3.8.1/envs/qiskit-ibm-runtime/lib/python3.8/site-packages/ibm_cloud_sdk_core/base_service.py\u001b[0m in \u001b[0;36mprepare_request\u001b[0;34m(self, method, url, headers, params, data, files, **kwargs)\u001b[0m\n\u001b[1;32m    402\u001b[0m         \u001b[0mrequest\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0;34m'data'\u001b[0m\u001b[0;34m]\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mdata\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    403\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 404\u001b[0;31m         \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mauthenticator\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mauthenticate\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mrequest\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    405\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    406\u001b[0m         \u001b[0;31m# Compress the request body if applicable\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;32m~/.pyenv/versions/3.8.1/envs/qiskit-ibm-runtime/lib/python3.8/site-packages/ibm_cloud_sdk_core/authenticators/iam_request_based_authenticator.py\u001b[0m in \u001b[0;36mauthenticate\u001b[0;34m(self, req)\u001b[0m\n\u001b[1;32m     61\u001b[0m         \"\"\"\n\u001b[1;32m     62\u001b[0m         \u001b[0mheaders\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mreq\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mget\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m'headers'\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 63\u001b[0;31m         \u001b[0mbearer_token\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtoken_manager\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mget_token\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m     64\u001b[0m         \u001b[0mheaders\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0;34m'Authorization'\u001b[0m\u001b[0;34m]\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;34m'Bearer {0}'\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mformat\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mbearer_token\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     65\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;32m~/.pyenv/versions/3.8.1/envs/qiskit-ibm-runtime/lib/python3.8/site-packages/ibm_cloud_sdk_core/token_managers/token_manager.py\u001b[0m in \u001b[0;36mget_token\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m     79\u001b[0m         \"\"\"\n\u001b[1;32m     80\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_is_token_expired\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 81\u001b[0;31m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mpaced_request_token\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m     82\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     83\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_token_needs_refresh\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;32m~/.pyenv/versions/3.8.1/envs/qiskit-ibm-runtime/lib/python3.8/site-packages/ibm_cloud_sdk_core/token_managers/token_manager.py\u001b[0m in \u001b[0;36mpaced_request_token\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    129\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    130\u001b[0m             \u001b[0;32mif\u001b[0m \u001b[0;32mnot\u001b[0m \u001b[0mrequest_active\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 131\u001b[0;31m                 \u001b[0mtoken_response\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mrequest_token\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    132\u001b[0m                 \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_save_token_info\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mtoken_response\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    133\u001b[0m                 \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mrequest_time\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;36m0\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;32m~/.pyenv/versions/3.8.1/envs/qiskit-ibm-runtime/lib/python3.8/site-packages/ibm_cloud_sdk_core/token_managers/iam_request_based_token_manager.py\u001b[0m in \u001b[0;36mrequest_token\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    113\u001b[0m             \u001b[0mauth_tuple\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mclient_id\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mclient_secret\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    114\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 115\u001b[0;31m         response = self._request(\n\u001b[0m\u001b[1;32m    116\u001b[0m             \u001b[0mmethod\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;34m'POST'\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    117\u001b[0m             \u001b[0murl\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0murl\u001b[0m \u001b[0;34m+\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mOPERATION_PATH\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;32mif\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0murl\u001b[0m \u001b[0;32melse\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0murl\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;32m~/.pyenv/versions/3.8.1/envs/qiskit-ibm-runtime/lib/python3.8/site-packages/ibm_cloud_sdk_core/token_managers/jwt_token_manager.py\u001b[0m in \u001b[0;36m_request\u001b[0;34m(self, method, url, headers, params, data, auth_tuple, **kwargs)\u001b[0m\n\u001b[1;32m     91\u001b[0m             \u001b[0mkwargs\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0;34m'verify'\u001b[0m\u001b[0;34m]\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;32mFalse\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     92\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 93\u001b[0;31m         response = requests.request(\n\u001b[0m\u001b[1;32m     94\u001b[0m             \u001b[0mmethod\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mmethod\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     95\u001b[0m             \u001b[0murl\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0murl\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;32m~/.pyenv/versions/3.8.1/envs/qiskit-ibm-runtime/lib/python3.8/site-packages/requests/api.py\u001b[0m in \u001b[0;36mrequest\u001b[0;34m(method, url, **kwargs)\u001b[0m\n\u001b[1;32m     59\u001b[0m     \u001b[0;31m# cases, and look like a memory leak in others.\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     60\u001b[0m     \u001b[0;32mwith\u001b[0m \u001b[0msessions\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mSession\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0msession\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 61\u001b[0;31m         \u001b[0;32mreturn\u001b[0m \u001b[0msession\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mrequest\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mmethod\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mmethod\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0murl\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0murl\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m**\u001b[0m\u001b[0mkwargs\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m     62\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     63\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;32m~/.pyenv/versions/3.8.1/envs/qiskit-ibm-runtime/lib/python3.8/site-packages/requests/sessions.py\u001b[0m in \u001b[0;36mrequest\u001b[0;34m(self, method, url, params, data, headers, cookies, files, auth, timeout, allow_redirects, proxies, hooks, stream, verify, cert, json)\u001b[0m\n\u001b[1;32m    527\u001b[0m         }\n\u001b[1;32m    528\u001b[0m         \u001b[0msend_kwargs\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mupdate\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0msettings\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 529\u001b[0;31m         \u001b[0mresp\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msend\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mprep\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m**\u001b[0m\u001b[0msend_kwargs\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    530\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    531\u001b[0m         \u001b[0;32mreturn\u001b[0m \u001b[0mresp\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;32m~/.pyenv/versions/3.8.1/envs/qiskit-ibm-runtime/lib/python3.8/site-packages/requests/sessions.py\u001b[0m in \u001b[0;36msend\u001b[0;34m(self, request, **kwargs)\u001b[0m\n\u001b[1;32m    643\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    644\u001b[0m         \u001b[0;31m# Send the request\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 645\u001b[0;31m         \u001b[0mr\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0madapter\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msend\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mrequest\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m**\u001b[0m\u001b[0mkwargs\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    646\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    647\u001b[0m         \u001b[0;31m# Total elapsed time of the request (approximately)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;32m~/.pyenv/versions/3.8.1/envs/qiskit-ibm-runtime/lib/python3.8/site-packages/requests/adapters.py\u001b[0m in \u001b[0;36msend\u001b[0;34m(self, request, stream, timeout, verify, cert, proxies)\u001b[0m\n\u001b[1;32m    517\u001b[0m                 \u001b[0;32mraise\u001b[0m \u001b[0mSSLError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0me\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mrequest\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mrequest\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    518\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 519\u001b[0;31m             \u001b[0;32mraise\u001b[0m \u001b[0mConnectionError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0me\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mrequest\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mrequest\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    520\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    521\u001b[0m         \u001b[0;32mexcept\u001b[0m \u001b[0mClosedPoolError\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0me\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
+      "\u001b[0;31mConnectionError\u001b[0m: HTTPSConnectionPool(host='iam.auth.quantum-computing.ibm.com', port=443): Max retries exceeded with url: /identity/token (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x1259bc6a0>: Failed to establish a new connection: [Errno 8] nodename nor servname provided, or not known'))"
+     ]
+    }
+   ],
    "source": [
-    "from qiskit_ibm_runtime import QiskitRuntimeService, Sampler\n",
+    "from qiskit_ibm_runtime import QiskitRuntimeService, Session, Options, Sampler\n",
     "from qiskit import QuantumCircuit\n",
     "\n",
     "service = QiskitRuntimeService()\n",
+    "options = Options(backend=\"ibmq_qasm_simulator\", optimization_level=1)\n",
     "\n",
     "bell = QuantumCircuit(2)\n",
     "bell.h(0)\n",
     "bell.cx(0, 1)\n",
-    "bell.measure_all()"
+    "bell.measure_all()\n",
+    "\n",
+    "with Session(service=service) as session:\n",
+    "    sampler = Sampler(session=session, options=options)"
    ]
   },
   {
@@ -65,20 +123,24 @@
    "id": "0403b43e",
    "metadata": {},
    "source": [
-    "## Write to & read from a session\n",
-    "\n",
-    "Running a job and returning the results are done by writing to and reading from the session. The session closes when the code exits the `with` block.\n",
+    "## Invoke the Sampler primitive\n",
     "\n",
+    "You can invoke the `Sampler` primitive program using the `run()` method of the `Sampler` instance you just created. The method returns a `IBMRuntimeJob` instance which you can use to query for things like job ID and status. The `result()` method of the job will return the result. \n",
     "\n",
-    "### Run the job & print results\n",
+    "You can invoke the `run()` method multiple times with the different inputs within a session.\n",
     "\n",
-    "Run the job, specifying your previously defined inputs and options. In this simple example, there is only one circuit and it does not have parameters.\n",
+    "### Specify program inputs\n",
     "\n",
-    "In each call, you will use `circuits` to specify which circuit to run and, if applicable,  `parameter_values` specifies which parameter to use with the specified circuit.\n",
+    "The `Sampler.run()` method takes in the following arguments:\n",
     "\n",
-    "In this example, we specified one circuit, `circuits=bell`, and we asked for the result for running the first (and only) circuit: `circuits=[0]`.\n",
+    "- **circuits**: A list of (parameterized) circuits that you want to investigate.\n",
+    "- **parameters**: A list of parameters for the parameterized circuits. It should be omitted if the circuits provided are not parameterized.\n",
+    "- **parameter_values**: An optional list of concrete parameters to be bound.\n",
+    "- **kwargs**: Additional options to overwrite the default values. \n",
+    "  \n",
+    "You can find more details in [the Sampler API reference](https://qiskit.org/documentation/partners/qiskit_ibm_runtime/stubs/qiskit_ibm_runtime.Sampler.html).\n",
     "\n",
-    "As you will see in later examples, if we had specified multiple circuits, such as `circuits=[pqc, pqc2]` when initializing the session, you could specify `circuits=[1]` or `circuits=[pqc2]` in each call to run the `pqc2` circuit."
+    "Example:"
    ]
   },
   {
@@ -91,16 +153,20 @@
      "name": "stdout",
      "output_type": "stream",
      "text": [
-      "SamplerResult(quasi_dists=[{'00': 0.5078125, '11': 0.4921875}], metadata=[{'header_metadata': None, 'shots': 1024}])\n"
+      "SamplerResult(quasi_dists=[{'11': 0.486328125, '00': 0.513671875}], metadata=[{'header_metadata': {}, 'shots': 1024}])\n"
      ]
     }
    ],
    "source": [
     "# executes a Bell circuit\n",
-    "with Sampler(circuits=bell, service=service, options={ \"backend\": \"ibmq_qasm_simulator\" }) as sampler:\n",
-    "    # pass indices of circuits\n",
-    "    result = sampler(circuits=[0], shots=1024)\n",
-    "    print(result)"
+    "with Session(service=service) as session:\n",
+    "    sampler = Sampler(session=session, options=options)\n",
+    "    job = sampler.run(circuits=bell)\n",
+    "    print(job.result())\n",
+    "\n",
+    "    # You can invoke run() multiple times.\n",
+    "    job = sampler.run(circuits=bell)\n",
+    "    print(job.result())"
    ]
   },
   {
@@ -115,7 +181,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 4,
+   "execution_count": 3,
    "id": "d51903c2",
    "metadata": {},
    "outputs": [
@@ -123,15 +189,16 @@
      "name": "stdout",
      "output_type": "stream",
      "text": [
-      "SamplerResult(quasi_dists=[{'00': 0.478515625, '11': 0.521484375}, {'00': 0.533203125, '11': 0.466796875}, {'00': 0.4853515625, '11': 0.5146484375}], metadata=[{'header_metadata': None, 'shots': 1024}, {'header_metadata': None, 'shots': 1024}, {'header_metadata': None, 'shots': 1024}])\n"
+      "SamplerResult(quasi_dists=[{'11': 0.484375, '00': 0.515625}, {'00': 0.5029296875, '11': 0.4970703125}, {'00': 0.4794921875, '11': 0.5205078125}], metadata=[{'header_metadata': {}, 'shots': 1024}, {'header_metadata': {}, 'shots': 1024}, {'header_metadata': {}, 'shots': 1024}])\n"
      ]
     }
    ],
    "source": [
-    "from qiskit_ibm_runtime import QiskitRuntimeService, Sampler\n",
+    "from qiskit_ibm_runtime import QiskitRuntimeService, Session, Options, Sampler\n",
     "from qiskit import QuantumCircuit\n",
     "\n",
     "service = QiskitRuntimeService()\n",
+    "options = Options(backend=\"ibmq_qasm_simulator\", optimization_level=1)\n",
     "\n",
     "bell = QuantumCircuit(2)\n",
     "bell.h(0)\n",
@@ -139,10 +206,10 @@
     "bell.measure_all()\n",
     "\n",
     "# executes three Bell circuits\n",
-    "with Sampler(circuits=[bell]*3, service=service, options={ \"backend\": \"ibmq_qasm_simulator\" }) as sampler:\n",
-    "    # alternatively you can also pass circuits as objects\n",
-    "    result = sampler(circuits=[bell]*3)\n",
-    "    print(result)"
+    "with Session(service=service) as session:\n",
+    "    sampler = Sampler(session=session, options=options)\n",
+    "    job = sampler.run(circuits=[bell]*3)\n",
+    "    print(job.result())"
    ]
   },
   {
@@ -159,7 +226,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 5,
+   "execution_count": 4,
    "id": "d213627d",
    "metadata": {},
    "outputs": [
@@ -167,16 +234,17 @@
      "name": "stdout",
      "output_type": "stream",
      "text": [
-      "SamplerResult(quasi_dists=[{'00': 0.1259765625, '10': 0.1005859375, '11': 0.408203125, '01': 0.365234375}, {'01': 0.025390625, '11': 0.3056640625, '00': 0.0693359375, '10': 0.599609375}, {'11': 0.03515625, '01': 0.67578125, '10': 0.0888671875, '00': 0.2001953125}], metadata=[{'header_metadata': None, 'shots': 1024}, {'header_metadata': None, 'shots': 1024}, {'header_metadata': None, 'shots': 1024}])\n"
+      "SamplerResult(quasi_dists=[{'01': 0.3828125, '11': 0.4150390625, '00': 0.1171875, '10': 0.0849609375}, {'01': 0.0419921875, '11': 0.3095703125, '00': 0.0751953125, '10': 0.5732421875}, {'01': 0.7080078125, '11': 0.03515625, '10': 0.08203125, '00': 0.1748046875}], metadata=[{'header_metadata': {}, 'shots': 1024}, {'header_metadata': {}, 'shots': 1024}, {'header_metadata': {}, 'shots': 1024}])\n"
      ]
     }
    ],
    "source": [
-    "from qiskit_ibm_runtime import QiskitRuntimeService, Sampler\n",
+    "from qiskit_ibm_runtime import QiskitRuntimeService, Session, Options, Sampler\n",
     "from qiskit import QuantumCircuit\n",
     "from qiskit.circuit.library import RealAmplitudes\n",
     "\n",
     "service = QiskitRuntimeService()\n",
+    "options = Options(backend=\"ibmq_qasm_simulator\", optimization_level=1)\n",
     "\n",
     "# parameterized circuit\n",
     "pqc = RealAmplitudes(num_qubits=2, reps=2)\n",
@@ -187,10 +255,12 @@
     "theta1 = [0, 1, 1, 2, 3, 5]\n",
     "theta2 = [1, 2, 3, 4, 5, 6]\n",
     "theta3 = [0, 1, 2, 3, 4, 5, 6, 7]\n",
+    "    \n",
+    "with Session(service) as session:\n",
+    "    sampler = Sampler(session=session, options=options)\n",
     "\n",
-    "with Sampler(circuits=[pqc, pqc2], service=service, options={ \"backend\": \"ibmq_qasm_simulator\" }) as sampler:\n",
-    "    result = sampler(circuits=[0, 0, 1], parameter_values=[theta1, theta2, theta3])\n",
-    "    print(result)"
+    "    job = sampler.run(circuits=[pqc, pqc, pqc2], parameter_values=[theta1, theta2, theta3])\n",
+    "    print(job.result())"
    ]
   },
   {
@@ -200,13 +270,21 @@
    "source": [
     "### Result\n",
     "\n",
-    "The results align with the parameter - circuit pairs specified previously.  For example, the first result (`{'11': 0.42578125, '00': 0.14453125, '10': 0.0888671875, '01': 0.3408203125}`) is the output of the parameter labeled `theta1` being sent to the first circuit."
+    "The results align with the parameter - circuit pairs specified previously.  For example, the first result (`{'00': 0.1376953125, '10': 0.095703125, '01': 0.359375, '11': 0.4072265625}`) is the output of the parameter labeled `theta1` being sent to the first circuit."
    ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "2c755e88",
+   "metadata": {},
+   "outputs": [],
+   "source": []
   }
  ],
  "metadata": {
   "kernelspec": {
-   "display_name": "Python 3 (ipykernel)",
+   "display_name": "Python 3.8.1 64-bit ('qiskit-ibm-runtime')",
    "language": "python",
    "name": "python3"
   },
@@ -220,7 +298,12 @@
    "name": "python",
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
-   "version": "3.8.8"
+   "version": "3.8.1"
+  },
+  "vscode": {
+   "interpreter": {
+    "hash": "fd361434a97303db55f7feadeb0d1409e986034dc8706e65a70496723772e70a"
+   }
   }
  },
  "nbformat": 4,
diff --git a/qiskit_ibm_runtime/__init__.py b/qiskit_ibm_runtime/__init__.py
index 70b01094..8fe3e232 100644
--- a/qiskit_ibm_runtime/__init__.py
+++ b/qiskit_ibm_runtime/__init__.py
@@ -236,6 +236,7 @@ from .runtime_job import RuntimeJob
 from .runtime_program import RuntimeProgram, ParameterNamespace
 from .runtime_options import RuntimeOptions
 from .utils.json import RuntimeEncoder, RuntimeDecoder
+from .session import Session  # pylint: disable=cyclic-import
 
 from .exceptions import *
 from .utils.utils import setup_logger
@@ -243,6 +244,7 @@ from .version import __version__
 
 from .estimator import Estimator
 from .sampler import Sampler
+from .options import Options
 
 # TODO remove when terra code is released
 from .qiskit.primitives import (
diff --git a/qiskit_ibm_runtime/estimator.py b/qiskit_ibm_runtime/estimator.py
index ce597340..a742a3ad 100644
--- a/qiskit_ibm_runtime/estimator.py
+++ b/qiskit_ibm_runtime/estimator.py
@@ -12,19 +12,30 @@
 
 """Estimator primitive."""
 
+from __future__ import annotations
+import copy
 from typing import Iterable, Optional, Dict, Sequence, Any, Union
 
 import numpy as np
 from qiskit.circuit import QuantumCircuit, Parameter
 from qiskit.quantum_info import SparsePauliOp
+from qiskit.opflow import PauliSumOp
+from qiskit.quantum_info.operators.base_operator import BaseOperator
+
+# pylint: disable=unused-import,cyclic-import
+import qiskit_ibm_runtime.session as session_pkg
 
 # TODO import BaseEstimator and EstimatorResult from terra once released
 from .qiskit.primitives import BaseEstimator, EstimatorResult
-from .exceptions import IBMInputValueError
-from .ibm_backend import IBMBackend
 from .qiskit_runtime_service import QiskitRuntimeService
-from .runtime_session import RuntimeSession
-from .utils.converters import hms_to_seconds
+from .program.result_decoder import ResultDecoder
+from .runtime_job import RuntimeJob
+from .utils.deprecation import deprecate_arguments, issue_deprecation_msg
+from .runtime_options import RuntimeOptions
+from .options import Options
+
+# pylint: disable=unused-import,cyclic-import
+from .session import Session
 
 
 class Estimator(BaseEstimator):
@@ -33,53 +44,12 @@ class Estimator(BaseEstimator):
     Qiskit Runtime Estimator primitive service estimates expectation values of quantum circuits and
     observables.
 
-    Estimator can be initialized with the following parameters.
-
-    * circuits: a (parameterized) :class:`~qiskit.circuit.QuantumCircuit` or
-        a list of (parameterized) :class:`~qiskit.circuit.QuantumCircuit`.
-
-    * observables: a list of :class:`~qiskit.quantum_info.SparsePauliOp`.
-
-    * parameters: a list of parameters of the quantum circuits.
-        (:class:`~qiskit.circuit.parametertable.ParameterView` or
-        a list of :class:`~qiskit.circuit.Parameter`) specifying the order
-        in which parameter values will be bound.
-
-    * skip_transpilation: Transpilation is skipped if set to True.
-        False by default.
-
-    * service: Optional instance of :class:`qiskit_ibm_runtime.QiskitRuntimeService` class,
-        defaults to `QiskitRuntimeService()` which tries to initialize your default saved account.
-
-    * options: Runtime options dictionary that control the execution environment.
-
-        * backend: Optional instance of :class:`qiskit_ibm_runtime.IBMBackend` class or
-            string name of backend, if not specified a backend will be selected
-            automatically (IBM Cloud only).
-        * image: the runtime image used to execute the program, specified in
-            the form of ``image_name:tag``. Not all accounts are
-            authorized to select a different image.
-        * log_level: logging level to set in the execution environment. The valid
-            log levels are: ``DEBUG``, ``INFO``, ``WARNING``, ``ERROR``, and ``CRITICAL``.
-            The default level is ``WARNING``.
-
-    The returned instance can be called repeatedly with the following parameters to
-    estimate expectation values.
-
-    * circuits: a (parameterized) :class:`~qiskit.circuit.QuantumCircuit` or
-        a list of (parameterized) :class:`~qiskit.circuit.QuantumCircuit` or
-        a list of circuit indices.
+    The :meth: `run` can be used to submit circuits, observables, and parameters
+    to the Estimator primitive.
 
-    * observables: a list of :class:`~qiskit.quantum_info.SparsePauliOp` or
-        a list of observable indices.
-
-    * parameter_values: An optional list of concrete parameters to be bound.
-
-    * circuit_indices: (DEPRECATED) A list of circuit indices.
-
-    * observable_indices: (DEPRECATED) A list of observable indices.
-
-    All the above lists should be of the same length.
+    You are encouraged to use :class:`~qiskit_ibm_runtime.Session` to open a session,
+    during which you can invoke one or more primitive programs. Jobs submitted within a session
+    are prioritized by the scheduler, and data is cached for efficiency.
 
     Example::
 
@@ -89,180 +59,202 @@ class Estimator(BaseEstimator):
         from qiskit_ibm_runtime import QiskitRuntimeService, Estimator
 
         service = QiskitRuntimeService(channel="ibm_cloud")
-        options = { "backend": "ibmq_qasm_simulator" }
 
         psi1 = RealAmplitudes(num_qubits=2, reps=2)
-        psi2 = RealAmplitudes(num_qubits=2, reps=3)
 
         H1 = SparsePauliOp.from_list([("II", 1), ("IZ", 2), ("XI", 3)])
         H2 = SparsePauliOp.from_list([("IZ", 1)])
         H3 = SparsePauliOp.from_list([("ZI", 1), ("ZZ", 1)])
 
-        with Estimator(
-            circuits=[psi1, psi2],
-            observables=[H1, H2, H3],
-            service=service,
-            options=options
-        ) as estimator:
+        with Session(service) as session:
+            estimator = Estimator(session=session)
+            estimator.options.backend = 'ibmq_qasm_simulator'
+
             theta1 = [0, 1, 1, 2, 3, 5]
-            theta2 = [0, 1, 1, 2, 3, 5, 8, 13]
-            theta3 = [1, 2, 3, 4, 5, 6]
 
             # calculate [ <psi1(theta1)|H1|psi1(theta1)> ]
-            # pass circuits and observables as indices
-            psi1_H1_result = estimator([0], [0], [theta1])
-            print(psi1_H1_result)
+            psi1_H1 = estimator.run(circuits=[psi1], observables=[H1], parameter_values=[theta1])
+            print(psi1_H1.result())
 
             # calculate [ <psi1(theta1)|H2|psi1(theta1)>, <psi1(theta1)|H3|psi1(theta1)> ]
-            # alternatively you can also pass circuits and observables as objects
-            psi1_H23_result = estimator([psi1, psi1], [H2, H3], [theta1]*2)
-            print(psi1_H23_result)
-
-            # calculate [ <psi2(theta2)|H2|psi2(theta2)> ]
-            psi2_H2_result = estimator([1], [1], [theta2])
-            print(psi2_H2_result)
-
-            # calculate [ <psi1(theta1)|H1|psi1(theta1)>, <psi1(theta3)|H1|psi1(theta3)> ]
-            psi1_H1_result2 = estimator([0, 0], [0, 0], [theta1, theta3])
-            print(psi1_H1_result2)
-
-            # calculate [ <psi1(theta1)|H1|psi1(theta1)>,
-            #             <psi2(theta2)|H2|psi2(theta2)>,
-            #             <psi1(theta3)|H3|psi1(theta3)> ]
-            psi12_H23_result = estimator([0, 1, 0], [0, 1, 2], [theta1, theta2, theta3])
-            print(psi12_H23_result)
+            psi1_H23 = estimator.run(
+                circuits=[psi1, psi1],
+                observables=[H2, H3],
+                parameter_values=[theta1]*2
+            )
+            print(psi1_H23.result())
     """
 
+    _PROGRAM_ID = "estimator"
+
     def __init__(
         self,
-        circuits: Union[QuantumCircuit, Iterable[QuantumCircuit]],
-        observables: Iterable[SparsePauliOp],
+        circuits: Optional[Union[QuantumCircuit, Iterable[QuantumCircuit]]] = None,
+        observables: Optional[Iterable[SparsePauliOp]] = None,
         parameters: Optional[Iterable[Iterable[Parameter]]] = None,
         service: Optional[QiskitRuntimeService] = None,
-        options: Optional[Dict] = None,
+        session: Optional[Session] = None,
+        options: Optional[Union[Dict, RuntimeOptions, Options]] = None,
         skip_transpilation: Optional[bool] = False,
-        transpilation_settings: Optional[Dict] = None,
-        resilience_settings: Optional[Dict] = None,
-        max_time: Optional[Union[int, str]] = None,
     ):
         """Initializes the Estimator primitive.
 
         Args:
-            circuits: a (parameterized) :class:`~qiskit.circuit.QuantumCircuit` or
+            circuits: (DEPRECATED) A (parameterized) :class:`~qiskit.circuit.QuantumCircuit` or
                 a list of (parameterized) :class:`~qiskit.circuit.QuantumCircuit`.
 
-            observables: a list of :class:`~qiskit.quantum_info.SparsePauliOp`
+            observables: (DEPRECATED) A list of :class:`~qiskit.quantum_info.SparsePauliOp`
 
-            parameters: a list of parameters of the quantum circuits.
+            parameters: (DEPRECATED) A list of parameters of the quantum circuits.
                 (:class:`~qiskit.circuit.parametertable.ParameterView` or
                 a list of :class:`~qiskit.circuit.Parameter`) specifying the order
                 in which parameter values will be bound.
 
-            service: Optional instance of :class:`qiskit_ibm_runtime.QiskitRuntimeService` class,
+            service: (DEPRECATED) Optional instance of
+                :class:`qiskit_ibm_runtime.QiskitRuntimeService` class,
                 defaults to `QiskitRuntimeService()` which tries to initialize your default
                 saved account.
 
-            options: Runtime options dictionary that control the execution environment.
-
-                * backend: Optional instance of :class:`qiskit_ibm_runtime.IBMBackend` class or
-                    string name of backend, if not specified a backend will be selected
-                    automatically (IBM Cloud only).
-                * image: the runtime image used to execute the program, specified in
-                    the form of ``image_name:tag``. Not all accounts are
-                    authorized to select a different image.
-                * log_level: logging level to set in the execution environment. The valid
-                    log levels are: ``DEBUG``, ``INFO``, ``WARNING``, ``ERROR``, and ``CRITICAL``.
-                    The default level is ``WARNING``.
-
-            skip_transpilation: Transpilation is skipped if set to True. False by default.
-
-            transpilation_settings: (EXPERIMENTAL setting, can break between releases without warning)
-                Qiskit transpiler settings. The transpilation process converts
-                operations in the circuit to those supported by the backend, swaps qubits with the
-                circuit to overcome limited qubit connectivity and some optimizations to reduce the
-                circuit's gate count where it can.
-
-                * skip_transpilation: Transpilation is skipped if set to True.
-                    False by default.
-
-                * optimization_settings:
-
-                    * level: How much optimization to perform on the circuits.
-                        Higher levels generate more optimized circuits,
-                        at the expense of longer transpilation times.
-                        * 0: no optimization
-                        * 1: light optimization
-                        * 2: heavy optimization
-                        * 3: even heavier optimization
-                        If ``None``, level 1 will be chosen as default.
-
-            resilience_settings: (EXPERIMENTAL setting, can break between releases without warning)
-                Using these settings allows you to build resilient algorithms by
-                leveraging the state of the art error suppression, mitigation and correction techniques.
-
-                * level: How much resilience to build against errors.
-                    Higher levels generate more accurate results,
-                    at the expense of longer processing times.
-                    * 0: no resilience
-                    * 1: light resilience
-                    If ``None``, level 0 will be chosen as default.
-
-            max_time: (EXPERIMENTAL setting, can break between releases without warning)
-                Maximum amount of time, a runtime session can be open before being
-                forcibly closed. Can be specified as seconds (int) or a string like "2h 30m 40s".
+            session: Session in which to call the sampler primitive. If ``None``, a new session
+                is created using the default saved account.
 
-        Raises:
-            IBMInputValueError: If an input value is invalid.
+            options: Primitive options, see :class:`Options` for detailed description.
+
+            skip_transpilation: (DEPRECATED) Transpilation is skipped if set to True. False by default.
+                Ignored ``skip_transpilation`` is also specified in ``options``.
         """
         super().__init__(
-            circuits=circuits if isinstance(circuits, Iterable) else [circuits],
+            circuits=circuits,
             observables=observables,
             parameters=parameters,
         )
-        self._skip_transpilation = skip_transpilation
-        if not service:
-            # try to initialize service with default saved account
-            service = QiskitRuntimeService()
-        self._service = service
-        if isinstance(options, dict) and "backend" in options:
-            backend = options.get("backend")
-            if isinstance(backend, IBMBackend):
-                del options["backend"]
-                options["backend_name"] = backend.name
-            elif isinstance(backend, str):
-                del options["backend"]
-                options["backend_name"] = backend
-            else:
-                raise IBMInputValueError(
-                    "'backend' property in 'options' should be either the string name of the "
-                    "backend or an instance of 'IBMBackend' class"
-                )
+
+        # TODO: Remove deprecation warnings if done in base class
+        if circuits or parameters or observables:
+            deprecate_arguments(
+                "circuits, parameters, and observables",
+                "0.7",
+                f"You can instead specify these inputs using the {self.__class__.__name__}.run method.",
+            )
+        if skip_transpilation:
+            deprecate_arguments(
+                "skip_transpilation",
+                "0.7",
+                "Instead, use the skip_transpilation keyword argument in transpilation_settings.",
+            )
+        if service:
+            deprecate_arguments(
+                "service", "0.7", "Please use the session parameter instead."
+            )
+
+        if options is None:
+            self.options = Options()
+        elif isinstance(options, Options):
+            self.options = copy.deepcopy(options)
+            skip_transpilation = self.options.transpilation.skip_transpilation
+        elif isinstance(options, RuntimeOptions):
+            self.options = options._to_new_options()
+        else:
+            self.options = Options._from_dict(options)
+            skip_transpilation = options.get("transpilation", {}).get(
+                "skip_transpilation", False
+            )
+        self.options.transpilation.skip_transpilation = skip_transpilation
+
+        self._initial_inputs = {
+            "circuits": circuits,
+            "observables": observables,
+            "parameters": parameters,
+        }
+
+        if session:
+            self._session = session
+        else:
+            if session_pkg._DEFAULT_SESSION is None:
+                session_pkg._DEFAULT_SESSION = Session(service=service)
+            self._session = session_pkg._DEFAULT_SESSION
+
+    def run(
+        self,
+        circuits: Union[QuantumCircuit, Sequence[QuantumCircuit]],
+        observables: Sequence[BaseOperator | PauliSumOp],
+        parameter_values: Optional[
+            Union[Sequence[float], Sequence[Sequence[float]]]
+        ] = None,
+        parameters: Sequence[Sequence[Parameter]] | None = None,
+        **kwargs: Any,
+    ) -> RuntimeJob:
+        """Submit a request to the estimator primitive program.
+
+        Args:
+            circuits: a (parameterized) :class:`~qiskit.circuit.QuantumCircuit` or
+                a list of (parameterized) :class:`~qiskit.circuit.QuantumCircuit`.
+
+            observables: A list of observable objects.
+
+            parameter_values: An optional list of concrete parameters to be bound.
+
+            parameters: A list of parameters of the quantum circuits
+                (:class:`~qiskit.circuit.parametertable.ParameterView` or
+                a list of :class:`~qiskit.circuit.Parameter`).
+                Defaults to ``[circ.parameters for circ in circuits]``.
+
+            **kwargs: Individual options to overwrite the default primitive options.
+
+        Returns:
+            Submitted job.
+
+        Raises:
+            ValueError: If the input values are invalid.
+        """
+        if isinstance(circuits, Iterable) and not all(
+            isinstance(inst, QuantumCircuit) for inst in circuits
+        ):
+            raise ValueError(
+                "The circuits parameter has to be instances of QuantumCircuit."
+            )
+
+        circ_count = 1 if isinstance(circuits, QuantumCircuit) else len(circuits)
+        obs_count = (
+            1
+            if isinstance(observables, (BaseOperator, PauliSumOp))
+            else len(observables)
+        )
+
         inputs = {
             "circuits": circuits,
+            "circuit_indices": list(range(circ_count)),
             "observables": observables,
+            "observable_indices": list(range(obs_count)),
             "parameters": parameters,
-            "skip_transpilation": self._skip_transpilation,
+            "parameter_values": parameter_values,
         }
-        if transpilation_settings:
-            inputs.update({"transpilation_settings": transpilation_settings})
-        if resilience_settings:
-            inputs.update({"resilience_settings": resilience_settings})
-        self._session = RuntimeSession(
-            service=self._service,
-            program_id="estimator",
+
+        combined = self.options._merge_options(kwargs)
+        inputs.update(Options._get_program_inputs(combined))
+
+        return self._session.run(
+            program_id=self._PROGRAM_ID,
             inputs=inputs,
-            options=options,
-            max_time=self.calculate_max_time(max_time=max_time),
+            options=Options._get_runtime_options(combined),
+            result_decoder=EstimatorResultDecoder,
         )
 
-    def calculate_max_time(self, max_time: Optional[Union[int, str]] = None) -> int:
-        """Calculate max_time in seconds from hour minute seconds string. Ex: 2h 30m 40s"""
-        try:
-            return hms_to_seconds(max_time) if isinstance(max_time, str) else max_time
-        except IBMInputValueError as input_value_error:
-            raise IBMInputValueError(
-                "Invalid value given for max_time.", input_value_error.message
-            )
+    def __call__(
+        self,
+        circuits: Sequence[int],
+        observables: Sequence[int],
+        parameter_values: Optional[
+            Union[Sequence[float], Sequence[Sequence[float]]]
+        ] = None,
+        **run_options: Any,
+    ) -> EstimatorResult:
+        issue_deprecation_msg(
+            msg="Calling an Estimator instance directly has been deprecated ",
+            version="0.7",
+            remedy="Please use qiskit_ibm_runtime.Session and Estimator.run() instead.",
+        )
+        return super().__call__(circuits, observables, parameter_values, **run_options)
 
     def _call(
         self,
@@ -295,18 +287,46 @@ class Estimator(BaseEstimator):
         Returns:
             An instance of :class:`qiskit.primitives.EstimatorResult`.
         """
-        self._session.write(
-            circuit_indices=circuits,
-            observable_indices=observables,
-            parameter_values=parameter_values,
-            run_options=run_options,
-        )
-        raw_result = self._session.read()
-        return EstimatorResult(
-            values=np.asarray(raw_result["values"]),
-            metadata=raw_result["metadata"],
-        )
+        inputs = {
+            "circuits": self._initial_inputs["circuits"],
+            "parameters": self._initial_inputs["parameters"],
+            "observables": self._initial_inputs["observables"],
+            "circuit_indices": circuits,
+            "parameter_values": parameter_values,
+            "observable_indices": observables,
+        }
+        combined = self.options._merge_options(run_options)
+        inputs.update(Options._get_program_inputs(combined))
+
+        return self._session.run(
+            program_id=self._PROGRAM_ID,
+            inputs=inputs,
+            options=Options._get_runtime_options(combined),
+            result_decoder=EstimatorResult,
+        ).result()
 
     def close(self) -> None:
         """Close the session and free resources"""
         self._session.close()
+
+    @property
+    def session(self) -> Session:
+        """Return session used by this primitive.
+
+        Returns:
+            Session used by this primitive.
+        """
+        return self._session
+
+
+class EstimatorResultDecoder(ResultDecoder):
+    """Class used to decode estimator results"""
+
+    @classmethod
+    def decode(cls, raw_result: str) -> EstimatorResult:
+        """Convert the result to EstimatorResult."""
+        decoded: Dict = super().decode(raw_result)
+        return EstimatorResult(
+            values=np.asarray(decoded["values"]),
+            metadata=decoded["metadata"],
+        )
diff --git a/qiskit_ibm_runtime/options.py b/qiskit_ibm_runtime/options.py
new file mode 100644
index 00000000..2432876e
--- /dev/null
+++ b/qiskit_ibm_runtime/options.py
@@ -0,0 +1,263 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2022.
+#
+# This code is licensed under the Apache License, Version 2.0. You may
+# obtain a copy of this license in the LICENSE.txt file in the root directory
+# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
+#
+# Any modifications or derivative works of this code must retain this
+# copyright notice, and modified files need to carry a notice indicating
+# that they have been altered from the originals.
+
+"""Primitive settings."""
+
+from typing import Optional, List, Dict, Union, Any
+from dataclasses import dataclass, asdict, field
+import copy
+
+from .utils.deprecation import issue_deprecation_msg
+
+
+@dataclass
+class Transpilation:
+    """Transpilation settings.
+
+    Args:
+        skip_transpilation: Whether to skip transpilation.
+
+        initial_layout: Initial position of virtual qubits on physical qubits.
+            See :function:`qiskit.compiler.transpile` for more information.
+
+        layout_method: Name of layout selection pass ('trivial', 'dense', 'noise_adaptive', 'sabre')
+
+        routing_method: Name of routing pass ('basic', 'lookahead', 'stochastic', 'sabre', 'none')
+
+        translation_method: Name of translation pass ('unroller', 'translator', 'synthesis')
+
+        approximation_degree (float): heuristic dial used for circuit approximation
+            (1.0=no approximation, 0.0=maximal approximation)
+
+        timing_constraints: An optional control hardware restriction on instruction time resolution.
+            A quantum computer backend may report a set of restrictions, namely:
+
+            - granularity: An integer value representing minimum pulse gate
+              resolution in units of ``dt``. A user-defined pulse gate should have
+              duration of a multiple of this granularity value.
+            - min_length: An integer value representing minimum pulse gate
+              length in units of ``dt``. A user-defined pulse gate should be longer
+              than this length.
+            - pulse_alignment: An integer value representing a time resolution of gate
+              instruction starting time. Gate instruction should start at time which
+              is a multiple of the alignment value.
+            - acquire_alignment: An integer value representing a time resolution of measure
+              instruction starting time. Measure instruction should start at time which
+              is a multiple of the alignment value.
+
+            This information will be provided by the backend configuration.
+            If the backend doesn't have any restriction on the instruction time allocation,
+            then ``timing_constraints`` is None and no adjustment will be performed.
+
+        seed_transpiler: Sets random seed for the stochastic parts of the transpiler
+    """
+
+    # TODO: Double check transpilation settings.
+
+    skip_transpilation: bool = False
+    initial_layout: Optional[Union[Dict, List]] = None  # TODO: Support Layout
+    layout_method: Optional[str] = None
+    routing_method: Optional[str] = None
+    translation_method: Optional[str] = None
+    approximation_degree: Optional[float] = None
+    timing_constraints: Optional[Dict[str, int]] = None
+    seed_transpiler: Optional[int] = None
+
+
+@dataclass
+class Resilience:
+    """Resilience settings."""
+
+    pass
+
+
+@dataclass
+class SimulatorOptions:
+    """Simulator options.
+
+    Args:
+        noise_model: Noise model, must have a to_dict() method.
+        seed_simulator: Random seed to control sampling.
+    """
+
+    def __init__(
+        self, noise_model: Any = None, seed_simulator: Optional[int] = None
+    ) -> None:
+        self.noise_model = noise_model
+        self.seed_simulator = seed_simulator
+
+    @property
+    def noise_model(self) -> Dict:
+        """Return the noise model.
+
+        Returns:
+            The noise model.
+        """
+        return self._noise_model
+
+    @noise_model.setter
+    def noise_model(self, noise_model: Any) -> None:
+        """Set the noise model.
+
+        Args:
+            noise_model: Noise model to use.
+
+        Raises:
+            ValueError: If the noise model doesn't have a ``to_dict()`` method.
+        """
+        if isinstance(noise_model, Dict):
+            self._noise_model = noise_model
+        else:
+            try:
+                self._noise_model = noise_model.to_dict()
+            except AttributeError:
+                raise ValueError(
+                    "Only noise models that have a to_dict() method are supported"
+                )
+
+
+@dataclass
+class Execution:
+    """Execution options."""
+
+    shots: int = 4000
+    qubit_lo_freq: Optional[List[float]] = None
+    meas_lo_freq: Optional[List[float]] = None
+    # TODO: need to be able to serialize schedule_los before we can support it
+    rep_delay: Optional[float] = None
+    init_qubits: bool = True
+
+
+@dataclass
+class Options:
+    """Primitive options.
+
+    Args:
+        optimization_level: How much optimization to perform on the circuits.
+            Higher levels generate more optimized circuits,
+            at the expense of longer transpilation times.
+
+            * 0: no optimization
+            * 1: light optimization
+            * 2: heavy optimization
+            * 3: even heavier optimization
+
+        resilience_level: How much resilience to build against errors.
+            Higher levels generate more accurate results,
+            at the expense of longer processing times.
+
+            * 0: no resilience
+            * 1: light resilience
+
+        log_level: logging level to set in the execution environment. The valid
+            log levels are: ``DEBUG``, ``INFO``, ``WARNING``, ``ERROR``, and ``CRITICAL``.
+            The default level is ``WARNING``.
+
+        backend: target backend to run on. This is required for ``ibm_quantum`` runtime.
+
+        log_level: logging level to set in the execution environment. The valid
+            log levels are: ``DEBUG``, ``INFO``, ``WARNING``, ``ERROR``, and ``CRITICAL``.
+            The default level is ``WARNING``.
+
+        transpilation: Transpilation options. See :class:`Transpilation`.
+
+        execution: Execution time options. See :class: `Execution`.
+    """
+
+    optimization_level: int = 1
+    resilience_level: int = 0
+    backend: str = None
+    log_level: str = "WARNING"
+    transpilation: Transpilation = field(default_factory=Transpilation)
+    execution: Execution = field(default_factory=Execution)
+    experimental: dict = None
+
+    def _merge_options(self, new_options: Optional[Dict] = None) -> Dict:
+        """Merge current options with the new ones.
+
+        Args:
+            new_options: New options to merge.
+
+        Returns:
+            Merged dictionary.
+        """
+
+        def _update_options(old: Dict, new: Dict) -> None:
+            if not new:
+                return
+            for key, val in old.items():
+                if key in new.keys():
+                    old[key] = new.pop(key)
+                if isinstance(val, Dict):
+                    _update_options(val, new)
+
+        # First combine options.
+        combined = copy.deepcopy(asdict(self))
+        _update_options(combined, new_options)
+        return combined
+
+    @classmethod
+    def _from_dict(cls, data: Dict) -> "Options":
+        data = copy.copy(data)
+        experimental = None
+        if "image" in data.keys():
+            issue_deprecation_msg(
+                msg="The 'image' option has been moved to the 'experimental' category",
+                version="0.7",
+                remedy="Please specify 'experimental':{'image': image} instead.",
+            )
+            experimental = {"image": data.pop("image")}
+        if "backend_name" in data.keys():
+            issue_deprecation_msg(
+                msg="The keyword 'backend_name' has been deprecated",
+                version="0.7",
+                remedy="Please use 'backend' instead.",
+            )
+            data["backend"] = data.pop("backend_name")
+        transp = Transpilation(**data.pop("transpilation", {}))
+        execution = Execution(**data.pop("execution", {}))
+        return cls(
+            experimental=experimental,
+            transpilation=transp,
+            execution=execution,
+            **data,
+        )
+
+    @staticmethod
+    def _get_program_inputs(options: Dict) -> Dict:
+        """Convert the input options to program compatible inputs.
+
+        Returns:
+            Inputs acceptable by primitive programs.
+        """
+        inputs = {}
+        inputs["transpilation_settings"] = options.get("transpilation", {})
+        inputs["transpilation_settings"].update(
+            {"optimization_settings": {"level": options.get("optimization_level")}}
+        )
+        inputs["resilience_settings"] = {"level": options.get("resilience_level")}
+        inputs["run_options"] = options.get("execution")
+        return inputs
+
+    @staticmethod
+    def _get_runtime_options(options: Dict) -> Dict:
+        """Extract runtime options.
+
+        Returns:
+            Runtime options.
+        """
+        experimental = options.get("experimental") or {}
+        return {
+            "backend_name": options.get("backend"),
+            "log_level": options.get("log_level"),
+            "image": experimental.get("image", None),
+        }
diff --git a/qiskit_ibm_runtime/qiskit/primitives/base_estimator.py b/qiskit_ibm_runtime/qiskit/primitives/base_estimator.py
index 09185a6b..3e4f94ff 100644
--- a/qiskit_ibm_runtime/qiskit/primitives/base_estimator.py
+++ b/qiskit_ibm_runtime/qiskit/primitives/base_estimator.py
@@ -135,8 +135,8 @@ class BaseEstimator(ABC):
 
     def __init__(
         self,
-        circuits: Iterable[QuantumCircuit] | QuantumCircuit,
-        observables: Iterable[SparsePauliOp] | SparsePauliOp,
+        circuits: Iterable[QuantumCircuit] | QuantumCircuit | None,
+        observables: Iterable[SparsePauliOp] | SparsePauliOp | None,
         parameters: Iterable[Iterable[Parameter]] | None = None,
     ):
         """
@@ -154,6 +154,11 @@ class BaseEstimator(ABC):
         Raises:
             QiskitError: For mismatch of circuits and parameters list.
         """
+        if circuits is None and observables is None:
+            self._circuits: tuple = tuple()
+            self._observables: tuple = tuple()
+            return
+
         if isinstance(circuits, QuantumCircuit):
             circuits = (circuits,)
         self._circuits = tuple(circuits)
@@ -185,9 +190,9 @@ class BaseEstimator(ABC):
 
     def __new__(  # pylint: disable=unused-argument
         cls,
-        circuits: Iterable[QuantumCircuit] | QuantumCircuit,
-        observables: Iterable[SparsePauliOp] | SparsePauliOp,
         *args: Any,
+        circuits: Iterable[QuantumCircuit] | QuantumCircuit | None = None,
+        observables: Iterable[SparsePauliOp] | SparsePauliOp | None = None,
         parameters: Iterable[Iterable[Parameter]] | None = None,
         **kwargs: Any,
     ) -> BaseEstimator:
diff --git a/qiskit_ibm_runtime/qiskit/primitives/base_sampler.py b/qiskit_ibm_runtime/qiskit/primitives/base_sampler.py
index b6a1f363..47b03a76 100644
--- a/qiskit_ibm_runtime/qiskit/primitives/base_sampler.py
+++ b/qiskit_ibm_runtime/qiskit/primitives/base_sampler.py
@@ -121,7 +121,7 @@ class BaseSampler(ABC):
 
     def __init__(
         self,
-        circuits: Iterable[QuantumCircuit] | QuantumCircuit,
+        circuits: Iterable[QuantumCircuit] | QuantumCircuit | None,
         parameters: Iterable[Iterable[Parameter]] | None = None,
     ):
         """
@@ -133,6 +133,10 @@ class BaseSampler(ABC):
         Raises:
             QiskitError: For mismatch of circuits and parameters list.
         """
+        if circuits is None:
+            self._circuits: tuple = tuple()
+            return
+
         if isinstance(circuits, QuantumCircuit):
             circuits = (circuits,)
         self._circuits = tuple(circuits)
@@ -153,13 +157,17 @@ class BaseSampler(ABC):
 
     def __new__(  # pylint: disable=unused-argument
         cls,
-        circuits: Iterable[QuantumCircuit] | QuantumCircuit,
         *args: Any,
+        circuits: Iterable[QuantumCircuit] | QuantumCircuit | None = None,
         parameters: Iterable[Iterable[Parameter]] | None = None,
         **kwargs: Any,
     ) -> BaseSampler:
 
         self = super().__new__(cls)
+        if circuits is None:
+            self._circuit_ids = []
+            return self
+
         if isinstance(circuits, Iterable):
             circuits = copy(circuits)
             self._circuit_ids = [id(circuit) for circuit in circuits]
diff --git a/qiskit_ibm_runtime/qiskit_runtime_service.py b/qiskit_ibm_runtime/qiskit_runtime_service.py
index cd495dc0..3be85a03 100644
--- a/qiskit_ibm_runtime/qiskit_runtime_service.py
+++ b/qiskit_ibm_runtime/qiskit_runtime_service.py
@@ -19,6 +19,7 @@ import warnings
 from datetime import datetime
 from collections import OrderedDict
 from typing import Dict, Callable, Optional, Union, List, Any, Type
+from dataclasses import asdict
 
 from qiskit.providers.backend import BackendV1 as Backend
 from qiskit.providers.provider import ProviderV1 as Provider
@@ -47,9 +48,10 @@ from .runtime_session import RuntimeSession  # pylint: disable=cyclic-import
 from .utils import RuntimeDecoder, to_base64_string, to_python_identifier
 from .utils.backend_decoder import configuration_from_server_data
 from .utils.hgp import to_instance_format, from_instance_format
-from .utils.utils import validate_job_tags
+from .utils.utils import validate_job_tags, validate_runtime_options
 from .api.client_parameters import ClientParameters
 from .runtime_options import RuntimeOptions
+from .utils.deprecation import deprecate_function
 
 logger = logging.getLogger(__name__)
 
@@ -90,7 +92,7 @@ class QiskitRuntimeService(Provider):
         }
 
         # Configure backend options
-        options = {'backend_name': "ibmq_qasm_simulator"}
+        options = {'backend': "ibmq_qasm_simulator"}
 
         # Execute the circuit using the "sampler" program.
         job = service.run(program_id="sampler",
@@ -851,8 +853,17 @@ class QiskitRuntimeService(Provider):
             program_id: Program ID.
             inputs: Program input parameters. These input values are passed
                 to the runtime program.
-            options: Runtime options that control the execution environment. See
-                :class:`RuntimeOptions` for all available options.
+            options: Runtime options that control the execution environment.
+                The use of :class:`RuntimeOptions` has been deprecated.
+
+                * backend: target backend to run on. This is required for ``ibm_quantum`` runtime.
+                * image: the runtime image used to execute the program, specified in
+                    the form of ``image_name:tag``. Not all accounts are
+                    authorized to select a different image.
+                * log_level: logging level to set in the execution environment. The valid
+                    log levels are: ``DEBUG``, ``INFO``, ``WARNING``, ``ERROR``, and ``CRITICAL``.
+                    The default level is ``WARNING``.
+
             callback: Callback function to be invoked for any interim results and final result.
                 The callback function will receive 2 positional parameters:
 
@@ -891,29 +902,29 @@ class QiskitRuntimeService(Provider):
             inputs = vars(inputs)
 
         if options is None:
-            options = RuntimeOptions()
-        if isinstance(options, dict):
-            options = RuntimeOptions(**options)
-
-        options.validate(channel=self.channel)
+            options = {}
+        elif isinstance(options, RuntimeOptions):
+            options = asdict(options)
+        options["backend"] = options.get("backend", options.get("backend_name", None))
+        validate_runtime_options(options=options, channel=self.channel)
 
         backend = None
         hgp_name = None
         if self._channel == "ibm_quantum":
             # Find the right hgp
-            hgp = self._get_hgp(instance=instance, backend_name=options.backend_name)
-            backend = hgp.backend(options.backend_name)
+            hgp = self._get_hgp(instance=instance, backend_name=options["backend"])
+            backend = hgp.backend(options["backend"])
             hgp_name = hgp.name
 
         result_decoder = result_decoder or ResultDecoder
         try:
             response = self._api_client.program_run(
                 program_id=program_id,
-                backend_name=options.backend_name,
+                backend_name=options["backend"],
                 params=inputs,
-                image=options.image,
+                image=options.get("image"),
                 hgp=hgp_name,
-                log_level=options.log_level,
+                log_level=options.get("log_level"),
                 session_id=session_id,
                 job_tags=job_tags,
                 max_execution_time=max_execution_time,
@@ -938,10 +949,15 @@ class QiskitRuntimeService(Provider):
             params=inputs,
             user_callback=callback,
             result_decoder=result_decoder,
-            image=options.image,
+            image=options.get("image"),
         )
         return job
 
+    @deprecate_function(
+        "QiskitRuntimeService.open",
+        "0.7",
+        "Instead, use the qiskit_ibm_runtime.Session class to create a runtime session.",
+    )
     def open(
         self,
         program_id: str,
diff --git a/qiskit_ibm_runtime/runtime_job.py b/qiskit_ibm_runtime/runtime_job.py
index f327c66f..7db7c112 100644
--- a/qiskit_ibm_runtime/runtime_job.py
+++ b/qiskit_ibm_runtime/runtime_job.py
@@ -117,7 +117,7 @@ class RuntimeJob:
         self._program_id = program_id
         self._status = JobStatus.INITIALIZING
         self._reason: Optional[str] = None
-        self._error_message = None  # type: Optional[str]
+        self._error_message: Optional[str] = None
         self._result_decoder = result_decoder
         self._image = image
         self._final_interim_results = False
diff --git a/qiskit_ibm_runtime/runtime_options.py b/qiskit_ibm_runtime/runtime_options.py
index a2879569..9a78cb67 100644
--- a/qiskit_ibm_runtime/runtime_options.py
+++ b/qiskit_ibm_runtime/runtime_options.py
@@ -15,9 +15,11 @@
 import re
 import logging
 from dataclasses import dataclass
-from typing import Optional
+from typing import Optional, Any
 
 from .exceptions import IBMInputValueError
+from .utils.deprecation import issue_deprecation_msg
+from .options import Options
 
 
 @dataclass
@@ -34,6 +36,15 @@ class RuntimeOptions:
             The default level is ``WARNING``.
     """
 
+    # pylint: disable=unused-argument
+    def __new__(cls, *args: Any, **kwargs: Any) -> "RuntimeOptions":
+        issue_deprecation_msg(
+            msg="The RuntimeOptions class has been deprecated",
+            version="0.7",
+            remedy="Please use qiskit_ibm_runtime.Options class instead.",
+        )
+        return super().__new__(cls)
+
     backend_name: Optional[str] = None
     image: Optional[str] = None
     log_level: Optional[str] = None
@@ -65,3 +76,10 @@ class RuntimeOptions:
                 f"{self.log_level} is not a valid log level. The valid log levels are: `DEBUG`, "
                 f"`INFO`, `WARNING`, `ERROR`, and `CRITICAL`."
             )
+
+    def _to_new_options(self) -> Options:
+        return Options(
+            backend=self.backend_name,
+            log_level=self.log_level,
+            experimental={"image": self.image},
+        )
diff --git a/qiskit_ibm_runtime/runtime_program.py b/qiskit_ibm_runtime/runtime_program.py
index a90a0ee4..d2e6bc5f 100644
--- a/qiskit_ibm_runtime/runtime_program.py
+++ b/qiskit_ibm_runtime/runtime_program.py
@@ -437,3 +437,7 @@ class ParameterNamespace(SimpleNamespace):
             "-" * len(header),
             params_str,
         )
+
+    def to_dict(self) -> Dict:
+        """Convert to dictionary."""
+        return self.__program_params
diff --git a/qiskit_ibm_runtime/sampler.py b/qiskit_ibm_runtime/sampler.py
index 6cb0d4e4..dad15e17 100644
--- a/qiskit_ibm_runtime/sampler.py
+++ b/qiskit_ibm_runtime/sampler.py
@@ -12,17 +12,24 @@
 
 """Sampler primitive."""
 
+from __future__ import annotations
 from typing import Dict, Iterable, Optional, Sequence, Any, Union
+import copy
 
 from qiskit.circuit import QuantumCircuit, Parameter
+import qiskit_ibm_runtime.session as session_pkg
 
 # TODO import BaseSampler and SamplerResult from terra once released
 from .qiskit.primitives import BaseSampler, SamplerResult
-from .exceptions import IBMInputValueError
-from .ibm_backend import IBMBackend
 from .qiskit_runtime_service import QiskitRuntimeService
-from .runtime_session import RuntimeSession
-from .utils.converters import hms_to_seconds
+from .options import Options
+from .runtime_options import RuntimeOptions
+from .program.result_decoder import ResultDecoder
+from .runtime_job import RuntimeJob
+from .utils.deprecation import deprecate_arguments, issue_deprecation_msg
+
+# pylint: disable=unused-import,cyclic-import
+from .session import Session
 
 
 class Sampler(BaseSampler):
@@ -31,213 +38,174 @@ class Sampler(BaseSampler):
     Qiskit Runtime Sampler primitive service calculates probabilities or quasi-probabilities
     of bitstrings from quantum circuits.
 
-    Sampler can be initialized with following parameters.
-
-    * circuits: a (parameterized) :class:`~qiskit.circuit.QuantumCircuit` or
-        a list of (parameterized) :class:`~qiskit.circuit.QuantumCircuit`.
-
-    * parameters: a list of parameters of the quantum circuits.
-        (:class:`~qiskit.circuit.parametertable.ParameterView` or
-        a list of :class:`~qiskit.circuit.Parameter`) specifying the order
-        in which parameter values will be bound.
-
-    * skip_transpilation: Transpilation is skipped if set to True.
-        False by default.
-
-    * service: Optional instance of :class:`qiskit_ibm_runtime.QiskitRuntimeService` class,
-        defaults to `QiskitRuntimeService()` which tries to initialize your default saved account.
-
-    * options: Runtime options dictionary that control the execution environment.
-
-        * backend: Optional instance of :class:`qiskit_ibm_runtime.IBMBackend` class or
-            string name of backend, if not specified a backend will be selected
-            automatically (IBM Cloud only).
-        * image: the runtime image used to execute the program, specified in
-            the form of ``image_name:tag``. Not all accounts are
-            authorized to select a different image.
-        * log_level: logging level to set in the execution environment. The valid
-            log levels are: ``DEBUG``, ``INFO``, ``WARNING``, ``ERROR``, and ``CRITICAL``.
-            The default level is ``WARNING``.
-
-    The returned instance can be called repeatedly with the following parameters to
-    calculate probabilities or quasi-probabilities.
-
-    * circuits: a (parameterized) :class:`~qiskit.circuit.QuantumCircuit` or
-        a list of (parameterized) :class:`~qiskit.circuit.QuantumCircuit` or a list of
-        circuit indices.
+    The :meth:`run` method can be used to submit circuits and parameters to the Sampler primitive.
 
-    * parameter_values: An optional list of concrete parameters to be bound.
-
-    * circuit_indices: (DEPRECATED) A list of circuit indices.
-
-    All the above lists should be of the same length.
+    You are encourage to use :class:`~qiskit_ibm_runtime.Session` to open a session,
+    during which you can invoke one or more primitive programs. Jobs sumitted within a session
+    are prioritized by the scheduler, and data is cached for efficiency.
 
     Example::
 
-        from qiskit import QuantumCircuit
-        from qiskit.circuit.library import RealAmplitudes
-
-        from qiskit_ibm_runtime import QiskitRuntimeService, Sampler
+        from qiskit.test.reference_circuits import ReferenceCircuits
+        from qiskit_ibm_runtime import QiskitRuntimeService, Session, Sampler
 
         service = QiskitRuntimeService(channel="ibm_cloud")
-        options = { "backend": "ibmq_qasm_simulator" }
-
-        bell = QuantumCircuit(2)
-        bell.h(0)
-        bell.cx(0, 1)
-        bell.measure_all()
-
-        # executes a Bell circuit
-        with Sampler(circuits=[bell], service=service, options=options) as sampler:
-            # pass circuits as indices
-            result = sampler(circuits=[0], parameter_values=[[]])
-            print(result)
-
-        # executes three Bell circuits
-        with Sampler(circuits=[bell]*3, service=service, options=options) as sampler:
-            # alternatively you can also pass circuits as objects
-            result = sampler(circuits=[bell]*3, parameter_values=[[]]*3)
-            print(result)
-
-        # parameterized circuit
-        pqc = RealAmplitudes(num_qubits=2, reps=2)
-        pqc.measure_all()
-        pqc2 = RealAmplitudes(num_qubits=2, reps=3)
-        pqc2.measure_all()
-
-        theta1 = [0, 1, 1, 2, 3, 5]
-        theta2 = [1, 2, 3, 4, 5, 6]
-        theta3 = [0, 1, 2, 3, 4, 5, 6, 7]
-
-        with Sampler(circuits=[pqc, pqc2], service=service, options=options) as sampler:
-            result = sampler(circuits=[0, 0, 1], parameter_values=[theta1, theta2, theta3])
-            print(result)
+        bell = ReferenceCircuits.bell()
+
+        with Session(service) as session:
+            sampler = Sampler(session=session)
+            sampler.options.resilience_level = 1
+
+            job = sampler.run(bell)
+            print(f"Job ID: {job.job_id}")
+            print(f"Job result: {job.result()}")
     """
 
+    _PROGRAM_ID = "sampler"
+
     def __init__(
         self,
-        circuits: Union[QuantumCircuit, Iterable[QuantumCircuit]],
+        circuits: Optional[Union[QuantumCircuit, Iterable[QuantumCircuit]]] = None,
         parameters: Optional[Iterable[Iterable[Parameter]]] = None,
         service: Optional[QiskitRuntimeService] = None,
-        options: Optional[Dict] = None,
+        session: Optional[Session] = None,
+        options: Optional[Union[Dict, RuntimeOptions, Options]] = None,
         skip_transpilation: Optional[bool] = False,
-        transpilation_settings: Optional[Dict] = None,
-        resilience_settings: Optional[Dict] = None,
-        max_time: Optional[Union[int, str]] = None,
     ):
         """Initializes the Sampler primitive.
 
         Args:
-            circuits: a (parameterized) :class:`~qiskit.circuit.QuantumCircuit` or
+            circuits: (DEPRECATED) A (parameterized) :class:`~qiskit.circuit.QuantumCircuit` or
                 a list of (parameterized) :class:`~qiskit.circuit.QuantumCircuit`.
 
-            parameters: A list of parameters of the quantum circuits
+            parameters: (DEPRECATED) A list of parameters of the quantum circuits
                 (:class:`~qiskit.circuit.parametertable.ParameterView` or
                 a list of :class:`~qiskit.circuit.Parameter`)
 
-            service: Optional instance of :class:`qiskit_ibm_runtime.QiskitRuntimeService` class,
+            service: (DEPRECATED) Optional instance of
+                :class:`qiskit_ibm_runtime.QiskitRuntimeService` class,
                 defaults to `QiskitRuntimeService()` which tries to initialize your default
                 saved account.
 
-            options: Runtime options dictionary that control the execution environment.
-
-                * backend: Optional instance of :class:`qiskit_ibm_runtime.IBMBackend` class or
-                    string name of backend, if not specified a backend will be selected
-                    automatically (IBM Cloud only).
-                * image: the runtime image used to execute the program, specified in
-                    the form of ``image_name:tag``. Not all accounts are
-                    authorized to select a different image.
-                * log_level: logging level to set in the execution environment. The valid
-                    log levels are: ``DEBUG``, ``INFO``, ``WARNING``, ``ERROR``, and ``CRITICAL``.
-                    The default level is ``WARNING``.
-
-            skip_transpilation: Transpilation is skipped if set to True. False by default.
-
-            transpilation_settings: (EXPERIMENTAL setting, can break between releases without warning)
-                Qiskit transpiler settings. The transpilation process converts
-                operations in the circuit to those supported by the backend, swaps qubits with the
-                circuit to overcome limited qubit connectivity and some optimizations to reduce the
-                circuit's gate count where it can.
-
-                * skip_transpilation: Transpilation is skipped if set to True.
-                    False by default.
-
-                * optimization_settings:
-
-                    * level: How much optimization to perform on the circuits.
-                        Higher levels generate more optimized circuits,
-                        at the expense of longer transpilation times.
-                        * 0: no optimization
-                        * 1: light optimization
-                        * 2: heavy optimization
-                        * 3: even heavier optimization
-                        If ``None``, level 1 will be chosen as default.
-
-            resilience_settings: (EXPERIMENTAL setting, can break between releases without warning)
-                Using these settings allows you to build resilient algorithms by
-                leveraging the state of the art error suppression, mitigation and correction techniques.
-
-                * level: How much resilience to build against errors.
-                    Higher levels generate more accurate results,
-                    at the expense of longer processing times.
-                    * 0: no resilience
-                    * 1: light resilience
-                    If ``None``, level 0 will be chosen as default.
-
-            max_time: (EXPERIMENTAL setting, can break between releases without warning)
-                Maximum amount of time, a runtime session can be open before being
-                forcibly closed. Can be specified as seconds (int) or a string like "2h 30m 40s".
+            session: Session in which to call the sampler primitive. If ``None``, a new session
+                is created using the default saved account.
 
-        Raises:
-            IBMInputValueError: If an input value is invalid.
+            options: Primitive options, see :class:`Options` for detailed description.
+
+            skip_transpilation (DEPRECATED): Transpilation is skipped if set to True. False by default.
+                Ignored ``skip_transpilation`` is also specified in ``options``.
         """
+        # TODO: Fix base classes once done
         super().__init__(
-            circuits=circuits if isinstance(circuits, Iterable) else [circuits],
+            circuits=circuits,
             parameters=parameters,
         )
-        self._skip_transpilation = skip_transpilation
-        if not service:
-            # try to initialize service with default saved account
-            service = QiskitRuntimeService()
-        self._service = service
-        if isinstance(options, dict) and "backend" in options:
-            backend = options.get("backend")
-            if isinstance(backend, IBMBackend):
-                del options["backend"]
-                options["backend_name"] = backend.name
-            elif isinstance(backend, str):
-                del options["backend"]
-                options["backend_name"] = backend
-            else:
-                raise IBMInputValueError(
-                    "'backend' property in 'options' should be either the string name of the "
-                    "backend or an instance of 'IBMBackend' class"
-                )
+
+        # TODO: Remove deprecation warnings if done in base class
+        if circuits or parameters:
+            deprecate_arguments(
+                "circuits and parameters",
+                "0.7",
+                f"You can instead specify these inputs using the {self.__class__.__name__}.run method.",
+            )
+        if skip_transpilation:
+            deprecate_arguments(
+                "skip_transpilation",
+                "0.7",
+                "Instead, use the skip_transpilation keyword argument in transpilation_settings.",
+            )
+        if service:
+            deprecate_arguments(
+                "service", "0.7", "Please use the session parameter instead."
+            )
+
+        if options is None:
+            self.options = Options()
+        elif isinstance(options, Options):
+            self.options = copy.deepcopy(options)
+            skip_transpilation = self.options.transpilation.skip_transpilation
+        elif isinstance(options, RuntimeOptions):
+            self.options = options._to_new_options()
+        else:
+            self.options = Options._from_dict(options)
+            skip_transpilation = options.get("transpilation", {}).get(
+                "skip_transpilation", False
+            )
+        self.options.transpilation.skip_transpilation = skip_transpilation
+
+        self._initial_inputs = {"circuits": circuits, "parameters": parameters}
+        if session:
+            self._session = session
+        else:
+            if session_pkg._DEFAULT_SESSION is None:
+                session_pkg._DEFAULT_SESSION = Session(service=service)
+            self._session = session_pkg._DEFAULT_SESSION
+
+    def run(
+        self,
+        circuits: Union[QuantumCircuit, Sequence[QuantumCircuit]],
+        parameter_values: Sequence[Sequence[float]] | None = None,
+        parameters: Sequence[Sequence[Parameter]] | None = None,
+        **kwargs: Any,
+    ) -> RuntimeJob:
+        """Submit a request to the sampler primitive program.
+
+        Args:
+            circuits: A (parameterized) :class:`~qiskit.circuit.QuantumCircuit` or
+                a list of (parameterized) :class:`~qiskit.circuit.QuantumCircuit`.
+
+            parameter_values: An optional list of concrete parameters to be bound.
+
+            parameters: A list of parameters of the quantum circuits
+                (:class:`~qiskit.circuit.parametertable.ParameterView` or
+                a list of :class:`~qiskit.circuit.Parameter`).
+                Defaults to ``[circ.parameters for circ in circuits]``.
+
+            **kwargs: Individual options to overwrite the default primitive options.
+
+        Returns:
+            Submitted job.
+
+        Raises:
+            ValueError: If the input values are invalid.
+        """
+        if isinstance(circuits, Iterable) and not all(
+            isinstance(inst, QuantumCircuit) for inst in circuits
+        ):
+            raise ValueError(
+                "The circuits parameter has to be instances of QuantumCircuit."
+            )
+
+        circ_count = 1 if isinstance(circuits, QuantumCircuit) else len(circuits)
+
         inputs = {
             "circuits": circuits,
             "parameters": parameters,
-            "skip_transpilation": self._skip_transpilation,
+            "circuit_indices": list(range(circ_count)),
+            "parameter_values": parameter_values,
         }
-        if transpilation_settings:
-            inputs.update({"transpilation_settings": transpilation_settings})
-        if resilience_settings:
-            inputs.update({"resilience_settings": resilience_settings})
-        self._session = RuntimeSession(
-            service=self._service,
-            program_id="sampler",
+        combined = self.options._merge_options(kwargs)
+        inputs.update(Options._get_program_inputs(combined))
+
+        return self._session.run(
+            program_id=self._PROGRAM_ID,
             inputs=inputs,
-            options=options,
-            max_time=self.calculate_max_time(max_time=max_time),
+            options=Options._get_runtime_options(combined),
+            result_decoder=SamplerResultDecoder,
         )
 
-    def calculate_max_time(self, max_time: Optional[Union[int, str]] = None) -> int:
-        """Calculate max_time in seconds from hour minute seconds string. Ex: 2h 30m 40s"""
-        try:
-            return hms_to_seconds(max_time) if isinstance(max_time, str) else max_time
-        except IBMInputValueError as input_value_error:
-            raise IBMInputValueError(
-                "Invalid value given for max_time.", input_value_error.message
-            )
+    def __call__(
+        self,
+        circuits: Sequence[int | QuantumCircuit],
+        parameter_values: Sequence[Sequence[float]] | None = None,
+        **run_options: Any,
+    ) -> SamplerResult:
+        issue_deprecation_msg(
+            msg="Calling a Sampler instance directly has been deprecated ",
+            version="0.7",
+            remedy="Please use qiskit_ibm_runtime.Session and Sampler.run() instead.",
+        )
+        return super().__call__(circuits, parameter_values, **run_options)
 
     def _call(
         self,
@@ -268,17 +236,45 @@ class Sampler(BaseSampler):
         Returns:
             An instance of :class:`qiskit.primitives.SamplerResult`.
         """
-        self._session.write(
-            circuit_indices=circuits,
-            parameter_values=parameter_values,
-            run_options=run_options,
-        )
-        raw_result = self._session.read()
-        return SamplerResult(
-            quasi_dists=raw_result["quasi_dists"],
-            metadata=raw_result["metadata"],
-        )
+
+        inputs = {
+            "circuits": self._initial_inputs["circuits"],
+            "parameters": self._initial_inputs["parameters"],
+            "circuit_indices": circuits,
+            "parameter_values": parameter_values,
+        }
+        combined = self.options._merge_options(run_options)
+        inputs.update(Options._get_program_inputs(combined))
+
+        return self._session.run(
+            program_id=self._PROGRAM_ID,
+            inputs=inputs,
+            options=Options._get_runtime_options(combined),
+            result_decoder=SamplerResultDecoder,
+        ).result()
 
     def close(self) -> None:
         """Close the session and free resources"""
         self._session.close()
+
+    @property
+    def session(self) -> Session:
+        """Return session used by this primitive.
+
+        Returns:
+            Session used by this primitive.
+        """
+        return self._session
+
+
+class SamplerResultDecoder(ResultDecoder):
+    """Class used to decode sampler results."""
+
+    @classmethod
+    def decode(cls, raw_result: str) -> SamplerResult:
+        """Convert the result to SamplerResult."""
+        decoded: Dict = super().decode(raw_result)
+        return SamplerResult(
+            quasi_dists=decoded["quasi_dists"],
+            metadata=decoded["metadata"],
+        )
diff --git a/qiskit_ibm_runtime/session.py b/qiskit_ibm_runtime/session.py
new file mode 100644
index 00000000..015820b0
--- /dev/null
+++ b/qiskit_ibm_runtime/session.py
@@ -0,0 +1,166 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2022.
+#
+# This code is licensed under the Apache License, Version 2.0. You may
+# obtain a copy of this license in the LICENSE.txt file in the root directory
+# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
+#
+# Any modifications or derivative works of this code must retain this
+# copyright notice, and modified files need to carry a notice indicating
+# that they have been altered from the originals.
+
+"""Qiskit Runtime flexible session."""
+
+from typing import Dict, Optional, Type, Union
+from types import TracebackType
+from functools import wraps
+
+from qiskit_ibm_runtime import QiskitRuntimeService
+from .runtime_job import RuntimeJob
+from .runtime_program import ParameterNamespace
+from .program.result_decoder import ResultDecoder
+from .utils.converters import hms_to_seconds
+
+
+def _active_session(func):  # type: ignore
+    """Decorator used to ensure the session is active."""
+
+    @wraps(func)
+    def _wrapper(self, *args, **kwargs):  # type: ignore
+        if not self._active:
+            raise RuntimeError("The session is closed.")
+        return func(self, *args, **kwargs)
+
+    return _wrapper
+
+
+class Session:
+    """Class for creating a flexible Qiskit Runtime session.
+
+    The ``Session`` class allows you to open a "session" with the Qiskit Runtime service.
+    Jobs submitted during a session get prioritized scheduling. This class allows
+    you to submit jobs to one or more of the primitives. For example::
+
+        from qiskit.test.reference_circuits import ReferenceCircuits
+
+        with Session() as session:
+            sampler = session.sampler()
+            sampler.options.backend = "ibmq_qasm_simulator"
+            sampler.settings.transpilation.optimization_level = 1
+            job = sampler.run(circ)
+            print(f"Sampler job ID: {job.job_id}")
+            print(f"Sampler job result:" {job.result()})
+    """
+
+    def __init__(
+        self,
+        service: Optional[QiskitRuntimeService] = None,
+        max_time: Optional[Union[int, str]] = None,
+    ):
+        """Session constructor.
+        Args:
+            service: Optional instance of :class:`qiskit_ibm_runtime.QiskitRuntimeService` class,
+                defaults to `QiskitRuntimeService()` which tries to initialize
+                your default saved account.
+            max_time: (EXPERIMENTAL setting, can break between releases without warning)
+                Maximum amount of time, a runtime session can be open before being
+                forcibly closed. Can be specified as seconds (int) or a string like "2h 30m 40s".
+        """
+
+        self._service = service or QiskitRuntimeService()
+        self._session_id: Optional[str] = None
+        self._active = True
+
+        self._max_time = (
+            max_time
+            if max_time is None or isinstance(max_time, int)
+            else hms_to_seconds(max_time, "Invalid max_time value: ")
+        )
+
+    @_active_session
+    def run(
+        self,
+        program_id: str,
+        inputs: Union[Dict, ParameterNamespace],
+        options: Optional[Dict] = None,
+        result_decoder: Optional[Type[ResultDecoder]] = None,
+    ) -> RuntimeJob:
+        """Run a program in the session.
+
+        Args:
+            program_id: Program ID.
+            inputs: Program input parameters. These input values are passed
+                to the runtime program.
+            options: Runtime options that control the execution environment.
+
+                * backend: target backend to run on. This is required for ``ibm_quantum`` runtime.
+                * image: the runtime image used to execute the program, specified in
+                    the form of ``image_name:tag``. Not all accounts are
+                    authorized to select a different image.
+                * log_level: logging level to set in the execution environment. The valid
+                    log levels are: ``DEBUG``, ``INFO``, ``WARNING``, ``ERROR``, and ``CRITICAL``.
+                    The default level is ``WARNING``.
+
+        Returns:
+            Submitted job.
+        """
+
+        # TODO: Cache data when server supports it.
+
+        # TODO: Do we really need to specify a None max time if session has started?
+        max_time = self._max_time if not self._session_id else None
+
+        job = self._service.run(
+            program_id=program_id,
+            options=options,
+            inputs=inputs,
+            session_id=self._session_id,
+            start_session=self._session_id is None,
+            max_execution_time=max_time,
+            result_decoder=result_decoder,
+        )
+
+        if self._session_id is None:
+            self._session_id = job.job_id
+
+        return job
+
+    def close(self) -> None:
+        """Close the session."""
+        self._active = False
+        if self._session_id:
+            self._service._api_client.close_session(self._session_id)
+
+    @property
+    def session_id(self) -> str:
+        """Return the session ID.
+
+        Returns:
+            Session ID.
+        """
+        return self._session_id
+
+    @property
+    def service(self) -> QiskitRuntimeService:
+        """Return service associated with this session.
+
+        Returns:
+            :class:`qiskit_ibm_runtime.QiskitRuntimeService` associated with this session.
+        """
+        return self._service
+
+    def __enter__(self) -> "Session":
+        return self
+
+    def __exit__(
+        self,
+        exc_type: Optional[Type[BaseException]],
+        exc_val: Optional[BaseException],
+        exc_tb: Optional[TracebackType],
+    ) -> None:
+        self.close()
+
+
+# Default session
+_DEFAULT_SESSION: Optional[Session] = None
diff --git a/qiskit_ibm_runtime/utils/converters.py b/qiskit_ibm_runtime/utils/converters.py
index 5caad839..dab4a1ff 100644
--- a/qiskit_ibm_runtime/utils/converters.py
+++ b/qiskit_ibm_runtime/utils/converters.py
@@ -137,11 +137,12 @@ def duration_difference(date_time: datetime) -> str:
     return time_str
 
 
-def hms_to_seconds(hms: str) -> int:
+def hms_to_seconds(hms: str, msg_prefix: str = "") -> int:
     """Convert duration specified as hours minutes seconds to seconds.
 
     Args:
         hms: The string input duration (in hours minutes seconds). Ex: 2h 10m 20s
+        msg_prefix: Additional message to prefix the error.
 
     Returns:
         Total seconds (int) in the duration.
@@ -158,4 +159,4 @@ def hms_to_seconds(hms: str) -> int:
             timedelta(hours=hours, minutes=minutes, seconds=seconds).total_seconds()
         )
     except parser.ParserError as parser_error:
-        raise IBMInputValueError(str(parser_error))
+        raise IBMInputValueError(msg_prefix + str(parser_error))
diff --git a/qiskit_ibm_runtime/utils/deprecation.py b/qiskit_ibm_runtime/utils/deprecation.py
new file mode 100644
index 00000000..0913a04f
--- /dev/null
+++ b/qiskit_ibm_runtime/utils/deprecation.py
@@ -0,0 +1,85 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2022.
+#
+# This code is licensed under the Apache License, Version 2.0. You may
+# obtain a copy of this license in the LICENSE.txt file in the root directory
+# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
+#
+# Any modifications or derivative works of this code must retain this
+# copyright notice, and modified files need to carry a notice indicating
+# that they have been altered from the originals.
+
+"""Deprecation utilities"""
+
+from typing import Callable, Any
+import functools
+import warnings
+
+
+def deprecate_function(
+    deprecated: str, version: str, remedy: str, stacklevel: int = 2
+) -> Callable:
+    """Emit a warning prior to calling decorated function.
+    Args:
+        deprecated: Function being deprecated.
+        version: First release the function is deprecated.
+        remedy: User action to take.
+        stacklevel: The warning stackevel to use.
+
+    Returns:
+        The decorated, deprecated callable.
+    """
+
+    def decorator(func: Callable) -> Callable:
+        @functools.wraps(func)
+        def wrapper(*args: Any, **kwargs: Any) -> Callable:
+            issue_deprecation_msg(
+                f"The {deprecated} method is deprecated",
+                version,
+                remedy,
+                stacklevel + 1,
+            )
+            return func(*args, **kwargs)
+
+        return wrapper
+
+    return decorator
+
+
+def deprecate_arguments(
+    deprecated: str, version: str, remedy: str, stacklevel: int = 2
+) -> None:
+    """Emit a warning about deprecated keyword arguments.
+
+    Args:
+        deprecated: Keyword arguments being deprecated.
+        version: First release the function is deprecated.
+        remedy: User action to take.
+        stacklevel: The warning stackevel to use.
+    """
+    issue_deprecation_msg(
+        f"The {deprecated} keyword arguments are deprecated",
+        version,
+        remedy,
+        stacklevel + 1,
+    )
+
+
+def issue_deprecation_msg(
+    msg: str, version: str, remedy: str, stacklevel: int = 2
+) -> None:
+    """Emit a deprecation warning.
+
+    Args:
+        msg: Deprecation message.
+        version: First release the function is deprecated.
+        remedy: User action to take.
+        stacklevel: The warning stackevel to use.
+    """
+    warnings.warn(
+        f"{msg} as of qiskit-ibm-runtime {version} "
+        f"and will be removed no sooner than 3 months after the release date. {remedy}",
+        DeprecationWarning,
+        stacklevel=stacklevel + 1,  # Increment to account for this function.
+    )
diff --git a/qiskit_ibm_runtime/utils/utils.py b/qiskit_ibm_runtime/utils/utils.py
index 2459a276..df67350f 100644
--- a/qiskit_ibm_runtime/utils/utils.py
+++ b/qiskit_ibm_runtime/utils/utils.py
@@ -25,6 +25,8 @@ import requests
 from ibm_cloud_sdk_core.authenticators import IAMAuthenticator
 from ibm_platform_services import ResourceControllerV2
 
+from ..exceptions import IBMInputValueError
+
 
 def validate_job_tags(
     job_tags: Optional[List[str]], exception: Type[Exception]
@@ -45,6 +47,33 @@ def validate_job_tags(
         raise exception("job_tags needs to be a list of strings.")
 
 
+def validate_runtime_options(options: Dict, channel: str) -> None:
+    """Validate runtime options.
+
+    Args:
+        options: Runtime options to validate.
+
+    Raises:
+        IBMInputValueError: If input values are invalid.
+    """
+    if options.get("image") and not re.match(
+        "[a-zA-Z0-9]+([/.\\-_][a-zA-Z0-9]+)*:[a-zA-Z0-9]+([.\\-_][a-zA-Z0-9]+)*$",
+        options["image"],
+    ):
+        raise IBMInputValueError('"image" needs to be in form of image_name:tag')
+
+    if channel == "ibm_quantum" and not options.get("backend"):
+        raise IBMInputValueError('"backend" is required for ``ibm_quantum`` runtime.')
+
+    if options.get("log_level") and not isinstance(
+        logging.getLevelName(options["log_level"].upper()), int
+    ):
+        raise IBMInputValueError(
+            f"{options['log_level']} is not a valid log level. The valid log levels are: "
+            "`DEBUG`, `INFO`, `WARNING`, `ERROR`, and `CRITICAL`."
+        )
+
+
 def get_iam_api_url(cloud_url: str) -> str:
     """Computes the IAM API URL for the given IBM Cloud URL."""
     parsed_url = urlparse(cloud_url)


Test patch:
diff --git a/test/integration/test_estimator.py b/test/integration/test_estimator.py
index 45f591cd..5b361ec4 100644
--- a/test/integration/test_estimator.py
+++ b/test/integration/test_estimator.py
@@ -15,7 +15,7 @@
 from qiskit.circuit.library import RealAmplitudes
 from qiskit.quantum_info import SparsePauliOp
 
-from qiskit_ibm_runtime import Estimator, EstimatorResult, BaseEstimator
+from qiskit_ibm_runtime import Estimator, EstimatorResult, BaseEstimator, Session
 
 from ..decorators import run_integration_test
 from ..ibm_test_case import IBMIntegrationTestCase
@@ -25,7 +25,7 @@ class TestIntegrationEstimator(IBMIntegrationTestCase):
     """Integration tests for Estimator primitive."""
 
     @run_integration_test
-    def test_estimator_primitive(self, service):
+    def test_estimator_session(self, service):
         """Verify if estimator primitive returns expected results"""
 
         options = {"backend": "ibmq_qasm_simulator"}
@@ -38,6 +38,83 @@ class TestIntegrationEstimator(IBMIntegrationTestCase):
         H2 = SparsePauliOp.from_list([("IZ", 1)])
         H3 = SparsePauliOp.from_list([("ZI", 1), ("ZZ", 1)])
 
+        with Session(service=service) as session:
+            estimator = session.estimator(options=options)
+            self.assertIsInstance(estimator, BaseEstimator)
+
+            theta1 = [0, 1, 1, 2, 3, 5]
+            theta2 = [0, 1, 1, 2, 3, 5, 8, 13]
+            theta3 = [1, 2, 3, 4, 5, 6]
+
+            circuits1 = [psi1]
+            # calculate [ <psi1(theta1)|H1|psi1(theta1)> ]
+            job = estimator.run(
+                circuits=circuits1, observables=[H1], parameter_values=[theta1]
+            )
+            result1 = job.result()
+            self.assertIsInstance(result1, EstimatorResult)
+            self.assertEqual(len(result1.values), len(circuits1))
+            self.assertEqual(len(result1.metadata), len(circuits1))
+
+            circuits2 = circuits1 * 2
+            # calculate [ <psi1(theta1)|H2|psi1(theta1)>, <psi1(theta1)|H3|psi1(theta1)> ]
+            job = estimator.run(
+                circuits=circuits2, observables=[H2, H3], parameter_values=[theta1] * 2
+            )
+            result2 = job.result()
+            self.assertIsInstance(result2, EstimatorResult)
+            self.assertEqual(len(result2.values), len(circuits2))
+            self.assertEqual(len(result2.metadata), len(circuits2))
+
+            circuits3 = [psi2]
+            # calculate [ <psi2(theta2)|H2|psi2(theta2)> ]
+            job = estimator.run(
+                circuits=circuits3, observables=[H2], parameter_values=[theta2]
+            )
+            result3 = job.result()
+            self.assertIsInstance(result3, EstimatorResult)
+            self.assertEqual(len(result3.values), len(circuits3))
+            self.assertEqual(len(result3.metadata), len(circuits3))
+
+            # calculate [ <psi1(theta1)|H1|psi1(theta1)>, <psi1(theta3)|H1|psi1(theta3)> ]
+            job = estimator.run(
+                circuits=circuits2,
+                observables=[H1, H1],
+                parameter_values=[theta1, theta3],
+            )
+            result4 = job.result()
+            self.assertIsInstance(result4, EstimatorResult)
+            self.assertEqual(len(result4.values), len(circuits2))
+            self.assertEqual(len(result4.metadata), len(circuits2))
+
+            circuits5 = [psi1, psi2, psi1]
+            # calculate [ <psi1(theta1)|H1|psi1(theta1)>,
+            #             <psi2(theta2)|H2|psi2(theta2)>,
+            #             <psi1(theta3)|H3|psi1(theta3)> ]
+            job = estimator.run(
+                circuits=circuits5,
+                observables=[H1, H2, H3],
+                parameter_values=[theta1, theta2, theta3],
+            )
+            result5 = job.result()
+            self.assertIsInstance(result5, EstimatorResult)
+            self.assertEqual(len(result5.values), len(circuits5))
+            self.assertEqual(len(result5.metadata), len(circuits5))
+
+    @run_integration_test
+    def test_estimator_primitive(self, service):
+        """Test to verify that estimator as a primitive still works."""
+
+        options = {"backend": "ibmq_qasm_simulator"}
+
+        psi1 = RealAmplitudes(num_qubits=2, reps=2)
+        psi2 = RealAmplitudes(num_qubits=2, reps=3)
+
+        # pylint: disable=invalid-name
+        H1 = SparsePauliOp.from_list([("II", 1), ("IZ", 2), ("XI", 3)])
+        H2 = SparsePauliOp.from_list([("IZ", 1)])
+        H3 = SparsePauliOp.from_list([("ZI", 1), ("ZZ", 1)])
+
         with Estimator(
             circuits=[psi1, psi2],
             observables=[H1, H2, H3],
diff --git a/test/integration/test_sampler.py b/test/integration/test_sampler.py
index 16b5cf88..fd1c5a13 100644
--- a/test/integration/test_sampler.py
+++ b/test/integration/test_sampler.py
@@ -12,10 +12,12 @@
 
 """Integration tests for Sampler primitive."""
 
+from qiskit.circuit import QuantumCircuit, Gate
 from qiskit.circuit.library import RealAmplitudes
 from qiskit.test.reference_circuits import ReferenceCircuits
 
-from qiskit_ibm_runtime import Sampler, BaseSampler, SamplerResult
+from qiskit_ibm_runtime import Sampler, BaseSampler, SamplerResult, Session
+from qiskit_ibm_runtime.exceptions import RuntimeJobFailureError
 
 from ..decorators import run_integration_test
 from ..ibm_test_case import IBMIntegrationTestCase
@@ -24,32 +26,39 @@ from ..ibm_test_case import IBMIntegrationTestCase
 class TestIntegrationIBMSampler(IBMIntegrationTestCase):
     """Integration tests for Sampler primitive."""
 
+    def setUp(self) -> None:
+        super().setUp()
+        self.bell = ReferenceCircuits.bell()
+        self.options = {"backend": "ibmq_qasm_simulator"}
+
     @run_integration_test
-    def test_sampler_primitive_non_parameterized_circuits(self, service):
+    def test_sampler_non_parameterized_single_circuit(self, service):
         """Verify if sampler primitive returns expected results for non-parameterized circuits."""
 
-        options = {"backend": "ibmq_qasm_simulator"}
-
-        bell = ReferenceCircuits.bell()
-
-        # executes a Bell circuit
-        with Sampler(circuits=bell, service=service, options=options) as sampler:
+        # Execute a Bell circuit
+        with Session(service=service) as session:
+            sampler = session.sampler(options=self.options)
             self.assertIsInstance(sampler, BaseSampler)
-
-            circuits0 = [0]
-            result = sampler(circuits=circuits0, parameter_values=[[]])
+            job = sampler.run(circuits=self.bell)
+            result = job.result()
             self.assertIsInstance(result, SamplerResult)
-            self.assertEqual(len(result.quasi_dists), len(circuits0))
-            self.assertEqual(len(result.metadata), len(circuits0))
+            self.assertEqual(len(result.quasi_dists), 1)
+            self.assertEqual(len(result.metadata), 1)
             self.assertAlmostEqual(result.quasi_dists[0]["11"], 0.5, delta=0.05)
             self.assertAlmostEqual(result.quasi_dists[0]["00"], 0.5, delta=0.05)
+            self.assertTrue(session.session_id)
 
-        # executes three Bell circuits
-        with Sampler(circuits=[bell] * 3, service=service, options=options) as sampler:
+    @run_integration_test
+    def test_sampler_non_parameterized_circuits(self, service):
+        """Test sampler with multiple non-parameterized circuits."""
+        # Execute three Bell circuits
+        with Session(service=service) as session:
+            sampler = session.sampler(options=self.options)
             self.assertIsInstance(sampler, BaseSampler)
+            circuits = [self.bell] * 3
 
-            circuits1 = [0, 1, 2]
-            result1 = sampler(circuits=circuits1, parameter_values=[[]] * 3)
+            circuits1 = circuits
+            result1 = sampler.run(circuits=circuits1).result()
             self.assertIsInstance(result1, SamplerResult)
             self.assertEqual(len(result1.quasi_dists), len(circuits1))
             self.assertEqual(len(result1.metadata), len(circuits1))
@@ -57,8 +66,8 @@ class TestIntegrationIBMSampler(IBMIntegrationTestCase):
                 self.assertAlmostEqual(result1.quasi_dists[i]["11"], 0.5, delta=0.05)
                 self.assertAlmostEqual(result1.quasi_dists[i]["00"], 0.5, delta=0.05)
 
-            circuits2 = [0, 2]
-            result2 = sampler(circuits=circuits2, parameter_values=[[]] * 2)
+            circuits2 = [circuits[0], circuits[2]]
+            result2 = sampler.run(circuits=circuits2).result()
             self.assertIsInstance(result2, SamplerResult)
             self.assertEqual(len(result2.quasi_dists), len(circuits2))
             self.assertEqual(len(result2.metadata), len(circuits2))
@@ -66,8 +75,8 @@ class TestIntegrationIBMSampler(IBMIntegrationTestCase):
                 self.assertAlmostEqual(result2.quasi_dists[i]["11"], 0.5, delta=0.05)
                 self.assertAlmostEqual(result2.quasi_dists[i]["00"], 0.5, delta=0.05)
 
-            circuits3 = [1, 2]
-            result3 = sampler(circuits=circuits3, parameter_values=[[]] * 2)
+            circuits3 = [circuits[1], circuits[2]]
+            result3 = sampler.run(circuits=circuits3).result()
             self.assertIsInstance(result3, SamplerResult)
             self.assertEqual(len(result3.quasi_dists), len(circuits3))
             self.assertEqual(len(result3.metadata), len(circuits3))
@@ -79,7 +88,60 @@ class TestIntegrationIBMSampler(IBMIntegrationTestCase):
     def test_sampler_primitive_parameterized_circuits(self, service):
         """Verify if sampler primitive returns expected results for parameterized circuits."""
 
-        options = {"backend": "ibmq_qasm_simulator"}
+        # parameterized circuit
+        pqc = RealAmplitudes(num_qubits=2, reps=2)
+        pqc.measure_all()
+        pqc2 = RealAmplitudes(num_qubits=2, reps=3)
+        pqc2.measure_all()
+
+        theta1 = [0, 1, 1, 2, 3, 5]
+        theta2 = [1, 2, 3, 4, 5, 6]
+        theta3 = [0, 1, 2, 3, 4, 5, 6, 7]
+
+        with Session(service=service) as session:
+            sampler = session.sampler(options=self.options)
+            self.assertIsInstance(sampler, BaseSampler)
+
+            circuits0 = [pqc, pqc, pqc2]
+            result = sampler.run(
+                circuits=circuits0,
+                parameter_values=[theta1, theta2, theta3],
+            ).result()
+            self.assertIsInstance(result, SamplerResult)
+            self.assertEqual(len(result.quasi_dists), len(circuits0))
+            self.assertEqual(len(result.metadata), len(circuits0))
+
+    @run_integration_test
+    def test_sampler_skip_transpile(self, service):
+        """Test skip transpilation option."""
+        circ = QuantumCircuit(1, 1)
+        custom_gate = Gate("my_custom_gate", 1, [3.14, 1])
+        circ.append(custom_gate, [0])
+        circ.measure(0, 0)
+
+        with Session(service=service) as session:
+            sampler = session.sampler(
+                options=self.options,
+                transpilation_settings={"skip_transpilation": True},
+            )
+            with self.assertRaises(RuntimeJobFailureError) as err:
+                sampler.run(circuits=circ).result()
+                # If transpilation not skipped the error would be something about cannot expand.
+                self.assertIn("invalid instructions", err.exception.message)
+
+    @run_integration_test
+    def test_sampler_optimization_level(self, service):
+        """Test transpiler optimization level is properly mapped."""
+        with Session(service=service) as session:
+            sampler = session.sampler(options=self.options)
+            sampler.settings.transpilation.optimization_level = 3
+            result = sampler.run(self.bell).result()
+            self.assertAlmostEqual(result.quasi_dists[0]["11"], 0.5, delta=0.05)
+            self.assertAlmostEqual(result.quasi_dists[0]["00"], 0.5, delta=0.05)
+
+    @run_integration_test
+    def test_sampler_primitive_as_session(self, service):
+        """Verify Sampler as a session still works."""
 
         # parameterized circuit
         pqc = RealAmplitudes(num_qubits=2, reps=2)
@@ -91,7 +153,9 @@ class TestIntegrationIBMSampler(IBMIntegrationTestCase):
         theta2 = [1, 2, 3, 4, 5, 6]
         theta3 = [0, 1, 2, 3, 4, 5, 6, 7]
 
-        with Sampler(circuits=[pqc, pqc2], service=service, options=options) as sampler:
+        with Sampler(
+            circuits=[pqc, pqc2], service=service, options=self.options
+        ) as sampler:
             self.assertIsInstance(sampler, BaseSampler)
 
             circuits0 = [pqc, pqc, pqc2]
diff --git a/test/unit/test_ibm_primitives.py b/test/unit/test_ibm_primitives.py
new file mode 100644
index 00000000..4bcb3f2e
--- /dev/null
+++ b/test/unit/test_ibm_primitives.py
@@ -0,0 +1,197 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2021.
+#
+# This code is licensed under the Apache License, Version 2.0. You may
+# obtain a copy of this license in the LICENSE.txt file in the root directory
+# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
+#
+# Any modifications or derivative works of this code must retain this
+# copyright notice, and modified files need to carry a notice indicating
+# that they have been altered from the originals.
+
+"""Tests for primitive classes."""
+
+import sys
+from unittest.mock import MagicMock, patch
+from dataclasses import asdict
+
+from qiskit_ibm_runtime import Sampler, Estimator, Options, RuntimeOptions
+from ..ibm_test_case import IBMTestCase
+
+
+class TestPrimitives(IBMTestCase):
+    """Class for testing the Sampler class."""
+
+    def test_skip_transpilation(self):
+        """Test skip_transpilation is hornored."""
+        primitives = [Sampler, Estimator]
+        for cls in primitives:
+            with self.subTest(primitive=cls):
+                inst = cls(session=MagicMock(), skip_transpilation=True)
+                self.assertTrue(inst.options.transpilation.skip_transpilation)
+
+    def test_skip_transpilation_overwrite(self):
+        """Test overwriting skip_transpilation."""
+        options = Options()
+        options.transpilation.skip_transpilation = False
+        primitives = [Sampler, Estimator]
+        for cls in primitives:
+            with self.subTest(primitive=cls):
+                inst = cls(
+                    session=MagicMock(), options=options, skip_transpilation=True
+                )
+                self.assertFalse(inst.options.transpilation.skip_transpilation)
+
+    def test_dict_options(self):
+        """Test passing a dictionary as options."""
+        options_vars = [
+            {},
+            {
+                "resilience_level": 1,
+                "transpilation": {"seed_transpiler": 24},
+                "execution": {"shots": 100, "init_qubits": True},
+            },
+            {"transpilation": {}},
+        ]
+        primitives = [Sampler, Estimator]
+        for cls in primitives:
+            for options in options_vars:
+                with self.subTest(primitive=cls, options=options):
+                    inst = cls(session=MagicMock(), options=options)
+                    expected = asdict(Options())
+                    self._update_dict(expected, options)
+                    self.assertDictEqual(expected, asdict(inst.options))
+
+    def test_runtime_options(self):
+        """Test passing in runtime options."""
+        options_vars = [
+            {"backend_name": "foo", "image": "foo:bar"},
+            RuntimeOptions(backend_name="foo", image="foo:bar"),
+        ]
+        primitives = [Sampler, Estimator]
+        for cls in primitives:
+            for options in options_vars:
+                with self.subTest(primitive=cls, options=options):
+                    inst = cls(session=MagicMock(), options=options)
+                    if not isinstance(options, dict):
+                        options = asdict(options)
+                    self.assertEqual(options["backend_name"], inst.options.backend)
+                    self.assertEqual(
+                        options["image"], inst.options.experimental["image"]
+                    )
+
+    @patch("qiskit_ibm_runtime.estimator.Session")
+    @patch("qiskit_ibm_runtime.sampler.Session")
+    def test_default_session(self, *_):
+        """Test a session is created if not passed in."""
+        sampler = Sampler()
+        self.assertIsNotNone(sampler.session)
+        estimator = Estimator()
+        self.assertEqual(estimator.session, sampler.session)
+
+    def test_run_inputs_default(self):
+        """Test run using default options."""
+        session = MagicMock()
+        options_vars = [
+            (Options(resilience_level=9), {"resilience_settings": {"level": 9}}),
+            (
+                Options(optimization_level=8),
+                {"transpilation_settings": {"optimization_settings": {"level": 8}}},
+            ),
+            (
+                {"transpilation": {"seed_transpiler": 24}, "execution": {"shots": 100}},
+                {
+                    "transpilation_settings": {"seed_transpiler": 24},
+                    "run_options": {"shots": 100},
+                },
+            ),
+        ]
+        primitives = [Sampler, Estimator]
+        for cls in primitives:
+            for options, expected in options_vars:
+                with self.subTest(primitive=cls, options=options):
+                    inst = cls(session=session, options=options)
+                    inst.run(MagicMock(), MagicMock())
+                    if sys.version_info >= (3, 8):
+                        inputs = session.run.call_args.kwargs["inputs"]
+                    else:
+                        _, kwargs = session.run.call_args
+                        inputs = kwargs["inputs"]
+
+                    self._assert_dict_paritally_equal(inputs, expected)
+
+    def test_run_inputs_updated_default(self):
+        """Test run using updated default options."""
+        session = MagicMock()
+        primitives = [Sampler, Estimator]
+        for cls in primitives:
+            with self.subTest(primitive=cls):
+                inst = cls(session=session)
+                inst.options.resilience_level = 1
+                inst.options.optimization_level = 2
+                inst.options.execution.shots = 3
+                inst.run(MagicMock(), MagicMock())
+                if sys.version_info >= (3, 8):
+                    inputs = session.run.call_args.kwargs["inputs"]
+                else:
+                    _, kwargs = session.run.call_args
+                    inputs = kwargs["inputs"]
+                self._assert_dict_paritally_equal(
+                    inputs,
+                    {
+                        "resilience_settings": {"level": 1},
+                        "transpilation_settings": {
+                            "optimization_settings": {"level": 2}
+                        },
+                        "run_options": {"shots": 3},
+                    },
+                )
+
+    def test_run_inputs_overwrite(self):
+        """Test run using overwritten options."""
+        session = MagicMock()
+        options_vars = [
+            ({"resilience_level": 9}, {"resilience_settings": {"level": 9}}),
+            ({"shots": 200}, {"run_options": {"shots": 200}}),
+            (
+                {"optimization_level": 8},
+                {"transpilation_settings": {"optimization_settings": {"level": 8}}},
+            ),
+            (
+                {"seed_transpiler": 24, "optimization_level": 8},
+                {
+                    "transpilation_settings": {
+                        "optimization_settings": {"level": 8},
+                        "seed_transpiler": 24,
+                    }
+                },
+            ),
+        ]
+        primitives = [Sampler, Estimator]
+        for cls in primitives:
+            for options, expected in options_vars:
+                with self.subTest(primitive=cls, options=options):
+                    inst = cls(session=session)
+                    inst.run(MagicMock(), MagicMock(), **options)
+                    if sys.version_info >= (3, 8):
+                        inputs = session.run.call_args.kwargs["inputs"]
+                    else:
+                        _, kwargs = session.run.call_args
+                        inputs = kwargs["inputs"]
+                    self._assert_dict_paritally_equal(inputs, expected)
+                    self.assertDictEqual(asdict(inst.options), asdict(Options()))
+
+    def _update_dict(self, dict1, dict2):
+        for key, val in dict1.items():
+            if isinstance(val, dict):
+                self._update_dict(val, dict2.pop(key, {}))
+            elif key in dict2.keys():
+                dict1[key] = dict2.pop(key)
+
+    def _assert_dict_paritally_equal(self, dict1, dict2):
+        for key, val in dict2.items():
+            if isinstance(val, dict):
+                self._assert_dict_paritally_equal(dict1.get(key), val)
+            elif key in dict1:
+                self.assertEqual(val, dict1[key])