diff --git a/src/firebase_functions/identity_fn.py b/src/firebase_functions/identity_fn.py
index d2afead..0246054 100644
--- a/src/firebase_functions/identity_fn.py
+++ b/src/firebase_functions/identity_fn.py
@@ -11,7 +11,7 @@
 # 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.
-"""Cloud functions to handle Eventarc events."""
+"""Cloud functions to handle Firebase Authentication and Identity Platform events."""
 
 # pylint: disable=protected-access,cyclic-import
 import dataclasses as _dataclasses
@@ -20,6 +20,7 @@
 import typing as _typing
 from enum import Enum
 
+import cloudevents.http as _ce
 from flask import (
     Request as _Request,
 )
@@ -27,8 +28,10 @@
     Response as _Response,
 )
 
+import firebase_functions.core as _core
 import firebase_functions.options as _options
 import firebase_functions.private.util as _util
+from firebase_functions.options import AuthOptions  # noqa: F401
 
 
 @_dataclasses.dataclass(frozen=True)
@@ -116,7 +119,7 @@ class AuthMultiFactorSettings:
 @_dataclasses.dataclass(frozen=True)
 class AuthUserRecord:
     """
-    The UserRecord passed to auth blocking functions from the identity platform.
+    The UserRecord passed to auth blocking and event functions from the identity platform.
     """
 
     uid: str
@@ -472,3 +475,120 @@ def before_user_created_wrapped(request: _Request) -> _Response:
         return before_user_created_wrapped
 
     return before_user_created_decorator
+
+
+@_dataclasses.dataclass(frozen=True)
+class AuthEvent(_core.CloudEvent[AuthUserRecord]):
+    """
+    A CloudEvent that contains an AuthUserRecord when a user is created or deleted.
+    """
+
+    project: str | None = None
+    """
+    The project identifier.
+    """
+
+    tenant_id: str | None = None
+    """
+    The ID of the Identity Platform tenant associated with the event, if applicable.
+    """
+
+
+AuthEventCallable = _typing.Callable[[AuthEvent], None]
+"""
+The type of the callable for 'on_user_created' and 'on_user_deleted' Auth events.
+"""
+
+
+@_util.copy_func_kwargs(_options.AuthOptions)
+def on_user_created(
+    **kwargs,
+) -> _typing.Callable[[AuthEventCallable], AuthEventCallable]:
+    """
+    Handles an event that is triggered when a user is created in Firebase Authentication.
+
+    Example:
+
+    .. code-block:: python
+
+      from firebase_functions import identity_fn
+
+      @identity_fn.on_user_created()
+      def example(event: identity_fn.AuthEvent) -> None:
+          pass
+
+    :param \\*\\*kwargs: Options.
+    :type \\*\\*kwargs: as :exc:`firebase_functions.options.AuthOptions`
+    :rtype: :exc:`typing.Callable`
+            \\[ \\[ :exc:`firebase_functions.identity_fn.AuthEvent` \\], `None` \\]
+            A function that takes an AuthEvent and returns ``None``.
+    """
+    options = _options.AuthOptions(**kwargs)
+
+    def on_user_created_decorator(func: AuthEventCallable):
+        from firebase_functions.private._identity_fn import (
+            event_type_user_created,
+            user_event_handler,
+        )
+
+        @_functools.wraps(func)
+        def on_user_created_wrapped(raw: _ce.CloudEvent):
+            return user_event_handler(func, raw)
+
+        _util.set_func_endpoint_attr(
+            on_user_created_wrapped,
+            options._endpoint(
+                func_name=func.__name__,
+                event_type=event_type_user_created,
+            ),
+        )
+        return on_user_created_wrapped
+
+    return on_user_created_decorator
+
+
+@_util.copy_func_kwargs(_options.AuthOptions)
+def on_user_deleted(
+    **kwargs,
+) -> _typing.Callable[[AuthEventCallable], AuthEventCallable]:
+    """
+    Handles an event that is triggered when a user is deleted in Firebase Authentication.
+
+    Example:
+
+    .. code-block:: python
+
+      from firebase_functions import identity_fn
+
+      @identity_fn.on_user_deleted()
+      def example(event: identity_fn.AuthEvent) -> None:
+          pass
+
+    :param \\*\\*kwargs: Options.
+    :type \\*\\*kwargs: as :exc:`firebase_functions.options.AuthOptions`
+    :rtype: :exc:`typing.Callable`
+            \\[ \\[ :exc:`firebase_functions.identity_fn.AuthEvent` \\], `None` \\]
+            A function that takes an AuthEvent and returns ``None``.
+    """
+    options = _options.AuthOptions(**kwargs)
+
+    def on_user_deleted_decorator(func: AuthEventCallable):
+        from firebase_functions.private._identity_fn import (
+            event_type_user_deleted,
+            user_event_handler,
+        )
+
+        @_functools.wraps(func)
+        def on_user_deleted_wrapped(raw: _ce.CloudEvent):
+            return user_event_handler(func, raw)
+
+        _util.set_func_endpoint_attr(
+            on_user_deleted_wrapped,
+            options._endpoint(
+                func_name=func.__name__,
+                event_type=event_type_user_deleted,
+            ),
+        )
+        return on_user_deleted_wrapped
+
+    return on_user_deleted_decorator
diff --git a/src/firebase_functions/options.py b/src/firebase_functions/options.py
index b4c0dff..2e2786b 100644
--- a/src/firebase_functions/options.py
+++ b/src/firebase_functions/options.py
@@ -1055,6 +1055,38 @@ def _required_apis(self) -> list[_manifest.ManifestRequiredApi]:
         ]
 
 
+@_dataclasses.dataclass(frozen=True, kw_only=True)
+class AuthOptions(EventHandlerOptions):
+    """
+    Options that can be set on an Auth event trigger.
+    Internal use only.
+    """
+
+    tenant_id: str | Expression[str] | None = None
+    """
+    An optional tenant ID to scope the trigger to.
+    """
+
+    def _endpoint(
+        self,
+        **kwargs,
+    ) -> _manifest.ManifestEndpoint:
+        assert kwargs["event_type"] is not None
+        event_filters: dict[str, str | Expression[str]] = {}
+        if self.tenant_id is not None:
+            event_filters["tenantid"] = self.tenant_id
+
+        endpoint = super()._endpoint(
+            **kwargs,
+            event_filters=event_filters,
+        )
+        assert endpoint.eventTrigger is not None
+        # Firebase Auth event triggers are global by nature, so Eventarc requires
+        # the trigger region to be "global" even when the Cloud Function is regional.
+        endpoint.eventTrigger["region"] = "global"
+        return endpoint
+
+
 @_dataclasses.dataclass(frozen=True, kw_only=True)
 class FirestoreOptions(RuntimeOptions):
     """
diff --git a/src/firebase_functions/private/_identity_fn.py b/src/firebase_functions/private/_identity_fn.py
index 0e09c04..1699748 100644
--- a/src/firebase_functions/private/_identity_fn.py
+++ b/src/firebase_functions/private/_identity_fn.py
@@ -11,14 +11,16 @@
 # 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.
-"""Cloud functions to handle Eventarc events."""
+"""Cloud functions to handle Firebase Authentication and Identity Platform events."""
 
 # pylint: disable=protected-access
 import datetime as _dt
 import json as _json
+import re as _re
 import time as _time
 import typing as _typing
 
+import cloudevents.http as _ce
 from flask import (
     Request as _Request,
 )
@@ -369,3 +371,250 @@ def before_operation_handler(
             exception = HttpsError(FunctionsErrorCode.INTERNAL, "INTERNAL")
         status = exception._http_error_code.status
         return _make_response(_jsonify(error=exception._as_dict()), status)
+
+
+event_type_user_created = "google.firebase.auth.user.v2.created"
+event_type_user_deleted = "google.firebase.auth.user.v2.deleted"
+# Matches both absolute (/projects/...) and relative (projects/...) CloudEvent source paths.
+_PROJECT_ID_REGEX = _re.compile(r"(?:^|/)projects/([^/]+)")
+_EPOCH_START_UTC = _dt.datetime.fromtimestamp(0, tz=_dt.timezone.utc)
+
+
+def _ensure_utc(value: _dt.datetime) -> _dt.datetime:
+    """Normalizes a naive or offset-aware datetime to timezone-aware UTC."""
+    if value.tzinfo is None:
+        return value.replace(tzinfo=_dt.timezone.utc)
+    return value.astimezone(_dt.timezone.utc)
+
+
+def _parse_auth_timestamp(value: _typing.Any) -> _dt.datetime | None:
+    """
+    Normalizes Auth event timestamps to timezone-aware UTC datetimes.
+    Supports epoch milliseconds (int, float, or numeric str) and ISO-8601 strings.
+    """
+    if isinstance(value, _dt.datetime):
+        return _ensure_utc(value)
+    if isinstance(value, str) and value.isdigit():
+        value = int(value)
+    # Guard against bool because bool is a subclass of int in Python.
+    if isinstance(value, int | float) and not isinstance(value, bool):
+        try:
+            return _dt.datetime.fromtimestamp(value / 1000.0, tz=_dt.timezone.utc)
+        except (OverflowError, OSError, ValueError):
+            return None
+    if isinstance(value, str) and value:
+        for parser in (_util.timestamp_conversion, _dt.datetime.fromisoformat):
+            try:
+                return _ensure_utc(parser(value))
+            except ValueError:
+                continue
+    return None
+
+
+def _auth_user_info_from_ce_data(info_data: dict[str, _typing.Any]):
+    """
+    Converts a single entry from the CloudEvent 'providerData' array into an AuthUserInfo.
+    """
+    from firebase_functions.identity_fn import AuthUserInfo
+
+    return AuthUserInfo(
+        uid=info_data.get("uid", ""),
+        provider_id=info_data.get("providerId", ""),
+        display_name=info_data.get("displayName"),
+        email=info_data.get("email"),
+        # Auth Emulator sends 'photoURL'; production Eventarc sends 'photoUrl'.
+        photo_url=info_data.get("photoURL") or info_data.get("photoUrl"),
+        phone_number=info_data.get("phoneNumber"),
+    )
+
+
+def _auth_user_metadata_from_ce_data(meta_data: dict[str, _typing.Any] | None):
+    """
+    Converts the CloudEvent 'metadata' dict into an AuthUserMetadata object.
+    """
+    from firebase_functions.identity_fn import AuthUserMetadata
+
+    if not meta_data or not isinstance(meta_data, dict):
+        # Default creation_time to UTC epoch 0 when metadata is omitted (e.g. partial
+        # deletion payloads) so AuthUserMetadata.creation_time remains non-optional.
+        return AuthUserMetadata(creation_time=_EPOCH_START_UTC, last_sign_in_time=None)
+
+    # Production Eventarc sends 'createTime'; Auth Emulator sends 'creationTime'.
+    raw_creation = meta_data.get("createTime") or meta_data.get("creationTime")
+    raw_last_sign_in = meta_data.get("lastSignInTime")
+    return AuthUserMetadata(
+        creation_time=_parse_auth_timestamp(raw_creation) or _EPOCH_START_UTC,
+        last_sign_in_time=_parse_auth_timestamp(raw_last_sign_in),
+    )
+
+
+def _auth_multi_factor_info_from_ce_data(factor_data: dict[str, _typing.Any]):
+    """
+    Converts a single enrolled MFA factor dict into an AuthMultiFactorInfo object.
+    """
+    from firebase_functions.identity_fn import AuthMultiFactorInfo
+
+    raw_enrollment_time = factor_data.get("enrollmentTime")
+    phone_number = factor_data.get("phoneNumber")
+    raw_factor_id = factor_data.get("factorId", "")
+    # Default SMS factors that omit an explicit factorId to "phone".
+    factor_id = raw_factor_id or ("phone" if phone_number else "")
+    return AuthMultiFactorInfo(
+        uid=factor_data.get("uid", ""),
+        display_name=factor_data.get("displayName"),
+        factor_id=factor_id,
+        enrollment_time=_parse_auth_timestamp(raw_enrollment_time),
+        phone_number=phone_number,
+    )
+
+
+def _auth_multi_factor_settings_from_ce_data(mfa_data: dict[str, _typing.Any] | None):
+    """
+    Converts the CloudEvent 'multiFactor' dict into AuthMultiFactorSettings,
+    returning None when no second factors are enrolled.
+    """
+    if not mfa_data or not isinstance(mfa_data, dict):
+        return None
+
+    from firebase_functions.identity_fn import AuthMultiFactorSettings
+
+    raw_factors = mfa_data.get("enrolledFactors")
+    if not isinstance(raw_factors, list):
+        raw_factors = []
+    enrolled_factors = [
+        _auth_multi_factor_info_from_ce_data(factor)
+        for factor in raw_factors
+        if isinstance(factor, dict)
+    ]
+    if not enrolled_factors:
+        return None
+    return AuthMultiFactorSettings(enrolled_factors=enrolled_factors)
+
+
+def _unwrap_auth_event_data(raw_data: _typing.Any) -> dict[str, _typing.Any]:
+    """
+    Extracts the inner user dictionary from an Eventarc AuthEventData payload.
+    Raises ValueError if the payload is missing, null, or does not contain a user dictionary.
+    """
+    if not raw_data or not isinstance(raw_data, dict):
+        raise ValueError("Auth event data payload must be a non-empty dictionary.")
+    # 'on_user_created' wraps the user in 'value'; 'on_user_deleted' wraps it in 'oldValue'.
+    # Check isinstance(..., dict) so {'value': None, 'oldValue': {...}} does not shadow 'oldValue'.
+    for envelope_key in ("value", "oldValue"):
+        if isinstance(raw_data.get(envelope_key), dict) and raw_data[envelope_key]:
+            return raw_data[envelope_key]
+    # If either envelope key is present with a non-dict (e.g. None) or empty dict, reject it.
+    if any(k in raw_data for k in ("value", "oldValue")):
+        raise ValueError("Auth event data envelope contains an empty or null user record.")
+    # Otherwise, assume raw_data is already the top-level user dict (e.g. synthetic tests).
+    return raw_data
+
+
+def _parse_custom_claims(user_data: dict[str, _typing.Any]) -> dict[str, _typing.Any] | None:
+    """
+    Extracts customClaims from a user dict, supporting both pre-parsed dicts
+    and serialized JSON object strings.
+    """
+    raw_claims = user_data.get("customClaims")
+    # Eventarc and Auth Emulator send a parsed dict; Identity Toolkit REST responses
+    # sometimes serialize custom claims as a JSON string.
+    if isinstance(raw_claims, dict):
+        return raw_claims
+    if isinstance(raw_claims, str) and raw_claims:
+        try:
+            parsed_claims = _json.loads(raw_claims)
+            if isinstance(parsed_claims, dict):
+                return parsed_claims
+        except _json.JSONDecodeError:
+            return None
+    return None
+
+
+def _auth_user_record_from_ce_data(raw_data: _typing.Any):
+    """
+    Unpacks and normalizes an AuthUserRecord from a v2 Eventarc CloudEvent data payload.
+    """
+    from firebase_functions.identity_fn import AuthUserRecord
+
+    user_data = _unwrap_auth_event_data(raw_data)
+    uid = user_data.get("uid")
+    if not isinstance(uid, str) or not uid:
+        raise ValueError("Auth event user record is missing a valid 'uid'.")
+
+    raw_provider_data = user_data.get("providerData")
+    if not isinstance(raw_provider_data, list):
+        raw_provider_data = []
+    provider_data = [
+        _auth_user_info_from_ce_data(info) for info in raw_provider_data if isinstance(info, dict)
+    ]
+
+    return AuthUserRecord(
+        uid=uid,
+        email=user_data.get("email"),
+        email_verified=bool(user_data.get("emailVerified", False)),
+        display_name=user_data.get("displayName"),
+        # Auth Emulator sends 'photoURL'; production Eventarc sends 'photoUrl'.
+        photo_url=user_data.get("photoURL") or user_data.get("photoUrl"),
+        phone_number=user_data.get("phoneNumber"),
+        disabled=bool(user_data.get("disabled", False)),
+        metadata=_auth_user_metadata_from_ce_data(user_data.get("metadata")),
+        provider_data=provider_data,
+        password_hash=user_data.get("passwordHash"),
+        password_salt=user_data.get("passwordSalt"),
+        custom_claims=_parse_custom_claims(user_data),
+        tenant_id=user_data.get("tenantId") or None,
+        tokens_valid_after_time=_parse_auth_timestamp(user_data.get("tokensValidAfterTime")),
+        multi_factor=_auth_multi_factor_settings_from_ce_data(user_data.get("multiFactor")),
+    )
+
+
+def _auth_event_from_ce(raw: _ce.CloudEvent):
+    """
+    Constructs a strongly-typed AuthEvent from a raw Eventarc CloudEvent,
+    extracting the project ID, tenant ID, and unpacked AuthUserRecord.
+    """
+    from firebase_functions.identity_fn import AuthEvent
+
+    event_attributes = raw._get_attributes()
+    event_data: _typing.Any = raw.get_data()
+    user_record = _auth_user_record_from_ce_data(event_data)
+
+    # Extract project ID from the CloudEvent source attribute
+    # (e.g. "//identitytoolkit.googleapis.com/projects/my-project").
+    source = event_attributes.get("source", "")
+    project: str | None = None
+    if source:
+        match = _PROJECT_ID_REGEX.search(source)
+        if match:
+            project = match.group(1)
+
+    # Check the CloudEvent 'tenantid' attribute (Eventarc standard), falling back
+    # to the user record's tenant_id (used by the Auth Emulator).
+    tenant_id: str | None = event_attributes.get("tenantid") or user_record.tenant_id or None
+
+    raw_time = event_attributes.get("time")
+    event_time = _parse_auth_timestamp(raw_time) or _dt.datetime.now(_dt.timezone.utc)
+
+    return AuthEvent(
+        specversion=event_attributes["specversion"],
+        id=event_attributes["id"],
+        source=source,
+        type=event_attributes["type"],
+        time=event_time,
+        data=user_record,
+        subject=event_attributes.get("subject"),
+        project=project,
+        tenant_id=tenant_id,
+    )
+
+
+def user_event_handler(
+    func: _typing.Callable,
+    raw: _ce.CloudEvent,
+) -> None:
+    """
+    Runtime handler invoked by functions-framework for v2 Auth triggers
+    (on_user_created and on_user_deleted).
+    """
+    event = _auth_event_from_ce(raw)
+    _with_init(func)(event)
diff --git a/src/firebase_functions/private/manifest.py b/src/firebase_functions/private/manifest.py
index 7672a9f..a482610 100644
--- a/src/firebase_functions/private/manifest.py
+++ b/src/firebase_functions/private/manifest.py
@@ -64,6 +64,7 @@ class EventTrigger(_typing.TypedDict):
     channel: _typing_extensions.NotRequired[str]
     eventType: _typing_extensions.Required[str]
     retry: _typing_extensions.Required[bool | _params.Expression[bool] | _util.Sentinel]
+    region: _typing_extensions.NotRequired[str]
 
 
 class RetryConfigBase(_typing.TypedDict):
diff --git a/tests/test_identity_fn.py b/tests/test_identity_fn.py
index 19421a3..4ba4f0f 100644
--- a/tests/test_identity_fn.py
+++ b/tests/test_identity_fn.py
@@ -2,14 +2,17 @@
 Identity function tests.
 """
 
+import datetime as dt
 import unittest
 from unittest.mock import MagicMock, Mock, patch
 
+import yaml
+from cloudevents.http import CloudEvent
 from flask import Flask, Request
 from werkzeug.test import EnvironBuilder
 
-from firebase_functions import core, identity_fn
-from firebase_functions.private import _identity_fn
+from firebase_functions import core, identity_fn, params
+from firebase_functions.private import _identity_fn, serving
 
 token_verifier_mock = MagicMock()
 token_verifier_mock.verify_auth_blocking_token = Mock(
@@ -28,6 +31,10 @@ class TestIdentity(unittest.TestCase):
     Identity function tests.
     """
 
+    def tearDown(self) -> None:
+        core._init_callback = None
+        core._did_init = False
+
     def test_calls_init_function(self):
         hello = None
 
@@ -53,3 +60,269 @@ def init():
                 decorated_func(request)
 
         self.assertEqual("world", hello)
+
+    def test_on_user_created_endpoint_metadata(self) -> None:
+        @identity_fn.on_user_created()
+        def example(_event: identity_fn.AuthEvent) -> None:
+            pass
+
+        endpoint = getattr(example, "__firebase_endpoint__", None)
+        self.assertIsNotNone(endpoint)
+        assert endpoint is not None
+        self.assertEqual(endpoint.entryPoint, "example")
+        self.assertEqual(
+            endpoint.eventTrigger,
+            {
+                "eventType": "google.firebase.auth.user.v2.created",
+                "eventFilters": {},
+                "retry": False,
+                "region": "global",
+            },
+        )
+
+    def test_on_user_deleted_endpoint_metadata(self) -> None:
+        @identity_fn.on_user_deleted()
+        def example(_event: identity_fn.AuthEvent) -> None:
+            pass
+
+        endpoint = getattr(example, "__firebase_endpoint__", None)
+        self.assertIsNotNone(endpoint)
+        assert endpoint is not None
+        self.assertEqual(endpoint.entryPoint, "example")
+        self.assertEqual(
+            endpoint.eventTrigger,
+            {
+                "eventType": "google.firebase.auth.user.v2.deleted",
+                "eventFilters": {},
+                "retry": False,
+                "region": "global",
+            },
+        )
+
+    def test_on_user_created_tenant_id_filter(self) -> None:
+        @identity_fn.on_user_created(tenant_id="my-tenant")
+        def example(_event: identity_fn.AuthEvent) -> None:
+            pass
+
+        endpoint = getattr(example, "__firebase_endpoint__", None)
+        assert endpoint is not None
+        self.assertEqual(endpoint.eventTrigger["eventFilters"], {"tenantid": "my-tenant"})
+
+    def test_on_user_created_tenant_id_expression(self) -> None:
+        tenant_param = params.StringParam("MY_TENANT")
+
+        @identity_fn.on_user_created(tenant_id=tenant_param)
+        def example(_event: identity_fn.AuthEvent) -> None:
+            pass
+
+        endpoint = getattr(example, "__firebase_endpoint__", None)
+        assert endpoint is not None
+        self.assertEqual(endpoint.eventTrigger["eventFilters"], {"tenantid": tenant_param})
+
+        rendered_yaml = yaml.safe_load(serving.functions_as_yaml({"example": example}))
+        self.assertEqual(
+            rendered_yaml["endpoints"]["example"]["eventTrigger"]["eventFilters"],
+            {"tenantid": "{{ params.MY_TENANT }}"},
+        )
+
+    def test_on_user_created_unpacks_eventarc_value_envelope(self) -> None:
+        captured_event: identity_fn.AuthEvent | None = None
+
+        @identity_fn.on_user_created()
+        def example(event: identity_fn.AuthEvent) -> None:
+            nonlocal captured_event
+            captured_event = event
+
+        raw_ce = CloudEvent(
+            attributes={
+                "specversion": "1.0",
+                "id": "event-id-123",
+                "source": "//identitytoolkit.googleapis.com/projects/my-project",
+                "type": "google.firebase.auth.user.v2.created",
+                "time": "2023-01-01T00:00:00Z",
+                "tenantid": "my-tenant",
+            },
+            data={
+                "value": {
+                    "uid": "1qEveruhwnbiG1B9taHIBMmOVE83",
+                    "email": "testuser@example.com",
+                    "emailVerified": True,
+                    "displayName": "Test User",
+                    "photoUrl": "http://example.com/photo.jpg",
+                    "phoneNumber": "+15555550100",
+                    "disabled": False,
+                    "metadata": {
+                        "createTime": "2023-01-01T00:00:00Z",
+                        "lastSignInTime": "2023-01-02T12:30:00Z",
+                    },
+                    "providerData": [
+                        {
+                            "uid": "google-uid",
+                            "providerId": "google.com",
+                            "displayName": "Google User",
+                            "email": "testuser@example.com",
+                            "photoUrl": "http://example.com/google.jpg",
+                        }
+                    ],
+                    "customClaims": {"admin": True},
+                    "tenantId": "my-tenant",
+                }
+            },
+        )
+
+        example(raw_ce)  # type: ignore[arg-type]
+        self.assertIsNotNone(captured_event)
+        assert captured_event is not None
+        self.assertEqual(captured_event.project, "my-project")
+        self.assertEqual(captured_event.tenant_id, "my-tenant")
+        self.assertIsNotNone(captured_event.data)
+        assert captured_event.data is not None
+        self.assertEqual(captured_event.data.uid, "1qEveruhwnbiG1B9taHIBMmOVE83")
+        self.assertEqual(captured_event.data.email, "testuser@example.com")
+        self.assertTrue(captured_event.data.email_verified)
+        self.assertEqual(captured_event.data.display_name, "Test User")
+        self.assertEqual(captured_event.data.photo_url, "http://example.com/photo.jpg")
+        self.assertEqual(captured_event.data.phone_number, "+15555550100")
+        self.assertFalse(captured_event.data.disabled)
+        self.assertEqual(
+            captured_event.data.metadata.creation_time,
+            dt.datetime(2023, 1, 1, 0, 0, 0, tzinfo=dt.timezone.utc),
+        )
+        self.assertEqual(
+            captured_event.data.metadata.last_sign_in_time,
+            dt.datetime(2023, 1, 2, 12, 30, 0, tzinfo=dt.timezone.utc),
+        )
+        self.assertEqual(len(captured_event.data.provider_data), 1)
+        self.assertEqual(captured_event.data.provider_data[0].provider_id, "google.com")
+        self.assertEqual(captured_event.data.custom_claims, {"admin": True})
+        self.assertEqual(captured_event.data.tenant_id, "my-tenant")
+
+    def test_on_user_deleted_unpacks_old_value_envelope(self) -> None:
+        captured_event: identity_fn.AuthEvent | None = None
+
+        @identity_fn.on_user_deleted()
+        def example(event: identity_fn.AuthEvent) -> None:
+            nonlocal captured_event
+            captured_event = event
+
+        raw_ce = CloudEvent(
+            attributes={
+                "specversion": "1.0",
+                "id": "event-id-del",
+                "source": "//identitytoolkit.googleapis.com/projects/my-project",
+                "type": "google.firebase.auth.user.v2.deleted",
+                "time": "2023-01-01T00:00:00Z",
+            },
+            data={
+                "value": None,
+                "oldValue": {
+                    "uid": "deleted-user-uid",
+                    "email": "deleted@example.com",
+                    "photoURL": "http://example.com/legacy-photo.jpg",
+                },
+            },
+        )
+
+        example(raw_ce)  # type: ignore[arg-type]
+        self.assertIsNotNone(captured_event)
+        assert captured_event is not None
+        self.assertEqual(captured_event.project, "my-project")
+        self.assertIsNone(captured_event.tenant_id)
+        assert captured_event.data is not None
+        self.assertEqual(captured_event.data.uid, "deleted-user-uid")
+        self.assertEqual(captured_event.data.email, "deleted@example.com")
+        self.assertEqual(captured_event.data.photo_url, "http://example.com/legacy-photo.jpg")
+
+    def test_raises_on_null_or_unrecognized_data(self) -> None:
+        captured_events: list[identity_fn.AuthEvent] = []
+
+        @identity_fn.on_user_created()
+        def example(event: identity_fn.AuthEvent) -> None:
+            captured_events.append(event)
+
+        invalid_payloads: list[object] = [
+            None,
+            {},
+            {"value": None},
+            {"value": {}},
+            {"unrecognized": "payload"},
+            {"value": {"email": "missing-uid@example.com"}},
+        ]
+        for invalid_data in invalid_payloads:
+            raw_invalid_ce = CloudEvent(
+                attributes={
+                    "specversion": "1.0",
+                    "id": "event-invalid",
+                    "source": "//identitytoolkit.googleapis.com/projects/my-project",
+                    "type": "google.firebase.auth.user.v2.created",
+                    "time": "2023-01-01T00:00:00Z",
+                },
+                data=invalid_data,
+            )
+            with self.assertRaises(ValueError):
+                example(raw_invalid_ce)  # type: ignore[arg-type]
+
+        self.assertEqual(len(captured_events), 0)
+
+    def test_parse_auth_timestamp_handles_out_of_range_numbers(self) -> None:
+        self.assertIsNone(_identity_fn._parse_auth_timestamp(1e20))
+        self.assertIsNone(_identity_fn._parse_auth_timestamp("99999999999999999999"))
+        self.assertIsNone(_identity_fn._parse_auth_timestamp(float("inf")))
+        self.assertIsNone(_identity_fn._parse_auth_timestamp(float("nan")))
+
+    def test_on_user_created_mfa_and_string_custom_claims(self) -> None:
+        captured_event: identity_fn.AuthEvent | None = None
+
+        @identity_fn.on_user_created()
+        def example(event: identity_fn.AuthEvent) -> None:
+            nonlocal captured_event
+            captured_event = event
+
+        raw_ce = CloudEvent(
+            attributes={
+                "specversion": "1.0",
+                "id": "event-mfa",
+                "source": "//identitytoolkit.googleapis.com/projects/my-project",
+                "type": "google.firebase.auth.user.v2.created",
+                "time": "2023-01-01T02:00:00+02:00",
+            },
+            data={
+                "value": {
+                    "uid": "mfa-user",
+                    "customClaims": '{"role": "editor"}',
+                    "multiFactor": {
+                        "enrolledFactors": [
+                            {
+                                "uid": "factor-1",
+                                "displayName": "Personal Phone",
+                                "phoneNumber": "+15555550199",
+                                "enrollmentTime": "2023-01-01T00:00:00",
+                            },
+                            {
+                                "uid": "factor-2",
+                                "displayName": "Authenticator App",
+                                "factorId": "totp",
+                            },
+                        ]
+                    },
+                }
+            },
+        )
+
+        example(raw_ce)  # type: ignore[arg-type]
+        assert captured_event is not None
+        self.assertEqual(
+            captured_event.time,
+            dt.datetime(2023, 1, 1, 0, 0, 0, tzinfo=dt.timezone.utc),
+        )
+        self.assertEqual(captured_event.data.custom_claims, {"role": "editor"})
+        assert captured_event.data.multi_factor is not None
+        factors = captured_event.data.multi_factor.enrolled_factors
+        self.assertEqual(len(factors), 2)
+        self.assertEqual(factors[0].factor_id, "phone")
+        self.assertEqual(factors[0].phone_number, "+15555550199")
+        self.assertEqual(
+            factors[0].enrollment_time,
+            dt.datetime(2023, 1, 1, 0, 0, 0, tzinfo=dt.timezone.utc),
+        )
+        self.assertEqual(factors[1].factor_id, "totp")
diff --git a/tests/test_options.py b/tests/test_options.py
index 676381e..ba6af3b 100644
--- a/tests/test_options.py
+++ b/tests/test_options.py
@@ -381,3 +381,26 @@ def sample(_event):
         "crashlytics.newFatalIssue",
         expect_app_id="app-123",
     )
+
+
+def test_auth_options_preserved_in_endpoint():
+    from firebase_functions import identity_fn
+
+    @identity_fn.on_user_created(
+        secrets=[ALERT_SECRET],
+        region="europe-west1",
+        max_instances=1,
+        retry=True,
+        tenant_id="tenant-123",
+    )
+    def sample(_event):
+        return None
+
+    endpoint = sample.__firebase_endpoint__
+    assert endpoint.region == ["europe-west1"]
+    assert endpoint.maxInstances == 1
+    assert endpoint.secretEnvironmentVariables == [{"key": "GITLAB_PERSONAL_ACCESS_TOKEN"}]
+    assert endpoint.eventTrigger["retry"] is True
+    assert endpoint.eventTrigger["region"] == "global"
+    assert endpoint.eventTrigger["eventType"] == "google.firebase.auth.user.v2.created"
+    assert endpoint.eventTrigger["eventFilters"]["tenantid"] == "tenant-123"