diff --git a/packages/google-cloud-firestore/google/cloud/firestore/__init__.py b/packages/google-cloud-firestore/google/cloud/firestore/__init__.py
index 47eac898520d..57fd2c45be0b 100644
--- a/packages/google-cloud-firestore/google/cloud/firestore/__init__.py
+++ b/packages/google-cloud-firestore/google/cloud/firestore/__init__.py
@@ -19,6 +19,17 @@
 
 __version__ = package_version.__version__
 
+from google.cloud.firestore_v1.bson import (
+    BSONBinary,
+    BSONDecimal128,
+    BSONInt32,
+    BSONMaxKey,
+    BSONMinKey,
+    BSONObjectID,
+    BSONRegex,
+    BSONTimestamp,
+)
+
 from typing import List
 
 from google.cloud.firestore_v1 import (
@@ -92,6 +103,14 @@
     "async_transactional",
     "AsyncTransaction",
     "AsyncWriteBatch",
+    "BSONBinary",
+    "BSONDecimal128",
+    "BSONInt32",
+    "BSONMaxKey",
+    "BSONMinKey",
+    "BSONObjectID",
+    "BSONRegex",
+    "BSONTimestamp",
     "Client",
     "CountAggregation",
     "CollectionGroup",
diff --git a/packages/google-cloud-firestore/google/cloud/firestore_v1/__init__.py b/packages/google-cloud-firestore/google/cloud/firestore_v1/__init__.py
index 38593d61832f..0cac575e2af7 100644
--- a/packages/google-cloud-firestore/google/cloud/firestore_v1/__init__.py
+++ b/packages/google-cloud-firestore/google/cloud/firestore_v1/__init__.py
@@ -46,6 +46,16 @@
 from google.cloud.firestore_v1.base_pipeline import SubPipeline
 from google.cloud.firestore_v1.base_query import And, FieldFilter, Or
 from google.cloud.firestore_v1.batch import WriteBatch
+from google.cloud.firestore_v1.bson import (
+    BSONBinary,
+    BSONDecimal128,
+    BSONInt32,
+    BSONMaxKey,
+    BSONMinKey,
+    BSONObjectID,
+    BSONRegex,
+    BSONTimestamp,
+)
 from google.cloud.firestore_v1.client import Client
 from google.cloud.firestore_v1.collection import CollectionReference
 from google.cloud.firestore_v1.document import DocumentReference
@@ -147,6 +157,14 @@
     "async_transactional",
     "AsyncTransaction",
     "AsyncWriteBatch",
+    "BSONBinary",
+    "BSONDecimal128",
+    "BSONInt32",
+    "BSONMaxKey",
+    "BSONMinKey",
+    "BSONObjectID",
+    "BSONRegex",
+    "BSONTimestamp",
     "Client",
     "CountAggregation",
     "CollectionGroup",
diff --git a/packages/google-cloud-firestore/google/cloud/firestore_v1/_helpers.py b/packages/google-cloud-firestore/google/cloud/firestore_v1/_helpers.py
index 04cdd05f0c45..96bad421b8c0 100644
--- a/packages/google-cloud-firestore/google/cloud/firestore_v1/_helpers.py
+++ b/packages/google-cloud-firestore/google/cloud/firestore_v1/_helpers.py
@@ -49,6 +49,17 @@
 from google.cloud.firestore_v1.types.write import DocumentTransform
 from google.cloud.firestore_v1.vector import Vector
 
+from google.cloud.firestore_v1.bson import (
+    BSONBinary,
+    BSONDecimal128,
+    BSONInt32,
+    BSONMaxKey,
+    BSONMinKey,
+    BSONObjectID,
+    BSONRegex,
+    BSONTimestamp,
+)
+
 if TYPE_CHECKING:  # pragma: NO COVER
     from google.cloud.firestore_v1 import DocumentSnapshot
 
@@ -218,7 +229,7 @@ def encode_value(value) -> types.document.Value:
         value_pb = document.ArrayValue(values=value_list)
         return document.Value(array_value=value_pb)
 
-    if isinstance(value, Vector):
+    if hasattr(value, "to_map_value") and callable(getattr(value, "to_map_value")):
         return encode_value(value.to_map_value())
 
     if isinstance(value, dict):
@@ -415,11 +426,33 @@ def decode_dict(value_fields, client) -> Union[dict, Vector]:
     value_fields_pb = getattr(value_fields, "_pb", value_fields)
     res = {key: decode_value(value, client) for key, value in value_fields_pb.items()}
 
-    if res.get("__type__", None) == "__vector__":
+    type_tag = res.get("__type__", None)
+    if type_tag == "__vector__":
         # Vector data type is represented as mapping.
         # {"__type__":"__vector__", "value": [1.0, 2.0, 3.0]}.
         values = cast(Sequence[float], res["value"])
         return Vector(values)
+    elif "__oid__" in res:
+        return BSONObjectID(res["__oid__"])
+    elif "__decimal128__" in res:
+        return BSONDecimal128(res["__decimal128__"])
+    elif "__int__" in res:
+        return BSONInt32(res["__int__"])
+    elif "__regex__" in res:
+        val = res["__regex__"]
+        return BSONRegex(pattern=val["pattern"], flags=val.get("options", ""))
+    elif "__request_timestamp__" in res:
+        val = res["__request_timestamp__"]
+        return BSONTimestamp(seconds=val["seconds"], increment=val["increment"])
+    elif "__binary__" in res:
+        raw = res["__binary__"]
+        if isinstance(raw, bytes) and len(raw) > 0:
+            return BSONBinary(sub_type=raw[0], data=raw[1:])
+        return BSONBinary(sub_type=0, data=b"")
+    elif "__min__" in res:
+        return BSONMinKey()
+    elif "__max__" in res:
+        return BSONMaxKey()
 
     return res
 
diff --git a/packages/google-cloud-firestore/google/cloud/firestore_v1/bson.py b/packages/google-cloud-firestore/google/cloud/firestore_v1/bson.py
new file mode 100644
index 000000000000..c44b60169057
--- /dev/null
+++ b/packages/google-cloud-firestore/google/cloud/firestore_v1/bson.py
@@ -0,0 +1,201 @@
+# -*- coding: utf-8 -*-
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+"""BSON data type support for Cloud Firestore MongoDB compatibility workloads."""
+
+from __future__ import annotations
+
+from typing import Any, Dict, Union
+
+
+class BSONObjectID:
+    """A class representing BSON ObjectID in Python.
+
+    Stored as a 24-character hexadecimal string.
+    """
+
+    def __init__(self, value: str):
+        if not isinstance(value, str) or len(value) != 24 or not all(c in "0123456789abcdefABCDEF" for c in value):
+            raise ValueError("BSONObjectID must be a 24-character hexadecimal string.")
+        self._value = value.lower()
+
+    @property
+    def value(self) -> str:
+        return self._value
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, BSONObjectID):
+            return False
+        return self._value == other._value
+
+    def __repr__(self) -> str:
+        return f"BSONObjectID('{self._value}')"
+
+    def to_map_value(self) -> Dict[str, Any]:
+        return {"__oid__": self._value}
+
+
+class BSONDecimal128:
+    """A class representing BSON 128-bit Decimal in Python."""
+
+    def __init__(self, value: Union[str, int, float]):
+        self._value = str(value)
+
+    @property
+    def value(self) -> str:
+        return self._value
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, BSONDecimal128):
+            return False
+        return self._value == other._value
+
+    def __repr__(self) -> str:
+        return f"BSONDecimal128('{self._value}')"
+
+    def to_map_value(self) -> Dict[str, Any]:
+        return {"__decimal128__": self._value}
+
+
+class BSONInt32:
+    """A class representing BSON 32-bit Integer in Python."""
+
+    def __init__(self, value: int):
+        self._value = int(value)
+
+    @property
+    def value(self) -> int:
+        return self._value
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, BSONInt32):
+            return False
+        return self._value == other._value
+
+    def __repr__(self) -> str:
+        return f"BSONInt32({self._value})"
+
+    def to_map_value(self) -> Dict[str, Any]:
+        return {"__int__": self._value}
+
+
+class BSONRegex:
+    """A class representing BSON Regular Expression in Python."""
+
+    def __init__(self, pattern: str, flags: str = ""):
+        self._pattern = str(pattern)
+        self._flags = str(flags) if flags is not None else ""
+
+    @property
+    def pattern(self) -> str:
+        return self._pattern
+
+    @property
+    def flags(self) -> str:
+        return self._flags
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, BSONRegex):
+            return False
+        return self._pattern == other._pattern and self._flags == other._flags
+
+    def __repr__(self) -> str:
+        return f"BSONRegex(pattern='{self._pattern}', flags='{self._flags}')"
+
+    def to_map_value(self) -> Dict[str, Any]:
+        return {"__regex__": {"pattern": self._pattern, "options": self._flags}}
+
+
+class BSONTimestamp:
+    """A class representing BSON Timestamp in Python."""
+
+    def __init__(self, seconds: int, increment: int):
+        self._seconds = int(seconds)
+        self._increment = int(increment)
+
+    @property
+    def seconds(self) -> int:
+        return self._seconds
+
+    @property
+    def increment(self) -> int:
+        return self._increment
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, BSONTimestamp):
+            return False
+        return self._seconds == other._seconds and self._increment == other._increment
+
+    def __repr__(self) -> str:
+        return f"BSONTimestamp(seconds={self._seconds}, increment={self._increment})"
+
+    def to_map_value(self) -> Dict[str, Any]:
+        return {"__request_timestamp__": {"seconds": self._seconds, "increment": self._increment}}
+
+
+class BSONBinary:
+    """A class representing BSON Binary data in Python."""
+
+    def __init__(self, sub_type: int, data: bytes):
+        self._sub_type = int(sub_type)
+        self._data = bytes(data)
+
+    @property
+    def sub_type(self) -> int:
+        return self._sub_type
+
+    @property
+    def data(self) -> bytes:
+        return self._data
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, BSONBinary):
+            return False
+        return self._sub_type == other._sub_type and self._data == other._data
+
+    def __repr__(self) -> str:
+        return f"BSONBinary(sub_type={self._sub_type}, data={self._data!r})"
+
+    def to_map_value(self) -> Dict[str, Any]:
+        return {
+            "__binary__": bytes([self._sub_type]) + self._data,
+        }
+
+
+class BSONMinKey:
+    """A class representing BSON MinKey in Python."""
+
+    def __eq__(self, other: object) -> bool:
+        return isinstance(other, BSONMinKey)
+
+    def __repr__(self) -> str:
+        return "BSONMinKey()"
+
+    def to_map_value(self) -> Dict[str, Any]:
+        return {"__min__": None}
+
+
+class BSONMaxKey:
+    """A class representing BSON MaxKey in Python."""
+
+    def __eq__(self, other: object) -> bool:
+        return isinstance(other, BSONMaxKey)
+
+    def __repr__(self) -> str:
+        return "BSONMaxKey()"
+
+    def to_map_value(self) -> Dict[str, Any]:
+        return {"__max__": None}
+
diff --git a/packages/google-cloud-firestore/tests/system/test__helpers.py b/packages/google-cloud-firestore/tests/system/test__helpers.py
index 018eea55bfc1..3334b446f66e 100644
--- a/packages/google-cloud-firestore/tests/system/test__helpers.py
+++ b/packages/google-cloud-firestore/tests/system/test__helpers.py
@@ -20,7 +20,7 @@
 EMULATOR_CREDS = EmulatorCreds()
 FIRESTORE_EMULATOR = os.environ.get(_FIRESTORE_EMULATOR_HOST) is not None
 FIRESTORE_OTHER_DB = os.environ.get("SYSTEM_TESTS_DATABASE", "system-tests-named-db")
-FIRESTORE_ENTERPRISE_DB = os.environ.get("ENTERPRISE_DATABASE", "enterprise-db-native")
+FIRESTORE_ENTERPRISE_DB = os.environ.get("ENTERPRISE_DATABASE", "enterprise-db-native-2")
 
 # To eliminate test duplication, we use the default database for the
 # core test suites. The named database is ONLY tested explicitly in dedicated
diff --git a/packages/google-cloud-firestore/tests/system/test_system.py b/packages/google-cloud-firestore/tests/system/test_system.py
index 1e6ffca06e9a..50a5fc04c6f4 100644
--- a/packages/google-cloud-firestore/tests/system/test_system.py
+++ b/packages/google-cloud-firestore/tests/system/test_system.py
@@ -30,6 +30,16 @@
 )
 from google.cloud._helpers import _datetime_to_pb_timestamp
 from google.oauth2 import service_account
+from google.cloud.firestore import (
+    BSONBinary,
+    BSONDecimal128,
+    BSONInt32,
+    BSONMaxKey,
+    BSONMinKey,
+    BSONObjectID,
+    BSONRegex,
+    BSONTimestamp,
+)
 from test__helpers import (
     EMULATOR_CREDS,
     ENTERPRISE_MODE_ERROR,
@@ -3872,3 +3882,29 @@ def test_large_document_pipeline(client, cleanup, database, method):
 
     results = list(method_under_test())
     assert [doc.data() for doc in results] == [{"payload": large_payload}]
+
+#@pytest.mark.skip(reason="Temporarily skipped. Requires backend BSON / MongoDB feature flag.")
+@pytest.mark.parametrize("database", [FIRESTORE_ENTERPRISE_DB], indirect=True)
+def test_bson_document_writes_and_reads(client, cleanup, database):
+    """Test standard write and read operations for BSON types on Enterprise DB."""
+    collection_id = "bson_docs_" + UNIQUE_RESOURCE_ID
+    doc_ref = client.collection(collection_id).document("bson_doc")
+    cleanup(doc_ref.delete)
+
+    bson_payload = {
+        "_id": BSONObjectID("507f191e810c19729de860ea"),
+        "price": BSONDecimal128("199.99"),
+        "qty": BSONInt32(50),
+        "pattern": BSONRegex(pattern="^prod.*", flags="i"),
+        "ts": BSONTimestamp(seconds=1710000000, increment=2),
+        "binary_data": BSONBinary(sub_type=1, data=b"binary_payload"),
+        "min_key": BSONMinKey(),
+        "max_key": BSONMaxKey(),
+    }
+
+    doc_ref.set(bson_payload)
+
+    snapshot = doc_ref.get()
+    assert snapshot.exists
+    assert snapshot.to_dict() == bson_payload
+
diff --git a/packages/google-cloud-firestore/tests/system/test_system_async.py b/packages/google-cloud-firestore/tests/system/test_system_async.py
index ac2451043cef..e658c4e6eecf 100644
--- a/packages/google-cloud-firestore/tests/system/test_system_async.py
+++ b/packages/google-cloud-firestore/tests/system/test_system_async.py
@@ -33,6 +33,16 @@
 )
 from google.cloud._helpers import _datetime_to_pb_timestamp
 from google.oauth2 import service_account
+from google.cloud.firestore import (
+    BSONBinary,
+    BSONDecimal128,
+    BSONInt32,
+    BSONMaxKey,
+    BSONMinKey,
+    BSONObjectID,
+    BSONRegex,
+    BSONTimestamp,
+)
 from test__helpers import (
     EMULATOR_CREDS,
     ENTERPRISE_MODE_ERROR,
@@ -3746,3 +3756,29 @@ async def test_large_document_pipeline_async(client, cleanup, database, method):
         results = [doc async for doc in pipeline.stream()]
 
     assert [doc.data() for doc in results] == [{"payload": large_payload}]
+
+
+# @pytest.mark.skip(reason="Temporarily skipped. Requires backend BSON / MongoDB feature flag.")
+@pytest.mark.parametrize("database", [FIRESTORE_ENTERPRISE_DB], indirect=True)
+async def test_bson_document_writes_and_reads_async(client, cleanup, database):
+    """Test async write and read operations for BSON types on Enterprise DB."""
+    collection_id = "bson_docs_async_" + UNIQUE_RESOURCE_ID
+    doc_ref = client.collection(collection_id).document("bson_doc")
+    cleanup(doc_ref.delete)
+
+    bson_payload = {
+        "_id": BSONObjectID("507f191e810c19729de860ea"),
+        "price": BSONDecimal128("299.99"),
+        "qty": BSONInt32(100),
+        "pattern": BSONRegex(pattern="^async.*", flags="m"),
+        "ts": BSONTimestamp(seconds=1720000000, increment=1),
+        "binary_data": BSONBinary(sub_type=1, data=b"async_binary_payload"),
+        "min_key": BSONMinKey(),
+        "max_key": BSONMaxKey(),
+    }
+
+    await doc_ref.set(bson_payload)
+
+    snapshot = await doc_ref.get()
+    assert snapshot.exists
+    assert snapshot.to_dict() == bson_payload
diff --git a/packages/google-cloud-firestore/tests/unit/v1/test_bson.py b/packages/google-cloud-firestore/tests/unit/v1/test_bson.py
new file mode 100644
index 000000000000..ea27087b9d79
--- /dev/null
+++ b/packages/google-cloud-firestore/tests/unit/v1/test_bson.py
@@ -0,0 +1,143 @@
+# -*- coding: utf-8 -*-
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+import pytest
+
+from google.cloud.firestore_v1._helpers import decode_dict, decode_value, encode_dict, encode_value
+from google.cloud.firestore_v1.bson import (
+    BSONBinary,
+    BSONDecimal128,
+    BSONInt32,
+    BSONMaxKey,
+    BSONMinKey,
+    BSONObjectID,
+    BSONRegex,
+    BSONTimestamp,
+)
+
+
+def test_bson_object_id():
+    oid_str = "507f191e810c19729de860ea"
+    oid = BSONObjectID(oid_str)
+    assert oid.value == oid_str
+    assert repr(oid) == f"BSONObjectID('{oid_str}')"
+    assert oid == BSONObjectID(oid_str)
+
+    # Invalid hex string length or characters
+    with pytest.raises(ValueError):
+        BSONObjectID("invalid")
+    with pytest.raises(ValueError):
+        BSONObjectID("507f191e810c19729de860eg")  # 'g' is invalid hex
+
+    # Encoding / Decoding roundtrip
+    encoded = encode_value(oid)
+    decoded = decode_value(encoded, client=None)
+    assert decoded == oid
+
+
+def test_bson_decimal128():
+    dec_val = "123.45678901234567890"
+    dec = BSONDecimal128(dec_val)
+    assert dec.value == dec_val
+    assert repr(dec) == f"BSONDecimal128('{dec_val}')"
+    assert dec == BSONDecimal128(dec_val)
+
+    encoded = encode_value(dec)
+    decoded = decode_value(encoded, client=None)
+    assert decoded == dec
+
+
+def test_bson_int32():
+    val = 42
+    b_int = BSONInt32(val)
+    assert b_int.value == val
+    assert repr(b_int) == "BSONInt32(42)"
+    assert b_int == BSONInt32(val)
+
+    encoded = encode_value(b_int)
+    decoded = decode_value(encoded, client=None)
+    assert decoded == b_int
+
+
+def test_bson_regex():
+    reg = BSONRegex(pattern="^test.*", flags="i")
+    assert reg.pattern == "^test.*"
+    assert reg.flags == "i"
+    assert repr(reg) == "BSONRegex(pattern='^test.*', flags='i')"
+    assert reg == BSONRegex(pattern="^test.*", flags="i")
+
+    encoded = encode_value(reg)
+    decoded = decode_value(encoded, client=None)
+    assert decoded == reg
+
+
+def test_bson_timestamp():
+    ts = BSONTimestamp(seconds=1600000000, increment=1)
+    assert ts.seconds == 1600000000
+    assert ts.increment == 1
+    assert repr(ts) == "BSONTimestamp(seconds=1600000000, increment=1)"
+    assert ts == BSONTimestamp(seconds=1600000000, increment=1)
+
+    encoded = encode_value(ts)
+    decoded = decode_value(encoded, client=None)
+    assert decoded == ts
+
+
+def test_bson_binary():
+    bin_data = BSONBinary(sub_type=128, data=b"\x01\x02\x03\x04")
+    assert bin_data.sub_type == 128
+    assert bin_data.data == b"\x01\x02\x03\x04"
+    assert repr(bin_data) == "BSONBinary(sub_type=128, data=b'\\x01\\x02\\x03\\x04')"
+    assert bin_data == BSONBinary(sub_type=128, data=b"\x01\x02\x03\x04")
+
+    encoded = encode_value(bin_data)
+    decoded = decode_value(encoded, client=None)
+    assert decoded == bin_data
+
+
+def test_bson_min_key():
+    min_key = BSONMinKey()
+    assert repr(min_key) == "BSONMinKey()"
+    assert min_key == BSONMinKey()
+
+    encoded = encode_value(min_key)
+    decoded = decode_value(encoded, client=None)
+    assert decoded == min_key
+
+
+def test_bson_max_key():
+    max_key = BSONMaxKey()
+    assert repr(max_key) == "BSONMaxKey()"
+    assert max_key == BSONMaxKey()
+
+    encoded = encode_value(max_key)
+    decoded = decode_value(encoded, client=None)
+    assert decoded == max_key
+
+
+def test_bson_nested_dict_encoding_decoding():
+    data = {
+        "_id": BSONObjectID("507f191e810c19729de860ea"),
+        "price": BSONDecimal128("99.99"),
+        "quantity": BSONInt32(10),
+        "regex": BSONRegex(pattern="abc", flags="m"),
+        "ts": BSONTimestamp(seconds=1700000000, increment=5),
+        "bin": BSONBinary(sub_type=0, data=b"hello"),
+        "min": BSONMinKey(),
+        "max": BSONMaxKey(),
+    }
+
+    encoded_dict = encode_dict(data)
+    decoded_dict = decode_dict(encoded_dict, client=None)
+    assert decoded_dict == data
+