diff --git a/cirq-google/cirq_google/engine/abstract_engine.py b/cirq-google/cirq_google/engine/abstract_engine.py
new file mode 100644
index 00000000000..4d6dea9cf20
--- /dev/null
+++ b/cirq-google/cirq_google/engine/abstract_engine.py
@@ -0,0 +1,139 @@
+# Copyright 2021 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.
+# coverage: ignore
+"""Interface for Engine objects.
+
+This class is an abstract class which all Engine implementations
+(production API or locally simulated) should follow.
+"""
+
+import abc
+import datetime
+from typing import Dict, List, Optional, Sequence, Set, Union
+
+import cirq
+from cirq_google.engine import abstract_job, abstract_program, abstract_processor
+from cirq_google.engine.client import quantum
+from cirq_google.serialization import Serializer
+
+VALID_DATE_TYPE = Union[datetime.datetime, datetime.date]
+
+
+class AbstractEngine(abc.ABC):
+    """An abstract object representing a collection of quantum processors.
+
+    Each processor within the AbstractEngine can be referenced by a string
+    identifier through the get_processor interface.
+
+    The Engine interface also includes convenience methods to access
+    programs, jobs, and sampler.
+
+    This is an abstract interface and inheritors must implement the abstract methods.
+
+    """
+
+    @abc.abstractmethod
+    def get_program(self, program_id: str) -> abstract_program.AbstractProgram:
+        """Returns an existing AbstractProgram given an identifier.
+
+        Args:
+            program_id: Unique ID of the program.
+
+        Returns:
+            An AbstractProgram object for the program.
+        """
+
+    @abc.abstractmethod
+    def list_programs(
+        self,
+        created_before: Optional[VALID_DATE_TYPE] = None,
+        created_after: Optional[VALID_DATE_TYPE] = None,
+        has_labels: Optional[Dict[str, str]] = None,
+    ) -> List[abstract_program.AbstractProgram]:
+        """Returns a list of previously executed quantum programs.
+
+        Args:
+            created_after: retrieve programs that were created after this date
+                or time.
+            created_before: retrieve programs that were created before this date
+                or time.
+            has_labels: retrieve programs that have labels on them specified by
+                this dict. If the value is set to `*`, programs having the label
+                regardless of the label value will be returned. For example, to
+                query programs that have the shape label and have the color
+                label with value red can be queried using
+                `{'color': 'red', 'shape': '*'}`
+        """
+
+    @abc.abstractmethod
+    def list_jobs(
+        self,
+        created_before: Optional[VALID_DATE_TYPE] = None,
+        created_after: Optional[VALID_DATE_TYPE] = None,
+        has_labels: Optional[Dict[str, str]] = None,
+        execution_states: Optional[Set[quantum.enums.ExecutionStatus.State]] = None,
+    ) -> List[abstract_job.AbstractJob]:
+        """Returns the list of jobs that match the specified criteria.
+
+        All historical jobs can be retrieved using this method and filtering
+        options are available too, to narrow down the search based on:
+          * creation time
+          * job labels
+          * execution states
+
+        Args:
+            created_after: retrieve jobs that were created after this date
+                or time.
+            created_before: retrieve jobs that were created before this date
+                or time.
+            has_labels: retrieve jobs that have labels on them specified by
+                this dict. If the value is set to `*`, jobs having the label
+                regardless of the label value will be returned. For example, to
+                query programs that have the shape label and have the color
+                label with value red can be queried using
+
+                {'color': 'red', 'shape':'*'}
+
+            execution_states: retrieve jobs that have an execution state  that
+                 is contained in `execution_states`. See
+                 `quantum.enums.ExecutionStatus.State` enum for accepted values.
+        """
+
+    @abc.abstractmethod
+    def list_processors(self) -> Sequence[abstract_processor.AbstractProcessor]:
+        """Returns all processors in this engine visible to the user."""
+
+    @abc.abstractmethod
+    def get_processor(self, processor_id: str) -> abstract_processor.AbstractProcessor:
+        """Returns an EngineProcessor for a Quantum Engine processor.
+
+        Args:
+            processor_id: The processor unique identifier.
+
+        Returns:
+            A EngineProcessor for the processor.
+        """
+
+    @abc.abstractmethod
+    def get_sampler(
+        self, processor_id: Union[str, List[str]], gate_set: Serializer
+    ) -> cirq.Sampler:
+        """Returns a sampler backed by the engine.
+
+        Args:
+            processor_id: String identifier, or list of string identifiers,
+                determining which processors may be used when sampling.
+            gate_set: Determines how to serialize circuits when requesting
+                samples.
+        """
diff --git a/cirq-google/cirq_google/engine/abstract_job.py b/cirq-google/cirq_google/engine/abstract_job.py
new file mode 100644
index 00000000000..322e2ed2b5f
--- /dev/null
+++ b/cirq-google/cirq_google/engine/abstract_job.py
@@ -0,0 +1,202 @@
+# Copyright 2021 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.
+"""A helper for jobs that have been created on the Quantum Engine."""
+
+import abc
+from typing import Dict, Iterator, List, Optional, overload, Tuple, TYPE_CHECKING
+
+import cirq
+import cirq_google.engine.client.quantum as quantum
+
+
+if TYPE_CHECKING:
+    import datetime
+    import cirq_google.engine.calibration as calibration
+    import cirq_google.engine.calibration_result as calibration_result
+    import cirq_google.engine.abstract_engine as abstract_engine
+    import cirq_google.engine.abstract_processor as abstract_processor
+    import cirq_google.engine.abstract_program as abstract_program
+
+
+class AbstractJob(abc.ABC):
+    """An abstract object representing a quantum job execution.
+
+    This represents the state of a possibly asynchronous Job being
+    executed by a simulator, the cloud Engine service, or other means.
+
+    This is an abstract interface that implementers of services or mocks
+    should implement.  It generally represents the execution of a circuit
+    using a set of parameters called a sweep.  It can also represent the
+    execution of a batch job (a list of circuit/sweep pairs) or the
+    execution of a calibration request.
+
+    This job may be in a variety of states. It may be scheduling, it may be
+    executing on a machine, or it may have entered a terminal state
+    (either succeeding or failing).
+
+    `AbstractJob`s can be iterated over, returning `Result`s. These
+    `Result`s can also be accessed by index. Note that this will block
+    until the results are returned.
+
+    """
+
+    @abc.abstractmethod
+    def engine(self) -> 'abstract_engine.AbstractEngine':
+        """Returns the parent `AbstractEngine` object."""
+
+    @abc.abstractmethod
+    def id(self) -> str:
+        """Returns the id of this job."""
+
+    @abc.abstractmethod
+    def program(self) -> 'abstract_program.AbstractProgram':
+        """Returns the parent `AbstractProgram`object."""
+
+    @abc.abstractmethod
+    def create_time(self) -> 'datetime.datetime':
+        """Returns when the job was created."""
+
+    @abc.abstractmethod
+    def update_time(self) -> 'datetime.datetime':
+        """Returns when the job was last updated."""
+
+    @abc.abstractmethod
+    def description(self) -> str:
+        """Returns the description of the job."""
+
+    @abc.abstractmethod
+    def set_description(self, description: str) -> 'AbstractJob':
+        """Sets the description of the job.
+
+        Params:
+            description: The new description for the job.
+
+        Returns:
+             This `AbstractJob`.
+        """
+
+    @abc.abstractmethod
+    def labels(self) -> Dict[str, str]:
+        """Returns the labels of the job."""
+
+    @abc.abstractmethod
+    def set_labels(self, labels: Dict[str, str]) -> 'AbstractJob':
+        """Sets (overwriting) the labels for a previously created quantum job.
+
+        Params:
+            labels: The entire set of new job labels.
+
+        Returns:
+             This `AbstractJob`.
+        """
+
+    @abc.abstractmethod
+    def add_labels(self, labels: Dict[str, str]) -> 'AbstractJob':
+        """Adds new labels to a previously created quantum job.
+
+        Params:
+            labels: New labels to add to the existing job labels.
+
+        Returns:
+             This `AbstractJob`.
+        """
+
+    @abc.abstractmethod
+    def remove_labels(self, keys: List[str]) -> 'AbstractJob':
+        """Removes labels with given keys.
+
+        Params:
+            label_keys: Label keys to remove from the existing job labels.
+
+        Returns:
+            This `AbstractJob`.
+        """
+
+    @abc.abstractmethod
+    def processor_ids(self) -> List[str]:
+        """Returns the processor ids provided when the job was created."""
+
+    @abc.abstractmethod
+    def execution_status(self) -> quantum.enums.ExecutionStatus.State:
+        """Return the execution status of the job."""
+
+    @abc.abstractmethod
+    def failure(self) -> Optional[Tuple[str, str]]:
+        """Return failure code and message of the job if present."""
+
+    @abc.abstractmethod
+    def get_repetitions_and_sweeps(self) -> Tuple[int, List[cirq.Sweep]]:
+        """Returns the repetitions and sweeps for the job.
+
+        Returns:
+            A tuple of the repetition count and list of sweeps.
+        """
+
+    @abc.abstractmethod
+    def get_processor(self) -> Optional['abstract_processor.AbstractProcessor']:
+        """Returns the AbstractProcessor for the processor the job is/was run on,
+        if available, else None."""
+
+    @abc.abstractmethod
+    def get_calibration(self) -> Optional['calibration.Calibration']:
+        """Returns the recorded calibration at the time when the job was run, if
+        one was captured, else None."""
+
+    @abc.abstractmethod
+    def cancel(self) -> Optional[bool]:
+        """Cancel the job."""
+
+    @abc.abstractmethod
+    def delete(self) -> Optional[bool]:
+        """Deletes the job and result, if any."""
+
+    @abc.abstractmethod
+    def batched_results(self) -> List[List[cirq.Result]]:
+        """Returns the job results, blocking until the job is complete.
+
+        This method is intended for batched jobs.  Instead of flattening
+        results into a single list, this will return a List[Result]
+        for each circuit in the batch.
+        """
+
+    @abc.abstractmethod
+    def results(self) -> List[cirq.Result]:
+        """Returns the job results, blocking until the job is complete."""
+
+    @abc.abstractmethod
+    def calibration_results(self) -> List['calibration_result.CalibrationResult']:
+        """Returns the results of a run_calibration() call.
+
+        This function will fail if any other type of results were returned.
+        """
+
+    def __iter__(self) -> Iterator[cirq.Result]:
+        return iter(self.results())
+
+    # pylint: disable=function-redefined
+    @overload
+    def __getitem__(self, item: int) -> cirq.Result:
+        pass
+
+    @overload
+    def __getitem__(self, item: slice) -> List[cirq.Result]:
+        pass
+
+    def __getitem__(self, item):
+        return self.results()[item]
+
+    # pylint: enable=function-redefined
+
+    def __len__(self) -> int:
+        return len(self.results())
diff --git a/cirq-google/cirq_google/engine/abstract_job_test.py b/cirq-google/cirq_google/engine/abstract_job_test.py
new file mode 100644
index 00000000000..fe723d6a655
--- /dev/null
+++ b/cirq-google/cirq_google/engine/abstract_job_test.py
@@ -0,0 +1,99 @@
+# Copyright 2021 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 Dict, List, TYPE_CHECKING
+from cirq_google.engine.abstract_job import AbstractJob
+
+if TYPE_CHECKING:
+    import datetime
+    import cirq_google.engine.abstract_engine as abstract_engine
+    import cirq_google.engine.abstract_processor as abstract_processor
+    import cirq_google.engine.abstract_program as abstract_program
+
+
+class MockJob(AbstractJob):
+    def engine(self) -> 'abstract_engine.AbstractEngine':
+        pass
+
+    def id(self) -> str:
+        pass
+
+    def program(self) -> 'abstract_program.AbstractProgram':
+        pass
+
+    def create_time(self) -> 'datetime.datetime':
+        pass
+
+    def update_time(self) -> 'datetime.datetime':
+        pass
+
+    def description(self) -> str:
+        pass
+
+    def set_description(self, description: str) -> 'AbstractJob':
+        pass
+
+    def labels(self) -> Dict[str, str]:
+        pass
+
+    def set_labels(self, labels: Dict[str, str]) -> 'AbstractJob':
+        pass
+
+    def add_labels(self, labels: Dict[str, str]) -> 'AbstractJob':
+        pass
+
+    def remove_labels(self, keys: List[str]) -> 'AbstractJob':
+        pass
+
+    def processor_ids(self):
+        pass
+
+    def execution_status(self):
+        pass
+
+    def failure(self):
+        pass
+
+    def get_repetitions_and_sweeps(self):
+        pass
+
+    def get_processor(self):
+        pass
+
+    def get_calibration(self):
+        pass
+
+    def cancel(self) -> None:
+        pass
+
+    def delete(self) -> None:
+        pass
+
+    def batched_results(self):
+        pass
+
+    def results(self):
+        return list(range(5))
+
+    def calibration_results(self):
+        pass
+
+
+def test_instantiation_and_iteration():
+    job = MockJob()
+    assert len(job) == 5
+    assert job[3] == 3
+    count = 0
+    for num in job:
+        assert num == count
+        count += 1
diff --git a/cirq-google/cirq_google/engine/abstract_processor.py b/cirq-google/cirq_google/engine/abstract_processor.py
new file mode 100644
index 00000000000..61222c3e734
--- /dev/null
+++ b/cirq-google/cirq_google/engine/abstract_processor.py
@@ -0,0 +1,433 @@
+# Copyright 2021 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.
+# coverage: ignore
+"""Abstract interface for a quantum processor.
+
+This interface can run circuits, sweeps, batches, or calibration
+requests.  Inheritors of this interface should implement all
+methods.
+"""
+
+import abc
+import datetime
+
+from typing import Dict, Iterable, List, Optional, Sequence, TYPE_CHECKING, Union
+
+import cirq
+import cirq_google.api.v2 as v2
+import cirq_google.engine.calibration as calibration
+import cirq_google.engine.client.quantum as quantum
+
+if TYPE_CHECKING:
+    import cirq_google
+    import cirq_google.engine.abstract_engine as abstract_engine
+    import cirq_google.engine.abstract_job as abstract_job
+    import cirq_google.engine.abstract_program as abstract_program
+    import cirq_google.serialization.serializer as serializer
+
+
+class AbstractProcessor(abc.ABC):
+    """An abstract interface for a quantum processor.
+
+    This quantum processor has the ability to execute single circuits
+    (via the run method), parameter sweeps (via run_sweep), batched
+    lists of circuits (via run_batch), and calibration
+    requests (via run_calibration).  Running circuits can also be
+    done using the `cirq.Sampler` by calling get_sampler.
+
+    The processor interface also includes methods to create, list,
+    and remove reservations on the processor for dedicated access.
+    The processor can also list calibration metrics for the processor
+    given a time period.
+
+    This is an abstract class.  Inheritors should implement abstract methods.
+    """
+
+    def run(
+        self,
+        program: cirq.Circuit,
+        program_id: Optional[str] = None,
+        job_id: Optional[str] = None,
+        param_resolver: cirq.ParamResolver = None,
+        repetitions: int = 1,
+        gate_set: Optional['serializer.Serializer'] = None,
+        program_description: Optional[str] = None,
+        program_labels: Optional[Dict[str, str]] = None,
+        job_description: Optional[str] = None,
+        job_labels: Optional[Dict[str, str]] = None,
+    ) -> cirq.Result:
+        """Runs the supplied Circuit on this processor.
+
+        Args:
+            program: The Circuit to execute. If a circuit is
+                provided, a moment by moment schedule will be used.
+            program_id: A user-provided identifier for the program. This must
+                be unique within the Google Cloud project being used. If this
+                parameter is not provided, a random id of the format
+                'prog-################YYMMDD' will be generated, where # is
+                alphanumeric and YYMMDD is the current year, month, and day.
+            job_id: Job identifier to use. If this is not provided, a random id
+                of the format 'job-################YYMMDD' will be generated,
+                where # is alphanumeric and YYMMDD is the current year, month,
+                and day.
+            param_resolver: Parameters to run with the program.
+            repetitions: The number of repetitions to simulate.
+            gate_set: The gate set used to serialize the circuit. The gate set
+                must be supported by the selected processor.
+            program_description: An optional description to set on the program.
+            program_labels: Optional set of labels to set on the program.
+            job_description: An optional description to set on the job.
+            job_labels: Optional set of labels to set on the job.
+        Returns:
+            A single Result for this run.
+        """
+
+    @abc.abstractmethod
+    def run_sweep(
+        self,
+        program: cirq.Circuit,
+        program_id: Optional[str] = None,
+        job_id: Optional[str] = None,
+        params: cirq.Sweepable = None,
+        repetitions: int = 1,
+        gate_set: Optional['serializer.Serializer'] = None,
+        program_description: Optional[str] = None,
+        program_labels: Optional[Dict[str, str]] = None,
+        job_description: Optional[str] = None,
+        job_labels: Optional[Dict[str, str]] = None,
+    ) -> 'abstract_job.AbstractJob':
+        """Runs the supplied Circuit on this processor.
+
+        In contrast to run, this runs across multiple parameter sweeps, and
+        does not block until a result is returned.
+        Args:
+            program: The Circuit to execute. If a circuit is
+                provided, a moment by moment schedule will be used.
+            program_id: A user-provided identifier for the program. This must
+                be unique within the Google Cloud project being used. If this
+                parameter is not provided, a random id of the format
+                'prog-################YYMMDD' will be generated, where # is
+                alphanumeric and YYMMDD is the current year, month, and day.
+            job_id: Job identifier to use. If this is not provided, a random id
+                of the format 'job-################YYMMDD' will be generated,
+                where # is alphanumeric and YYMMDD is the current year, month,
+                and day.
+            params: Parameters to run with the program.
+            repetitions: The number of circuit repetitions to run.
+            gate_set: The gate set used to serialize the circuit. The gate set
+                must be supported by the selected processor.
+            program_description: An optional description to set on the program.
+            program_labels: Optional set of labels to set on the program.
+            job_description: An optional description to set on the job.
+            job_labels: Optional set of labels to set on the job.
+        Returns:
+            An AbstractJob. If this is iterated over it returns a list of
+            `cirq.Result`, one for each parameter sweep.
+        """
+
+    @abc.abstractmethod
+    def run_batch(
+        self,
+        programs: Sequence[cirq.AbstractCircuit],
+        program_id: Optional[str] = None,
+        job_id: Optional[str] = None,
+        params_list: List[cirq.Sweepable] = None,
+        repetitions: int = 1,
+        gate_set: Optional['serializer.Serializer'] = None,
+        program_description: Optional[str] = None,
+        program_labels: Optional[Dict[str, str]] = None,
+        job_description: Optional[str] = None,
+        job_labels: Optional[Dict[str, str]] = None,
+    ) -> 'abstract_job.AbstractJob':
+        """Runs the supplied Circuits on this processor.
+
+        This will combine each Circuit provided in `programs` into
+        a BatchProgram.  Each circuit will pair with the associated
+        parameter sweep provided in the `params_list`.  The number of
+        programs is required to match the number of sweeps.
+        This method does not block until a result is returned.  However,
+        no results will be available until the entire batch is complete.
+        Args:
+            programs: The Circuits to execute as a batch.
+            program_id: A user-provided identifier for the program. This must
+                be unique within the Google Cloud project being used. If this
+                parameter is not provided, a random id of the format
+                'prog-################YYMMDD' will be generated, where # is
+                alphanumeric and YYMMDD is the current year, month, and day.
+            job_id: Job identifier to use. If this is not provided, a random id
+                of the format 'job-################YYMMDD' will be generated,
+                where # is alphanumeric and YYMMDD is the current year, month,
+                and day.
+            params_list: Parameter sweeps to use with the circuits. The number
+                of sweeps should match the number of circuits and will be
+                paired in order with the circuits. If this is None, it is
+                assumed that the circuits are not parameterized and do not
+                require sweeps.
+            repetitions: Number of circuit repetitions to run.  Each sweep value
+                of each circuit in the batch will run with the same repetitions.
+            gate_set: The gate set used to serialize the circuit. The gate set
+                must be supported by the selected processor.
+            program_description: An optional description to set on the program.
+            program_labels: Optional set of labels to set on the program.
+            job_description: An optional description to set on the job.
+            job_labels: Optional set of labels to set on the job.
+        Returns:
+            An AbstractJob. If this is iterated over it returns a list of
+            `cirq.Result`. All Results for the first circuit are listed
+            first, then the Results for the second, etc. The Results
+            for a circuit are listed in the order imposed by the associated
+            parameter sweep.
+        """
+
+    @abc.abstractmethod
+    def run_calibration(
+        self,
+        layers: List['cirq_google.CalibrationLayer'],
+        program_id: Optional[str] = None,
+        job_id: Optional[str] = None,
+        gate_set: Optional['serializer.Serializer'] = None,
+        program_description: Optional[str] = None,
+        program_labels: Optional[Dict[str, str]] = None,
+        job_description: Optional[str] = None,
+        job_labels: Optional[Dict[str, str]] = None,
+    ) -> 'abstract_job.AbstractJob':
+        """Runs the specified calibrations on the processor.
+
+        Each calibration will be specified by a `CalibrationLayer`
+        that contains the type of the calibrations to run, a `Circuit`
+        to optimize, and any arguments needed by the calibration routine.
+        Arguments and circuits needed for each layer will vary based on the
+        calibration type.  However, the typical calibration routine may
+        require a single moment defining the gates to optimize, for example.
+        Note: this is an experimental API and is not yet fully supported
+        for all users.
+        Args:
+            layers: The layers of calibration to execute as a batch.
+            program_id: A user-provided identifier for the program. This must
+                be unique within the Google Cloud project being used. If this
+                parameter is not provided, a random id of the format
+                'calibration-################YYMMDD' will be generated,
+                where # is alphanumeric and YYMMDD is the current year, month,
+                and day.
+            job_id: Job identifier to use. If this is not provided, a random id
+                of the format 'calibration-################YYMMDD' will be
+                generated, where # is alphanumeric and YYMMDD is the current
+                year, month, and day.
+            gate_set: The gate set used to serialize the circuit. The gate set
+                must be supported by the selected processor.
+            program_description: An optional description to set on the program.
+            program_labels: Optional set of labels to set on the program.
+            job_description: An optional description to set on the job.
+            job_labels: Optional set of labels to set on the job.  By defauly,
+                this will add a 'calibration' label to the job.
+        Returns:
+            An AbstractJob whose results can be retrieved by calling
+            calibration_results().
+        """
+
+    @abc.abstractmethod
+    def get_sampler(self, gate_set: Optional['serializer.Serializer']) -> cirq.Sampler:
+        """Returns a sampler backed by the processor.
+
+        Args:
+            gate_set: Determines how to serialize circuits if needed.
+        """
+
+    @abc.abstractmethod
+    def engine(self) -> Optional['abstract_engine.AbstractEngine']:
+        """Returns the parent Engine object.
+
+        Returns:
+            The program's parent Engine.
+        """
+
+    @abc.abstractmethod
+    def health(self) -> str:
+        """Returns the current health of processor."""
+
+    @abc.abstractmethod
+    def expected_down_time(self) -> 'Optional[datetime.datetime]':
+        """Returns the start of the next expected down time of the processor, if
+        set."""
+
+    @abc.abstractmethod
+    def expected_recovery_time(self) -> 'Optional[datetime.datetime]':
+        """Returns the expected the processor should be available, if set."""
+
+    @abc.abstractmethod
+    def supported_languages(self) -> List[str]:
+        """Returns the list of processor supported program languages."""
+
+    @abc.abstractmethod
+    def get_device_specification(self) -> Optional[v2.device_pb2.DeviceSpecification]:
+        """Returns a device specification proto for use in determining
+        information about the device.
+
+        Returns:
+            Device specification proto if present.
+        """
+
+    @abc.abstractmethod
+    def get_device(self, gate_sets: Iterable['serializer.Serializer']) -> cirq.Device:
+        """Returns a `Device` created from the processor's device specification.
+
+        This method queries the processor to retrieve the device specification,
+        which is then use to create a `Device` that will validate
+        that operations are supported and use the correct qubits.
+
+        Args:
+            gate_sets: An iterable of serializers that can be used in the device.
+
+        Returns:
+            A `cirq.Devive` representing the processor.
+        """
+
+    @abc.abstractmethod
+    def list_calibrations(
+        self,
+        earliest_timestamp: Optional[Union[datetime.datetime, datetime.date, int]] = None,
+        latest_timestamp: Optional[Union[datetime.datetime, datetime.date, int]] = None,
+    ) -> List[calibration.Calibration]:
+        """Retrieve metadata about a specific calibration run.
+
+        Args:
+            earliest_timestamp: The earliest timestamp of a calibration
+                to return in UTC.
+            latest_timestamp: The latest timestamp of a calibration to
+                return in UTC.
+
+        Returns:
+            The list of calibration data with the most recent first.
+        """
+
+    @abc.abstractmethod
+    def get_calibration(self, calibration_timestamp_seconds: int) -> calibration.Calibration:
+        """Retrieve metadata about a specific calibration run.
+
+        Args:
+            calibration_timestamp_seconds: The timestamp of the calibration in
+                seconds since epoch.
+
+        Returns:
+            The calibration data.
+        """
+
+    @abc.abstractmethod
+    def get_current_calibration(
+        self,
+    ) -> Optional[calibration.Calibration]:
+        """Returns metadata about the current calibration for a processor.
+
+        Returns:
+            The calibration data or None if there is no current calibration.
+        """
+
+    @abc.abstractmethod
+    def create_reservation(
+        self,
+        start_time: datetime.datetime,
+        end_time: datetime.datetime,
+        whitelisted_users: Optional[List[str]] = None,
+    ) -> quantum.types.QuantumReservation:
+        """Creates a reservation on this processor.
+
+        Args:
+            start_time: the starting date/time of the reservation.
+            end_time: the ending date/time of the reservation.
+            whitelisted_users: a list of emails that are allowed
+              to send programs during this reservation (in addition to users
+              with permission "quantum.reservations.use" on the project).
+        """
+
+    @abc.abstractmethod
+    def remove_reservation(self, reservation_id: str) -> None:
+        """Removes a reservation on this processor."""
+
+    @abc.abstractmethod
+    def get_reservation(self, reservation_id: str) -> quantum.types.QuantumReservation:
+        """Retrieve a reservation given its id."""
+
+    @abc.abstractmethod
+    def update_reservation(
+        self,
+        reservation_id: str,
+        start_time: datetime.datetime = None,
+        end_time: datetime.datetime = None,
+        whitelisted_users: List[str] = None,
+    ):
+        """Updates a reservation with new information.
+
+        Updates a reservation with a new start date, end date, or
+        list of additional users.  For each field, it the argument is left as
+        None, it will not be updated.
+        """
+
+    @abc.abstractmethod
+    def list_reservations(
+        self,
+        from_time: Union[None, datetime.datetime, datetime.timedelta],
+        to_time: Union[None, datetime.datetime, datetime.timedelta],
+    ) -> List[quantum.types.QuantumReservation]:
+        """Retrieves the reservations from a processor.
+
+        Only reservations from this processor and project will be
+        returned. The schedule may be filtered by starting and ending time.
+
+        Args:
+            from_time: Filters the returned reservations to only include entries
+                that end no earlier than the given value. Specified either as an
+                absolute time (datetime.datetime) or as a time relative to now
+                (datetime.timedelta). Defaults to now (a relative time of 0).
+                Set to None to omit this filter.
+            to_time: Filters the returned reservations to only include entries
+                that start no later than the given value. Specified either as an
+                absolute time (datetime.datetime) or as a time relative to now
+                (datetime.timedelta). Defaults to two weeks from now (a relative
+                time of two weeks). Set to None to omit this filter.
+
+        Returns:
+            A list of reservations.
+        """
+
+    @abc.abstractmethod
+    def get_schedule(
+        self,
+        from_time: Union[None, datetime.datetime, datetime.timedelta] = datetime.timedelta(),
+        to_time: Union[None, datetime.datetime, datetime.timedelta] = datetime.timedelta(weeks=2),
+        time_slot_type: Optional[quantum.enums.QuantumTimeSlot.TimeSlotType] = None,
+    ) -> List[quantum.enums.QuantumTimeSlot]:
+        """Retrieves the schedule for a processor.
+
+        The schedule may be filtered by time.
+
+        Time slot type will be supported in the future.
+
+        Args:
+            from_time: Filters the returned schedule to only include entries
+                that end no earlier than the given value. Specified either as an
+                absolute time (datetime.datetime) or as a time relative to now
+                (datetime.timedelta). Defaults to now (a relative time of 0).
+                Set to None to omit this filter.
+            to_time: Filters the returned schedule to only include entries
+                that start no later than the given value. Specified either as an
+                absolute time (datetime.datetime) or as a time relative to now
+                (datetime.timedelta). Defaults to two weeks from now (a relative
+                time of two weeks). Set to None to omit this filter.
+            time_slot_type: Filters the returned schedule to only include
+                entries with a given type (e.g. maintenance, open swim).
+                Defaults to None. Set to None to omit this filter.
+
+        Returns:
+            Schedule time slots.
+        """
diff --git a/cirq-google/cirq_google/engine/abstract_program.py b/cirq-google/cirq_google/engine/abstract_program.py
new file mode 100644
index 00000000000..4a9c80b8936
--- /dev/null
+++ b/cirq-google/cirq_google/engine/abstract_program.py
@@ -0,0 +1,191 @@
+# Copyright 2021 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.
+# coverage: ignore
+"""An interface for quantum programs.
+
+The quantum program represents a circuit (or other execution) that,
+when combined with a run context, will become a quantum job.
+"""
+
+
+import abc
+import datetime
+from typing import Dict, List, Optional, Sequence, Set, TYPE_CHECKING, Union
+import cirq
+import cirq_google.engine.client.quantum as quantum
+
+if TYPE_CHECKING:
+    import cirq_google.engine.abstract_job as abstract_job
+    import cirq_google.engine.abstract_engine as abstract_engine
+
+
+class AbstractProgram(abc.ABC):
+    """An abstract object representing a quantum program.
+
+    This program generally wraps a `Circuit` with additional metadata.
+    When combined with an appropriate RunContext, this becomes a
+    Job that can run on either an Engine service or simulator.
+    Programs can also be a batch (list of circuits) or calibration
+    requests.
+
+    This is an abstract class that inheritors should implement.
+    """
+
+    @abc.abstractmethod
+    def engine(self) -> 'abstract_engine.AbstractEngine':
+        """Returns the parent Engine object.
+
+        Returns:
+            The program's parent Engine.
+        """
+
+    @abc.abstractmethod
+    def get_job(self, job_id: str) -> 'abstract_job.AbstractJob':
+        """Returns an AbstractJob for an existing id.
+
+        Args:
+            job_id: Unique ID of the job within the parent program.
+
+        Returns:
+            A AbstractJob for this program.
+        """
+
+    @abc.abstractmethod
+    def list_jobs(
+        self,
+        created_before: Optional[Union[datetime.datetime, datetime.date]] = None,
+        created_after: Optional[Union[datetime.datetime, datetime.date]] = None,
+        has_labels: Optional[Dict[str, str]] = None,
+        execution_states: Optional[Set[quantum.enums.ExecutionStatus.State]] = None,
+    ) -> Sequence['abstract_job.AbstractJob']:
+        """Returns the list of jobs for this program.
+
+        Args:
+            project_id: A project_id of the parent Google Cloud Project.
+            program_id: Unique ID of the program within the parent project.
+            created_after: retrieve jobs that were created after this date
+                or time.
+            created_before: retrieve jobs that were created after this date
+                or time.
+            has_labels: retrieve jobs that have labels on them specified by
+                this dict. If the value is set to `*`, jobs having the label
+                regardless of the label value will be returned. For example, to
+                query programs that have the shape label and have the color
+                label with value red can be queried using
+
+                {'color': 'red', 'shape':'*'}
+
+            execution_states: retrieve jobs that have an execution state  that
+                is contained in `execution_states`. See
+                `quantum.enums.ExecutionStatus.State` enum for accepted values.
+
+        Returns:
+            A sequence of `AbstractJob` objects that satisfy the constraints.
+        """
+
+    @abc.abstractmethod
+    def create_time(self) -> 'datetime.datetime':
+        """Returns when the program was created."""
+
+    @abc.abstractmethod
+    def update_time(self) -> 'datetime.datetime':
+        """Returns when the program was last updated."""
+
+    @abc.abstractmethod
+    def description(self) -> str:
+        """Returns the description of the program."""
+
+    @abc.abstractmethod
+    def set_description(self, description: str) -> 'AbstractProgram':
+        """Sets the description of the program.
+
+        Params:
+            description: The new description for the program.
+
+        Returns:
+             This AbstractProgram.
+        """
+
+    @abc.abstractmethod
+    def labels(self) -> Dict[str, str]:
+        """Returns the labels of the program."""
+
+    @abc.abstractmethod
+    def set_labels(self, labels: Dict[str, str]) -> 'AbstractProgram':
+        """Sets (overwriting) the labels for a previously created quantum program.
+
+        Params:
+            labels: The entire set of new program labels.
+
+        Returns:
+             This AbstractProgram.
+        """
+
+    @abc.abstractmethod
+    def add_labels(self, labels: Dict[str, str]) -> 'AbstractProgram':
+        """Adds new labels to a previously created quantum program.
+
+        Params:
+            labels: New labels to add to the existing program labels.
+
+        Returns:
+             This AbstractProgram.
+        """
+
+    @abc.abstractmethod
+    def remove_labels(self, keys: List[str]) -> 'AbstractProgram':
+        """Removes labels with given keys from the labels of a previously
+        created quantum program.
+
+        Params:
+            label_keys: Label keys to remove from the existing program labels.
+
+        Returns:
+             This AbstractProgram.
+        """
+
+    @abc.abstractmethod
+    def get_circuit(self, program_num: Optional[int] = None) -> cirq.Circuit:
+        """Returns the cirq Circuit for the program. This is only
+        supported if the program was created with the V2 protos.
+
+        Args:
+            program_num: if this is a batch program, the index of the circuit in
+                the batch.  This argument is zero-indexed. Negative values
+                indexing from the end of the list.
+
+        Returns:
+            The program's cirq Circuit.
+        """
+
+    @abc.abstractmethod
+    def batch_size(self) -> int:
+        """Returns the number of programs in a batch program.
+
+        Raises:
+            ValueError: if the program created was not a batch program.
+        """
+
+    @abc.abstractmethod
+    def delete(self, delete_jobs: bool = False) -> None:
+        """Deletes a previously created quantum program.
+
+        Params:
+            delete_jobs: If True will delete all the program's jobs, other this
+                will fail if the program contains any jobs.
+        """
+
+    @abc.abstractmethod
+    def delete_job(self, job_id: str) -> None:
+        """Removes a child job from this program."""