diff --git a/cirq-google/cirq_google/__init__.py b/cirq-google/cirq_google/__init__.py
index 2c4429812e8..34204be43cc 100644
--- a/cirq-google/cirq_google/__init__.py
+++ b/cirq-google/cirq_google/__init__.py
@@ -78,6 +78,7 @@
     EngineResult,
     ProtoVersion,
     QuantumEngineSampler,
+    ProcessorSampler,
     ValidatingSampler,
     get_engine,
     get_engine_calibration,
diff --git a/cirq-google/cirq_google/engine/__init__.py b/cirq-google/cirq_google/engine/__init__.py
index f4c73a3f0d6..0bbcbdde75a 100644
--- a/cirq-google/cirq_google/engine/__init__.py
+++ b/cirq-google/cirq_google/engine/__init__.py
@@ -80,3 +80,5 @@
 )
 
 from cirq_google.engine.engine_result import EngineResult
+
+from cirq_google.engine.processor_sampler import ProcessorSampler
diff --git a/cirq-google/cirq_google/engine/abstract_processor.py b/cirq-google/cirq_google/engine/abstract_processor.py
index f7cbe6e933c..6bada185eef 100644
--- a/cirq-google/cirq_google/engine/abstract_processor.py
+++ b/cirq-google/cirq_google/engine/abstract_processor.py
@@ -30,7 +30,7 @@
 from cirq_google.engine import calibration, util
 
 if TYPE_CHECKING:
-    import cirq_google
+    import cirq_google as cg
     import cirq_google.engine.abstract_engine as abstract_engine
     import cirq_google.engine.abstract_job as abstract_job
     import cirq_google.serialization.serializer as serializer
@@ -97,7 +97,7 @@ def run(
     @util.deprecated_gate_set_parameter
     def run_sweep(
         self,
-        program: cirq.Circuit,
+        program: cirq.AbstractCircuit,
         program_id: Optional[str] = None,
         job_id: Optional[str] = None,
         params: cirq.Sweepable = None,
@@ -196,7 +196,7 @@ def run_batch(
     @util.deprecated_gate_set_parameter
     def run_calibration(
         self,
-        layers: List['cirq_google.CalibrationLayer'],
+        layers: List['cg.CalibrationLayer'],
         program_id: Optional[str] = None,
         job_id: Optional[str] = None,
         gate_set: Optional['serializer.Serializer'] = None,
@@ -241,7 +241,9 @@ def run_calibration(
 
     @abc.abstractmethod
     @util.deprecated_gate_set_parameter
-    def get_sampler(self, gate_set: Optional['serializer.Serializer'] = None) -> cirq.Sampler:
+    def get_sampler(
+        self, gate_set: Optional['serializer.Serializer'] = None
+    ) -> 'cg.ProcessorSampler':
         """Returns a sampler backed by the processor.
 
         Args:
diff --git a/cirq-google/cirq_google/engine/engine_processor.py b/cirq-google/cirq_google/engine/engine_processor.py
index 9edb77c038b..c10e2bfd1c6 100644
--- a/cirq-google/cirq_google/engine/engine_processor.py
+++ b/cirq-google/cirq_google/engine/engine_processor.py
@@ -25,13 +25,14 @@
     abstract_processor,
     calibration,
     calibration_layer,
-    engine_sampler,
+    processor_sampler,
     util,
 )
 from cirq_google.serialization import serializable_gate_set, serializer
 from cirq_google.serialization import gate_sets as gs
 
 if TYPE_CHECKING:
+    import cirq_google as cg
     import cirq_google.engine.engine as engine_base
     import cirq_google.engine.abstract_job as abstract_job
 
@@ -105,7 +106,7 @@ def engine(self) -> 'engine_base.Engine':
     @util.deprecated_gate_set_parameter
     def get_sampler(
         self, gate_set: Optional[serializer.Serializer] = None
-    ) -> engine_sampler.QuantumEngineSampler:
+    ) -> 'cg.engine.ProcessorSampler':
         """Returns a sampler backed by the engine.
 
         Args:
@@ -117,9 +118,7 @@ def get_sampler(
             that will send circuits to the Quantum Computing Service
             when sampled.1
         """
-        return engine_sampler.QuantumEngineSampler(
-            engine=self.engine(), processor_id=self.processor_id
-        )
+        return processor_sampler.ProcessorSampler(processor=self)
 
     @util.deprecated_gate_set_parameter
     def run_batch(
diff --git a/cirq-google/cirq_google/engine/processor_sampler.py b/cirq-google/cirq_google/engine/processor_sampler.py
new file mode 100644
index 00000000000..57f50e8a511
--- /dev/null
+++ b/cirq-google/cirq_google/engine/processor_sampler.py
@@ -0,0 +1,76 @@
+# Copyright 2022 The Cirq Developers
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from typing import List, Optional, Sequence, TYPE_CHECKING, Union, cast
+
+import cirq
+
+if TYPE_CHECKING:
+    import cirq_google as cg
+
+
+class ProcessorSampler(cirq.Sampler):
+    """A wrapper around AbstractProcessor to implement the cirq.Sampler interface."""
+
+    def __init__(self, *, processor: 'cg.engine.AbstractProcessor'):
+        """Inits ProcessorSampler.
+
+        Args:
+            processor: AbstractProcessor instance to use.
+        """
+        self._processor = processor
+
+    def run_sweep(
+        self, program: 'cirq.AbstractCircuit', params: cirq.Sweepable, repetitions: int = 1
+    ) -> Sequence['cg.EngineResult']:
+        job = self._processor.run_sweep(program=program, params=params, repetitions=repetitions)
+        return job.results()
+
+    def run_batch(
+        self,
+        programs: Sequence[cirq.AbstractCircuit],
+        params_list: Optional[List[cirq.Sweepable]] = None,
+        repetitions: Union[int, List[int]] = 1,
+    ) -> Sequence[Sequence['cg.EngineResult']]:
+        """Runs the supplied circuits.
+
+        In order to gain a speedup from using this method instead of other run
+        methods, the following conditions must be satisfied:
+            1. All circuits must measure the same set of qubits.
+            2. The number of circuit repetitions must be the same for all
+               circuits. That is, the `repetitions` argument must be an integer,
+               or else a list with identical values.
+        """
+        if isinstance(repetitions, List) and len(programs) != len(repetitions):
+            raise ValueError(
+                'len(programs) and len(repetitions) must match. '
+                f'Got {len(programs)} and {len(repetitions)}.'
+            )
+        if isinstance(repetitions, int) or len(set(repetitions)) == 1:
+            # All repetitions are the same so batching can be done efficiently
+            if isinstance(repetitions, List):
+                repetitions = repetitions[0]
+            job = self._processor.run_batch(
+                programs=programs, params_list=params_list, repetitions=repetitions
+            )
+            return job.batched_results()
+        # Varying number of repetitions so no speedup
+        return cast(
+            Sequence[Sequence['cg.EngineResult']],
+            super().run_batch(programs, params_list, repetitions),
+        )
+
+    @property
+    def processor(self) -> 'cg.engine.AbstractProcessor':
+        return self._processor
diff --git a/cirq-google/cirq_google/engine/processor_sampler_test.py b/cirq-google/cirq_google/engine/processor_sampler_test.py
new file mode 100644
index 00000000000..82047cf45c1
--- /dev/null
+++ b/cirq-google/cirq_google/engine/processor_sampler_test.py
@@ -0,0 +1,110 @@
+# Copyright 2022 The Cirq Developers
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from unittest import mock
+
+import pytest
+
+import cirq
+import cirq_google as cg
+
+
+@pytest.mark.parametrize('circuit', [cirq.Circuit(), cirq.FrozenCircuit()])
+def test_run_circuit(circuit):
+    processor = mock.Mock()
+    sampler = cg.ProcessorSampler(processor=processor)
+    params = [cirq.ParamResolver({'a': 1})]
+    sampler.run_sweep(circuit, params, 5)
+    processor.run_sweep.assert_called_with(params=params, program=circuit, repetitions=5)
+
+
+def test_run_batch():
+    processor = mock.Mock()
+    sampler = cg.ProcessorSampler(processor=processor)
+    a = cirq.LineQubit(0)
+    circuit1 = cirq.Circuit(cirq.X(a))
+    circuit2 = cirq.Circuit(cirq.Y(a))
+    params1 = [cirq.ParamResolver({'t': 1})]
+    params2 = [cirq.ParamResolver({'t': 2})]
+    circuits = [circuit1, circuit2]
+    params_list = [params1, params2]
+    sampler.run_batch(circuits, params_list, 5)
+    processor.run_batch.assert_called_with(
+        params_list=params_list, programs=circuits, repetitions=5
+    )
+
+
+def test_run_batch_identical_repetitions():
+    processor = mock.Mock()
+    sampler = cg.ProcessorSampler(processor=processor)
+    a = cirq.LineQubit(0)
+    circuit1 = cirq.Circuit(cirq.X(a))
+    circuit2 = cirq.Circuit(cirq.Y(a))
+    params1 = [cirq.ParamResolver({'t': 1})]
+    params2 = [cirq.ParamResolver({'t': 2})]
+    circuits = [circuit1, circuit2]
+    params_list = [params1, params2]
+    sampler.run_batch(circuits, params_list, [5, 5])
+    processor.run_batch.assert_called_with(
+        params_list=params_list, programs=circuits, repetitions=5
+    )
+
+
+def test_run_batch_bad_number_of_repetitions():
+    processor = mock.Mock()
+    sampler = cg.ProcessorSampler(processor=processor)
+    a = cirq.LineQubit(0)
+    circuit1 = cirq.Circuit(cirq.X(a))
+    circuit2 = cirq.Circuit(cirq.Y(a))
+    params1 = [cirq.ParamResolver({'t': 1})]
+    params2 = [cirq.ParamResolver({'t': 2})]
+    circuits = [circuit1, circuit2]
+    params_list = [params1, params2]
+    with pytest.raises(ValueError, match='2 and 3'):
+        sampler.run_batch(circuits, params_list, [5, 5, 5])
+
+
+def test_run_batch_differing_repetitions():
+    processor = mock.Mock()
+    job = mock.Mock()
+    job.results.return_value = []
+    processor.run_sweep.return_value = job
+    sampler = cg.ProcessorSampler(processor=processor)
+    a = cirq.LineQubit(0)
+    circuit1 = cirq.Circuit(cirq.X(a))
+    circuit2 = cirq.Circuit(cirq.Y(a))
+    params1 = [cirq.ParamResolver({'t': 1})]
+    params2 = [cirq.ParamResolver({'t': 2})]
+    circuits = [circuit1, circuit2]
+    params_list = [params1, params2]
+    repetitions = [1, 2]
+    sampler.run_batch(circuits, params_list, repetitions)
+    processor.run_sweep.assert_called_with(params=params2, program=circuit2, repetitions=2)
+    processor.run_batch.assert_not_called()
+
+
+def test_processor_sampler_processor_property():
+    processor = mock.Mock()
+    sampler = cg.ProcessorSampler(processor=processor)
+    assert sampler.processor is processor
+
+
+def test_with_local_processor():
+    sampler = cg.ProcessorSampler(
+        processor=cg.engine.SimulatedLocalProcessor(processor_id='my-fancy-processor')
+    )
+    r = sampler.run(cirq.Circuit(cirq.measure(cirq.LineQubit(0), key='z')))
+    assert isinstance(r, cg.EngineResult)
+    assert r.job_id == 'projects/fake_project/processors/my-fancy-processor/job/2'
+    assert r.measurements['z'] == [[0]]
diff --git a/cirq-google/cirq_google/engine/qcs_notebook.py b/cirq-google/cirq_google/engine/qcs_notebook.py
index f26bdf0be9f..30bab26cb7f 100644
--- a/cirq-google/cirq_google/engine/qcs_notebook.py
+++ b/cirq-google/cirq_google/engine/qcs_notebook.py
@@ -19,7 +19,7 @@
 
 from cirq_google import (
     PhasedFSimEngineSimulator,
-    QuantumEngineSampler,
+    ProcessorSampler,
     Sycamore,
     SQRT_ISWAP_INV_PARAMETERS,
     PhasedFSimCharacterization,
@@ -30,7 +30,7 @@
 @dataclasses.dataclass
 class QCSObjectsForNotebook:
     device: cirq.Device
-    sampler: Union[PhasedFSimEngineSimulator, QuantumEngineSampler]
+    sampler: Union[PhasedFSimEngineSimulator, ProcessorSampler]
     signed_in: bool
 
     @property
@@ -80,7 +80,7 @@ def get_qcs_objects_for_notebook(
             print(f"Authentication failed: {exc}")
 
     # Attempt to connect to the Quantum Engine API, and use a simulator if unable to connect.
-    sampler: Union[PhasedFSimEngineSimulator, QuantumEngineSampler]
+    sampler: Union[PhasedFSimEngineSimulator, ProcessorSampler]
     try:
         engine = get_engine(project_id)
         if processor_id:
diff --git a/cirq-google/cirq_google/engine/simulated_local_processor.py b/cirq-google/cirq_google/engine/simulated_local_processor.py
index e8fa7ccbc43..abbf8bf1f63 100644
--- a/cirq-google/cirq_google/engine/simulated_local_processor.py
+++ b/cirq-google/cirq_google/engine/simulated_local_processor.py
@@ -26,8 +26,10 @@
 from cirq_google.engine.simulated_local_job import SimulatedLocalJob
 from cirq_google.engine.simulated_local_program import SimulatedLocalProgram
 from cirq_google.serialization.circuit_serializer import CIRCUIT_SERIALIZER
+from cirq_google.engine.processor_sampler import ProcessorSampler
 
 if TYPE_CHECKING:
+    import cirq_google as cg
     from cirq_google.serialization.serializer import Serializer
 
 VALID_LANGUAGES = [
@@ -161,7 +163,7 @@ def list_calibrations(
 
     @util.deprecated_gate_set_parameter
     def get_sampler(self, gate_set: Optional['Serializer'] = None) -> cirq.Sampler:
-        return self._sampler
+        return ProcessorSampler(processor=self)
 
     def supported_languages(self) -> List[str]:
         return VALID_LANGUAGES
@@ -256,7 +258,7 @@ def run(
         program_labels: Optional[Dict[str, str]] = None,
         job_description: Optional[str] = None,
         job_labels: Optional[Dict[str, str]] = None,
-    ) -> cirq.Result:
+    ) -> 'cg.EngineResult':
         """Runs the supplied Circuit on this processor.
 
         Args:
diff --git a/cirq-google/cirq_google/json_test_data/spec.py b/cirq-google/cirq_google/json_test_data/spec.py
index ffe41b148e6..5ae602c1cdc 100644
--- a/cirq-google/cirq_google/json_test_data/spec.py
+++ b/cirq-google/cirq_google/json_test_data/spec.py
@@ -51,6 +51,7 @@
         'SerializingArg',
         'THETA_ZETA_GAMMA_FLOQUET_PHASED_FSIM_CHARACTERIZATION',
         'QuantumEngineSampler',
+        'ProcessorSampler',
         'ValidatingSampler',
         'CouldNotPlaceError',
         # Abstract:
diff --git a/cirq-google/cirq_google/workflow/processor_record.py b/cirq-google/cirq_google/workflow/processor_record.py
index 197f5edd036..d6638e5f312 100644
--- a/cirq-google/cirq_google/workflow/processor_record.py
+++ b/cirq-google/cirq_google/workflow/processor_record.py
@@ -34,7 +34,7 @@ def get_processor(self) -> 'cg.engine.AbstractProcessor':
         This is the primary method that descendants must implement.
         """
 
-    def get_sampler(self) -> 'cirq.Sampler':
+    def get_sampler(self) -> 'cg.ProcessorSampler':
         """Return a `cirq.Sampler` for the processor specified by this class.
 
         The default implementation delegates to `self.get_processor()`.
diff --git a/cirq-google/cirq_google/workflow/processor_record_test.py b/cirq-google/cirq_google/workflow/processor_record_test.py
index 5a0f0c696db..d631f52138d 100644
--- a/cirq-google/cirq_google/workflow/processor_record_test.py
+++ b/cirq-google/cirq_google/workflow/processor_record_test.py
@@ -118,18 +118,37 @@ def test_simulated_backend_with_bad_local_device():
 def test_simulated_backend_descriptive_name():
     p = cg.SimulatedProcessorWithLocalDeviceRecord('rainbow')
     assert str(p) == 'rainbow-simulator'
-    assert isinstance(p.get_sampler(), cg.ValidatingSampler)
-    assert isinstance(p.get_sampler()._sampler, cirq.Simulator)
+    assert isinstance(p.get_sampler(), cg.engine.ProcessorSampler)
+
+    # The actual simulator hiding behind the indirection is
+    # p.get_sampler() -> ProcessorSampler
+    # p.get_sampler().processor._sampler -> Validating Sampler
+    # p.get_sampler().processor._sampler._sampler -> The actual simulator
+    assert isinstance(p.get_sampler().processor._sampler._sampler, cirq.Simulator)
 
     p = cg.SimulatedProcessorWithLocalDeviceRecord('rainbow', noise_strength=1e-3)
     assert str(p) == 'rainbow-depol(1.000e-03)'
-    assert isinstance(p.get_sampler()._sampler, cirq.DensityMatrixSimulator)
+    assert isinstance(p.get_sampler().processor._sampler._sampler, cirq.DensityMatrixSimulator)
 
     p = cg.SimulatedProcessorWithLocalDeviceRecord('rainbow', noise_strength=float('inf'))
     assert str(p) == 'rainbow-zeros'
-    assert isinstance(p.get_sampler()._sampler, cirq.ZerosSampler)
+    assert isinstance(p.get_sampler().processor._sampler._sampler, cirq.ZerosSampler)
 
 
 def test_sampler_equality():
     p = cg.SimulatedProcessorWithLocalDeviceRecord('rainbow')
     assert p.get_sampler().__class__ == p.get_processor().get_sampler().__class__
+
+
+def test_engine_result():
+    proc_rec = cg.SimulatedProcessorWithLocalDeviceRecord('rainbow')
+
+    proc = proc_rec.get_processor()
+    samp = proc_rec.get_sampler()
+
+    circ = cirq.Circuit(cirq.measure(cirq.GridQubit(5, 4)))
+
+    res1 = proc.run(circ)
+    assert isinstance(res1, cg.EngineResult)
+    res2 = samp.run(circ)
+    assert isinstance(res2, cg.EngineResult)