diff --git a/packages/google-cloud-storage/google/cloud/storage/_grpc_conversions.py b/packages/google-cloud-storage/google/cloud/storage/_grpc_conversions.py
index 9b86a0a4db44..2071b3cb07e0 100644
--- a/packages/google-cloud-storage/google/cloud/storage/_grpc_conversions.py
+++ b/packages/google-cloud-storage/google/cloud/storage/_grpc_conversions.py
@@ -87,4 +87,14 @@ def blob_to_proto(blob):
             retain_until_time=retain_until_time_proto,
         )
 
+    contexts = getattr(blob, "contexts", None)
+    if contexts:
+        custom_contexts = {}
+        for key, payload in contexts.custom.items():
+            custom_contexts[key] = _storage_v2.ObjectCustomContextPayload(
+                value=payload.value
+            )
+
+        resource_params["contexts"] = _storage_v2.ObjectContexts(custom=custom_contexts)
+
     return _storage_v2.Object(**resource_params)
diff --git a/packages/google-cloud-storage/google/cloud/storage/blob.py b/packages/google-cloud-storage/google/cloud/storage/blob.py
index c6fbcf4c12b7..8299c737d1d0 100644
--- a/packages/google-cloud-storage/google/cloud/storage/blob.py
+++ b/packages/google-cloud-storage/google/cloud/storage/blob.py
@@ -105,6 +105,7 @@
     "name",
     "retention",
     "storageClass",
+    "contexts",
 )
 _READ_LESS_THAN_SIZE = (
     "Size {:d} was specified but the file-like object only had {:d} bytes remaining."
@@ -3849,6 +3850,7 @@ def compose(
         if_metageneration_match=None,
         if_source_generation_match=None,
         retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
+        destination_contexts=None,
     ):
         """Concatenate source blobs into this one.
 
@@ -3908,6 +3910,12 @@ def compose(
             Change the value to ``DEFAULT_RETRY`` or another `google.api_core.retry.Retry` object
             to enable retries regardless of generation precondition setting.
             See [Configuring Retries](https://cloud.google.com/python/docs/reference/storage/latest/retry_timeout).
+
+        :type destination_contexts: :class:`~google.cloud.storage.blob.ObjectContexts`
+        :param destination_contexts:
+            (Optional) New contexts to set for the destination object.
+            See: https://docs.cloud.google.com/storage/docs/use-object-contexts#manage_object_contexts_during_object_operations
+
         """
         with create_trace_span(name="Storage.Blob.compose"):
             sources_len = len(sources)
@@ -3959,6 +3967,14 @@ def compose(
 
                 source_objects.append(source_object)
 
+            if destination_contexts is not None:
+                if isinstance(destination_contexts, ObjectContexts):
+                    self.contexts = destination_contexts
+                else:
+                    raise ValueError(
+                        "destination_contexts must be an ObjectContexts object"
+                    )
+
             request = {
                 "sourceObjects": source_objects,
                 "destination": self._properties.copy(),
@@ -3998,6 +4014,7 @@ def rewrite(
         if_source_metageneration_not_match=None,
         timeout=_DEFAULT_TIMEOUT,
         retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
+        destination_contexts=None,
     ):
         """Rewrite source blob into this one.
 
@@ -4081,6 +4098,11 @@ def rewrite(
             to enable retries regardless of generation precondition setting.
             See [Configuring Retries](https://cloud.google.com/python/docs/reference/storage/latest/retry_timeout).
 
+        :type destination_contexts: :class:`~google.cloud.storage.blob.ObjectContexts` or dict
+        :param destination_contexts:
+            (Optional) New contexts to set for the destination object.
+            See: https://docs.cloud.google.com/storage/docs/use-object-contexts#manage_object_contexts_during_object_operations
+
         :rtype: tuple
         :returns: ``(token, bytes_rewritten, total_bytes)``, where ``token``
                   is a rewrite token (``None`` if the rewrite is complete),
@@ -4126,6 +4148,14 @@ def rewrite(
                 if_source_metageneration_not_match=if_source_metageneration_not_match,
             )
 
+            if destination_contexts is not None:
+                if isinstance(destination_contexts, ObjectContexts):
+                    self.contexts = destination_contexts
+                else:
+                    raise ValueError(
+                        "destination_contexts must be an ObjectContexts object"
+                    )
+
             path = f"{source.path}/rewriteTo{self.path}"
             api_response = client._post_resource(
                 path,
@@ -5008,6 +5038,29 @@ def retention(self):
         info = self._properties.get("retention", {})
         return Retention.from_api_repr(info, self)
 
+    @property
+    def contexts(self):
+        """Retrieve the contexts for this object.
+
+        :rtype: :class:`ObjectContexts`
+        :returns: an instance for managing the object's contexts.
+        """
+        info = self._properties.get("contexts", {})
+        return ObjectContexts.from_api_repr(info, self)
+
+    @contexts.setter
+    def contexts(self, value):
+        """Update the contexts for this object.
+
+        :type value: :class:`ObjectContexts` or dict or None
+        :param value: the new contexts for the object.
+        """
+        if value is None:
+            self._properties["contexts"] = None
+        else:
+            self._properties["contexts"] = value
+        self._patch_property("contexts", value)
+
     @property
     def soft_delete_time(self):
         """If this object has been soft-deleted, returns the time at which it became soft-deleted.
@@ -5300,3 +5353,145 @@ def retention_expiration_time(self):
         retention_expiration_time = self.get("retentionExpirationTime")
         if retention_expiration_time is not None:
             return _rfc3339_nanos_to_datetime(retention_expiration_time)
+
+
+class ObjectCustomContextPayload(dict):
+    """Payload for a custom context.
+
+    :type value: str or ``NoneType``
+    :param value: (Optional) The value of the custom context.
+
+    :type create_time: :class:`datetime.datetime` or ``NoneType``
+    :param create_time: (Optional) Creation time of the custom context.
+
+    :type update_time: :class:`datetime.datetime` or ``NoneType``
+    :param update_time: (Optional) Last update time of the custom context.
+    """
+
+    def __init__(self, value=None, create_time=None, update_time=None):
+        data = {"value": value}
+        if create_time is not None:
+            data["createTime"] = _datetime_to_rfc3339(create_time)
+        if update_time is not None:
+            data["updateTime"] = _datetime_to_rfc3339(update_time)
+        super(ObjectCustomContextPayload, self).__init__(data)
+        self._contexts = None
+
+    @property
+    def value(self):
+        """The value of the custom context.
+
+        :rtype: str or ``NoneType``
+        :returns: The value of the custom context.
+        """
+        return self.get("value")
+
+    @value.setter
+    def value(self, value):
+        self["value"] = value
+        if hasattr(self, "_contexts") and self._contexts and self._contexts.blob:
+            self._contexts.blob._patch_property("contexts", self._contexts)
+
+    @property
+    def create_time(self):
+        """Creation time of the custom context.
+
+        :rtype: :class:`datetime.datetime` or ``NoneType``
+        :returns: Datetime object parsed from RFC3339 valid timestamp.
+        """
+        create_time = self.get("createTime")
+        if create_time is not None:
+            return _rfc3339_nanos_to_datetime(create_time)
+
+    @property
+    def update_time(self):
+        """Last update time of the custom context.
+
+        :rtype: :class:`datetime.datetime` or ``NoneType``
+        :returns: Datetime object parsed from RFC3339 valid timestamp.
+        """
+        update_time = self.get("updateTime")
+        if update_time is not None:
+            return _rfc3339_nanos_to_datetime(update_time)
+
+
+class ObjectContexts(dict):
+    """Container for an object's contexts.
+
+    :type blob: :class:`Blob`
+    :param blob: blob for which these contexts apply to.
+
+    :type custom: dict or ``NoneType``
+    :param custom: (Optional) Custom contexts mapping.
+    """
+
+    def __init__(self, blob, custom=None):
+        data = {}
+        if custom is not None:
+            if not isinstance(custom, dict):
+                raise ValueError(
+                    "custom must be a dictionary mapping keys to ObjectCustomContextPayload instances"
+                )
+            for payload in custom.values():
+                if not isinstance(payload, ObjectCustomContextPayload):
+                    raise ValueError(
+                        "All values in custom must be ObjectCustomContextPayload instances"
+                    )
+            data["custom"] = custom
+        super(ObjectContexts, self).__init__(data)
+        self._blob = blob
+        if custom is not None:
+            for payload in custom.values():
+                payload._contexts = self
+
+    @classmethod
+    def from_api_repr(cls, resource, blob):
+        """Factory: construct instance from resource.
+
+        :type resource: dict
+        :param resource: mapping as returned from API call.
+
+        :type blob: :class:`Blob`
+        :param blob: Blob for which these contexts apply to.
+
+        :rtype: :class:`ObjectContexts`
+        :returns: ObjectContexts instance created from resource.
+        """
+        instance = cls(blob)
+        custom = {}
+        for key, payload_resource in resource.get("custom", {}).items():
+            payload = ObjectCustomContextPayload()
+            payload.update(payload_resource)
+            payload._contexts = instance
+            custom[key] = payload
+        instance["custom"] = custom
+        return instance
+
+    @property
+    def blob(self):
+        """Blob for which these contexts apply to.
+
+        :rtype: :class:`Blob`
+        :returns: the instance's blob.
+        """
+        return self._blob
+
+    @property
+    def custom(self):
+        """Custom contexts mapping.
+
+        :rtype: dict
+        :returns: Mapping of keys to :class:`ObjectCustomContextPayload` instances.
+        """
+        if "custom" not in self:
+            self["custom"] = {}
+        return self["custom"]
+
+    @custom.setter
+    def custom(self, value):
+        if not isinstance(value, dict):
+            raise ValueError(
+                "custom must be a dictionary mapping keys to ObjectCustomContextPayload instances"
+            )
+        self["custom"] = value
+        self.blob._patch_property("contexts", self)
diff --git a/packages/google-cloud-storage/google/cloud/storage/bucket.py b/packages/google-cloud-storage/google/cloud/storage/bucket.py
index 6fd690cf38b2..7ed8b375cd89 100644
--- a/packages/google-cloud-storage/google/cloud/storage/bucket.py
+++ b/packages/google-cloud-storage/google/cloud/storage/bucket.py
@@ -42,6 +42,7 @@
 from google.cloud.storage._signing import generate_signed_url_v2, generate_signed_url_v4
 from google.cloud.storage.acl import BucketACL, DefaultObjectACL
 from google.cloud.storage.blob import Blob, _quote
+from google.cloud.storage.blob import ObjectContexts
 from google.cloud.storage.constants import (
     _DEFAULT_TIMEOUT,
     ARCHIVE_STORAGE_CLASS,
@@ -1423,6 +1424,7 @@ def list_blobs(
         include_folders_as_prefixes=None,
         soft_deleted=None,
         page_size=None,
+        filter_=None,
     ):
         """Return an iterator used to find blobs in the bucket.
 
@@ -1516,6 +1518,11 @@ def list_blobs(
             Note ``soft_deleted`` and ``versions`` cannot be set to True simultaneously. See:
             https://cloud.google.com/storage/docs/soft-delete
 
+        :type filter_: str
+        :param filter_:
+            (Optional) Filter string used to filter objects. See:
+            https://docs.cloud.google.com/storage/docs/listing-objects#filter-by-object-contexts-syntax
+
         :type page_size: int
         :param page_size:
             (Optional) Maximum number of blobs to return in each page.
@@ -1545,6 +1552,7 @@ def list_blobs(
                 match_glob=match_glob,
                 include_folders_as_prefixes=include_folders_as_prefixes,
                 soft_deleted=soft_deleted,
+                filter_=filter_,
             )
 
     def list_notifications(
@@ -1972,6 +1980,7 @@ def copy_blob(
         if_source_metageneration_not_match=None,
         timeout=_DEFAULT_TIMEOUT,
         retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
+        destination_contexts=None,
     ):
         """Copy the given blob to the given bucket, optionally with a new name.
 
@@ -2065,6 +2074,10 @@ def copy_blob(
             to enable retries regardless of generation precondition setting.
             See [Configuring Retries](https://cloud.google.com/python/docs/reference/storage/latest/retry_timeout).
 
+        :type destination_contexts: :class:`~google.cloud.storage.blob.ObjectContexts` or dict
+        :param destination_contexts:
+            (Optional) New contexts to set for the destination object.
+            See: https://docs.cloud.google.com/storage/docs/use-object-contexts#manage_object_contexts_during_object_operations
         :rtype: :class:`google.cloud.storage.blob.Blob`
         :returns: The new Blob.
         """
@@ -2094,10 +2107,22 @@ def copy_blob(
                 new_name = blob.name
 
             new_blob = Blob(bucket=destination_bucket, name=new_name)
+
+            if destination_contexts is not None:
+                if isinstance(destination_contexts, ObjectContexts):
+                    new_blob.contexts = destination_contexts
+                else:
+                    raise ValueError(
+                        "destination_contexts must be an ObjectContexts object"
+                    )
+                request_body = new_blob._properties.copy()
+            else:
+                request_body = None
+
             api_path = blob.path + "/copyTo" + new_blob.path
             copy_result = client._post_resource(
                 api_path,
-                None,
+                request_body,
                 query_params=query_params,
                 timeout=timeout,
                 retry=retry,
diff --git a/packages/google-cloud-storage/google/cloud/storage/client.py b/packages/google-cloud-storage/google/cloud/storage/client.py
index 528b2255f451..ca0e4d9b7721 100644
--- a/packages/google-cloud-storage/google/cloud/storage/client.py
+++ b/packages/google-cloud-storage/google/cloud/storage/client.py
@@ -1291,6 +1291,7 @@ def list_blobs(
         match_glob=None,
         include_folders_as_prefixes=None,
         soft_deleted=None,
+        filter_=None,
     ):
         """Return an iterator used to find blobs in the bucket.
 
@@ -1400,6 +1401,10 @@ def list_blobs(
                 Note ``soft_deleted`` and ``versions`` cannot be set to True simultaneously. See:
                 https://cloud.google.com/storage/docs/soft-delete
 
+            filter_ (str):
+                (Optional) Filter string used to filter objects. See:
+                https://docs.cloud.google.com/storage/docs/listing-objects#filter-by-object-contexts-syntax
+
         Returns:
             Iterator of all :class:`~google.cloud.storage.blob.Blob`
             in this bucket matching the arguments. The RPC call
@@ -1443,6 +1448,9 @@ def list_blobs(
             if soft_deleted is not None:
                 extra_params["softDeleted"] = soft_deleted
 
+            if filter_ is not None:
+                extra_params["filter"] = filter_
+
             if bucket.user_project is not None:
                 extra_params["userProject"] = bucket.user_project
 
diff --git a/packages/google-cloud-storage/tests/system/test_blob.py b/packages/google-cloud-storage/tests/system/test_blob.py
index 60a2fa2568b2..1917f1844cf7 100644
--- a/packages/google-cloud-storage/tests/system/test_blob.py
+++ b/packages/google-cloud-storage/tests/system/test_blob.py
@@ -853,6 +853,33 @@ def test_blob_compose_new_blob(shared_bucket, blobs_to_delete):
     assert destination.download_as_bytes() == payload_1 + payload_2
 
 
+def test_blob_compose_w_destination_contexts(shared_bucket, blobs_to_delete):
+    from google.cloud.storage.blob import ObjectContexts, ObjectCustomContextPayload
+
+    payload_1 = b"AAA\n"
+    source_1 = shared_bucket.blob("source-1-contexts")
+    source_1.upload_from_string(payload_1)
+    blobs_to_delete.append(source_1)
+
+    payload_2 = b"BBB\n"
+    source_2 = shared_bucket.blob("source-2-contexts")
+    source_2.upload_from_string(payload_2)
+    blobs_to_delete.append(source_2)
+
+    destination = shared_bucket.blob("destination-contexts")
+    destination.content_type = "text/plain"
+
+    context_payload = ObjectCustomContextPayload(value="bar")
+    contexts = ObjectContexts(None, custom={"foo": context_payload})
+
+    destination.compose([source_1, source_2], destination_contexts=contexts)
+    blobs_to_delete.append(destination)
+
+    assert destination.download_as_bytes() == payload_1 + payload_2
+    destination.reload()
+    assert destination.contexts.custom["foo"].value == "bar"
+
+
 def test_blob_compose_new_blob_wo_content_type(shared_bucket, blobs_to_delete):
     payload_1 = b"AAA\n"
     source_1 = shared_bucket.blob("source-1")
@@ -1014,6 +1041,30 @@ def test_blob_rewrite_new_blob_add_key(shared_bucket, blobs_to_delete, file_data
     assert dest.download_as_bytes() == source_data
 
 
+def test_blob_rewrite_w_destination_contexts(shared_bucket, blobs_to_delete, file_data):
+    from google.cloud.storage.blob import ObjectContexts, ObjectCustomContextPayload
+
+    info = file_data["simple"]
+    source = shared_bucket.blob(uuid.uuid4().hex)
+    source.upload_from_filename(info["path"])
+    blobs_to_delete.append(source)
+    source_data = source.download_as_bytes()
+
+    dest = shared_bucket.blob(uuid.uuid4().hex)
+    context_payload = ObjectCustomContextPayload(value="bar")
+    contexts = ObjectContexts(None, custom={"foo": context_payload})
+
+    token, rewritten, total = dest.rewrite(source, destination_contexts=contexts)
+    blobs_to_delete.append(dest)
+
+    assert token is None
+    assert rewritten == len(source_data)
+    assert total == len(source_data)
+    assert dest.download_as_bytes() == source_data
+    dest.reload()
+    assert dest.contexts.custom["foo"].value == "bar"
+
+
 def test_blob_rewrite_rotate_key(shared_bucket, blobs_to_delete, file_data):
     blob_name = "rotating-keys"
     info = file_data["simple"]
@@ -1209,3 +1260,67 @@ def test_blob_download_as_bytes_single_shot_download(
 
     result_single_shot_download = blob.download_as_bytes(single_shot_download=True)
     assert result_single_shot_download == payload
+
+
+def test_blob_contexts(shared_bucket, blobs_to_delete):
+    from google.cloud.storage.blob import ObjectContexts, ObjectCustomContextPayload
+
+    blob_name = f"ObjectContexts-{uuid.uuid4().hex}"
+    blob = shared_bucket.blob(blob_name)
+
+    # 1. Create with contexts
+    custom = {
+        "k1": ObjectCustomContextPayload(value="v1"),
+        "k2": ObjectCustomContextPayload(value="v2"),
+    }
+    blob.contexts = ObjectContexts(blob, custom=custom)
+    blob.upload_from_string(b"foo")
+    blobs_to_delete.append(blob)
+
+    blob.reload()
+    assert blob.contexts.custom["k1"].value == "v1"
+    assert blob.contexts.custom["k2"].value == "v2"
+
+    # 2. Patch: update one, delete one
+    blob.contexts.custom["k1"].value = "v1-updated"
+    blob.contexts.custom["k2"].value = None
+    blob.patch()
+
+    blob.reload()
+    assert blob.contexts.custom["k1"].value == "v1-updated"
+    assert "k2" not in blob.contexts.custom or blob.contexts.custom["k2"].value is None
+
+    # 3. Clear all
+    blob.contexts = None
+    blob.patch()
+
+    blob.reload()
+    assert not blob.contexts.custom
+
+
+def test_blob_contexts_custom_setter(shared_bucket, blobs_to_delete):
+    from google.cloud.storage.blob import ObjectContexts, ObjectCustomContextPayload
+
+    blob_name = f"ObjectContextsCustomSetter-{uuid.uuid4().hex}"
+    blob = shared_bucket.blob(blob_name)
+    blob.upload_from_string(b"foo")
+    blobs_to_delete.append(blob)
+
+    # 1. Use custom setter to assign dictionary of payloads
+    custom = {
+        "k1": ObjectCustomContextPayload(value="v1"),
+        "k2": ObjectCustomContextPayload(value="v2"),
+    }
+    blob.contexts.custom = custom
+    blob.patch()
+
+    blob.reload()
+    assert blob.contexts.custom["k1"].value == "v1"
+    assert blob.contexts.custom["k2"].value == "v2"
+
+    # 2. Update one key.value and patch
+    blob.contexts.custom["k1"].value = "v1-updated"
+    blob.patch()
+
+    blob.reload()
+    assert blob.contexts.custom["k1"].value == "v1-updated"
diff --git a/packages/google-cloud-storage/tests/system/test_bucket.py b/packages/google-cloud-storage/tests/system/test_bucket.py
index 88ceea5cc8fa..efe38203136a 100644
--- a/packages/google-cloud-storage/tests/system/test_bucket.py
+++ b/packages/google-cloud-storage/tests/system/test_bucket.py
@@ -360,6 +360,37 @@ def test_bucket_copy_blob(
     assert copied_contents == payload
 
 
+def test_bucket_copy_blob_w_destination_contexts(
+    storage_client,
+    buckets_to_delete,
+    blobs_to_delete,
+):
+    from google.cloud.storage.blob import ObjectContexts, ObjectCustomContextPayload
+
+    payload = b"DEADBEEF"
+    bucket_name = _helpers.unique_name("copy-blob-contexts")
+    created = _helpers.retry_429_503(storage_client.create_bucket)(bucket_name)
+    buckets_to_delete.append(created)
+    assert created.name == bucket_name
+
+    blob = created.blob("CloudLogo")
+    blob.upload_from_string(payload)
+    blobs_to_delete.append(blob)
+
+    context_payload = ObjectCustomContextPayload(value="bar")
+    contexts = ObjectContexts(None, custom={"foo": context_payload})
+
+    new_blob = _helpers.retry_bad_copy(created.copy_blob)(
+        blob, created, "CloudLogoCopy", destination_contexts=contexts
+    )
+    blobs_to_delete.append(new_blob)
+
+    copied_contents = new_blob.download_as_bytes()
+    assert copied_contents == payload
+    new_blob.reload()
+    assert new_blob.contexts.custom["foo"].value == "bar"
+
+
 def test_bucket_copy_blob_w_user_project(
     storage_client,
     buckets_to_delete,
@@ -731,6 +762,35 @@ def test_bucket_list_blobs_w_match_glob(
         assert [blob.name for blob in blobs] == expected_names
 
 
+@_helpers.retry_failures
+def test_bucket_list_blobs_w_filter(
+    storage_client,
+    buckets_to_delete,
+    blobs_to_delete,
+):
+    from google.cloud.storage.blob import ObjectContexts, ObjectCustomContextPayload
+
+    bucket_name = _helpers.unique_name("w-filter")
+    bucket = _helpers.retry_429_503(storage_client.create_bucket)(bucket_name)
+    buckets_to_delete.append(bucket)
+
+    payload = b"helloworld"
+    blob_names = ["foo", "bar", "baz"]
+    for name in blob_names:
+        blob = bucket.blob(name)
+        blob.upload_from_string(payload)
+        if name == "bar":
+            custom = {"target": ObjectCustomContextPayload(value="match")}
+            blob.contexts = ObjectContexts(blob, custom=custom)
+            blob.patch()
+        blobs_to_delete.append(blob)
+
+    # List with filter matching only 'bar'
+    blob_iter = bucket.list_blobs(filter_='contexts."target"="match"')
+    blobs = list(blob_iter)
+    assert [blob.name for blob in blobs] == ["bar"]
+
+
 def test_bucket_list_blobs_include_managed_folders(
     storage_client,
     buckets_to_delete,
diff --git a/packages/google-cloud-storage/tests/system/test_zonal.py b/packages/google-cloud-storage/tests/system/test_zonal.py
index c87bc583386a..d2cb6ae435c2 100644
--- a/packages/google-cloud-storage/tests/system/test_zonal.py
+++ b/packages/google-cloud-storage/tests/system/test_zonal.py
@@ -18,6 +18,10 @@
     _DEFAULT_FLUSH_INTERVAL_BYTES,
     AsyncAppendableObjectWriter,
 )
+from google.cloud.storage.blob import (
+    ObjectContexts,
+    ObjectCustomContextPayload,
+)
 from google.cloud.storage.asyncio.async_grpc_client import AsyncGrpcClient
 from google.cloud.storage.asyncio.async_multi_range_downloader import (
     AsyncMultiRangeDownloader,
@@ -431,6 +435,36 @@ async def _run():
     event_loop.run_until_complete(_run())
 
 
+@pytest.mark.asyncio
+async def test_write_blob_with_contexts(storage_client, blobs_to_delete):
+    async_client = await create_async_grpc_client()
+    blob_name = f"ObjectContextsGrpc-{uuid.uuid4().hex}"
+
+    bucket = storage_client.bucket(_ZONAL_BUCKET)
+    blob = bucket.blob(blob_name)
+    blob.contexts = ObjectContexts(
+        blob, custom={"foo": ObjectCustomContextPayload(value="bar")}
+    )
+    writer = AsyncAppendableObjectWriter.from_blob(async_client, blob)
+    await writer.open()
+    await writer.append(b"grpc-test")
+    await writer.close(finalize_on_close=True)
+
+    try:
+        blobs = list(
+            storage_client.list_blobs(_ZONAL_BUCKET, filter_='contexts."foo"="bar"')
+        )
+        names = [b.name for b in blobs]
+        assert blob_name in names
+
+        # Assert contexts via gRPC GetObject
+        obj_proto = await async_client.get_object(_ZONAL_BUCKET, blob_name)
+        assert "foo" in obj_proto.contexts.custom
+        assert obj_proto.contexts.custom["foo"].value == "bar"
+    finally:
+        blobs_to_delete.append(storage_client.bucket(_ZONAL_BUCKET).blob(blob_name))
+
+
 def test_read_unfinalized_appendable_object(
     storage_client, blobs_to_delete, event_loop, grpc_client_direct
 ):
diff --git a/packages/google-cloud-storage/tests/unit/asyncio/test_async_write_object_stream.py b/packages/google-cloud-storage/tests/unit/asyncio/test_async_write_object_stream.py
index 274b9ad187e3..a73828345aa7 100644
--- a/packages/google-cloud-storage/tests/unit/asyncio/test_async_write_object_stream.py
+++ b/packages/google-cloud-storage/tests/unit/asyncio/test_async_write_object_stream.py
@@ -21,6 +21,7 @@
 
 from google.cloud import _storage_v2
 from google.cloud.storage import Blob, Bucket
+from google.cloud.storage.blob import ObjectContexts, ObjectCustomContextPayload
 from google.cloud.storage.asyncio.async_write_object_stream import (
     _AsyncWriteObjectStream,
 )
@@ -197,6 +198,8 @@ async def test_open_new_object_with_blob_sync_attrs(
             "mode": "Locked",
             "retain_until_time": retain_until_time,
         }
+        payload = ObjectCustomContextPayload(value="context-value")
+        mock_blob.contexts = ObjectContexts(mock_blob, custom={"context-key": payload})
 
         stream = _AsyncWriteObjectStream(mock_client, BUCKET, OBJECT, blob=mock_blob)
         await stream.open()
@@ -225,6 +228,8 @@ async def test_open_new_object_with_blob_sync_attrs(
         assert int(resource.retention.retain_until_time.timestamp()) == int(
             retain_until_time.timestamp()
         )
+        assert "context-key" in resource.contexts.custom
+        assert resource.contexts.custom["context-key"].value == "context-value"
 
     @pytest.mark.asyncio
     async def test_open_already_open_raises(self, mock_client):
diff --git a/packages/google-cloud-storage/tests/unit/test__grpc_conversions.py b/packages/google-cloud-storage/tests/unit/test__grpc_conversions.py
index fdfde54310ad..b5029873eb74 100644
--- a/packages/google-cloud-storage/tests/unit/test__grpc_conversions.py
+++ b/packages/google-cloud-storage/tests/unit/test__grpc_conversions.py
@@ -134,3 +134,28 @@ def test_blob_to_proto_retention():
     assert int(proto.retention.retain_until_time.timestamp()) == int(
         retain_until_time.timestamp()
     )
+
+
+def test_blob_to_proto_contexts():
+    blob = mock.Mock(
+        spec=["name", "bucket", "contexts", "custom_time", "acl", "retention"]
+    )
+    blob.name = "blob-name"
+    blob.bucket.name = "bucket-name"
+
+    from google.cloud.storage.blob import ObjectContexts, ObjectCustomContextPayload
+
+    create_time = datetime.datetime(2025, 1, 1, tzinfo=datetime.timezone.utc)
+    payload = ObjectCustomContextPayload(value="val", create_time=create_time)
+    blob.contexts = ObjectContexts(blob, custom={"key": payload})
+
+    blob.custom_time = None
+    blob.acl = None
+    blob.retention = None
+    for attr in _grpc_conversions._BLOB_ATTR_TO_PROTO_FIELD:
+        setattr(blob, attr, None)
+
+    proto = _grpc_conversions.blob_to_proto(blob)
+
+    assert "key" in proto.contexts.custom
+    assert proto.contexts.custom["key"].value == "val"
diff --git a/packages/google-cloud-storage/tests/unit/test_blob.py b/packages/google-cloud-storage/tests/unit/test_blob.py
index a218f011dd17..82016ac9d4ef 100644
--- a/packages/google-cloud-storage/tests/unit/test_blob.py
+++ b/packages/google-cloud-storage/tests/unit/test_blob.py
@@ -4874,6 +4874,51 @@ def test_compose_w_metageneration_match(self):
             _target_object=destination,
         )
 
+    def test_compose_w_destination_contexts(self):
+        from google.cloud.storage.blob import (
+            ObjectContexts,
+            ObjectCustomContextPayload,
+        )
+
+        source_1_name = "source-1"
+        source_2_name = "source-2"
+        destination_name = "destination"
+        api_response = {"contexts": {"custom": {"foo": {"value": "bar"}}}}
+        client = mock.Mock(spec=["_post_resource"])
+        client._post_resource.return_value = api_response
+        bucket = _Bucket(client=client)
+        source_1 = self._make_one(source_1_name, bucket=bucket)
+        source_2 = self._make_one(source_2_name, bucket=bucket)
+        destination = self._make_one(destination_name, bucket=bucket)
+
+        payload = ObjectCustomContextPayload(value="bar")
+        contexts = ObjectContexts(None, custom={"foo": payload})
+
+        destination.compose(
+            sources=[source_1, source_2],
+            destination_contexts=contexts,
+        )
+
+        self.assertEqual(destination.contexts.custom["foo"].value, "bar")
+
+        expected_path = f"/b/name/o/{destination_name}/compose"
+        expected_data = {
+            "sourceObjects": [
+                {"name": source_1.name, "generation": source_1.generation},
+                {"name": source_2.name, "generation": source_2.generation},
+            ],
+            "destination": {"contexts": {"custom": {"foo": {"value": "bar"}}}},
+        }
+        expected_query_params = {}
+        client._post_resource.assert_called_once_with(
+            expected_path,
+            expected_data,
+            query_params=expected_query_params,
+            timeout=self._get_default_timeout(),
+            retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
+            _target_object=destination,
+        )
+
     def test_rewrite_w_response_wo_resource(self):
         source_name = "source"
         dest_name = "dest"
@@ -5272,6 +5317,56 @@ def test_rewrite_same_name_w_kms_key_w_version(self):
             _target_object=dest,
         )
 
+    def test_rewrite_w_destination_contexts(self):
+        from google.cloud.storage.blob import (
+            ObjectContexts,
+            ObjectCustomContextPayload,
+        )
+
+        blob_name = "blob"
+        bytes_rewritten = object_size = 42
+        api_response = {
+            "totalBytesRewritten": bytes_rewritten,
+            "objectSize": object_size,
+            "done": True,
+            "resource": {
+                "etag": "DEADBEEF",
+                "contexts": {"custom": {"foo": {"value": "bar"}}},
+            },
+        }
+        client = mock.Mock(spec=["_post_resource"])
+        client._post_resource.return_value = api_response
+        bucket = _Bucket(client=client)
+        source = self._make_one(blob_name, bucket=bucket)
+        dest = self._make_one(blob_name, bucket=bucket)
+
+        payload = ObjectCustomContextPayload(value="bar")
+        contexts = ObjectContexts(None, custom={"foo": payload})
+
+        token, rewritten, size = dest.rewrite(
+            source,
+            destination_contexts=contexts,
+        )
+
+        self.assertIsNone(token)
+        self.assertEqual(rewritten, bytes_rewritten)
+        self.assertEqual(size, object_size)
+        self.assertEqual(dest.contexts.custom["foo"].value, "bar")
+
+        expected_path = f"/b/name/o/{blob_name}/rewriteTo/b/name/o/{blob_name}"
+        expected_data = {"contexts": {"custom": {"foo": {"value": "bar"}}}}
+        expected_query_params = {}
+        expected_headers = {}
+        client._post_resource.assert_called_once_with(
+            expected_path,
+            expected_data,
+            query_params=expected_query_params,
+            headers=expected_headers,
+            timeout=self._get_default_timeout(),
+            retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
+            _target_object=dest,
+        )
+
     def _update_storage_class_multi_pass_helper(self, **kw):
         blob_name = "blob-name"
         storage_class = "NEARLINE"
@@ -6205,6 +6300,118 @@ def test_object_lock_retention_configuration_setter(self):
         self.assertIsNone(blob.retention.retain_until_time)
         self.assertIn("retention", blob._changes)
 
+    def test_object_contexts_payload_ctor(self):
+        from google.cloud.storage.blob import ObjectCustomContextPayload
+
+        create_time = datetime.datetime(2025, 1, 1, tzinfo=_UTC)
+        update_time = datetime.datetime(2025, 1, 2, tzinfo=_UTC)
+        payload = ObjectCustomContextPayload(
+            value="foo", create_time=create_time, update_time=update_time
+        )
+        self.assertEqual(payload.value, "foo")
+        self.assertEqual(payload.create_time, create_time)
+        self.assertEqual(payload.update_time, update_time)
+
+    def test_object_contexts_ctor(self):
+        from google.cloud.storage.blob import (
+            Blob,
+            ObjectContexts,
+            ObjectCustomContextPayload,
+        )
+
+        blob = mock.Mock(spec=Blob)
+        custom = {"key": ObjectCustomContextPayload(value="val")}
+        contexts = ObjectContexts(blob, custom=custom)
+        self.assertIs(contexts.blob, blob)
+        self.assertEqual(contexts.custom, custom)
+
+    def test_object_contexts_from_api_repr(self):
+        from google.cloud.storage.blob import Blob, ObjectContexts
+
+        blob = mock.Mock(spec=Blob)
+        resource = {
+            "custom": {
+                "key": {
+                    "value": "val",
+                    "createTime": "2025-01-01T00:00:00Z",
+                    "updateTime": "2025-01-02T00:00:00Z",
+                }
+            }
+        }
+        contexts = ObjectContexts.from_api_repr(resource, blob)
+        self.assertIs(contexts.blob, blob)
+        self.assertIn("key", contexts.custom)
+        payload = contexts.custom["key"]
+        self.assertEqual(payload.value, "val")
+        self.assertEqual(
+            payload.create_time, datetime.datetime(2025, 1, 1, tzinfo=_UTC)
+        )
+        self.assertEqual(
+            payload.update_time, datetime.datetime(2025, 1, 2, tzinfo=_UTC)
+        )
+
+    def test_object_contexts_property(self):
+        from google.cloud.storage.blob import (
+            Blob,
+            ObjectContexts,
+            ObjectCustomContextPayload,
+        )
+
+        bucket = mock.Mock()
+        bucket.name = "b"
+        bucket.__getitem__ = mock.Mock(
+            side_effect=lambda x: "b" if x in (0, -1) else None
+        )
+        blob = Blob("blob-name", bucket=bucket)
+        self.assertIsInstance(blob.contexts, ObjectContexts)
+        self.assertEqual(blob.contexts.custom, {})
+
+        custom = {"key": ObjectCustomContextPayload(value="val")}
+        blob.contexts = ObjectContexts(blob, custom=custom)
+        self.assertEqual(blob.contexts.custom, custom)
+
+        blob.contexts = None
+        self.assertIsNone(blob._properties["contexts"])
+
+    def test_patch_contexts(self):
+        from google.cloud.storage.blob import (
+            Blob,
+            ObjectContexts,
+            ObjectCustomContextPayload,
+        )
+        from google.cloud.storage.bucket import Bucket
+
+        client = self._make_client(project="p")
+        bucket = Bucket(client, name="b")
+        blob = Blob("blob-name", bucket=bucket)
+
+        custom = {"key": ObjectCustomContextPayload(value="val")}
+        blob.contexts = ObjectContexts(blob, custom=custom)
+
+        with mock.patch.object(client, "_patch_resource") as mocked:
+            blob.patch()
+            mocked.assert_called_once()
+            args, kwargs = mocked.call_args
+            sent_resource = args[1]
+            self.assertEqual(sent_resource["contexts"]["custom"]["key"]["value"], "val")
+
+    def test_patch_contexts_none(self):
+        from google.cloud.storage.blob import Blob
+        from google.cloud.storage.bucket import Bucket
+
+        client = self._make_client(project="p")
+        bucket = Bucket(client, name="b")
+        blob = Blob("blob-name", bucket=bucket)
+
+        blob.contexts = None
+
+        with mock.patch.object(client, "_patch_resource") as mocked:
+            blob.patch()
+            mocked.assert_called_once()
+            args, kwargs = mocked.call_args
+            sent_resource = args[1]
+            self.assertIsNone(sent_resource["contexts"])
+
 
 class Test__quote(unittest.TestCase):
     @staticmethod
diff --git a/packages/google-cloud-storage/tests/unit/test_bucket.py b/packages/google-cloud-storage/tests/unit/test_bucket.py
index 76c0eb5104c0..f0e996eb424b 100644
--- a/packages/google-cloud-storage/tests/unit/test_bucket.py
+++ b/packages/google-cloud-storage/tests/unit/test_bucket.py
@@ -1221,6 +1221,7 @@ def test_list_blobs_w_defaults(self):
         expected_include_folders_as_prefixes = None
         soft_deleted = None
         page_size = None
+        filter_ = None
         client.list_blobs.assert_called_once_with(
             bucket,
             max_results=expected_max_results,
@@ -1239,6 +1240,7 @@ def test_list_blobs_w_defaults(self):
             include_folders_as_prefixes=expected_include_folders_as_prefixes,
             soft_deleted=soft_deleted,
             page_size=page_size,
+            filter_=filter_,
         )
 
     def test_list_blobs_w_explicit(self):
@@ -1299,6 +1301,7 @@ def test_list_blobs_w_explicit(self):
         expected_include_folders_as_prefixes = include_folders_as_prefixes
         expected_soft_deleted = soft_deleted
         expected_page_size = page_size
+        expected_filter = None
         other_client.list_blobs.assert_called_once_with(
             bucket,
             max_results=expected_max_results,
@@ -1317,6 +1320,54 @@ def test_list_blobs_w_explicit(self):
             include_folders_as_prefixes=expected_include_folders_as_prefixes,
             soft_deleted=expected_soft_deleted,
             page_size=expected_page_size,
+            filter_=expected_filter,
+        )
+
+    def test_list_blobs_w_filter(self):
+        name = "name"
+        filter_ = 'contexts."foo"="bar"'
+        bucket = self._make_one(client=None, name=name)
+        other_client = self._make_client()
+        other_client.list_blobs = mock.Mock(spec=[])
+
+        iterator = bucket.list_blobs(filter_=filter_, client=other_client)
+
+        self.assertIs(iterator, other_client.list_blobs.return_value)
+
+        expected_page_token = None
+        expected_max_results = None
+        expected_prefix = None
+        expected_delimiter = None
+        expected_match_glob = None
+        expected_start_offset = None
+        expected_end_offset = None
+        expected_include_trailing_delimiter = None
+        expected_versions = None
+        expected_projection = "noAcl"
+        expected_fields = None
+        expected_include_folders_as_prefixes = None
+        expected_soft_deleted = None
+        expected_page_size = None
+        expected_filter = filter_
+        other_client.list_blobs.assert_called_once_with(
+            bucket,
+            max_results=expected_max_results,
+            page_token=expected_page_token,
+            prefix=expected_prefix,
+            delimiter=expected_delimiter,
+            start_offset=expected_start_offset,
+            end_offset=expected_end_offset,
+            include_trailing_delimiter=expected_include_trailing_delimiter,
+            versions=expected_versions,
+            projection=expected_projection,
+            fields=expected_fields,
+            timeout=self._get_default_timeout(),
+            retry=DEFAULT_RETRY,
+            match_glob=expected_match_glob,
+            include_folders_as_prefixes=expected_include_folders_as_prefixes,
+            soft_deleted=expected_soft_deleted,
+            page_size=expected_page_size,
+            filter_=expected_filter,
         )
 
     def test_list_notifications_w_defaults(self):
@@ -2296,6 +2347,51 @@ def test_copy_blob_w_name_and_user_project(self):
             _target_object=new_blob,
         )
 
+    def test_copy_blob_w_destination_contexts(self):
+        from google.cloud.storage.blob import (
+            ObjectContexts,
+            ObjectCustomContextPayload,
+        )
+
+        source_name = "source"
+        dest_name = "dest"
+        blob_name = "blob-name"
+        new_name = "new_name"
+        api_response = {"contexts": {"custom": {"foo": {"value": "bar"}}}}
+        client = mock.Mock(spec=["_post_resource"])
+        client._post_resource.return_value = api_response
+        source = self._make_one(client=client, name=source_name)
+        dest = self._make_one(client=client, name=dest_name)
+        blob = self._make_blob(source_name, blob_name)
+
+        payload = ObjectCustomContextPayload(value="bar")
+        contexts = ObjectContexts(None, custom={"foo": payload})
+
+        new_blob = source.copy_blob(
+            blob,
+            dest,
+            new_name,
+            destination_contexts=contexts,
+        )
+
+        self.assertIs(new_blob.bucket, dest)
+        self.assertEqual(new_blob.name, new_name)
+        self.assertEqual(new_blob.contexts.custom["foo"].value, "bar")
+
+        expected_path = "/b/{}/o/{}/copyTo/b/{}/o/{}".format(
+            source_name, blob_name, dest_name, new_name
+        )
+        expected_data = {"contexts": {"custom": {"foo": {"value": "bar"}}}}
+        expected_query_params = {}
+        client._post_resource.assert_called_once_with(
+            expected_path,
+            expected_data,
+            query_params=expected_query_params,
+            timeout=self._get_default_timeout(),
+            retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
+            _target_object=new_blob,
+        )
+
     def test_move_blob_w_no_retry_timeout_and_generation_match(self):
         source_name = "source"
         blob_name = "blob-name"
diff --git a/packages/google-cloud-storage/tests/unit/test_client.py b/packages/google-cloud-storage/tests/unit/test_client.py
index be6e7273e3b5..27b1f65b3868 100644
--- a/packages/google-cloud-storage/tests/unit/test_client.py
+++ b/packages/google-cloud-storage/tests/unit/test_client.py
@@ -2251,6 +2251,50 @@ def test_list_blobs_w_explicit_w_user_project(self):
             retry=retry,
         )
 
+    def test_list_blobs_w_filter(self):
+        from google.cloud.storage.bucket import _blobs_page_start, _item_to_blob
+
+        project = "PROJECT"
+        bucket_name = "name"
+        filter_ = 'contexts."foo"="bar"'
+        credentials = _make_credentials()
+        client = self._make_one(project=project, credentials=credentials)
+        client._list_resource = mock.Mock(spec=[])
+        client._bucket_arg_to_bucket = mock.Mock(spec=[])
+        bucket = client._bucket_arg_to_bucket.return_value = mock.Mock(
+            spec=["path", "user_project"],
+        )
+        bucket.path = f"/b/{bucket_name}"
+        bucket.user_project = None
+
+        iterator = client.list_blobs(bucket_or_name=bucket_name, filter_=filter_)
+
+        self.assertIs(iterator, client._list_resource.return_value)
+        self.assertIs(iterator.bucket, bucket)
+        self.assertEqual(iterator.prefixes, set())
+
+        expected_path = f"/b/{bucket_name}/o"
+        expected_item_to_value = _item_to_blob
+        expected_page_token = None
+        expected_max_results = None
+        expected_extra_params = {
+            "projection": "noAcl",
+            "filter": filter_,
+        }
+        expected_page_start = _blobs_page_start
+        expected_page_size = None
+        client._list_resource.assert_called_once_with(
+            expected_path,
+            expected_item_to_value,
+            page_token=expected_page_token,
+            max_results=expected_max_results,
+            extra_params=expected_extra_params,
+            page_start=expected_page_start,
+            page_size=expected_page_size,
+            timeout=self._get_default_timeout(),
+            retry=DEFAULT_RETRY,
+        )
+
     def test_list_buckets_wo_project(self):
         from google.cloud.exceptions import BadRequest