diff --git a/CHANGELOG.md b/CHANGELOG.md
index fb1f4dea3..1f11e953c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,6 +8,9 @@ Changelog
 
 ### Improvements and Changes
 
+- Both `get_qc` and `QPU` now accept an `endpoint_id` argument which is used to engage
+  against a specific QCS [quantum processor endpoint](https://docs.api.qcs.rigetti.com/#tag/endpoints).
+
 ### Bugfixes
 
 - Allow `np.ndarray` when writing QAM memory. Disallow non-integer and non-float types.
diff --git a/docs/source/advanced_usage.rst b/docs/source/advanced_usage.rst
index 4c2c21a01..2bb7e570e 100644
--- a/docs/source/advanced_usage.rst
+++ b/docs/source/advanced_usage.rst
@@ -82,6 +82,30 @@ Below is an example that demonstrates how to use pyQuil in a multithreading scen
         print(f"Results for program {i}:\n{result}\n")
 
 
+Alternative QPU Endpoints
+~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Rigetti QCS supports alternative endpoints for access to a QPU architecture, useful for very particular cases.
+Generally, this is useful to call "mock" or test endpoints, which simulate the results of execution for the
+purposes of integration testing without the need for an active reservation or contention with other users.
+See the `QCS API Docs <https://docs.api.qcs.rigetti.com/#tag/endpoints>`_ for more information on QPU Endpoints.
+
+To be able to call these endpoints using pyQuil, enter the ``endpoint_id`` of your desired endpoint in one
+of the sites where ``quantum_processor_id`` is used:
+
+.. code:: python
+
+    # Option 1
+    qc = get_qc("Aspen-9", endpoint_id="my_endpoint")
+
+    # Option 2
+    qam = QPU("Aspen-9", endpoint_id="my_endpoint")
+
+After doing so, for all intents and purposes - compilation, optimization, etc - your program will behave the same
+as when using "default" endpoint for a given quantum processor, except that it will be executed by an
+alternate QCS service, and the results of execution should not be treated as correct or meaningful.
+
+
 Using Qubit Placeholders
 ~~~~~~~~~~~~~~~~~~~~~~~~
 
diff --git a/poetry.lock b/poetry.lock
index 366b3f9d2..0c1874881 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -903,21 +903,6 @@ python-versions = "*"
 freezegun = ">0.3"
 pytest = ">=3.0.0"
 
-[[package]]
-name = "pytest-httpx"
-version = "0.9.0"
-description = "Send responses to httpx."
-category = "dev"
-optional = false
-python-versions = ">=3.6"
-
-[package.dependencies]
-httpx = ">=0.15.0,<0.16.0"
-pytest = ">=6.0.0,<7.0.0"
-
-[package.extras]
-testing = ["pytest-asyncio (>=0.14.0,<0.15.0)", "pytest-cov (>=2.0.0,<3.0.0)"]
-
 [[package]]
 name = "pytest-mock"
 version = "3.6.1"
@@ -1076,6 +1061,17 @@ urllib3 = ">=1.21.1,<1.27"
 security = ["pyOpenSSL (>=0.14)", "cryptography (>=1.3.4)"]
 socks = ["PySocks (>=1.5.6,!=1.5.7)", "win-inet-pton"]
 
+[[package]]
+name = "respx"
+version = "0.15.1"
+description = "A utility for mocking out the Python HTTPX and HTTP Core libraries."
+category = "dev"
+optional = false
+python-versions = ">=3.6"
+
+[package.dependencies]
+httpx = ">=0.15"
+
 [[package]]
 name = "retry"
 version = "0.9.2"
@@ -1415,7 +1411,7 @@ latex = ["ipython"]
 [metadata]
 lock-version = "1.1"
 python-versions = "^3.7"
-content-hash = "c059d449facf631a118d653669a4f0891da7d2a38e5402232bd5310045997975"
+content-hash = "050a9333fe1cc8e432323d9a4ca369d5b984da16a5b286a63d5ae10445cff067"
 
 [metadata.files]
 alabaster = [
@@ -1912,10 +1908,6 @@ pytest-freezegun = [
     {file = "pytest-freezegun-0.4.2.zip", hash = "sha256:19c82d5633751bf3ec92caa481fb5cffaac1787bd485f0df6436fd6242176949"},
     {file = "pytest_freezegun-0.4.2-py2.py3-none-any.whl", hash = "sha256:5318a6bfb8ba4b709c8471c94d0033113877b3ee02da5bfcd917c1889cde99a7"},
 ]
-pytest-httpx = [
-    {file = "pytest_httpx-0.9.0-py3-none-any.whl", hash = "sha256:f337cc19176600ed0615c5ec8390646306857d35ee49e5b662e68dfc0fc17dce"},
-    {file = "pytest_httpx-0.9.0.tar.gz", hash = "sha256:7bfcc9344e13f441068181fb909fc86a545f25b4343ad98de55c1327ab336bfd"},
-]
 pytest-mock = [
     {file = "pytest-mock-3.6.1.tar.gz", hash = "sha256:40217a058c52a63f1042f0784f62009e976ba824c418cced42e88d5f40ab0e62"},
     {file = "pytest_mock-3.6.1-py3-none-any.whl", hash = "sha256:30c2f2cc9759e76eee674b81ea28c9f0b94f8f0445a1b87762cadf774f0df7e3"},
@@ -2072,6 +2064,10 @@ requests = [
     {file = "requests-2.25.1-py2.py3-none-any.whl", hash = "sha256:c210084e36a42ae6b9219e00e48287def368a26d03a048ddad7bfee44f75871e"},
     {file = "requests-2.25.1.tar.gz", hash = "sha256:27973dd4a904a4f13b263a19c866c13b92a39ed1c964655f025f3f8d3d75b804"},
 ]
+respx = [
+    {file = "respx-0.15.1-py2.py3-none-any.whl", hash = "sha256:07b69af4f127e6651ab0fd104a484bcb9d98b901b25234d4158851ff5a37e34a"},
+    {file = "respx-0.15.1.tar.gz", hash = "sha256:d3438b7ec2edb5a4f575c0ca5a51c37b9a55c42c6693d178a1917aeca290de7f"},
+]
 retry = [
     {file = "retry-0.9.2-py2.py3-none-any.whl", hash = "sha256:ccddf89761fa2c726ab29391837d4327f819ea14d244c232a1d24c67a2f98606"},
     {file = "retry-0.9.2.tar.gz", hash = "sha256:f8bfa8b99b69c4506d6f5bd3b0aabf77f98cdb17f3c9fc3f5ca820033336fba4"},
diff --git a/pyproject.toml b/pyproject.toml
index 90fa3b279..b86dc74cf 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -45,13 +45,13 @@ black = "^20.8b1"
 flake8 = "^3.8.1"
 pytest = "^6.2.2"
 pytest-cov = "^2.11.1"
-pytest-httpx = "^0.9"
 mypy = "0.740"
 pytest-xdist = "^2.2.1"
 pytest-rerunfailures = "^9.1.1"
 pytest-timeout = "^1.4.2"
 pytest-mock = "^3.6.1"
 pytest-freezegun = "^0.4.2"
+respx = "^0.15"
 
 [tool.poetry.extras]
 latex = ["ipython"]
diff --git a/pyquil/api/_engagement_manager.py b/pyquil/api/_engagement_manager.py
index 9e6b2ea8a..5ff22523b 100644
--- a/pyquil/api/_engagement_manager.py
+++ b/pyquil/api/_engagement_manager.py
@@ -15,13 +15,14 @@
 ##############################################################################
 import threading
 from datetime import datetime
-from typing import Dict, Optional, TYPE_CHECKING
+from typing import Dict, NamedTuple, Optional, TYPE_CHECKING
 
 from dateutil.parser import parse as parsedate
 from dateutil.tz import tzutc
 from qcs_api_client.client import QCSClientConfiguration
 from qcs_api_client.models import EngagementWithCredentials, CreateEngagementRequest
 from qcs_api_client.operations.sync import create_engagement
+from qcs_api_client.types import UNSET
 
 from pyquil.api._qcs_client import qcs_client
 
@@ -29,6 +30,11 @@
     import httpx
 
 
+class EngagementCacheKey(NamedTuple):
+    quantum_processor_id: str
+    endpoint_id: Optional[str]
+
+
 class EngagementManager:
     """
     Fetches (and caches) engagements for use when accessing a QPU.
@@ -44,28 +50,36 @@ def __init__(self, *, client_configuration: QCSClientConfiguration) -> None:
         :param client_configuration: Client configuration, used for refreshing engagements.
         """
         self._client_configuration = client_configuration
-        self._cached_engagements: Dict[str, EngagementWithCredentials] = {}
+        self._cached_engagements: Dict[EngagementCacheKey, EngagementWithCredentials] = {}
         self._lock = threading.Lock()
 
-    def get_engagement(self, *, quantum_processor_id: str, request_timeout: float = 10.0) -> EngagementWithCredentials:
+    def get_engagement(
+        self, *, quantum_processor_id: str, request_timeout: float = 10.0, endpoint_id: Optional[str] = None
+    ) -> EngagementWithCredentials:
         """
-        Gets an engagement for the given quantum processor. If an engagement was already fetched previously and
-        remains valid, it will be returned instead of creating a new engagement.
+        Gets an engagement for the given quantum processor endpoint.
+
+        If an engagement was already fetched previously and remains valid, it will be returned instead
+        of creating a new engagement.
 
         :param quantum_processor_id: Quantum processor being engaged.
         :param request_timeout: Timeout for request, in seconds.
+        :param endpoint_id: Optional ID of the endpoint to use for engagement. If provided, it must
+            correspond to an endpoint serving the provided Quantum Processor.
         :return: Fetched or cached engagement.
         """
+        key = EngagementCacheKey(quantum_processor_id, endpoint_id)
+
         with self._lock:
-            if not self._engagement_valid(self._cached_engagements.get(quantum_processor_id)):
+            if not self._engagement_valid(self._cached_engagements.get(key)):
                 with qcs_client(
                     client_configuration=self._client_configuration, request_timeout=request_timeout
                 ) as client:  # type: httpx.Client
-                    request = CreateEngagementRequest(quantum_processor_id=quantum_processor_id)
-                    self._cached_engagements[quantum_processor_id] = create_engagement(
-                        client=client, json_body=request
-                    ).parsed
-            return self._cached_engagements[quantum_processor_id]
+                    request = CreateEngagementRequest(
+                        quantum_processor_id=quantum_processor_id, endpoint_id=endpoint_id or UNSET
+                    )
+                    self._cached_engagements[key] = create_engagement(client=client, json_body=request).parsed
+            return self._cached_engagements[key]
 
     @staticmethod
     def _engagement_valid(engagement: Optional[EngagementWithCredentials]) -> bool:
diff --git a/pyquil/api/_qpu.py b/pyquil/api/_qpu.py
index 76a98ed74..a9bdeaf8b 100644
--- a/pyquil/api/_qpu.py
+++ b/pyquil/api/_qpu.py
@@ -115,6 +115,7 @@ def __init__(
         timeout: float = 10.0,
         client_configuration: Optional[QCSClientConfiguration] = None,
         engagement_manager: Optional[EngagementManager] = None,
+        endpoint_id: Optional[str] = None,
     ) -> None:
         """
         A connection to the QPU.
@@ -123,10 +124,9 @@ def __init__(
         :param priority: The priority with which to insert jobs into the QPU queue. Lower integers
             correspond to higher priority.
         :param timeout: Time limit for requests, in seconds.
-        :param client_configuration: Optional client configuration. If none is provided, a default
-            one will be loaded.
-        :param engagement_manager: Optional engagement manager. If none is provided, a default one
-            will be created.
+        :param client_configuration: Optional client configuration. If none is provided, a default one will be loaded.
+        :param endpoint_id: Optional endpoint ID to be used for engagement.
+        :param engagement_manager: Optional engagement manager. If none is provided, a default one will be created.
         """
         super().__init__()
 
@@ -136,6 +136,7 @@ def __init__(
         engagement_manager = engagement_manager or EngagementManager(client_configuration=client_configuration)
         self._qpu_client = QPUClient(
             quantum_processor_id=quantum_processor_id,
+            endpoint_id=endpoint_id,
             engagement_manager=engagement_manager,
             request_timeout=timeout,
         )
diff --git a/pyquil/api/_qpu_client.py b/pyquil/api/_qpu_client.py
index c4d56864e..b65cb2aa7 100644
--- a/pyquil/api/_qpu_client.py
+++ b/pyquil/api/_qpu_client.py
@@ -15,7 +15,7 @@
 ##############################################################################
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Dict, cast, Tuple, Union, List, Any
+from typing import Dict, Optional, cast, Tuple, Union, List, Any
 
 import rpcq
 from dateutil.parser import parse as parsedate
@@ -104,6 +104,7 @@ def __init__(
         *,
         quantum_processor_id: str,
         engagement_manager: EngagementManager,
+        endpoint_id: Optional[str] = None,
         request_timeout: float = 10.0,
     ) -> None:
         """
@@ -114,6 +115,7 @@ def __init__(
         :param request_timeout: Timeout for requests, in seconds.
         """
         self.quantum_processor_id = quantum_processor_id
+        self._endpoint_id = endpoint_id
         self._engagement_manager = engagement_manager
         self.timeout = request_timeout
 
@@ -157,6 +159,7 @@ def get_buffers(self, request: GetBuffersRequest) -> GetBuffersResponse:
     @retry(exceptions=TimeoutError, tries=2)  # type: ignore
     def _rpcq_request(self, method_name: str, *args: Any, **kwargs: Any) -> Any:
         engagement = self._engagement_manager.get_engagement(
+            endpoint_id=self._endpoint_id,
             quantum_processor_id=self.quantum_processor_id,
             request_timeout=self.timeout,
         )
diff --git a/pyquil/api/_quantum_computer.py b/pyquil/api/_quantum_computer.py
index 29b5d58cc..c9e31b588 100644
--- a/pyquil/api/_quantum_computer.py
+++ b/pyquil/api/_quantum_computer.py
@@ -725,6 +725,7 @@ def get_qc(
     compiler_timeout: float = 10.0,
     execution_timeout: float = 10.0,
     client_configuration: Optional[QCSClientConfiguration] = None,
+    endpoint_id: Optional[str] = None,
     engagement_manager: Optional[EngagementManager] = None,
 ) -> QuantumComputer:
     """
@@ -794,9 +795,12 @@ def get_qc(
     :param compiler_timeout: Time limit for compilation requests, in seconds.
     :param execution_timeout: Time limit for execution requests, in seconds.
     :param client_configuration: Optional client configuration. If none is provided, a default one will be loaded.
+    :param endpoint_id: Optional quantum processor endpoint ID, as used in the `QCS API Docs`_.
     :param engagement_manager: Optional engagement manager. If none is provided, a default one will be created.
 
     :return: A pre-configured QuantumComputer
+
+    .. _QCS API Docs: https://docs.api.qcs.rigetti.com/#tag/endpoints
     """
 
     client_configuration = client_configuration or QCSClientConfiguration.load()
@@ -862,6 +866,7 @@ def get_qc(
             quantum_processor_id=quantum_processor.quantum_processor_id,
             timeout=execution_timeout,
             client_configuration=client_configuration,
+            endpoint_id=endpoint_id,
             engagement_manager=engagement_manager,
         )
         compiler = QPUCompiler(
diff --git a/test/unit/test_engagement_manager.py b/test/unit/test_engagement_manager.py
index 2c0c2c81f..63c0b1ac8 100644
--- a/test/unit/test_engagement_manager.py
+++ b/test/unit/test_engagement_manager.py
@@ -14,47 +14,93 @@
 #    limitations under the License.
 ##############################################################################
 import json
+from typing import List, Optional
 
-from pytest_httpx import HTTPXMock
-from qcs_api_client.models import EngagementWithCredentials, EngagementCredentials
-
-from pyquil.api._engagement_manager import EngagementManager
+import httpx
+import respx
 from pyquil.api import QCSClientConfiguration
+from pyquil.api._engagement_manager import EngagementManager
+from qcs_api_client.models import EngagementCredentials, EngagementWithCredentials
 
+DEFAULT_ENDPOINT_ID = "some-endpoint"
 
+@respx.mock
 def test_get_engagement__refreshes_engagement_when_cached_engagement_expired(
-    client_configuration: QCSClientConfiguration, httpx_mock: HTTPXMock
+    client_configuration: QCSClientConfiguration,
 ):
     engagement_manager = EngagementManager(client_configuration=client_configuration)
     cache_engagement(
-        engagement_manager, expired_engagement(quantum_processor_id="some-processor"), client_configuration, httpx_mock
+        engagement_manager, expired_engagement(quantum_processor_id="some-processor"), client_configuration,
     )
-    httpx_mock.add_response(
-        method="POST",
+    respx.post(
         url=f"{client_configuration.profile.api_url}/v1/engagements",
-        match_content=json.dumps({"quantumProcessorId": "some-processor"}).encode(),
-        json=unexpired_engagement(quantum_processor_id="some-processor").to_dict(),
+        json={"quantumProcessorId": "some-processor"},
+    ).respond(json=unexpired_engagement(quantum_processor_id="some-processor").to_dict())
+
+    engagement = engagement_manager.get_engagement(quantum_processor_id="some-processor")
+
+    assert engagement == unexpired_engagement(quantum_processor_id="some-processor")
+
+
+@respx.mock
+def test_get_engagement__refreshes_engagement_when_cached_engagement_expired__using_endpoint_id(
+    client_configuration: QCSClientConfiguration,
+):
+    """
+    Assert that endpoint ID is correctly used to engage against an endpoint when the cached engagement has expired.
+    """
+    engagement_manager = EngagementManager(client_configuration=client_configuration)
+    cache_engagement(
+        engagement_manager,
+        expired_engagement(quantum_processor_id="some-processor", endpoint_id="custom-endpoint"),
+        client_configuration,
     )
+    respx.post(
+        url=f"{client_configuration.profile.api_url}/v1/engagements",
+        json={"quantumProcessorId": "some-processor", "endpointId": "custom-endpoint"},
+    ).respond(json=unexpired_engagement(quantum_processor_id="some-processor").to_dict())
 
     engagement = engagement_manager.get_engagement(quantum_processor_id="some-processor")
 
     assert engagement == unexpired_engagement(quantum_processor_id="some-processor")
 
 
+@respx.mock
 def test_get_engagement__reuses_engagement_when_cached_engagement_unexpired(
-    client_configuration: QCSClientConfiguration, httpx_mock: HTTPXMock
+    client_configuration: QCSClientConfiguration,
 ):
     engagement_manager = EngagementManager(client_configuration=client_configuration)
     cached_engagement = cache_engagement(
         engagement_manager,
         unexpired_engagement(quantum_processor_id="some-processor"),
         client_configuration,
-        httpx_mock,
     )
-    network_calls_before = len(httpx_mock.get_requests())
+    network_calls_before = respx.calls.call_count
 
     engagement = engagement_manager.get_engagement(quantum_processor_id="some-processor")
-    network_calls_after = len(httpx_mock.get_requests())
+    network_calls_after = respx.calls.call_count
+
+    assert network_calls_before == network_calls_after
+    assert engagement is cached_engagement
+
+
+@respx.mock
+def test_get_engagement__reuses_engagement_when_cached_engagement_unexpired__using_endpoint_id(
+    client_configuration: QCSClientConfiguration,
+):
+    """
+    Assert that endpoint ID is correctly used to engage against an endpoint.
+    """
+    engagement_manager = EngagementManager(client_configuration=client_configuration)
+    cached_engagement = cache_engagement(
+        engagement_manager,
+        unexpired_engagement(quantum_processor_id="some-processor", endpoint_id="custom-endpoint"),
+        client_configuration,
+    )
+    network_calls_before = respx.calls.call_count
+
+    engagement = engagement_manager.get_engagement(quantum_processor_id="some-processor", endpoint_id="custom-endpoint")
+    network_calls_after = respx.calls.call_count
 
     assert network_calls_before == network_calls_after
     assert engagement is cached_engagement
@@ -64,22 +110,42 @@ def cache_engagement(
     engagement_manager: EngagementManager,
     engagement: EngagementWithCredentials,
     client_configuration: QCSClientConfiguration,
-    httpx_mock: HTTPXMock,
 ) -> EngagementWithCredentials:
-    httpx_mock.add_response(
-        method="POST",
-        url=f"{client_configuration.profile.api_url}/v1/engagements",
-        match_content=json.dumps({"quantumProcessorId": engagement.quantum_processor_id}).encode(),
-        json=engagement.to_dict(),
-    )
+    mock_engagement(client_configuration=client_configuration, engagement=engagement)
 
-    cached_engagement = engagement_manager.get_engagement(quantum_processor_id=engagement.quantum_processor_id)
+    if engagement.endpoint_id == DEFAULT_ENDPOINT_ID:
+        endpoint_id = None
+    else:
+        endpoint_id = engagement.endpoint_id
+
+    cached_engagement = engagement_manager.get_engagement(
+        quantum_processor_id=engagement.quantum_processor_id, endpoint_id=endpoint_id
+    )
 
     assert cached_engagement == engagement
     return cached_engagement
 
 
-def make_engagement(*, quantum_processor_id: str, expires_at: str) -> EngagementWithCredentials:
+def mock_engagement(engagement: EngagementWithCredentials, *, client_configuration: QCSClientConfiguration):
+    """
+    Apply and respond with an engagement when it matches.
+    """
+
+    if engagement.endpoint_id == DEFAULT_ENDPOINT_ID:
+        respx.post(
+            url=f"{client_configuration.profile.api_url}/v1/engagements",
+            json={"quantumProcessorId": engagement.quantum_processor_id}
+        ).respond(json=engagement.to_dict())
+
+    respx.post(
+        url=f"{client_configuration.profile.api_url}/v1/engagements",
+        json={"endpointId": engagement.endpoint_id}
+    ).respond(json=engagement.to_dict())
+
+
+def make_engagement(
+    *, quantum_processor_id: str, endpoint_id: Optional[str] = None, expires_at: str
+) -> EngagementWithCredentials:
     return EngagementWithCredentials(
         address="tcp://example.com/qpu",
         credentials=EngagementCredentials(
@@ -87,7 +153,7 @@ def make_engagement(*, quantum_processor_id: str, expires_at: str) -> Engagement
             client_secret="client-secret-123",
             server_public="server-public-123",
         ),
-        endpoint_id="some-endpoint",
+        endpoint_id=endpoint_id or DEFAULT_ENDPOINT_ID,
         expires_at=expires_at,
         quantum_processor_id=quantum_processor_id,
         user_id="some-user",
@@ -95,9 +161,13 @@ def make_engagement(*, quantum_processor_id: str, expires_at: str) -> Engagement
     )
 
 
-def unexpired_engagement(*, quantum_processor_id: str) -> EngagementWithCredentials:
-    return make_engagement(quantum_processor_id=quantum_processor_id, expires_at="9999-01-01T00:00:00Z")
+def unexpired_engagement(*, quantum_processor_id: str, endpoint_id: Optional[str] = None) -> EngagementWithCredentials:
+    return make_engagement(
+        quantum_processor_id=quantum_processor_id, endpoint_id=endpoint_id, expires_at="9999-01-01T00:00:00Z"
+    )
 
 
-def expired_engagement(*, quantum_processor_id: str) -> EngagementWithCredentials:
-    return make_engagement(quantum_processor_id=quantum_processor_id, expires_at="1970-01-01T00:00:00Z")
+def expired_engagement(*, quantum_processor_id: str, endpoint_id: Optional[str] = None) -> EngagementWithCredentials:
+    return make_engagement(
+        quantum_processor_id=quantum_processor_id, endpoint_id=endpoint_id, expires_at="1970-01-01T00:00:00Z"
+    )
diff --git a/test/unit/test_qpu_client.py b/test/unit/test_qpu_client.py
index ada3725f8..704256f2b 100644
--- a/test/unit/test_qpu_client.py
+++ b/test/unit/test_qpu_client.py
@@ -14,9 +14,11 @@
 #    limitations under the License.
 ##############################################################################
 from datetime import datetime, timedelta
+from typing import Optional, Union
 from unittest import mock
 
 import pytest
+from qcs_api_client.types import UNSET, Unset
 import rpcq
 from dateutil.tz import tzutc
 from pytest_mock import MockerFixture
@@ -132,13 +134,21 @@ def test_fetches_engagement_for_quantum_processor_on_request(
         engagement_manager=mock_engagement_manager,
         request_timeout=3.14,
     )
-    mock_engagement_manager.get_engagement.return_value = engagement(
-            quantum_processor_id=processor_id,
+
+    def mock_get_engagement(
+        quantum_processor_id: str, request_timeout: float = 10.0, endpoint_id: Optional[str] = None
+    ) -> EngagementWithCredentials:
+        assert quantum_processor_id == processor_id
+        assert request_timeout == qpu_client.timeout
+        return engagement(
+            quantum_processor_id="some-processor",
             seconds_left=9999,
             credentials=engagement_credentials,
             port=1234,
         )
 
+    mock_engagement_manager.get_engagement.side_effect = mock_get_engagement
+
     patch_rpcq_client(mocker=mocker, return_value="")
 
     request = RunProgramRequest(
@@ -149,6 +159,7 @@ def test_fetches_engagement_for_quantum_processor_on_request(
     )
     qpu_client.run_program(request)
     mock_engagement_manager.get_engagement.assert_called_once_with(
+        endpoint_id=None,
         quantum_processor_id=processor_id,
         request_timeout=qpu_client.timeout,
     )
@@ -246,8 +257,8 @@ def test_run_program__retries_on_timeout(
     # ASSERT
     # Engagement should be fetched twice, once per RPC call
     mock_engagement_manager.get_engagement.assert_has_calls([
-        mocker.call(quantum_processor_id='some-processor', request_timeout=1.0),
-        mocker.call(quantum_processor_id='some-processor', request_timeout=1.0),
+        mocker.call(quantum_processor_id='some-processor', request_timeout=1.0, endpoint_id=None),
+        mocker.call(quantum_processor_id='some-processor', request_timeout=1.0, endpoint_id=None),
     ])
     # RPC call should happen twice since the first one times out
     qpu_request = QPURequest(**request_kwargs)  # Thing QPUClient gives to rpcq.Client
diff --git a/test/unit/test_quantum_computer.py b/test/unit/test_quantum_computer.py
index f1c7dbff1..fd8207f51 100644
--- a/test/unit/test_quantum_computer.py
+++ b/test/unit/test_quantum_computer.py
@@ -5,6 +5,8 @@
 import networkx as nx
 import numpy as np
 import pytest
+import respx
+
 from pyquil import Program, list_quantum_computers
 from pyquil.api import QCSClientConfiguration
 from pyquil.api._quantum_computer import (
@@ -30,6 +32,7 @@
 from pyquil.pyqvm import PyQVM
 from pyquil.quantum_processor import NxQuantumProcessor
 from pyquil.quilbase import Declare, MemoryReference
+from qcs_api_client.models.instruction_set_architecture import InstructionSetArchitecture
 from rpcq.messages import ParameterAref
 
 
@@ -202,7 +205,7 @@ def test_run(client_configuration: QCSClientConfiguration):
             MEASURE(2, MemoryReference("ro", 2)),
         ).wrap_in_numshots_loop(1000)
     )
-    bitstrings = result.readout_data.get('ro')
+    bitstrings = result.readout_data.get("ro")
 
     assert bitstrings.shape == (1000, 3)
     parity = np.sum(bitstrings, axis=1) % 3
@@ -221,7 +224,7 @@ def test_run_pyqvm_noiseless(client_configuration: QCSClientConfiguration):
     for q in range(3):
         prog += MEASURE(q, ro[q])
     result = qc.run(prog.wrap_in_numshots_loop(1000))
-    bitstrings = result.readout_data.get('ro')
+    bitstrings = result.readout_data.get("ro")
 
     assert bitstrings.shape == (1000, 3)
     parity = np.sum(bitstrings, axis=1) % 3
@@ -240,7 +243,7 @@ def test_run_pyqvm_noisy(client_configuration: QCSClientConfiguration):
     for q in range(3):
         prog += MEASURE(q, ro[q])
     result = qc.run(prog.wrap_in_numshots_loop(1000))
-    bitstrings = result.readout_data.get('ro')
+    bitstrings = result.readout_data.get("ro")
 
     assert bitstrings.shape == (1000, 3)
     parity = np.sum(bitstrings, axis=1) % 3
@@ -266,7 +269,7 @@ def test_readout_symmetrization(client_configuration: QCSClientConfiguration):
     prog.wrap_in_numshots_loop(1000)
 
     result_1 = qc.run(prog)
-    bitstrings_1 = result_1.readout_data.get('ro')
+    bitstrings_1 = result_1.readout_data.get("ro")
     avg0_us = np.mean(bitstrings_1[:, 0])
     avg1_us = 1 - np.mean(bitstrings_1[:, 1])
     diff_us = avg1_us - avg0_us
@@ -422,7 +425,7 @@ def test_qc_run(client_configuration: QCSClientConfiguration):
                 MEASURE(0, ("ro", 0)),
             ).wrap_in_numshots_loop(3)
         )
-    ).readout_data.get('ro')
+    ).readout_data.get("ro")
     assert bs.shape == (3, 1)
 
 
@@ -472,7 +475,7 @@ def test_run_with_parameters(client_configuration: QCSClientConfiguration, param
     ).wrap_in_numshots_loop(1000)
 
     executable.write_memory(region_name="theta", value=param)
-    bitstrings = qc.run(executable).readout_data.get('ro')
+    bitstrings = qc.run(executable).readout_data.get("ro")
 
     assert bitstrings.shape == (1000, 1)
     assert all([bit == 1 for bit in bitstrings])
@@ -556,7 +559,7 @@ def test_get_qvm_with_topology_2(client_configuration: QCSClientConfiguration):
                 MEASURE(7, ("ro", 2)),
             ).wrap_in_numshots_loop(5)
         )
-    ).readout_data.get('ro')
+    ).readout_data.get("ro")
     assert results.shape == (5, 3)
     assert all(r[0] == 1 for r in results)
 
@@ -575,7 +578,7 @@ def test_noisy(client_configuration: QCSClientConfiguration):
         MEASURE(0, ("ro", 0)),
     ).wrap_in_numshots_loop(10000)
     qc = get_qc("1q-qvm", noisy=True, client_configuration=client_configuration)
-    result = qc.run(qc.compile(p)).readout_data.get('ro')
+    result = qc.run(qc.compile(p)).readout_data.get("ro")
     assert result.mean() < 1.0
 
 
@@ -831,3 +834,18 @@ def test_qc_expectation_on_qvm(client_configuration: QCSClientConfiguration, dum
     assert np.isclose(results[2][0].expectation, 1.0, atol=0.01)
     assert np.isclose(results[2][0].std_err, 0)
     assert results[2][0].total_counts == 20000
+
+
+@respx.mock
+def test_get_qc_endpoint_id(client_configuration: QCSClientConfiguration, qcs_aspen8_isa: InstructionSetArchitecture):
+    """
+    Assert that get_qc passes a specified ``endpoint_id`` through to its QPU when constructed
+    for a live quantum processor.
+    """
+    respx.get(
+        url=f"{client_configuration.profile.api_url}/v1/quantumProcessors/test/instructionSetArchitecture",
+    ).respond(json=qcs_aspen8_isa.to_dict())
+
+    qc = get_qc("test", endpoint_id="test-endpoint")
+
+    assert qc.qam._qpu_client._endpoint_id == "test-endpoint"
diff --git a/test/unit/test_qvm_client.py b/test/unit/test_qvm_client.py
index 57fe1270b..5254a54ce 100644
--- a/test/unit/test_qvm_client.py
+++ b/test/unit/test_qvm_client.py
@@ -14,11 +14,10 @@
 #    limitations under the License.
 ##############################################################################
 import json
-import time
 from typing import Any, Dict
 
 import httpx
-from pytest_httpx import HTTPXMock, to_response
+import respx
 from qcs_api_client.client import QCSClientConfiguration
 
 from pyquil.api._qvm_client import (
@@ -41,48 +40,34 @@ def test_init__sets_base_url_and_timeout(client_configuration: QCSClientConfigur
     assert qvm_client.timeout == 3.14
 
 
-def test_sets_timeout_on_requests(client_configuration: QCSClientConfiguration, httpx_mock: HTTPXMock):
-    qvm_client = QVMClient(client_configuration=client_configuration, request_timeout=0.1)
-
-    def assert_timeout(request: httpx.Request, ext: Dict[str, Any]):
-        assert ext["timeout"] == httpx.Timeout(qvm_client.timeout).as_dict()
-        return to_response(data="1.2.3 [abc123]")
-
-    httpx_mock.add_callback(assert_timeout)
-
-    qvm_client.get_version()
-
-
-def test_get_version__returns_version(client_configuration: QCSClientConfiguration, httpx_mock: HTTPXMock):
+@respx.mock
+def test_get_version__returns_version(client_configuration: QCSClientConfiguration):
     qvm_client = QVMClient(client_configuration=client_configuration)
 
-    httpx_mock.add_response(
+    respx.post(
         url=client_configuration.profile.applications.pyquil.qvm_url,
-        match_content=json.dumps({"type": "version"}).encode(),
-        data="1.2.3 [abc123]",
-    )
+        json={"type": "version"},
+    ).respond(status_code=200, text="1.2.3 [abc123]")
 
     assert qvm_client.get_version() == "1.2.3"
 
 
-def test_run_program__returns_results(client_configuration: QCSClientConfiguration, httpx_mock: HTTPXMock):
+@respx.mock
+def test_run_program__returns_results(client_configuration: QCSClientConfiguration):
     qvm_client = QVMClient(client_configuration=client_configuration)
 
-    httpx_mock.add_response(
+    respx.post(
         url=client_configuration.profile.applications.pyquil.qvm_url,
-        match_content=json.dumps(
-            {
-                "type": "multishot",
-                "compiled-quil": "some-program",
-                "addresses": {"ro": True},
-                "trials": 1,
-                "measurement-noise": (3.14, 1.61, 6.28),
-                "gate-noise": (1.0, 2.0, 3.0),
-                "rng-seed": 314,
-            },
-        ).encode(),
-        json={"ro": [[1, 0, 1]]},
-    )
+        json={
+            "type": "multishot",
+            "compiled-quil": "some-program",
+            "addresses": {"ro": True},
+            "trials": 1,
+            "measurement-noise": (3.14, 1.61, 6.28),
+            "gate-noise": (1.0, 2.0, 3.0),
+            "rng-seed": 314,
+        },
+    ).respond(status_code=200, json={"ro": [[1, 0, 1]]})
 
     request = RunProgramRequest(
         program="some-program",
@@ -95,24 +80,22 @@ def test_run_program__returns_results(client_configuration: QCSClientConfigurati
     assert qvm_client.run_program(request) == RunProgramResponse(results={"ro": [[1, 0, 1]]})
 
 
-def test_run_and_measure_program__returns_results(client_configuration: QCSClientConfiguration, httpx_mock: HTTPXMock):
+@respx.mock
+def test_run_and_measure_program__returns_results(client_configuration: QCSClientConfiguration):
     qvm_client = QVMClient(client_configuration=client_configuration)
 
-    httpx_mock.add_response(
+    respx.post(
         url=client_configuration.profile.applications.pyquil.qvm_url,
-        match_content=json.dumps(
-            {
-                "type": "multishot-measure",
-                "compiled-quil": "some-program",
-                "qubits": [0, 1, 2],
-                "trials": 1,
-                "measurement-noise": (3.14, 1.61, 6.28),
-                "gate-noise": (1.0, 2.0, 3.0),
-                "rng-seed": 314,
-            },
-        ).encode(),
-        json=[[1, 0, 1]],
-    )
+        json={
+            "type": "multishot-measure",
+            "compiled-quil": "some-program",
+            "qubits": [0, 1, 2],
+            "trials": 1,
+            "measurement-noise": (3.14, 1.61, 6.28),
+            "gate-noise": (1.0, 2.0, 3.0),
+            "rng-seed": 314,
+        },
+    ).respond(status_code=200, json=[[1, 0, 1]])
 
     request = RunAndMeasureProgramRequest(
         program="some-program",
@@ -125,19 +108,20 @@ def test_run_and_measure_program__returns_results(client_configuration: QCSClien
     assert qvm_client.run_and_measure_program(request) == RunAndMeasureProgramResponse(results=[[1, 0, 1]])
 
 
-def test_measure_expectation__returns_expectation(client_configuration: QCSClientConfiguration, httpx_mock: HTTPXMock):
+@respx.mock
+def test_measure_expectation__returns_expectation(client_configuration: QCSClientConfiguration):
     qvm_client = QVMClient(client_configuration=client_configuration)
 
-    httpx_mock.add_response(
+    respx.post(
         url=client_configuration.profile.applications.pyquil.qvm_url,
-        match_content=json.dumps(
-            {
-                "type": "expectation",
-                "state-preparation": "some-program",
-                "operators": ["some-op-program"],
-                "rng-seed": 314,
-            },
-        ).encode(),
+        json={
+            "type": "expectation",
+            "state-preparation": "some-program",
+            "operators": ["some-op-program"],
+            "rng-seed": 314,
+        },
+    ).respond(
+        status_code=200,
         json=[0.161],
     )
 
@@ -149,22 +133,20 @@ def test_measure_expectation__returns_expectation(client_configuration: QCSClien
     assert qvm_client.measure_expectation(request) == MeasureExpectationResponse(expectations=[0.161])
 
 
-def test_get_wavefunction__returns_wavefunction(client_configuration: QCSClientConfiguration, httpx_mock: HTTPXMock):
+@respx.mock
+def test_get_wavefunction__returns_wavefunction(client_configuration: QCSClientConfiguration):
     qvm_client = QVMClient(client_configuration=client_configuration)
 
-    httpx_mock.add_response(
+    respx.post(
         url=client_configuration.profile.applications.pyquil.qvm_url,
-        match_content=json.dumps(
-            {
+        json={
                 "type": "wavefunction",
                 "compiled-quil": "some-program",
                 "measurement-noise": (3.14, 1.61, 6.28),
                 "gate-noise": (1.0, 2.0, 3.0),
                 "rng-seed": 314,
             },
-        ).encode(),
-        data=b"some-wavefunction",
-    )
+    ).respond(status_code=200, text="some-wavefunction")
 
     request = GetWavefunctionRequest(
         program="some-program",