diff --git a/api/src/opentrons/protocol_api/core/engine/instrument.py b/api/src/opentrons/protocol_api/core/engine/instrument.py
index dd6c2ff1bf0..7772fcbeb15 100644
--- a/api/src/opentrons/protocol_api/core/engine/instrument.py
+++ b/api/src/opentrons/protocol_api/core/engine/instrument.py
@@ -84,6 +84,7 @@ def aspirate(
         volume: float,
         rate: float,
         flow_rate: float,
+        in_place: bool,
     ) -> None:
         """Aspirate a given volume of liquid from the specified location.
         Args:
@@ -92,29 +93,44 @@ def aspirate(
             well_core: The well to aspirate from, if applicable.
             rate: Not used in this core.
             flow_rate: The flow rate in µL/s to aspirate at.
+            in_place: whether this is a in-place command.
         """
         if well_core is None:
-            raise NotImplementedError(
-                "InstrumentCore.aspirate with well_core value of None not implemented"
+            if not in_place:
+                self._engine_client.move_to_coordinates(
+                    pipette_id=self._pipette_id,
+                    coordinates=DeckPoint(
+                        x=location.point.x, y=location.point.y, z=location.point.z
+                    ),
+                    minimum_z_height=None,
+                    force_direct=False,
+                    speed=None,
+                )
+
+            self._engine_client.aspirate_in_place(
+                pipette_id=self._pipette_id, volume=volume, flow_rate=flow_rate
             )
 
-        well_name = well_core.get_name()
-        labware_id = well_core.labware_id
+        else:
+            well_name = well_core.get_name()
+            labware_id = well_core.labware_id
 
-        well_location = self._engine_client.state.geometry.get_relative_well_location(
-            labware_id=labware_id,
-            well_name=well_name,
-            absolute_point=location.point,
-        )
+            well_location = (
+                self._engine_client.state.geometry.get_relative_well_location(
+                    labware_id=labware_id,
+                    well_name=well_name,
+                    absolute_point=location.point,
+                )
+            )
 
-        self._engine_client.aspirate(
-            pipette_id=self._pipette_id,
-            labware_id=labware_id,
-            well_name=well_name,
-            well_location=well_location,
-            volume=volume,
-            flow_rate=flow_rate,
-        )
+            self._engine_client.aspirate(
+                pipette_id=self._pipette_id,
+                labware_id=labware_id,
+                well_name=well_name,
+                well_location=well_location,
+                volume=volume,
+                flow_rate=flow_rate,
+            )
 
         self._protocol_core.set_last_location(location=location, mount=self.get_mount())
 
@@ -125,6 +141,7 @@ def dispense(
         volume: float,
         rate: float,
         flow_rate: float,
+        in_place: bool,
     ) -> None:
         """Dispense a given volume of liquid into the specified location.
         Args:
@@ -133,61 +150,93 @@ def dispense(
             well_core: The well to dispense to, if applicable.
             rate: Not used in this core.
             flow_rate: The flow rate in µL/s to dispense at.
+            in_place: whether this is a in-place command.
         """
         if well_core is None:
-            raise NotImplementedError(
-                "InstrumentCore.dispense with well_core value of None not implemented"
-            )
+            if not in_place:
+                self._engine_client.move_to_coordinates(
+                    pipette_id=self._pipette_id,
+                    coordinates=DeckPoint(
+                        x=location.point.x, y=location.point.y, z=location.point.z
+                    ),
+                    minimum_z_height=None,
+                    force_direct=False,
+                    speed=None,
+                )
 
-        well_name = well_core.get_name()
-        labware_id = well_core.labware_id
+            self._engine_client.dispense_in_place(
+                pipette_id=self._pipette_id, volume=volume, flow_rate=flow_rate
+            )
+        else:
+            well_name = well_core.get_name()
+            labware_id = well_core.labware_id
 
-        well_location = self._engine_client.state.geometry.get_relative_well_location(
-            labware_id=labware_id, well_name=well_name, absolute_point=location.point
-        )
+            well_location = (
+                self._engine_client.state.geometry.get_relative_well_location(
+                    labware_id=labware_id,
+                    well_name=well_name,
+                    absolute_point=location.point,
+                )
+            )
 
-        self._engine_client.dispense(
-            pipette_id=self._pipette_id,
-            labware_id=labware_id,
-            well_name=well_name,
-            well_location=well_location,
-            volume=volume,
-            flow_rate=flow_rate,
-        )
+            self._engine_client.dispense(
+                pipette_id=self._pipette_id,
+                labware_id=labware_id,
+                well_name=well_name,
+                well_location=well_location,
+                volume=volume,
+                flow_rate=flow_rate,
+            )
 
         self._protocol_core.set_last_location(location=location, mount=self.get_mount())
 
     def blow_out(
-        self, location: Location, well_core: Optional[WellCore], move_to_well: bool
+        self, location: Location, well_core: Optional[WellCore], in_place: bool
     ) -> None:
         """Blow liquid out of the tip.
 
         Args:
             location: The location to blow out into.
             well_core: The well to blow out into.
-            move_to_well: Unused by engine core.
+            in_place: whether this is a in-place command.
         """
+        flow_rate = self.get_blow_out_flow_rate(1.0)
         if well_core is None:
-            raise NotImplementedError("In-place blow-out is not implemented")
+            if not in_place:
+                self._engine_client.move_to_coordinates(
+                    pipette_id=self._pipette_id,
+                    coordinates=DeckPoint(
+                        x=location.point.x, y=location.point.y, z=location.point.z
+                    ),
+                    force_direct=False,
+                    minimum_z_height=None,
+                    speed=None,
+                )
 
-        well_name = well_core.get_name()
-        labware_id = well_core.labware_id
+            self._engine_client.blow_out_in_place(
+                pipette_id=self._pipette_id, flow_rate=flow_rate
+            )
+        else:
+            well_name = well_core.get_name()
+            labware_id = well_core.labware_id
 
-        well_location = self._engine_client.state.geometry.get_relative_well_location(
-            labware_id=labware_id,
-            well_name=well_name,
-            absolute_point=location.point,
-        )
+            well_location = (
+                self._engine_client.state.geometry.get_relative_well_location(
+                    labware_id=labware_id,
+                    well_name=well_name,
+                    absolute_point=location.point,
+                )
+            )
 
-        self._engine_client.blow_out(
-            pipette_id=self._pipette_id,
-            labware_id=labware_id,
-            well_name=well_name,
-            well_location=well_location,
-            # TODO(jbl 2022-11-07) PAPIv2 does not have an argument for rate and
-            #   this also needs to be refactored along with other flow rate related issues
-            flow_rate=self.get_blow_out_flow_rate(),
-        )
+            self._engine_client.blow_out(
+                pipette_id=self._pipette_id,
+                labware_id=labware_id,
+                well_name=well_name,
+                well_location=well_location,
+                # TODO(jbl 2022-11-07) PAPIv2 does not have an argument for rate and
+                #   this also needs to be refactored along with other flow rate related issues
+                flow_rate=flow_rate,
+            )
 
         self._protocol_core.set_last_location(location=location, mount=self.get_mount())
 
diff --git a/api/src/opentrons/protocol_api/core/instrument.py b/api/src/opentrons/protocol_api/core/instrument.py
index 929f1aef7ac..3f4ac9a859d 100644
--- a/api/src/opentrons/protocol_api/core/instrument.py
+++ b/api/src/opentrons/protocol_api/core/instrument.py
@@ -29,6 +29,7 @@ def aspirate(
         volume: float,
         rate: float,
         flow_rate: float,
+        in_place: bool,
     ) -> None:
         """Aspirate a given volume of liquid from the specified location.
         Args:
@@ -37,6 +38,7 @@ def aspirate(
             well_core: The well to aspirate from, if applicable.
             rate: The rate for how quickly to aspirate.
             flow_rate: The flow rate in µL/s to aspirate at.
+            in_place: Whether this is in-place.
         """
         ...
 
@@ -48,6 +50,7 @@ def dispense(
         volume: float,
         rate: float,
         flow_rate: float,
+        in_place: bool,
     ) -> None:
         """Dispense a given volume of liquid into the specified location.
         Args:
@@ -56,6 +59,7 @@ def dispense(
             well_core: The well to dispense to, if applicable.
             rate: The rate for how quickly to dispense.
             flow_rate: The flow rate in µL/s to dispense at.
+            in_place: Whether this is in-place.
         """
         ...
 
@@ -64,14 +68,14 @@ def blow_out(
         self,
         location: types.Location,
         well_core: Optional[WellCoreType],
-        move_to_well: bool,
+        in_place: bool,
     ) -> None:
         """Blow liquid out of the tip.
 
         Args:
             location: The location to blow out into.
             well_core: The well to blow out into.
-            move_to_well: If pipette should be moved before blow-out.
+            in_place: Whether this is in-place.
         """
         ...
 
diff --git a/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py b/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py
index c7c06fd82a3..71c213039b2 100644
--- a/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py
+++ b/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py
@@ -73,6 +73,7 @@ def aspirate(
         volume: float,
         rate: float,
         flow_rate: float,
+        in_place: bool,
     ) -> None:
         """Aspirate a given volume of liquid from the specified location.
         Args:
@@ -81,6 +82,7 @@ def aspirate(
             well_core: The well to aspirate from, if applicable.
             rate: The rate in µL/s to aspirate at.
             flow_rate: Not used in this core.
+            in_place: Whether we should move_to location.
         """
         if self.get_current_volume() == 0:
             # Make sure we're at the top of the labware and clear of any
@@ -100,7 +102,7 @@ def aspirate(
                     )
                 self.prepare_for_aspirate()
             self.move_to(location=location)
-        elif location != self._protocol_interface.get_last_location():
+        elif not in_place:
             self.move_to(location=location)
 
         self._protocol_interface.get_hardware().aspirate(self._mount, volume, rate)
@@ -112,6 +114,7 @@ def dispense(
         volume: float,
         rate: float,
         flow_rate: float,
+        in_place: bool,
     ) -> None:
         """Dispense a given volume of liquid into the specified location.
         Args:
@@ -120,8 +123,10 @@ def dispense(
             well_core: The well to dispense to, if applicable.
             rate: The rate in µL/s to dispense at.
             flow_rate: Not used in this core.
+            in_place: Whether we should move_to location.
         """
-        self.move_to(location=location)
+        if not in_place:
+            self.move_to(location=location)
 
         self._protocol_interface.get_hardware().dispense(self._mount, volume, rate)
 
@@ -129,16 +134,16 @@ def blow_out(
         self,
         location: types.Location,
         well_core: Optional[LegacyWellCore],
-        move_to_well: bool,
+        in_place: bool,
     ) -> None:
         """Blow liquid out of the tip.
 
         Args:
             location: The location to blow out into.
             well_core: Unused by legacy core.
-            move_to_well: If pipette should be moved before blow-out.
+            in_place: Whether we should move_to location.
         """
-        if move_to_well:
+        if not in_place:
             self.move_to(location=location)
         self._protocol_interface.get_hardware().blow_out(self._mount)
 
diff --git a/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py b/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py
index 1de8ace714f..ac71b590502 100644
--- a/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py
+++ b/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py
@@ -77,6 +77,7 @@ def aspirate(
         volume: float,
         rate: float,
         flow_rate: float,
+        in_place: bool,
     ) -> None:
         if self.get_current_volume() == 0:
             # Make sure we're at the top of the labware and clear of any
@@ -116,8 +117,10 @@ def dispense(
         volume: float,
         rate: float,
         flow_rate: float,
+        in_place: bool,
     ) -> None:
-        self.move_to(location=location, well_core=well_core)
+        if not in_place:
+            self.move_to(location=location, well_core=well_core)
         self._raise_if_no_tip(HardwareAction.DISPENSE.name)
         self._update_volume(self.get_current_volume() - volume)
 
@@ -125,9 +128,9 @@ def blow_out(
         self,
         location: types.Location,
         well_core: Optional[LegacyWellCore],
-        move_to_well: bool,
+        in_place: bool,
     ) -> None:
-        if move_to_well:
+        if not in_place:
             self.move_to(location=location, well_core=well_core)
         self._raise_if_no_tip(HardwareAction.BLOWOUT.name)
         self._update_volume(0)
diff --git a/api/src/opentrons/protocol_api/instrument_context.py b/api/src/opentrons/protocol_api/instrument_context.py
index f0d5c3fe1d1..5cc4352d099 100644
--- a/api/src/opentrons/protocol_api/instrument_context.py
+++ b/api/src/opentrons/protocol_api/instrument_context.py
@@ -23,8 +23,9 @@
 )
 
 from .core.common import InstrumentCore, ProtocolCore
+from .core.engine import ENGINE_CORE_API_VERSION
 from .config import Clearances
-from . import labware
+from . import labware, validation
 
 
 AdvancedLiquidHandling = Union[
@@ -173,29 +174,30 @@ def aspirate(
             )
         )
 
-        well: Optional[labware.Well]
-        last_location = self._protocol_core.get_last_location()
+        well: Optional[labware.Well] = None
+        move_to_location: types.Location
 
-        if isinstance(location, labware.Well):
-            move_to_location = location.bottom(z=self._well_bottom_clearances.aspirate)
-            well = location
-        elif isinstance(location, types.Location):
-            move_to_location = location
-            _, well = move_to_location.labware.get_parent_labware_and_well()
-        elif location is not None:
-            raise TypeError(
-                "location should be a Well or Location, but it is {}".format(location)
+        last_location = self._get_last_location_by_api_version()
+        try:
+            target = validation.validate_location(
+                location=location, last_location=last_location
             )
-        elif last_location:
-            move_to_location = last_location
-            _, well = move_to_location.labware.get_parent_labware_and_well()
-        else:
+        except validation.NoLocationError as e:
             raise RuntimeError(
                 "If aspirate is called without an explicit location, another"
                 " method that moves to a location (such as move_to or "
                 "dispense) must previously have been called so the robot "
                 "knows where it is."
+            ) from e
+
+        if isinstance(target, validation.WellTarget):
+            move_to_location = target.location or target.well.bottom(
+                z=self._well_bottom_clearances.aspirate
             )
+            well = target.well
+        if isinstance(target, validation.PointTarget):
+            move_to_location = target.location
+
         if self.api_version >= APIVersion(2, 11):
             instrument.validate_takes_liquid(
                 location=move_to_location,
@@ -221,6 +223,7 @@ def aspirate(
                 volume=c_vol,
                 rate=rate,
                 flow_rate=flow_rate,
+                in_place=target.in_place,
             )
 
         return self
@@ -277,34 +280,34 @@ def dispense(
                 volume, location if location else "current position", rate
             )
         )
-        well: Optional[labware.Well]
-        last_location = self._protocol_core.get_last_location()
+        well: Optional[labware.Well] = None
+        last_location = self._get_last_location_by_api_version()
 
-        if isinstance(location, labware.Well):
-            well = location
-            if well.parent._core.is_fixed_trash():
-                move_to_location = location.top()
-            else:
-                move_to_location = location.bottom(
-                    z=self._well_bottom_clearances.dispense
-                )
-        elif isinstance(location, types.Location):
-            move_to_location = location
-            _, well = move_to_location.labware.get_parent_labware_and_well()
-        elif location is not None:
-            raise TypeError(
-                f"location should be a Well or Location, but it is {location}"
+        try:
+            target = validation.validate_location(
+                location=location, last_location=last_location
             )
-        elif last_location:
-            move_to_location = last_location
-            _, well = move_to_location.labware.get_parent_labware_and_well()
-        else:
+        except validation.NoLocationError as e:
             raise RuntimeError(
                 "If dispense is called without an explicit location, another"
                 " method that moves to a location (such as move_to or "
                 "aspirate) must previously have been called so the robot "
                 "knows where it is."
-            )
+            ) from e
+
+        if isinstance(target, validation.WellTarget):
+            well = target.well
+            if target.location:
+                move_to_location = target.location
+            elif well.parent._core.is_fixed_trash():
+                move_to_location = target.well.top()
+            else:
+                move_to_location = target.well.bottom(
+                    z=self._well_bottom_clearances.dispense
+                )
+        if isinstance(target, validation.PointTarget):
+            move_to_location = target.location
+
         if self.api_version >= APIVersion(2, 11):
             instrument.validate_takes_liquid(
                 location=move_to_location,
@@ -331,6 +334,7 @@ def dispense(
                 location=move_to_location,
                 well_core=well._core if well is not None else None,
                 flow_rate=flow_rate,
+                in_place=target.in_place,
             )
 
         return self
@@ -432,50 +436,41 @@ def blow_out(
                               :py:meth:`dispense`)
         :returns: This instance
         """
+        well: Optional[labware.Well] = None
+        move_to_location: types.Location
 
-        well: Optional[labware.Well]
-        # TODO(jbl 2022-11-10) refactor this boolean out and make location optional when PE blow-out in place exists
-        move_to_well = True
-        last_location = self._protocol_core.get_last_location()
-
-        if isinstance(location, labware.Well):
-            if location.parent.is_tiprack:
-                _log.warning(
-                    "Blow_out being performed on a tiprack. "
-                    "Please re-check your code"
-                )
-            checked_loc = location.top()
-            well = location
-        elif isinstance(location, types.Location):
-            checked_loc = location
-            _, well = location.labware.get_parent_labware_and_well()
-        elif location is not None:
-            raise TypeError(
-                "location should be a Well or Location, but it is {}".format(location)
+        last_location = self._get_last_location_by_api_version()
+        try:
+            target = validation.validate_location(
+                location=location, last_location=last_location
             )
-        elif last_location:
-            checked_loc = last_location
-            _, well = checked_loc.labware.get_parent_labware_and_well()
-            # if no explicit location given but location cache exists,
-            # pipette blows out immediately at
-            # current location, no movement is needed
-            move_to_well = False
-        else:
+        except validation.NoLocationError as e:
             raise RuntimeError(
                 "If blow out is called without an explicit location, another"
                 " method that moves to a location (such as move_to or "
                 "dispense) must previously have been called so the robot "
                 "knows where it is."
-            )
+            ) from e
+
+        if isinstance(target, validation.WellTarget):
+            if target.well.parent.is_tiprack:
+                _log.warning(
+                    "Blow_out being performed on a tiprack. "
+                    "Please re-check your code"
+                )
+            move_to_location = target.location or target.well.top()
+            well = target.well
+        elif isinstance(target, validation.PointTarget):
+            move_to_location = target.location
 
         with publisher.publish_context(
             broker=self.broker,
-            command=cmds.blow_out(instrument=self, location=checked_loc),
+            command=cmds.blow_out(instrument=self, location=move_to_location),
         ):
             self._core.blow_out(
-                location=checked_loc,
+                location=move_to_location,
                 well_core=well._core if well is not None else None,
-                move_to_well=move_to_well,
+                in_place=target.in_place,
             )
 
         return self
@@ -1509,6 +1504,17 @@ def well_bottom_clearance(self) -> "Clearances":
         """
         return self._well_bottom_clearances
 
+    def _get_last_location_by_api_version(self) -> Optional[types.Location]:
+        """Get the last location accessed by this pipette, if any.
+
+        In pre-engine Protocol API versions, this call omits the pipette mount.
+        This is to preserve pre-existing, potentially buggy behavior.
+        """
+        if self._api_version >= ENGINE_CORE_API_VERSION:
+            return self._protocol_core.get_last_location(mount=self._core.get_mount())
+        else:
+            return self._protocol_core.get_last_location()
+
     def __repr__(self) -> str:
         return "<{}: {} in {}>".format(
             self.__class__.__name__,
diff --git a/api/src/opentrons/protocol_api/validation.py b/api/src/opentrons/protocol_api/validation.py
index d3b66b0d627..c4d7f897248 100644
--- a/api/src/opentrons/protocol_api/validation.py
+++ b/api/src/opentrons/protocol_api/validation.py
@@ -1,9 +1,22 @@
-from typing import Any, Dict, List, Optional, Sequence, Union, Tuple, Mapping
+from __future__ import annotations
+from typing import (
+    Any,
+    Dict,
+    List,
+    Optional,
+    Sequence,
+    Union,
+    Tuple,
+    Mapping,
+    NamedTuple,
+    TYPE_CHECKING,
+)
+
 from typing_extensions import TypeGuard
 
 from opentrons_shared_data.pipette.dev_types import PipetteNameType
 
-from opentrons.types import Mount, DeckSlotName
+from opentrons.types import Mount, DeckSlotName, Location
 from opentrons.hardware_control.modules.types import (
     ModuleModel,
     MagneticModuleModel,
@@ -13,6 +26,9 @@
     ThermocyclerStep,
 )
 
+if TYPE_CHECKING:
+    from .labware import Well
+
 
 def ensure_mount(mount: Union[str, Mount]) -> Mount:
     """Ensure that an input value represents a valid Mount."""
@@ -180,3 +196,69 @@ def ensure_valid_labware_offset_vector(
     if not all(isinstance(v, (float, int)) for v in offsets):
         raise TypeError("Offset values should be a number (int or float).")
     return offsets
+
+
+class WellTarget(NamedTuple):
+    """A movement target that is a well."""
+
+    well: Well
+    location: Optional[Location]
+    in_place: bool
+
+
+class PointTarget(NamedTuple):
+    """A movement to coordinates"""
+
+    location: Location
+    in_place: bool
+
+
+class NoLocationError(ValueError):
+    """Error representing that no location was supplied."""
+
+
+class LocationTypeError(TypeError):
+    """Error representing that the location supplied is of different expected type."""
+
+
+def validate_location(
+    location: Union[Location, Well, None], last_location: Optional[Location]
+) -> Union[WellTarget, PointTarget]:
+    """Validate a given location for a liquid handling command.
+
+    Args:
+        location: The input location.
+        last_location: The last location accessed by the pipette.
+
+    Returns:
+        A `WellTarget` if the input location represents a well.
+        A `PointTarget` if the input location is an x, y, z coordinate.
+
+    Raises:
+        NoLocationError: The is no input location and no cached loaction.
+        LocationTypeError: The location supplied is of unexpected type.
+    """
+    from .labware import Well
+
+    target_location = location or last_location
+
+    if target_location is None:
+        raise NoLocationError()
+
+    if not isinstance(target_location, (Location, Well)):
+        raise LocationTypeError(
+            f"location should be a Well or Location, but it is {location}"
+        )
+
+    in_place = target_location == last_location
+
+    if isinstance(target_location, Well):
+        return WellTarget(well=target_location, location=None, in_place=in_place)
+
+    _, well = target_location.labware.get_parent_labware_and_well()
+
+    return (
+        WellTarget(well=well, location=target_location, in_place=in_place)
+        if well is not None
+        else PointTarget(location=target_location, in_place=in_place)
+    )
diff --git a/api/src/opentrons/protocol_engine/clients/sync_client.py b/api/src/opentrons/protocol_engine/clients/sync_client.py
index 672e0ccf0d2..48e0903581b 100644
--- a/api/src/opentrons/protocol_engine/clients/sync_client.py
+++ b/api/src/opentrons/protocol_engine/clients/sync_client.py
@@ -261,6 +261,24 @@ def aspirate(
 
         return cast(commands.AspirateResult, result)
 
+    def aspirate_in_place(
+        self,
+        pipette_id: str,
+        volume: float,
+        flow_rate: float,
+    ) -> commands.AspirateInPlaceResult:
+        """Execute an ``AspirateInPlace`` command and return the result."""
+        request = commands.AspirateInPlaceCreate(
+            params=commands.AspirateInPlaceParams(
+                pipetteId=pipette_id,
+                volume=volume,
+                flowRate=flow_rate,
+            )
+        )
+        result = self._transport.execute_command(request=request)
+
+        return cast(commands.AspirateInPlaceResult, result)
+
     def dispense(
         self,
         pipette_id: str,
@@ -284,6 +302,23 @@ def dispense(
         result = self._transport.execute_command(request=request)
         return cast(commands.DispenseResult, result)
 
+    def dispense_in_place(
+        self,
+        pipette_id: str,
+        volume: float,
+        flow_rate: float,
+    ) -> commands.DispenseInPlaceResult:
+        """Execute a ``DispenseInPlace`` command and return the result."""
+        request = commands.DispenseInPlaceCreate(
+            params=commands.DispenseInPlaceParams(
+                pipetteId=pipette_id,
+                volume=volume,
+                flowRate=flow_rate,
+            )
+        )
+        result = self._transport.execute_command(request=request)
+        return cast(commands.DispenseInPlaceResult, result)
+
     def blow_out(
         self,
         pipette_id: str,
@@ -305,6 +340,21 @@ def blow_out(
         result = self._transport.execute_command(request=request)
         return cast(commands.BlowOutResult, result)
 
+    def blow_out_in_place(
+        self,
+        pipette_id: str,
+        flow_rate: float,
+    ) -> commands.BlowOutInPlaceResult:
+        """Execute a ``BlowOutInPlace`` command and return the result."""
+        request = commands.BlowOutInPlaceCreate(
+            params=commands.BlowOutInPlaceParams(
+                pipetteId=pipette_id,
+                flowRate=flow_rate,
+            )
+        )
+        result = self._transport.execute_command(request=request)
+        return cast(commands.BlowOutInPlaceResult, result)
+
     def touch_tip(
         self,
         pipette_id: str,
diff --git a/api/src/opentrons/protocol_engine/commands/__init__.py b/api/src/opentrons/protocol_engine/commands/__init__.py
index 0c463582f33..006218c45dd 100644
--- a/api/src/opentrons/protocol_engine/commands/__init__.py
+++ b/api/src/opentrons/protocol_engine/commands/__init__.py
@@ -46,6 +46,14 @@
     AspirateCommandType,
 )
 
+from .aspirate_in_place import (
+    AspirateInPlace,
+    AspirateInPlaceParams,
+    AspirateInPlaceCreate,
+    AspirateInPlaceResult,
+    AspirateInPlaceCommandType,
+)
+
 from .comment import (
     Comment,
     CommentParams,
@@ -215,6 +223,14 @@
     BlowOut,
 )
 
+from .blow_out_in_place import (
+    BlowOutInPlaceParams,
+    BlowOutInPlaceResult,
+    BlowOutInPlaceCreate,
+    BlowOutInPlaceImplementation,
+    BlowOutInPlace,
+)
+
 __all__ = [
     # command type unions
     "Command",
@@ -238,6 +254,12 @@
     "AspirateParams",
     "AspirateResult",
     "AspirateCommandType",
+    # aspirate in place command models
+    "AspirateInPlace",
+    "AspirateInPlaceCreate",
+    "AspirateInPlaceParams",
+    "AspirateInPlaceResult",
+    "AspirateInPlaceCommandType",
     # comment command models
     "Comment",
     "CommentParams",
@@ -358,6 +380,12 @@
     "BlowOutImplementation",
     "BlowOutParams",
     "BlowOut",
+    # blow out in place command models
+    "BlowOutInPlaceParams",
+    "BlowOutInPlaceResult",
+    "BlowOutInPlaceCreate",
+    "BlowOutInPlaceImplementation",
+    "BlowOutInPlace",
     # load liquid command models
     "LoadLiquid",
     "LoadLiquidCreate",
diff --git a/api/src/opentrons/protocol_engine/commands/aspirate_in_place.py b/api/src/opentrons/protocol_engine/commands/aspirate_in_place.py
new file mode 100644
index 00000000000..d9f7d56541a
--- /dev/null
+++ b/api/src/opentrons/protocol_engine/commands/aspirate_in_place.py
@@ -0,0 +1,101 @@
+"""Aspirate in place command request, result, and implementation models."""
+
+from __future__ import annotations
+from typing import TYPE_CHECKING, Optional, Type
+from typing_extensions import Literal
+
+from opentrons.hardware_control import HardwareControlAPI
+
+from .pipetting_common import (
+    PipetteIdMixin,
+    VolumeMixin,
+    FlowRateMixin,
+    BaseLiquidHandlingResult,
+)
+from .command import AbstractCommandImpl, BaseCommand, BaseCommandCreate
+from ..errors.exceptions import PipetteNotReadyToAspirateError
+
+if TYPE_CHECKING:
+    from ..execution import PipettingHandler
+    from ..state import StateView
+
+
+AspirateInPlaceCommandType = Literal["aspirateInPlace"]
+
+
+class AspirateInPlaceParams(PipetteIdMixin, VolumeMixin, FlowRateMixin):
+    """Payload required to aspirate in place."""
+
+    pass
+
+
+class AspirateInPlaceResult(BaseLiquidHandlingResult):
+    """Result data from the execution of a AspirateInPlace command."""
+
+    pass
+
+
+class AspirateInPlaceImplementation(
+    AbstractCommandImpl[AspirateInPlaceParams, AspirateInPlaceResult]
+):
+    """AspirateInPlace command implementation."""
+
+    def __init__(
+        self,
+        pipetting: PipettingHandler,
+        hardware_api: HardwareControlAPI,
+        state_view: StateView,
+        **kwargs: object,
+    ) -> None:
+        self._pipetting = pipetting
+        self._state_view = state_view
+        self._hardware_api = hardware_api
+
+    async def execute(self, params: AspirateInPlaceParams) -> AspirateInPlaceResult:
+        """Aspirate without moving the pipette."""
+        hw_pipette = self._state_view.pipettes.get_hardware_pipette(
+            pipette_id=params.pipetteId,
+            attached_pipettes=self._hardware_api.attached_instruments,
+        )
+
+        ready_to_aspirate = self._state_view.pipettes.get_is_ready_to_aspirate(
+            pipette_id=params.pipetteId,
+            pipette_config=hw_pipette.config,
+        )
+
+        if not ready_to_aspirate:
+            raise PipetteNotReadyToAspirateError(
+                "Pipette cannot aspirate in place because of a previous blow out."
+                " The first aspirate following a blow-out must be from a specific well"
+                " so the plunger can be reset in a known safe position."
+            )
+
+        with self._pipetting.set_flow_rate(
+            pipette=hw_pipette, aspirate_flow_rate=params.flowRate
+        ):
+            await self._hardware_api.aspirate(
+                mount=hw_pipette.mount, volume=params.volume
+            )
+
+        return AspirateInPlaceResult(volume=params.volume)
+
+
+class AspirateInPlace(BaseCommand[AspirateInPlaceParams, AspirateInPlaceResult]):
+    """AspirateInPlace command model."""
+
+    commandType: AspirateInPlaceCommandType = "aspirateInPlace"
+    params: AspirateInPlaceParams
+    result: Optional[AspirateInPlaceResult]
+
+    _ImplementationCls: Type[
+        AspirateInPlaceImplementation
+    ] = AspirateInPlaceImplementation
+
+
+class AspirateInPlaceCreate(BaseCommandCreate[AspirateInPlaceParams]):
+    """AspirateInPlace command request model."""
+
+    commandType: AspirateInPlaceCommandType = "aspirateInPlace"
+    params: AspirateInPlaceParams
+
+    _CommandCls: Type[AspirateInPlace] = AspirateInPlace
diff --git a/api/src/opentrons/protocol_engine/commands/blow_out_in_place.py b/api/src/opentrons/protocol_engine/commands/blow_out_in_place.py
new file mode 100644
index 00000000000..ec5effdc236
--- /dev/null
+++ b/api/src/opentrons/protocol_engine/commands/blow_out_in_place.py
@@ -0,0 +1,86 @@
+"""Blow-out in place command request, result, and implementation models."""
+
+from __future__ import annotations
+from typing import TYPE_CHECKING, Optional, Type
+from typing_extensions import Literal
+from pydantic import BaseModel
+
+from .pipetting_common import (
+    PipetteIdMixin,
+    FlowRateMixin,
+)
+from .command import AbstractCommandImpl, BaseCommand, BaseCommandCreate
+
+from opentrons.hardware_control import HardwareControlAPI
+
+
+if TYPE_CHECKING:
+    from ..execution import PipettingHandler
+    from ..state import StateView
+
+
+BlowOutInPlaceCommandType = Literal["blowOutInPlace"]
+
+
+class BlowOutInPlaceParams(PipetteIdMixin, FlowRateMixin):
+    """Payload required to blow-out in place."""
+
+    pass
+
+
+class BlowOutInPlaceResult(BaseModel):
+    """Result data from the execution of a BlowOutInPlace command."""
+
+    pass
+
+
+class BlowOutInPlaceImplementation(
+    AbstractCommandImpl[BlowOutInPlaceParams, BlowOutInPlaceResult]
+):
+    """BlowOutInPlace command implementation."""
+
+    def __init__(
+        self,
+        pipetting: PipettingHandler,
+        state_view: StateView,
+        hardware_api: HardwareControlAPI,
+        **kwargs: object,
+    ) -> None:
+        self._pipetting = pipetting
+        self._state_view = state_view
+        self._hardware_api = hardware_api
+
+    async def execute(self, params: BlowOutInPlaceParams) -> BlowOutInPlaceResult:
+        """Blow-out without moving the pipette."""
+        hw_pipette = self._state_view.pipettes.get_hardware_pipette(
+            pipette_id=params.pipetteId,
+            attached_pipettes=self._hardware_api.attached_instruments,
+        )
+
+        with self._pipetting.set_flow_rate(
+            pipette=hw_pipette, blow_out_flow_rate=params.flowRate
+        ):
+            await self._hardware_api.blow_out(mount=hw_pipette.mount)
+
+        return BlowOutInPlaceResult()
+
+
+class BlowOutInPlace(BaseCommand[BlowOutInPlaceParams, BlowOutInPlaceResult]):
+    """BlowOutInPlace command model."""
+
+    commandType: BlowOutInPlaceCommandType = "blowOutInPlace"
+    params: BlowOutInPlaceParams
+    result: Optional[BlowOutInPlaceResult]
+
+    _ImplementationCls: Type[
+        BlowOutInPlaceImplementation
+    ] = BlowOutInPlaceImplementation
+
+
+class BlowOutInPlaceCreate(BaseCommandCreate[BlowOutInPlaceParams]):
+    """BlowOutInPlace command request model."""
+
+    commandType: BlowOutInPlaceCommandType = "blowOutInPlace"
+    params: BlowOutInPlaceParams
+
+    _CommandCls: Type[BlowOutInPlace] = BlowOutInPlace
diff --git a/api/src/opentrons/protocol_engine/commands/command_unions.py b/api/src/opentrons/protocol_engine/commands/command_unions.py
index 67cc0b29ce5..6db76a48cf6 100644
--- a/api/src/opentrons/protocol_engine/commands/command_unions.py
+++ b/api/src/opentrons/protocol_engine/commands/command_unions.py
@@ -25,6 +25,14 @@
     AspirateCommandType,
 )
 
+from .aspirate_in_place import (
+    AspirateInPlace,
+    AspirateInPlaceParams,
+    AspirateInPlaceCreate,
+    AspirateInPlaceResult,
+    AspirateInPlaceCommandType,
+)
+
 from .comment import (
     Comment,
     CommentParams,
@@ -185,13 +193,23 @@
     BlowOutResult,
 )
 
+from .blow_out_in_place import (
+    BlowOutInPlaceParams,
+    BlowOutInPlace,
+    BlowOutInPlaceCreate,
+    BlowOutInPlaceCommandType,
+    BlowOutInPlaceResult,
+)
+
 Command = Union[
     Aspirate,
+    AspirateInPlace,
     Comment,
     Custom,
     Dispense,
     DispenseInPlace,
     BlowOut,
+    BlowOutInPlace,
     DropTip,
     Home,
     LoadLabware,
@@ -236,11 +254,13 @@
 
 CommandParams = Union[
     AspirateParams,
+    AspirateInPlaceParams,
     CommentParams,
     CustomParams,
     DispenseParams,
     DispenseInPlaceParams,
     BlowOutParams,
+    BlowOutInPlaceParams,
     DropTipParams,
     HomeParams,
     LoadLabwareParams,
@@ -286,11 +306,13 @@
 
 CommandType = Union[
     AspirateCommandType,
+    AspirateInPlaceCommandType,
     CommentCommandType,
     CustomCommandType,
     DispenseCommandType,
     DispenseInPlaceCommandType,
     BlowOutCommandType,
+    BlowOutInPlaceCommandType,
     DropTipCommandType,
     HomeCommandType,
     LoadLabwareCommandType,
@@ -335,11 +357,13 @@
 
 CommandCreate = Union[
     AspirateCreate,
+    AspirateInPlaceCreate,
     CommentCreate,
     CustomCreate,
     DispenseCreate,
     DispenseInPlaceCreate,
     BlowOutCreate,
+    BlowOutInPlaceCreate,
     DropTipCreate,
     HomeCreate,
     LoadLabwareCreate,
@@ -384,11 +408,13 @@
 
 CommandResult = Union[
     AspirateResult,
+    AspirateInPlaceResult,
     CommentResult,
     CustomResult,
     DispenseResult,
     DispenseInPlaceResult,
     BlowOutResult,
+    BlowOutInPlaceResult,
     DropTipResult,
     HomeResult,
     LoadLabwareResult,
diff --git a/api/src/opentrons/protocol_engine/errors/exceptions.py b/api/src/opentrons/protocol_engine/errors/exceptions.py
index 3a829ae5c44..d0e5d414bbe 100644
--- a/api/src/opentrons/protocol_engine/errors/exceptions.py
+++ b/api/src/opentrons/protocol_engine/errors/exceptions.py
@@ -207,3 +207,7 @@ class LocationIsOccupiedError(ProtocolEngineError):
 
 class FirmwareUpdateRequired(ProtocolEngineError):
     """An error raised when the firmware needs to be updated."""
+
+
+class PipetteNotReadyToAspirateError(ProtocolEngineError):
+    """An error raised when the pipette is not ready to aspirate."""
diff --git a/api/tests/opentrons/protocol_api/core/engine/test_instrument_core.py b/api/tests/opentrons/protocol_api/core/engine/test_instrument_core.py
index 1051252c323..709db0190dd 100644
--- a/api/tests/opentrons/protocol_api/core/engine/test_instrument_core.py
+++ b/api/tests/opentrons/protocol_api/core/engine/test_instrument_core.py
@@ -286,7 +286,12 @@ def test_aspirate_from_well(
     ).then_return(WellLocation(origin=WellOrigin.TOP, offset=WellOffset(x=3, y=2, z=1)))
 
     subject.aspirate(
-        location=location, well_core=well_core, volume=12.34, rate=5.6, flow_rate=7.8
+        location=location,
+        well_core=well_core,
+        volume=12.34,
+        rate=5.6,
+        flow_rate=7.8,
+        in_place=False,
     )
 
     decoy.verify(
@@ -304,13 +309,74 @@ def test_aspirate_from_well(
     )
 
 
+def test_aspirate_from_location(
+    decoy: Decoy,
+    mock_engine_client: EngineClient,
+    mock_protocol_core: ProtocolCore,
+    subject: InstrumentCore,
+) -> None:
+    """It should aspirate from coordinates."""
+    location = Location(point=Point(1, 2, 3), labware=None)
+    subject.aspirate(
+        volume=12.34,
+        rate=5.6,
+        flow_rate=7.8,
+        well_core=None,
+        location=location,
+        in_place=False,
+    )
+
+    decoy.verify(
+        mock_engine_client.move_to_coordinates(
+            pipette_id="abc123",
+            coordinates=DeckPoint(x=1, y=2, z=3),
+            minimum_z_height=None,
+            force_direct=False,
+            speed=None,
+        ),
+        mock_engine_client.aspirate_in_place(
+            pipette_id="abc123",
+            volume=12.34,
+            flow_rate=7.8,
+        ),
+        mock_protocol_core.set_last_location(location=location, mount=Mount.LEFT),
+    )
+
+
+def test_aspirate_in_place(
+    decoy: Decoy,
+    mock_engine_client: EngineClient,
+    mock_protocol_core: ProtocolCore,
+    subject: InstrumentCore,
+) -> None:
+    """It should aspirate in place."""
+    location = Location(point=Point(1, 2, 3), labware=None)
+    subject.aspirate(
+        volume=12.34,
+        rate=5.6,
+        flow_rate=7.8,
+        well_core=None,
+        location=location,
+        in_place=True,
+    )
+
+    decoy.verify(
+        mock_engine_client.aspirate_in_place(
+            pipette_id="abc123",
+            volume=12.34,
+            flow_rate=7.8,
+        ),
+        mock_protocol_core.set_last_location(location=location, mount=Mount.LEFT),
+    )
+
+
 def test_blow_out_to_well(
     decoy: Decoy,
     mock_engine_client: EngineClient,
     mock_protocol_core: ProtocolCore,
     subject: InstrumentCore,
 ) -> None:
-    """It should aspirate from a well."""
+    """It should blow out from a well."""
     location = Location(point=Point(1, 2, 3), labware=None)
 
     well_core = WellCore(
@@ -323,7 +389,7 @@ def test_blow_out_to_well(
         )
     ).then_return(WellLocation(origin=WellOrigin.TOP, offset=WellOffset(x=3, y=2, z=1)))
 
-    subject.blow_out(location=location, well_core=well_core, move_to_well=True)
+    subject.blow_out(location=location, well_core=well_core, in_place=False)
 
     decoy.verify(
         mock_engine_client.blow_out(
@@ -339,6 +405,55 @@ def test_blow_out_to_well(
     )
 
 
+def test_blow_to_coordinates(
+    decoy: Decoy,
+    mock_engine_client: EngineClient,
+    mock_protocol_core: ProtocolCore,
+    subject: InstrumentCore,
+) -> None:
+    """It should move to coordinate and blow out in place."""
+    location = Location(point=Point(1, 2, 3), labware=None)
+
+    subject.blow_out(location=location, well_core=None, in_place=False)
+
+    decoy.verify(
+        mock_engine_client.move_to_coordinates(
+            pipette_id="abc123",
+            coordinates=DeckPoint(x=1, y=2, z=3),
+            minimum_z_height=None,
+            speed=None,
+            force_direct=False,
+        ),
+        mock_engine_client.blow_out_in_place(
+            pipette_id="abc123",
+            flow_rate=6.7,
+        ),
+        mock_protocol_core.set_last_location(location=location, mount=Mount.LEFT),
+    )
+
+
+def test_blow_out_in_place(
+    decoy: Decoy,
+    mock_engine_client: EngineClient,
+    mock_protocol_core: ProtocolCore,
+    subject: InstrumentCore,
+) -> None:
+    """Should blow-out in place."""
+    location = Location(point=Point(1, 2, 3), labware=None)
+    subject.blow_out(
+        location=location,
+        well_core=None,
+        in_place=True,
+    )
+
+    decoy.verify(
+        mock_engine_client.blow_out_in_place(
+            pipette_id="abc123",
+            flow_rate=6.7,
+        ),
+    )
+
+
 def test_dispense_to_well(
     decoy: Decoy,
     mock_engine_client: EngineClient,
@@ -359,7 +474,12 @@ def test_dispense_to_well(
     ).then_return(WellLocation(origin=WellOrigin.TOP, offset=WellOffset(x=3, y=2, z=1)))
 
     subject.dispense(
-        location=location, well_core=well_core, volume=12.34, rate=5.6, flow_rate=6.0
+        location=location,
+        well_core=well_core,
+        volume=12.34,
+        rate=5.6,
+        flow_rate=6.0,
+        in_place=False,
     )
 
     decoy.verify(
@@ -377,6 +497,65 @@ def test_dispense_to_well(
     )
 
 
+def test_dispense_in_place(
+    decoy: Decoy,
+    mock_engine_client: EngineClient,
+    mock_protocol_core: ProtocolCore,
+    subject: InstrumentCore,
+) -> None:
+    """It should dispense in place."""
+    location = Location(point=Point(1, 2, 3), labware=None)
+    subject.dispense(
+        volume=12.34,
+        rate=5.6,
+        flow_rate=7.8,
+        well_core=None,
+        location=location,
+        in_place=True,
+    )
+
+    decoy.verify(
+        mock_engine_client.dispense_in_place(
+            pipette_id="abc123",
+            volume=12.34,
+            flow_rate=7.8,
+        ),
+    )
+
+
+def test_dispense_to_coordinates(
+    decoy: Decoy,
+    mock_engine_client: EngineClient,
+    mock_protocol_core: ProtocolCore,
+    subject: InstrumentCore,
+) -> None:
+    """It should dispense in place."""
+    location = Location(point=Point(1, 2, 3), labware=None)
+    subject.dispense(
+        volume=12.34,
+        rate=5.6,
+        flow_rate=7.8,
+        well_core=None,
+        location=location,
+        in_place=False,
+    )
+
+    decoy.verify(
+        mock_engine_client.move_to_coordinates(
+            pipette_id="abc123",
+            coordinates=DeckPoint(x=1, y=2, z=3),
+            minimum_z_height=None,
+            force_direct=False,
+            speed=None,
+        ),
+        mock_engine_client.dispense_in_place(
+            pipette_id="abc123",
+            volume=12.34,
+            flow_rate=7.8,
+        ),
+    )
+
+
 def test_initialization_sets_default_movement_speed(
     decoy: Decoy,
     subject: InstrumentCore,
diff --git a/api/tests/opentrons/protocol_api/test_instrument_context.py b/api/tests/opentrons/protocol_api/test_instrument_context.py
index 29a8a235619..e327f545de9 100644
--- a/api/tests/opentrons/protocol_api/test_instrument_context.py
+++ b/api/tests/opentrons/protocol_api/test_instrument_context.py
@@ -14,11 +14,19 @@
     Labware,
     Well,
     labware,
+    validation as mock_validation,
 )
+from opentrons.protocol_api.validation import WellTarget, PointTarget
 from opentrons.protocol_api.core.common import InstrumentCore, ProtocolCore
 from opentrons.types import Location, Mount, Point
 
 
+@pytest.fixture(autouse=True)
+def _mock_validation_module(decoy: Decoy, monkeypatch: pytest.MonkeyPatch) -> None:
+    for name, func in inspect.getmembers(mock_validation, inspect.isfunction):
+        monkeypatch.setattr(mock_validation, name, decoy.mock(func=func))
+
+
 @pytest.fixture(autouse=True)
 def _mock_instrument_support_module(
     decoy: Decoy, monkeypatch: pytest.MonkeyPatch
@@ -201,21 +209,109 @@ def test_pick_up_from_well_deprecated_args(
 
 
 def test_aspirate(
-    decoy: Decoy, mock_instrument_core: InstrumentCore, subject: InstrumentContext
+    decoy: Decoy,
+    mock_instrument_core: InstrumentCore,
+    subject: InstrumentContext,
+    mock_protocol_core: ProtocolCore,
 ) -> None:
     """It should aspirate to a well."""
     mock_well = decoy.mock(cls=Well)
     bottom_location = Location(point=Point(1, 2, 3), labware=mock_well)
+    input_location = Location(point=Point(2, 2, 2), labware=None)
+    last_location = Location(point=Point(9, 9, 9), labware=None)
+    decoy.when(mock_instrument_core.get_mount()).then_return(Mount.RIGHT)
 
+    decoy.when(mock_protocol_core.get_last_location(Mount.RIGHT)).then_return(
+        last_location
+    )
+    decoy.when(
+        mock_validation.validate_location(
+            location=input_location, last_location=last_location
+        )
+    ).then_return(WellTarget(well=mock_well, location=None, in_place=False))
     decoy.when(mock_well.bottom(z=1.0)).then_return(bottom_location)
     decoy.when(mock_instrument_core.get_aspirate_flow_rate(1.23)).then_return(5.67)
 
-    subject.aspirate(volume=42.0, location=mock_well, rate=1.23)
+    subject.aspirate(volume=42.0, location=input_location, rate=1.23)
 
     decoy.verify(
         mock_instrument_core.aspirate(
             location=bottom_location,
             well_core=mock_well._core,
+            in_place=False,
+            volume=42.0,
+            rate=1.23,
+            flow_rate=5.67,
+        ),
+        times=1,
+    )
+
+
+def test_aspirate_well_location(
+    decoy: Decoy,
+    mock_instrument_core: InstrumentCore,
+    subject: InstrumentContext,
+    mock_protocol_core: ProtocolCore,
+) -> None:
+    """It should aspirate to a well."""
+    mock_well = decoy.mock(cls=Well)
+    input_location = Location(point=Point(2, 2, 2), labware=mock_well)
+    last_location = Location(point=Point(9, 9, 9), labware=None)
+    decoy.when(mock_instrument_core.get_mount()).then_return(Mount.RIGHT)
+
+    decoy.when(mock_protocol_core.get_last_location(Mount.RIGHT)).then_return(
+        last_location
+    )
+    decoy.when(
+        mock_validation.validate_location(
+            location=input_location, last_location=last_location
+        )
+    ).then_return(WellTarget(well=mock_well, location=input_location, in_place=False))
+    decoy.when(mock_instrument_core.get_aspirate_flow_rate(1.23)).then_return(5.67)
+
+    subject.aspirate(volume=42.0, location=input_location, rate=1.23)
+
+    decoy.verify(
+        mock_instrument_core.aspirate(
+            location=input_location,
+            well_core=mock_well._core,
+            in_place=False,
+            volume=42.0,
+            rate=1.23,
+            flow_rate=5.67,
+        ),
+        times=1,
+    )
+
+
+def test_aspirate_from_coordinates(
+    decoy: Decoy,
+    mock_instrument_core: InstrumentCore,
+    subject: InstrumentContext,
+    mock_protocol_core: ProtocolCore,
+) -> None:
+    """It should aspirate from given coordinates."""
+    input_location = Location(point=Point(2, 2, 2), labware=None)
+    last_location = Location(point=Point(9, 9, 9), labware=None)
+    decoy.when(mock_instrument_core.get_mount()).then_return(Mount.RIGHT)
+
+    decoy.when(mock_protocol_core.get_last_location(Mount.RIGHT)).then_return(
+        last_location
+    )
+    decoy.when(
+        mock_validation.validate_location(
+            location=input_location, last_location=last_location
+        )
+    ).then_return(PointTarget(location=input_location, in_place=True))
+    decoy.when(mock_instrument_core.get_aspirate_flow_rate(1.23)).then_return(5.67)
+
+    subject.aspirate(volume=42.0, location=input_location, rate=1.23)
+
+    decoy.verify(
+        mock_instrument_core.aspirate(
+            location=input_location,
+            well_core=None,
+            in_place=True,
             volume=42.0,
             rate=1.23,
             flow_rate=5.67,
@@ -224,84 +320,132 @@ def test_aspirate(
     )
 
 
+def test_aspirate_raises_no_location(
+    decoy: Decoy,
+    mock_instrument_core: InstrumentCore,
+    subject: InstrumentContext,
+    mock_protocol_core: ProtocolCore,
+) -> None:
+    """Shound raise a RuntimeError error."""
+    decoy.when(mock_instrument_core.get_mount()).then_return(Mount.RIGHT)
+    decoy.when(mock_protocol_core.get_last_location(Mount.RIGHT)).then_return(None)
+
+    decoy.when(
+        mock_validation.validate_location(location=None, last_location=None)
+    ).then_raise(mock_validation.NoLocationError())
+    with pytest.raises(RuntimeError):
+        subject.aspirate(location=None)
+
+
 def test_blow_out_to_well(
-    decoy: Decoy, mock_instrument_core: InstrumentCore, subject: InstrumentContext
+    decoy: Decoy,
+    mock_instrument_core: InstrumentCore,
+    subject: InstrumentContext,
+    mock_protocol_core: ProtocolCore,
 ) -> None:
     """It should blow out to a well."""
     mock_well = decoy.mock(cls=Well)
     top_location = Location(point=Point(1, 2, 3), labware=mock_well)
+    input_location = Location(point=Point(2, 2, 2), labware=None)
+    last_location = Location(point=Point(9, 9, 9), labware=None)
+    decoy.when(mock_instrument_core.get_mount()).then_return(Mount.RIGHT)
 
+    decoy.when(mock_protocol_core.get_last_location(Mount.RIGHT)).then_return(
+        last_location
+    )
+    decoy.when(
+        mock_validation.validate_location(
+            location=input_location, last_location=last_location
+        )
+    ).then_return(WellTarget(well=mock_well, location=None, in_place=False))
     decoy.when(mock_well.top()).then_return(top_location)
-
-    subject.blow_out(location=mock_well)
+    subject.blow_out(location=input_location)
 
     decoy.verify(
         mock_instrument_core.blow_out(
-            location=top_location,
-            well_core=mock_well._core,
-            move_to_well=True,
+            location=top_location, well_core=mock_well._core, in_place=False
         ),
         times=1,
     )
 
 
-def test_blow_out_to_location(
-    decoy: Decoy, mock_instrument_core: InstrumentCore, subject: InstrumentContext
+def test_blow_out_to_well_location(
+    decoy: Decoy,
+    mock_instrument_core: InstrumentCore,
+    subject: InstrumentContext,
+    mock_protocol_core: ProtocolCore,
 ) -> None:
-    """It should blow out to a location."""
-    mock_location = decoy.mock(cls=Location)
+    """It should blow out to a well location."""
     mock_well = decoy.mock(cls=Well)
+    input_location = Location(point=Point(2, 2, 2), labware=None)
+    last_location = Location(point=Point(9, 9, 9), labware=None)
+    decoy.when(mock_instrument_core.get_mount()).then_return(Mount.RIGHT)
 
-    decoy.when(mock_location.labware.get_parent_labware_and_well()).then_return(
-        (None, mock_well)
+    decoy.when(mock_protocol_core.get_last_location(Mount.RIGHT)).then_return(
+        last_location
     )
-
-    subject.blow_out(location=mock_location)
+    decoy.when(
+        mock_validation.validate_location(
+            location=input_location, last_location=last_location
+        )
+    ).then_return(WellTarget(well=mock_well, location=input_location, in_place=False))
+    subject.blow_out(location=input_location)
 
     decoy.verify(
         mock_instrument_core.blow_out(
-            location=mock_location,
-            well_core=mock_well._core,
-            move_to_well=True,
+            location=input_location, well_core=mock_well._core, in_place=False
         ),
         times=1,
     )
 
 
-def test_blow_out_in_place(
+def test_blow_out_to_location(
     decoy: Decoy,
     mock_instrument_core: InstrumentCore,
-    mock_protocol_core: ProtocolCore,
     subject: InstrumentContext,
+    mock_protocol_core: ProtocolCore,
 ) -> None:
-    """It should blow out in place."""
+    """It should blow out to a location."""
     mock_well = decoy.mock(cls=Well)
-    location = Location(point=Point(1, 2, 3), labware=mock_well)
+    input_location = Location(point=Point(2, 2, 2), labware=mock_well)
+    last_location = Location(point=Point(9, 9, 9), labware=None)
+    point_target = PointTarget(location=input_location, in_place=True)
+    decoy.when(mock_instrument_core.get_mount()).then_return(Mount.RIGHT)
 
-    decoy.when(mock_protocol_core.get_last_location()).then_return(location)
+    decoy.when(mock_protocol_core.get_last_location(Mount.RIGHT)).then_return(
+        last_location
+    )
+    decoy.when(
+        mock_validation.validate_location(
+            location=input_location, last_location=last_location
+        )
+    ).then_return(point_target)
 
-    subject.blow_out()
+    subject.blow_out(location=input_location)
 
     decoy.verify(
         mock_instrument_core.blow_out(
-            location=location,
-            well_core=mock_well._core,
-            move_to_well=False,
+            location=input_location, well_core=None, in_place=True
         ),
         times=1,
     )
 
 
-def test_blow_out_no_location_cache_raises(
+def test_blow_out_raises_no_location(
     decoy: Decoy,
-    mock_protocol_core: ProtocolCore,
+    mock_instrument_core: InstrumentCore,
     subject: InstrumentContext,
+    mock_protocol_core: ProtocolCore,
 ) -> None:
-    """It should raise if no location or well is provided and the location cache returns None."""
-    decoy.when(mock_protocol_core.get_last_location()).then_return(None)
+    """Should raise a RuntimeError."""
+    decoy.when(mock_instrument_core.get_mount()).then_return(Mount.RIGHT)
+    decoy.when(mock_protocol_core.get_last_location(Mount.RIGHT)).then_return(None)
 
+    decoy.when(
+        mock_validation.validate_location(location=None, last_location=None)
+    ).then_raise(mock_validation.NoLocationError())
     with pytest.raises(RuntimeError):
-        subject.blow_out()
+        subject.blow_out(location=None)
 
 
 def test_pick_up_tip_from_labware(
@@ -492,81 +636,134 @@ def test_return_tip(
 
 
 def test_dispense_with_location(
-    decoy: Decoy, mock_instrument_core: InstrumentCore, subject: InstrumentContext
+    decoy: Decoy,
+    mock_instrument_core: InstrumentCore,
+    subject: InstrumentContext,
+    mock_protocol_core: ProtocolCore,
 ) -> None:
     """It should dispense to a given location."""
-    mock_well = decoy.mock(cls=Well)
-    location = Location(point=Point(1, 2, 3), labware=mock_well)
+    input_location = Location(point=Point(2, 2, 2), labware=None)
+    last_location = Location(point=Point(9, 9, 9), labware=None)
+    decoy.when(mock_instrument_core.get_mount()).then_return(Mount.RIGHT)
 
-    decoy.when(mock_instrument_core.get_dispense_flow_rate(1.0)).then_return(3.0)
+    decoy.when(mock_protocol_core.get_last_location(Mount.RIGHT)).then_return(
+        last_location
+    )
+    decoy.when(
+        mock_validation.validate_location(
+            location=input_location, last_location=last_location
+        )
+    ).then_return(PointTarget(location=input_location, in_place=True))
+    decoy.when(mock_instrument_core.get_dispense_flow_rate(1.23)).then_return(5.67)
 
-    subject.dispense(volume=42.0, location=location)
+    subject.dispense(volume=42.0, location=input_location, rate=1.23)
 
     decoy.verify(
         mock_instrument_core.dispense(
-            location=location,
-            well_core=mock_well._core,
+            location=input_location,
+            well_core=None,
+            in_place=True,
             volume=42.0,
-            rate=1.0,
-            flow_rate=3.0,
+            rate=1.23,
+            flow_rate=5.67,
         ),
         times=1,
     )
 
 
 def test_dispense_with_well_location(
-    decoy: Decoy, mock_instrument_core: InstrumentCore, subject: InstrumentContext
+    decoy: Decoy,
+    mock_instrument_core: InstrumentCore,
+    subject: InstrumentContext,
+    mock_protocol_core: ProtocolCore,
 ) -> None:
-    """It should dispense to a well."""
+    """It should dispense to a well location."""
     mock_well = decoy.mock(cls=Well)
+    input_location = Location(point=Point(2, 2, 2), labware=None)
+    last_location = Location(point=Point(9, 9, 9), labware=None)
+    decoy.when(mock_instrument_core.get_mount()).then_return(Mount.RIGHT)
 
-    decoy.when(mock_well.bottom(1.0)).then_return(
-        Location(point=Point(1, 2, 3), labware=mock_well)
+    decoy.when(mock_protocol_core.get_last_location(Mount.RIGHT)).then_return(
+        last_location
     )
+    decoy.when(
+        mock_validation.validate_location(
+            location=input_location, last_location=last_location
+        )
+    ).then_return(WellTarget(well=mock_well, location=input_location, in_place=False))
+    decoy.when(mock_instrument_core.get_dispense_flow_rate(1.23)).then_return(3.0)
 
-    decoy.when(mock_instrument_core.get_dispense_flow_rate(1.0)).then_return(3.0)
-
-    subject.dispense(volume=42.0, location=mock_well)
+    subject.dispense(volume=42.0, location=input_location, rate=1.23)
 
     decoy.verify(
         mock_instrument_core.dispense(
-            location=Location(point=Point(1, 2, 3), labware=mock_well),
+            location=input_location,
             well_core=mock_well._core,
+            in_place=False,
             volume=42.0,
-            rate=1.0,
+            rate=1.23,
             flow_rate=3.0,
         ),
         times=1,
     )
 
 
-def test_dispense_with_no_location(
+def test_dispense_with_well(
     decoy: Decoy,
     mock_instrument_core: InstrumentCore,
     subject: InstrumentContext,
     mock_protocol_core: ProtocolCore,
 ) -> None:
     """It should dispense to a well."""
-    decoy.when(mock_protocol_core.get_last_location()).then_return(
-        Location(point=Point(1, 2, 3), labware=None)
-    )
+    mock_well = decoy.mock(cls=Well)
+    bottom_location = Location(point=Point(1, 2, 3), labware=mock_well)
+    input_location = Location(point=Point(2, 2, 2), labware=None)
+    last_location = Location(point=Point(9, 9, 9), labware=None)
+    decoy.when(mock_instrument_core.get_mount()).then_return(Mount.RIGHT)
 
-    decoy.when(mock_instrument_core.get_dispense_flow_rate(1.0)).then_return(3.0)
+    decoy.when(mock_protocol_core.get_last_location(Mount.RIGHT)).then_return(
+        last_location
+    )
+    decoy.when(
+        mock_validation.validate_location(
+            location=input_location, last_location=last_location
+        )
+    ).then_return(WellTarget(well=mock_well, location=None, in_place=False))
+    decoy.when(mock_well.bottom(z=1.0)).then_return(bottom_location)
+    decoy.when(mock_instrument_core.get_dispense_flow_rate(1.23)).then_return(5.67)
 
-    subject.dispense(volume=42.0)
+    subject.dispense(volume=42.0, location=input_location, rate=1.23)
 
     decoy.verify(
         mock_instrument_core.dispense(
-            location=Location(point=Point(1, 2, 3), labware=None),
-            well_core=None,
+            location=bottom_location,
+            well_core=mock_well._core,
+            in_place=False,
             volume=42.0,
-            rate=1.0,
-            flow_rate=3.0,
+            rate=1.23,
+            flow_rate=5.67,
         ),
         times=1,
     )
 
 
+def test_dispense_raises_no_location(
+    decoy: Decoy,
+    mock_instrument_core: InstrumentCore,
+    subject: InstrumentContext,
+    mock_protocol_core: ProtocolCore,
+) -> None:
+    """Should raise a RuntimeError."""
+    decoy.when(mock_instrument_core.get_mount()).then_return(Mount.RIGHT)
+    decoy.when(mock_protocol_core.get_last_location(Mount.RIGHT)).then_return(None)
+
+    decoy.when(
+        mock_validation.validate_location(location=None, last_location=None)
+    ).then_raise(mock_validation.NoLocationError())
+    with pytest.raises(RuntimeError):
+        subject.dispense(location=None)
+
+
 def test_touch_tip(
     decoy: Decoy,
     mock_instrument_core: InstrumentCore,
diff --git a/api/tests/opentrons/protocol_api/test_validation.py b/api/tests/opentrons/protocol_api/test_validation.py
index b53315aea80..5fbc7f89741 100644
--- a/api/tests/opentrons/protocol_api/test_validation.py
+++ b/api/tests/opentrons/protocol_api/test_validation.py
@@ -1,10 +1,11 @@
 """Tests for Protocol API input validation."""
 from typing import List, Union, Optional, Dict
 
+from decoy import Decoy
 import pytest
 
 from opentrons_shared_data.pipette.dev_types import PipetteNameType
-from opentrons.types import Mount, DeckSlotName
+from opentrons.types import Mount, DeckSlotName, Location, Point
 from opentrons.hardware_control.modules.types import (
     ModuleModel,
     MagneticModuleModel,
@@ -13,7 +14,7 @@
     HeaterShakerModuleModel,
     ThermocyclerStep,
 )
-from opentrons.protocol_api import validation as subject
+from opentrons.protocol_api import validation as subject, Well, Labware
 
 
 @pytest.mark.parametrize(
@@ -250,3 +251,120 @@ def test_ensure_valid_labware_offset_vector(offset: Dict[str, float]) -> None:
     )
     with pytest.raises(TypeError):
         subject.ensure_valid_labware_offset_vector(offset)
+
+
+def test_validate_well_no_location(decoy: Decoy) -> None:
+    """Should return a WellTarget with no location."""
+    input_location = decoy.mock(cls=Well)
+    expected_result = subject.WellTarget(
+        well=input_location, location=None, in_place=False
+    )
+
+    result = subject.validate_location(location=input_location, last_location=None)
+
+    assert result == expected_result
+
+
+def test_validate_coordinates(decoy: Decoy) -> None:
+    """Should return a WellTarget with no location."""
+    input_location = Location(point=Point(x=1, y=1, z=2), labware=None)
+    expected_result = subject.PointTarget(location=input_location, in_place=False)
+
+    result = subject.validate_location(location=input_location, last_location=None)
+
+    assert result == expected_result
+
+
+def test_validate_in_place(decoy: Decoy) -> None:
+    """Should return an `in_place` PointTarget."""
+    input_last_location = Location(point=Point(x=1, y=1, z=2), labware=None)
+    expected_result = subject.PointTarget(location=input_last_location, in_place=True)
+
+    result = subject.validate_location(location=None, last_location=input_last_location)
+
+    assert result == expected_result
+
+
+def test_validate_location_with_well(decoy: Decoy) -> None:
+    """Should return a WellTarget with location."""
+    mock_well = decoy.mock(cls=Well)
+    input_location = Location(point=Point(x=1, y=1, z=1), labware=mock_well)
+    expected_result = subject.WellTarget(
+        well=mock_well, location=input_location, in_place=False
+    )
+
+    result = subject.validate_location(location=input_location, last_location=None)
+
+    assert result == expected_result
+
+
+def test_validate_last_location(decoy: Decoy) -> None:
+    """Should return a WellTarget with location."""
+    mock_well = decoy.mock(cls=Well)
+    input_last_location = Location(point=Point(x=1, y=1, z=1), labware=mock_well)
+    expected_result = subject.WellTarget(
+        well=mock_well, location=input_last_location, in_place=True
+    )
+
+    result = subject.validate_location(location=None, last_location=input_last_location)
+
+    assert result == expected_result
+
+
+def test_validate_location_matches_last_location(decoy: Decoy) -> None:
+    """Should return an in_place WellTarget."""
+    mock_well = decoy.mock(cls=Well)
+    input_last_location = Location(point=Point(x=1, y=1, z=1), labware=mock_well)
+    input_location = Location(point=Point(x=1, y=1, z=1), labware=mock_well)
+    expected_result = subject.WellTarget(
+        well=mock_well, location=input_last_location, in_place=True
+    )
+
+    result = subject.validate_location(
+        location=input_location, last_location=input_last_location
+    )
+
+    assert result == expected_result
+
+
+def test_validate_with_wrong_location_with_last_location() -> None:
+    """Should raise a LocationTypeError."""
+    with pytest.raises(subject.LocationTypeError):
+        subject.validate_location(
+            location=42,  # type: ignore[arg-type]
+            last_location=Location(point=Point(x=1, y=1, z=1), labware=None),
+        )
+
+
+def test_validate_with_wrong_location() -> None:
+    """Should raise a LocationTypeError."""
+    with pytest.raises(subject.LocationTypeError):
+        subject.validate_location(
+            location=42, last_location=None  # type: ignore[arg-type]
+        )
+
+
+def test_validate_raises_no_location_error() -> None:
+    """Should raise a NoLocationError."""
+    with pytest.raises(subject.NoLocationError):
+        subject.validate_location(location=None, last_location=None)
+
+
+def test_validate_with_labware(decoy: Decoy) -> None:
+    """Should return a PointTarget for a non-Well Location."""
+    mock_labware = decoy.mock(cls=Labware)
+    input_location = Location(point=Point(1, 1, 1), labware=mock_labware)
+
+    result = subject.validate_location(location=input_location, last_location=None)
+
+    assert result == subject.PointTarget(location=input_location, in_place=False)
+
+
+def test_validate_last_location_with_labware(decoy: Decoy) -> None:
+    """Should return a PointTarget for non-Well previous Location."""
+    mock_labware = decoy.mock(cls=Labware)
+    input_last_location = Location(point=Point(1, 1, 1), labware=mock_labware)
+
+    result = subject.validate_location(location=None, last_location=input_last_location)
+
+    assert result == subject.PointTarget(location=input_last_location, in_place=True)
diff --git a/api/tests/opentrons/protocol_api_old/core/simulator/test_instrument_context.py b/api/tests/opentrons/protocol_api_old/core/simulator/test_instrument_context.py
index 0cd28b887c6..71e7f7c3211 100644
--- a/api/tests/opentrons/protocol_api_old/core/simulator/test_instrument_context.py
+++ b/api/tests/opentrons/protocol_api_old/core/simulator/test_instrument_context.py
@@ -50,7 +50,12 @@ def test_dispense_no_tip(subject: InstrumentCore) -> None:
     location = Location(point=Point(1, 2, 3), labware=None)
     with pytest.raises(NoTipAttachedError, match="Cannot perform DISPENSE"):
         subject.dispense(
-            volume=1, rate=1, flow_rate=1, location=location, well_core=None
+            volume=1,
+            rate=1,
+            flow_rate=1,
+            location=location,
+            well_core=None,
+            in_place=False,
         )
 
 
@@ -69,7 +74,7 @@ def test_blow_out_no_tip(subject: InstrumentCore, labware: LabwareCore) -> None:
         subject.blow_out(
             location=Location(point=Point(1, 2, 3), labware=None),
             well_core=labware.get_well_core("A1"),
-            move_to_well=False,
+            in_place=True,
         )
 
 
@@ -115,6 +120,7 @@ def test_pick_up_tip_prep_after(
         volume=1,
         rate=1,
         flow_rate=1,
+        in_place=False,
     )
     subject.dispense(
         volume=1,
@@ -122,6 +128,7 @@ def test_pick_up_tip_prep_after(
         flow_rate=1,
         location=Location(point=Point(2, 2, 3), labware=None),
         well_core=labware.get_well_core("A2"),
+        in_place=False,
     )
 
     subject.drop_tip(location=None, well_core=tip_core, home_after=True)
@@ -140,6 +147,7 @@ def test_pick_up_tip_prep_after(
         volume=1,
         rate=1,
         flow_rate=1,
+        in_place=False,
     )
     subject.dispense(
         volume=1,
@@ -147,6 +155,7 @@ def test_pick_up_tip_prep_after(
         flow_rate=1,
         location=Location(point=Point(2, 2, 3), labware=None),
         well_core=labware.get_well_core("A2"),
+        in_place=False,
     )
 
     subject.drop_tip(location=None, well_core=tip_core, home_after=True)
@@ -176,6 +185,7 @@ def test_aspirate_too_much(
             volume=subject.get_max_volume() + 1,
             rate=1,
             flow_rate=1,
+            in_place=False,
         )
 
 
@@ -226,6 +236,7 @@ def _aspirate(i: InstrumentCore, labware: LabwareCore) -> None:
         volume=12,
         rate=10,
         flow_rate=10,
+        in_place=False,
     )
 
 
@@ -238,6 +249,7 @@ def _aspirate_dispense(i: InstrumentCore, labware: LabwareCore) -> None:
         volume=12,
         rate=10,
         flow_rate=10,
+        in_place=False,
     )
     i.dispense(
         volume=2,
@@ -245,6 +257,7 @@ def _aspirate_dispense(i: InstrumentCore, labware: LabwareCore) -> None:
         flow_rate=2,
         location=Location(point=Point(2, 2, 3), labware=None),
         well_core=labware.get_well_core("A2"),
+        in_place=False,
     )
 
 
@@ -257,11 +270,12 @@ def _aspirate_blowout(i: InstrumentCore, labware: LabwareCore) -> None:
         volume=11,
         rate=13,
         flow_rate=13,
+        in_place=False,
     )
     i.blow_out(
         location=Location(point=Point(1, 2, 3), labware=None),
         well_core=labware.get_well_core("A1"),
-        move_to_well=False,
+        in_place=True,
     )
 
 
diff --git a/api/tests/opentrons/protocol_engine/clients/test_sync_client.py b/api/tests/opentrons/protocol_engine/clients/test_sync_client.py
index 0b43c45d20e..75fc7b453f2 100644
--- a/api/tests/opentrons/protocol_engine/clients/test_sync_client.py
+++ b/api/tests/opentrons/protocol_engine/clients/test_sync_client.py
@@ -358,6 +358,35 @@ def test_aspirate(
     assert result == result_from_transport
 
 
+def test_aspirate_in_place(
+    decoy: Decoy,
+    transport: AbstractSyncTransport,
+    subject: SyncClient,
+) -> None:
+    """It should send an AspirateInPlaceCommand through the transport."""
+    request = commands.AspirateInPlaceCreate(
+        params=commands.AspirateInPlaceParams(
+            pipetteId="123",
+            volume=123.45,
+            flowRate=6.7,
+        )
+    )
+
+    result_from_transport = commands.AspirateInPlaceResult(volume=67.89)
+
+    decoy.when(transport.execute_command(request=request)).then_return(
+        result_from_transport
+    )
+
+    result = subject.aspirate_in_place(
+        pipette_id="123",
+        volume=123.45,
+        flow_rate=6.7,
+    )
+
+    assert result == result_from_transport
+
+
 def test_dispense(
     decoy: Decoy,
     transport: AbstractSyncTransport,
@@ -396,6 +425,33 @@ def test_dispense(
     assert result == response
 
 
+def test_dispense_in_place(
+    decoy: Decoy,
+    transport: AbstractSyncTransport,
+    subject: SyncClient,
+) -> None:
+    """It should execute a DispenceInPlace command."""
+    request = commands.DispenseInPlaceCreate(
+        params=commands.DispenseInPlaceParams(
+            pipetteId="123",
+            volume=10,
+            flowRate=2.0,
+        )
+    )
+
+    response = commands.DispenseInPlaceResult(volume=1)
+
+    decoy.when(transport.execute_command(request=request)).then_return(response)
+
+    result = subject.dispense_in_place(
+        pipette_id="123",
+        volume=10,
+        flow_rate=2.0,
+    )
+
+    assert result == response
+
+
 def test_touch_tip(
     decoy: Decoy,
     transport: AbstractSyncTransport,
@@ -745,6 +801,31 @@ def test_blow_out(
     assert result == response
 
 
+def test_blow_out_in_place(
+    decoy: Decoy,
+    transport: AbstractSyncTransport,
+    subject: SyncClient,
+) -> None:
+    """It should execute a blow_out command."""
+    request = commands.BlowOutInPlaceCreate(
+        params=commands.BlowOutInPlaceParams(
+            pipetteId="123",
+            flowRate=7.8,
+        )
+    )
+
+    response = commands.BlowOutInPlaceResult()
+
+    decoy.when(transport.execute_command(request=request)).then_return(response)
+
+    result = subject.blow_out_in_place(
+        pipette_id="123",
+        flow_rate=7.8,
+    )
+
+    assert result == response
+
+
 def test_heater_shaker_set_target_temperature(
     decoy: Decoy,
     transport: AbstractSyncTransport,
diff --git a/api/tests/opentrons/protocol_engine/commands/test_aspirate_in_place.py b/api/tests/opentrons/protocol_engine/commands/test_aspirate_in_place.py
new file mode 100644
index 00000000000..7aebea013d8
--- /dev/null
+++ b/api/tests/opentrons/protocol_engine/commands/test_aspirate_in_place.py
@@ -0,0 +1,151 @@
+"""Test aspirate-in-place commands."""
+import pytest
+from decoy import Decoy
+from typing import cast
+
+from opentrons.types import Mount
+from opentrons.hardware_control import API as HardwareAPI
+from opentrons.hardware_control.dev_types import PipetteDict
+
+from opentrons.protocol_engine.execution import PipettingHandler
+from opentrons.protocol_engine.commands.aspirate_in_place import (
+    AspirateInPlaceParams,
+    AspirateInPlaceResult,
+    AspirateInPlaceImplementation,
+)
+from opentrons.protocol_engine.errors.exceptions import PipetteNotReadyToAspirateError
+
+from opentrons.protocol_engine.state import (
+    StateStore,
+    HardwarePipette,
+)
+
+
+@pytest.fixture
+def hardware_api(decoy: Decoy) -> HardwareAPI:
+    """Get a mock in the shape of a HardwareAPI."""
+    return decoy.mock(cls=HardwareAPI)
+
+
+@pytest.fixture
+def state_store(decoy: Decoy) -> StateStore:
+    """Get a mock in the shape of a StateStore."""
+    return decoy.mock(cls=StateStore)
+
+
+@pytest.fixture
+def pipetting(decoy: Decoy) -> PipettingHandler:
+    """Get a mock in the shape of a PipettingHandler."""
+    return decoy.mock(cls=PipettingHandler)
+
+
+async def test_aspirate_in_place_implementation(
+    decoy: Decoy,
+    pipetting: PipettingHandler,
+    state_store: StateStore,
+    hardware_api: HardwareAPI,
+) -> None:
+    """It should aspirate in place."""
+    subject = AspirateInPlaceImplementation(
+        pipetting=pipetting,
+        hardware_api=hardware_api,
+        state_view=state_store,
+    )
+
+    data = AspirateInPlaceParams(
+        pipetteId="pipette-id-abc",
+        volume=123,
+        flowRate=1.234,
+    )
+
+    left_config = cast(PipetteDict, {"name": "p300_single", "pipette_id": "123"})
+    right_config = cast(PipetteDict, {"name": "p300_multi", "pipette_id": "abc"})
+
+    pipette_dict_by_mount = {Mount.LEFT: left_config, Mount.RIGHT: right_config}
+
+    hw_pipette_result = HardwarePipette(
+        mount=Mount.RIGHT,
+        config=right_config,
+    )
+
+    decoy.when(hardware_api.attached_instruments).then_return(pipette_dict_by_mount)
+
+    decoy.when(
+        state_store.pipettes.get_hardware_pipette(
+            pipette_id="pipette-id-abc",
+            attached_pipettes=pipette_dict_by_mount,
+        )
+    ).then_return(hw_pipette_result)
+
+    decoy.when(
+        state_store.pipettes.get_is_ready_to_aspirate(
+            pipette_id="pipette-id-abc",
+            pipette_config=hw_pipette_result.config,
+        )
+    ).then_return(True)
+
+    mock_flow_rate_context = decoy.mock(name="mock flow rate context")
+    decoy.when(
+        pipetting.set_flow_rate(
+            pipette=hw_pipette_result,
+            aspirate_flow_rate=1.234,
+        )
+    ).then_return(mock_flow_rate_context)
+
+    result = await subject.execute(params=data)
+
+    assert result == AspirateInPlaceResult(volume=123)
+
+
+async def test_handle_aspirate_in_place_request_not_ready_to_aspirate(
+    decoy: Decoy,
+    pipetting: PipettingHandler,
+    state_store: StateStore,
+    hardware_api: HardwareAPI,
+) -> None:
+    """Should raise an exception for not ready to aspirate."""
+    subject = AspirateInPlaceImplementation(
+        pipetting=pipetting,
+        hardware_api=hardware_api,
+        state_view=state_store,
+    )
+
+    data = AspirateInPlaceParams(
+        pipetteId="pipette-id-abc",
+        volume=123,
+        flowRate=1.234,
+    )
+
+    left_config = cast(PipetteDict, {"name": "p300_single", "pipette_id": "123"})
+    right_config = cast(PipetteDict, {"name": "p300_multi", "pipette_id": "abc"})
+
+    pipette_dict_by_mount = {Mount.LEFT: left_config, Mount.RIGHT: right_config}
+
+    hw_pipette_result = HardwarePipette(
+        mount=Mount.RIGHT,
+        config=right_config,
+    )
+
+    decoy.when(hardware_api.attached_instruments).then_return(pipette_dict_by_mount)
+
+    decoy.when(
+        state_store.pipettes.get_hardware_pipette(
+            pipette_id="pipette-id-abc",
+            attached_pipettes=pipette_dict_by_mount,
+        )
+    ).then_return(hw_pipette_result)
+
+    decoy.when(
+        state_store.pipettes.get_is_ready_to_aspirate(
+            pipette_id="pipette-id-abc",
+            pipette_config=hw_pipette_result.config,
+        )
+    ).then_return(False)
+
+    with pytest.raises(
+        PipetteNotReadyToAspirateError,
+        match="Pipette cannot aspirate in place because of a previous blow out."
+        " The first aspirate following a blow-out must be from a specific well"
+        " so the plunger can be reset in a known safe position.",
+    ):
+        await subject.execute(params=data)
diff --git a/api/tests/opentrons/protocol_engine/commands/test_blow_out_in_place.py b/api/tests/opentrons/protocol_engine/commands/test_blow_out_in_place.py
new file mode 100644
index 00000000000..685dcf5715b
--- /dev/null
+++ b/api/tests/opentrons/protocol_engine/commands/test_blow_out_in_place.py
@@ -0,0 +1,71 @@
+"""Test blow-out-in-place commands."""
+from decoy import Decoy
+from typing import cast
+
+from opentrons.protocol_engine.state import StateView, HardwarePipette
+from opentrons.protocol_engine.commands.blow_out_in_place import (
+    BlowOutInPlaceParams,
+    BlowOutInPlaceResult,
+    BlowOutInPlaceImplementation,
+)
+
+from opentrons.protocol_engine.execution import (
+    MovementHandler,
+    PipettingHandler,
+)
+from opentrons.types import Mount
+from opentrons.hardware_control.dev_types import PipetteDict
+from opentrons.hardware_control import HardwareControlAPI
+
+
+async def test_blow_out_in_place_implementation(
+    decoy: Decoy,
+    state_view: StateView,
+    hardware_api: HardwareControlAPI,
+    movement: MovementHandler,
+    pipetting: PipettingHandler,
+) -> None:
+    """Test BlowOut command execution."""
+    subject = BlowOutInPlaceImplementation(
+        state_view=state_view,
+        hardware_api=hardware_api,
+        pipetting=pipetting,
+    )
+
+    left_config = cast(PipetteDict, {"name": "p300_single", "pipette_id": "123"})
+    right_config = cast(PipetteDict, {"name": "p300_multi", "pipette_id": "abc"})
+
+    pipette_dict_by_mount = {Mount.LEFT: left_config, Mount.RIGHT: right_config}
+
+    left_pipette = HardwarePipette(mount=Mount.LEFT, config=left_config)
+
+    decoy.when(hardware_api.attached_instruments).then_return(pipette_dict_by_mount)
+    decoy.when(
+        state_view.pipettes.get_hardware_pipette(
+            pipette_id="pipette-id",
+            attached_pipettes=pipette_dict_by_mount,
+        )
+    ).then_return(HardwarePipette(mount=Mount.LEFT, config=left_config))
+
+    data = BlowOutInPlaceParams(
+        pipetteId="pipette-id",
+        flowRate=1.234,
+    )
+
+    mock_flow_rate_context = decoy.mock(name="mock flow rate context")
+    decoy.when(
+        pipetting.set_flow_rate(
+            pipette=HardwarePipette(mount=Mount.LEFT, config=left_config),
+            blow_out_flow_rate=1.234,
+        )
+    ).then_return(mock_flow_rate_context)
+
+    result = await subject.execute(data)
+
+    assert result == BlowOutInPlaceResult()
+
+    decoy.verify(
+        mock_flow_rate_context.__enter__(),
+        await hardware_api.blow_out(mount=left_pipette.mount),
+        mock_flow_rate_context.__exit__(None, None, None),
+    )
diff --git a/shared-data/command/schemas/7.json b/shared-data/command/schemas/7.json
index 4a55a392341..fbb06f67415 100644
--- a/shared-data/command/schemas/7.json
+++ b/shared-data/command/schemas/7.json
@@ -5,6 +5,9 @@
     {
       "$ref": "#/definitions/AspirateCreate"
     },
+    {
+      "$ref": "#/definitions/AspirateInPlaceCreate"
+    },
     {
       "$ref": "#/definitions/CommentCreate"
     },
@@ -20,6 +23,9 @@
     {
       "$ref": "#/definitions/BlowOutCreate"
     },
+    {
+      "$ref": "#/definitions/BlowOutInPlaceCreate"
+    },
     {
       "$ref": "#/definitions/DropTipCreate"
     },
@@ -268,6 +274,61 @@
       },
       "required": ["params"]
     },
+    "AspirateInPlaceParams": {
+      "title": "AspirateInPlaceParams",
+      "description": "Payload required to aspirate in place.",
+      "type": "object",
+      "properties": {
+        "flowRate": {
+          "title": "Flowrate",
+          "description": "Speed in \u00b5L/s configured for the pipette",
+          "exclusiveMinimum": 0,
+          "type": "number"
+        },
+        "volume": {
+          "title": "Volume",
+          "description": "Amount of liquid in uL. Must be greater than 0 and less than a pipette-specific maximum volume.",
+          "exclusiveMinimum": 0,
+          "type": "number"
+        },
+        "pipetteId": {
+          "title": "Pipetteid",
+          "description": "Identifier of pipette to use for liquid handling.",
+          "type": "string"
+        }
+      },
+      "required": ["flowRate", "volume", "pipetteId"]
+    },
+    "AspirateInPlaceCreate": {
+      "title": "AspirateInPlaceCreate",
+      "description": "AspirateInPlace command request model.",
+      "type": "object",
+      "properties": {
+        "commandType": {
+          "title": "Commandtype",
+          "default": "aspirateInPlace",
+          "enum": ["aspirateInPlace"],
+          "type": "string"
+        },
+        "params": {
+          "$ref": "#/definitions/AspirateInPlaceParams"
+        },
+        "intent": {
+          "description": "The reason the command was added. If not specified or `protocol`, the command will be treated as part of the protocol run itself, and added to the end of the existing command queue.\n\nIf `setup`, the command will be treated as part of run setup. A setup command may only be enqueued if the run has not started.\n\nUse setup commands for activities like pre-run calibration checks and module setup, like pre-heating.",
+          "allOf": [
+            {
+              "$ref": "#/definitions/CommandIntent"
+            }
+          ]
+        },
+        "key": {
+          "title": "Key",
+          "description": "A key value, unique in this run, that can be used to track the same logical command across multiple runs of the same protocol. If a value is not provided, one will be generated.",
+          "type": "string"
+        }
+      },
+      "required": ["params"]
+    },
     "CommentParams": {
       "title": "CommentParams",
       "description": "Payload required to annotate execution with a comment.",
@@ -544,6 +605,55 @@
       },
       "required": ["params"]
     },
+    "BlowOutInPlaceParams": {
+      "title": "BlowOutInPlaceParams",
+      "description": "Payload required to blow-out in place.",
+      "type": "object",
+      "properties": {
+        "flowRate": {
+          "title": "Flowrate",
+          "description": "Speed in \u00b5L/s configured for the pipette",
+          "exclusiveMinimum": 0,
+          "type": "number"
+        },
+        "pipetteId": {
+          "title": "Pipetteid",
+          "description": "Identifier of pipette to use for liquid handling.",
+          "type": "string"
+        }
+      },
+      "required": ["flowRate", "pipetteId"]
+    },
+    "BlowOutInPlaceCreate": {
+      "title": "BlowOutInPlaceCreate",
+      "description": "BlowOutInPlace command request model.",
+      "type": "object",
+      "properties": {
+        "commandType": {
+          "title": "Commandtype",
+          "default": "blowOutInPlace",
+          "enum": ["blowOutInPlace"],
+          "type": "string"
+        },
+        "params": {
+          "$ref": "#/definitions/BlowOutInPlaceParams"
+        },
+        "intent": {
+          "description": "The reason the command was added. If not specified or `protocol`, the command will be treated as part of the protocol run itself, and added to the end of the existing command queue.\n\nIf `setup`, the command will be treated as part of run setup. A setup command may only be enqueued if the run has not started.\n\nUse setup commands for activities like pre-run calibration checks and module setup, like pre-heating.",
+          "allOf": [
+            {
+              "$ref": "#/definitions/CommandIntent"
+            }
+          ]
+        },
+        "key": {
+          "title": "Key",
+          "description": "A key value, unique in this run, that can be used to track the same logical command across multiple runs of the same protocol. If a value is not provided, one will be generated.",
+          "type": "string"
+        }
+      },
+      "required": ["params"]
+    },
     "DropTipParams": {
       "title": "DropTipParams",
       "description": "Payload required to drop a tip in a specific well.",