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..c39acd1331da 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,22 @@ def blob_to_proto(blob):
             retain_until_time=retain_until_time_proto,
         )
 
+    contexts = getattr(blob, "contexts", None)
+    if contexts:
+        custom_contexts = contexts.get("custom")
+        if custom_contexts is not None:
+            custom_contexts_proto = {}
+            for key, payload in custom_contexts.items():
+                if payload is not None:
+                    custom_contexts_proto[key] = _storage_v2.ObjectCustomContextPayload(
+                        value=payload.get("value")
+                    )
+
+            resource_params["contexts"] = _storage_v2.ObjectContexts(
+                custom=custom_contexts_proto
+            )
+        else:
+            # Signal clearing of all custom contexts.
+            resource_params["contexts"] = _storage_v2.ObjectContexts(custom=None)
+
     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..711a973ab74f 100644
--- a/packages/google-cloud-storage/google/cloud/storage/blob.py
+++ b/packages/google-cloud-storage/google/cloud/storage/blob.py
@@ -101,6 +101,7 @@
     "crc32c",
     "customTime",
     "md5Hash",
+    "objectContexts",
     "metadata",
     "name",
     "retention",
@@ -232,6 +233,7 @@ def __init__(
         )
 
         self._encryption_key = encryption_key
+        self._contexts = None
 
         if kms_key_name is not None:
             self._properties["kmsKeyName"] = kms_key_name
@@ -239,6 +241,15 @@ def __init__(
         if generation is not None:
             self._properties["generation"] = generation
 
+    def _set_properties(self, value):
+        """Set the properties for the current object.
+
+        :type value: dict or :class:`google.cloud.storage.batch._FutureDict`
+        :param value: The properties to be set.
+        """
+        super()._set_properties(value)
+        self._contexts = None
+
     @property
     def bucket(self):
         """Bucket which contains the object.
@@ -5008,6 +5019,18 @@ def retention(self):
         info = self._properties.get("retention", {})
         return Retention.from_api_repr(info, self)
 
+    @property
+    def contexts(self):
+        """Retrieve the object contexts for this object.
+
+        :rtype: :class:`ObjectContexts`
+        :returns: an instance for managing the object's contexts.
+        """
+        if self._contexts is None:
+            info = self._properties.get("objectContexts", {})
+            self._contexts = ObjectContexts.from_api_repr(info, self)
+        return self._contexts
+
     @property
     def soft_delete_time(self):
         """If this object has been soft-deleted, returns the time at which it became soft-deleted.
@@ -5300,3 +5323,90 @@ 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 ObjectContexts(dict):
+    """Map 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) A map of custom contexts.
+    """
+
+    def __init__(self, blob, custom=None):
+        super().__init__({"custom": custom})
+        self._blob = blob
+
+    @classmethod
+    def from_api_repr(cls, resource, blob):
+        """Factory:  construct instance from resource.
+
+        :type blob: :class:`Blob`
+        :param blob: Blob for which these contexts apply to.
+
+        :type resource: dict
+        :param resource: mapping as returned from API call.
+
+        :rtype: :class:`ObjectContexts`
+        :returns: ObjectContexts created from resource.
+        """
+        instance = cls(blob)
+        if resource:
+            # Handle timestamps in the resource if present
+            custom = resource.get("custom")
+            if custom:
+                for payload in custom.values():
+                    if payload and "createTime" in payload:
+                        payload["create_time"] = _rfc3339_nanos_to_datetime(
+                            payload["createTime"]
+                        )
+                    if payload and "updateTime" in payload:
+                        payload["update_time"] = _rfc3339_nanos_to_datetime(
+                            payload["updateTime"]
+                        )
+            instance.update(resource)
+        return instance
+
+    @property
+    def blob(self):
+        """Blob for which these contexts apply to.
+
+        :rtype: :class:`Blob`
+        :returns: the instance's blob.
+        """
+        return self._blob
+
+    def set_custom_context(self, key, value):
+        """Set a custom context.
+
+        :type key: str
+        :param key: The key of the custom context.
+
+        :type value: str
+        :param value: The value of the custom context.
+        """
+        custom = self.get("custom")
+        if custom is None:
+            custom = {}
+            self["custom"] = custom
+        custom[key] = {"value": value}
+        self.blob._patch_property("objectContexts", self)
+
+    def delete_custom_context(self, key):
+        """Delete a custom context.
+
+        :type key: str
+        :param key: The key of the custom context to delete.
+        """
+        custom = self.get("custom")
+        if custom is not None:
+            custom[key] = None
+            self.blob._patch_property("objectContexts", self)
+
+    def clear_custom_contexts(self):
+        """Clear all custom contexts."""
+        self["custom"] = None
+        self.blob._patch_property("objectContexts", 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..c0faae46ddf9 100644
--- a/packages/google-cloud-storage/google/cloud/storage/bucket.py
+++ b/packages/google-cloud-storage/google/cloud/storage/bucket.py
@@ -1423,6 +1423,8 @@ 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.
 
@@ -1521,6 +1523,12 @@ def list_blobs(
             (Optional) Maximum number of blobs to return in each page.
             Defaults to a value set by the API.
 
+        :type filter_: str
+        :param filter_:
+            (Optional) A filter expression that filters objects listed in the response.
+            The expression must be specified in the GCS filter syntax.
+            See: https://cloud.google.com/storage/docs/json_api/v1/objects/list#filter
+
         :rtype: :class:`~google.api_core.page_iterator.Iterator`
         :returns: Iterator of all :class:`~google.cloud.storage.blob.Blob`
                   in this bucket matching the arguments.
@@ -1545,6 +1553,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(
diff --git a/packages/google-cloud-storage/google/cloud/storage/client.py b/packages/google-cloud-storage/google/cloud/storage/client.py
index 528b2255f451..e21d5cd8a983 100644
--- a/packages/google-cloud-storage/google/cloud/storage/client.py
+++ b/packages/google-cloud-storage/google/cloud/storage/client.py
@@ -1291,6 +1291,8 @@ 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 +1402,11 @@ 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) A filter expression that filters objects listed in the response.
+                The expression must be specified in the GCS filter syntax.
+                See: https://cloud.google.com/storage/docs/json_api/v1/objects/list#filter
+
         Returns:
             Iterator of all :class:`~google.cloud.storage.blob.Blob`
             in this bucket matching the arguments. The RPC call
@@ -1440,6 +1447,9 @@ def list_blobs(
             if include_folders_as_prefixes is not None:
                 extra_params["includeFoldersAsPrefixes"] = include_folders_as_prefixes
 
+            if filter_ is not None:
+                extra_params["filter"] = filter_
+
             if soft_deleted is not None:
                 extra_params["softDeleted"] = soft_deleted
 
diff --git a/packages/google-cloud-storage/tests/system/test_blob.py b/packages/google-cloud-storage/tests/system/test_blob.py
index 60a2fa2568b2..625e398fb6c9 100644
--- a/packages/google-cloud-storage/tests/system/test_blob.py
+++ b/packages/google-cloud-storage/tests/system/test_blob.py
@@ -1209,3 +1209,77 @@ 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):
+    blob_name = f"context-test-{uuid.uuid4().hex}"
+    blob = shared_bucket.blob(blob_name)
+    blob.upload_from_string(b"foo")
+    blobs_to_delete.append(blob)
+
+    # Set context
+    blob.contexts.set_custom_context("foo", "bar")
+    blob.patch()
+
+    assert blob.contexts["custom"]["foo"]["value"] == "bar"
+    assert "create_time" in blob.contexts["custom"]["foo"]
+
+    # Reload and check
+    blob.reload()
+    assert blob.contexts["custom"]["foo"]["value"] == "bar"
+
+    # Update context
+    blob.contexts.set_custom_context("foo", "baz")
+    blob.patch()
+    assert blob.contexts["custom"]["foo"]["value"] == "baz"
+
+    # Add another context
+    blob.contexts.set_custom_context("another", "value")
+    blob.patch()
+    assert blob.contexts["custom"]["another"]["value"] == "value"
+
+    # Delete one context
+    blob.contexts.delete_custom_context("foo")
+    blob.patch()
+    assert "foo" not in blob.contexts["custom"] or blob.contexts["custom"]["foo"] is None
+    assert blob.contexts["custom"]["another"]["value"] == "value"
+
+    # Clear all custom contexts
+    blob.contexts.clear_custom_contexts()
+    blob.patch()
+    assert blob.contexts["custom"] is None
+
+def test_list_blobs_with_filter(shared_bucket, blobs_to_delete):
+    suffix = uuid.uuid4().hex
+    blob1_name = f"filter-test-1-{suffix}"
+    blob2_name = f"filter-test-2-{suffix}"
+
+    blob1 = shared_bucket.blob(blob1_name)
+    blob1.contexts.set_custom_context("color", "red")
+    blob1.upload_from_string(b"red-content")
+    blobs_to_delete.append(blob1)
+
+    blob2 = shared_bucket.blob(blob2_name)
+    blob2.contexts.set_custom_context("color", "blue")
+    blob2.upload_from_string(b"blue-content")
+    blobs_to_delete.append(blob2)
+
+    # Filter for red
+    # The GCS filter syntax uses 'contexts' for the field name regardless of internal SDK representation.
+    filter_expr = f'contexts.custom.color.value="red" AND name="{blob1_name}"'
+    blobs = list(shared_bucket.list_blobs(filter_=filter_expr))
+
+    assert len(blobs) == 1
+    assert blobs[0].name == blob1_name
+
+    # Filter for blue
+    filter_expr = f'contexts.custom.color.value="blue" AND name="{blob2_name}"'
+    blobs = list(shared_bucket.list_blobs(filter_=filter_expr))
+
+    assert len(blobs) == 1
+    assert blobs[0].name == blob2_name
+
+    # Filter for non-existent value
+    filter_expr = f'contexts.custom.color.value="green" AND name.startsWith("filter-test-")'
+    blobs = list(shared_bucket.list_blobs(filter_=filter_expr))
+
+    assert len(blobs) == 0
diff --git a/packages/google-cloud-storage/tests/unit/test_bucket.py b/packages/google-cloud-storage/tests/unit/test_bucket.py
index 76c0eb5104c0..4166726b8c1f 100644
--- a/packages/google-cloud-storage/tests/unit/test_bucket.py
+++ b/packages/google-cloud-storage/tests/unit/test_bucket.py
@@ -1239,6 +1239,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_=None,
         )
 
     def test_list_blobs_w_explicit(self):
@@ -1254,6 +1255,7 @@ def test_list_blobs_w_explicit(self):
         include_folders_as_prefixes = True
         versions = True
         soft_deleted = True
+        filter_ = 'objectContexts.custom.foo.value="bar"'
         page_size = 2
         projection = "full"
         fields = "items/contentLanguage,nextPageToken"
@@ -1281,6 +1283,7 @@ def test_list_blobs_w_explicit(self):
             include_folders_as_prefixes=include_folders_as_prefixes,
             soft_deleted=soft_deleted,
             page_size=page_size,
+            filter_=filter_,
         )
 
         self.assertIs(iterator, other_client.list_blobs.return_value)
@@ -1298,6 +1301,7 @@ def test_list_blobs_w_explicit(self):
         expected_fields = fields
         expected_include_folders_as_prefixes = include_folders_as_prefixes
         expected_soft_deleted = soft_deleted
+        expected_filter = filter_
         expected_page_size = page_size
         other_client.list_blobs.assert_called_once_with(
             bucket,
@@ -1317,6 +1321,7 @@ 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_notifications_w_defaults(self):
diff --git a/packages/google-cloud-storage/tests/unit/test_client.py b/packages/google-cloud-storage/tests/unit/test_client.py
index be6e7273e3b5..3246161864ee 100644
--- a/packages/google-cloud-storage/tests/unit/test_client.py
+++ b/packages/google-cloud-storage/tests/unit/test_client.py
@@ -2179,6 +2179,7 @@ def test_list_blobs_w_explicit_w_user_project(self):
         include_trailing_delimiter = True
         include_folders_as_prefixes = True
         soft_deleted = False
+        filter_ = 'objectContexts.custom.foo.value="bar"'
         versions = True
         projection = "full"
         page_size = 2
@@ -2213,6 +2214,7 @@ def test_list_blobs_w_explicit_w_user_project(self):
             match_glob=match_glob,
             include_folders_as_prefixes=include_folders_as_prefixes,
             soft_deleted=soft_deleted,
+            filter_=filter_,
         )
 
         self.assertIs(iterator, client._list_resource.return_value)
@@ -2236,6 +2238,7 @@ def test_list_blobs_w_explicit_w_user_project(self):
             "userProject": user_project,
             "includeFoldersAsPrefixes": include_folders_as_prefixes,
             "softDeleted": soft_deleted,
+            "filter": filter_,
         }
         expected_page_start = _blobs_page_start
         expected_page_size = 2
diff --git a/packages/google-cloud-storage/tests/unit/test_object_contexts.py b/packages/google-cloud-storage/tests/unit/test_object_contexts.py
new file mode 100644
index 000000000000..43bb01b2c383
--- /dev/null
+++ b/packages/google-cloud-storage/tests/unit/test_object_contexts.py
@@ -0,0 +1,69 @@
+import unittest
+from mock import Mock
+
+class TestObjectContexts(unittest.TestCase):
+    def _getTargetClass(self):
+        from google.cloud.storage.blob import ObjectContexts
+        return ObjectContexts
+
+    def _make_one(self, *args, **kwargs):
+        return self._getTargetClass()(*args, **kwargs)
+
+    def test_ctor(self):
+        blob = Mock()
+        custom = {"foo": {"value": "bar"}}
+        contexts = self._make_one(blob, custom=custom)
+        self.assertEqual(contexts["custom"], custom)
+        self.assertIs(contexts.blob, blob)
+
+    def test_from_api_repr(self):
+        blob = Mock()
+        resource = {"custom": {"foo": {"value": "bar"}}}
+        contexts = self._getTargetClass().from_api_repr(resource, blob)
+        self.assertEqual(contexts["custom"], resource["custom"])
+        self.assertIs(contexts.blob, blob)
+
+    def test_set_custom_context(self):
+        blob = Mock()
+        contexts = self._make_one(blob)
+        contexts.set_custom_context("foo", "bar")
+        self.assertEqual(contexts["custom"], {"foo": {"value": "bar"}})
+        blob._patch_property.assert_called_with("objectContexts", contexts)
+
+    def test_delete_custom_context(self):
+        blob = Mock()
+        custom = {"foo": {"value": "bar"}}
+        contexts = self._make_one(blob, custom=custom)
+        contexts.delete_custom_context("foo")
+        self.assertIsNone(contexts["custom"]["foo"])
+        blob._patch_property.assert_called_with("objectContexts", contexts)
+
+    def test_clear_custom_contexts(self):
+        blob = Mock()
+        custom = {"foo": {"value": "bar"}}
+        contexts = self._make_one(blob, custom=custom)
+        contexts.clear_custom_contexts()
+        self.assertIsNone(contexts["custom"])
+        blob._patch_property.assert_called_with("objectContexts", contexts)
+
+    def test_from_api_repr_w_timestamps(self):
+        from datetime import datetime, timezone
+        blob = Mock()
+        resource = {
+            "custom": {
+                "foo": {
+                    "value": "bar",
+                    "createTime": "2026-01-01T00:00:00.000Z",
+                    "updateTime": "2026-01-01T00:00:01.000Z",
+                }
+            }
+        }
+        contexts = self._getTargetClass().from_api_repr(resource, blob)
+        self.assertEqual(
+            contexts["custom"]["foo"]["create_time"],
+            datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc),
+        )
+        self.assertEqual(
+            contexts["custom"]["foo"]["update_time"],
+            datetime(2026, 1, 1, 0, 0, 1, tzinfo=timezone.utc),
+        )