diff --git a/api/api/openapi.py b/api/api/openapi.py
index 2784dd1270b4..fb3d5d91f56a 100644
--- a/api/api/openapi.py
+++ b/api/api/openapi.py
@@ -165,19 +165,38 @@ def get_security_definition(
 
 class CohortSyncKeyAuthenticationExtension(OpenApiAuthenticationExtension):  # type: ignore[no-untyped-call]
     target_class = "cohorts.authentication.CohortSyncKeyAuthentication"
-    name = "Cohort Sync Key"
+    name = ["Cohort Sync Key", "Cohort Sync Key (Basic)"]
+
+    def get_security_requirement(
+        self, auto_schema: openapi.AutoSchema
+    ) -> list[dict[str, list[Any]]]:
+        # Separate entries: the caller sends the key with either scheme,
+        # not both at once.
+        return [{name: []} for name in self.name]
 
     def get_security_definition(
         self, auto_schema: openapi.AutoSchema | None = None
-    ) -> dict[str, Any]:
-        return {
-            "type": "http",
-            "scheme": "bearer",
-            "description": (
-                "For cohort sync endpoints called by an external cohort "
-                "source, such as Amplitude."
-            ),
-        }
+    ) -> list[dict[str, Any]]:
+        return [
+            {
+                "type": "http",
+                "scheme": "bearer",
+                "description": (
+                    "For cohort sync endpoints called by an external cohort "
+                    "source, such as Amplitude."
+                ),
+            },
+            {
+                "type": "http",
+                "scheme": "basic",
+                "description": (
+                    "For cohort sync endpoints called by an external cohort "
+                    "source that can only send Basic credentials, such as "
+                    "Mixpanel. The key is the password; the username is "
+                    "ignored."
+                ),
+            },
+        ]
 
 
 # Tag definitions controlling the order and display of sections in the Swagger UI.
diff --git a/api/cohorts/authentication.py b/api/cohorts/authentication.py
index 175a29b2f005..6ef0b9c78569 100644
--- a/api/cohorts/authentication.py
+++ b/api/cohorts/authentication.py
@@ -1,4 +1,4 @@
-import typing
+import base64
 from contextlib import suppress
 
 from django.contrib.auth.models import AnonymousUser
@@ -9,18 +9,36 @@
 
 
 class CohortSyncKeyAuthentication(authentication.BaseAuthentication):
+    """
+    Accepts a cohort sync key sent either as a Bearer token or as the
+    password of Basic credentials. Amplitude sends Bearer; Mixpanel's
+    webhook setup only offers a username/password form, so its customers
+    enter any username and the key as the password. The username is
+    ignored.
+    """
+
     def authenticate(
         self, request: Request
     ) -> tuple[AnonymousUser, CohortSyncKey] | None:
         header = request.headers.get("Authorization", "")
-        if not header.startswith("Bearer "):
+        if header.startswith("Bearer "):
+            raw_key = header.removeprefix("Bearer ")
+        elif header.startswith("Basic "):
+            try:
+                decoded = base64.b64decode(
+                    header.removeprefix("Basic "), validate=True
+                ).decode()
+            except ValueError:
+                # Covers malformed base64, header bytes outside ASCII, and
+                # decoded credentials that are not valid UTF-8.
+                raise exceptions.AuthenticationFailed("Invalid Basic credentials.")
+            # Split at the first colon, so a key containing colons survives.
+            _, _, raw_key = decoded.partition(":")
+        else:
             return None
 
         with suppress(CohortSyncKey.DoesNotExist):
-            key = typing.cast(
-                CohortSyncKey,
-                CohortSyncKey.objects.get_from_key(header.removeprefix("Bearer ")),
-            )
+            key = CohortSyncKey.objects.get_from_key(raw_key)
             if not key.has_expired:
                 # No person is acting here, so no user is returned: the key
                 # alone carries authority, and audit trails record the source
diff --git a/api/cohorts/migrations/0004_mixpanel_source.py b/api/cohorts/migrations/0004_mixpanel_source.py
new file mode 100644
index 000000000000..1cdb5e5260e4
--- /dev/null
+++ b/api/cohorts/migrations/0004_mixpanel_source.py
@@ -0,0 +1,33 @@
+# Generated by Django 5.2.16 on 2026-08-20 10:06
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("cohorts", "0003_cohort_sync_key"),
+        ("environments", "0039_use_no_ssrf_url_field"),
+        ("segments", "0032_add_segment_rules_data"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="cohort",
+            name="external_id",
+            field=models.CharField(blank=True, max_length=255, null=True),
+        ),
+        migrations.AlterField(
+            model_name="cohort",
+            name="source_type",
+            field=models.CharField(
+                choices=[
+                    ("csv", "CSV"),
+                    ("amplitude", "Amplitude"),
+                    ("mixpanel", "Mixpanel"),
+                ],
+                default="csv",
+                max_length=50,
+            ),
+        ),
+    ]
diff --git a/api/cohorts/models.py b/api/cohorts/models.py
index b15927b6fcd0..1efd43e478ff 100644
--- a/api/cohorts/models.py
+++ b/api/cohorts/models.py
@@ -1,5 +1,7 @@
+import typing
+
 from django.db import models
-from rest_framework_api_key.models import AbstractAPIKey
+from rest_framework_api_key.models import AbstractAPIKey, APIKeyManager
 
 from cohorts.constants import COHORT_SYSTEM_TRAIT_KEY_PREFIX
 from core.models import SoftDeleteExportableModel
@@ -8,6 +10,7 @@
 class CohortSourceType(models.TextChoices):
     CSV = "csv", "CSV"
     AMPLITUDE = "amplitude", "Amplitude"
+    MIXPANEL = "mixpanel", "Mixpanel"
 
 
 class Cohort(SoftDeleteExportableModel):
@@ -26,6 +29,11 @@ class Cohort(SoftDeleteExportableModel):
         choices=CohortSourceType.choices,
         default=CohortSourceType.CSV,
     )
+    # The cohort's identifier in the external source. Mixpanel pushes under
+    # its own cohort ID, so we store it to route later requests; Amplitude
+    # uses the ID we hand back at list creation, and CSV cohorts have no
+    # external system, so both leave this null.
+    external_id = models.CharField(max_length=255, null=True, blank=True)
     version = models.PositiveIntegerField(default=0)
     created_at = models.DateTimeField(auto_now_add=True)
     # Deletion drains memberships from the identity store first; the cohort is
@@ -48,7 +56,19 @@ class Meta:
         ]
 
 
+class CohortSyncKeyManager(APIKeyManager):
+    def get_from_key(self, key: str) -> "CohortSyncKey":
+        if "\x00" in key:
+            # A NUL can't travel in a raw header, but base64 credentials can
+            # decode to one, and the database driver refuses to build a query
+            # containing it. No real key holds one, so treat it as absent.
+            raise self.model.DoesNotExist("Key contains a NUL character.")
+        return typing.cast("CohortSyncKey", super().get_from_key(key))
+
+
 class CohortSyncKey(AbstractAPIKey):
+    objects: typing.ClassVar[CohortSyncKeyManager] = CohortSyncKeyManager()
+
     environment = models.ForeignKey(
         "environments.Environment",
         on_delete=models.CASCADE,
diff --git a/api/cohorts/serializers.py b/api/cohorts/serializers.py
index a5e5bc39915d..259adb351f49 100644
--- a/api/cohorts/serializers.py
+++ b/api/cohorts/serializers.py
@@ -60,6 +60,27 @@ class AmplitudeListSerializer(serializers.Serializer[None]):
     name = serializers.CharField(max_length=2000)
 
 
+class MixpanelMemberSerializer(serializers.Serializer[None]):
+    # Length mirrors CohortMembership.identifier.
+    mixpanel_distinct_id = serializers.CharField(max_length=2000)
+
+
+class MixpanelParametersSerializer(serializers.Serializer[None]):
+    mixpanel_cohort_id = serializers.CharField(max_length=255)
+    mixpanel_cohort_name = serializers.CharField(max_length=2000)
+    # An empty page is valid: a first sync of an empty cohort has no members.
+    members = MixpanelMemberSerializer(many=True, allow_empty=True)
+
+
+class MixpanelWebhookSerializer(serializers.Serializer[None]):
+    # "members" carries the full membership on the first sync;
+    # "add_members"/"remove_members" carry changes since the last sync.
+    action = serializers.ChoiceField(
+        choices=["members", "add_members", "remove_members"]
+    )
+    parameters = MixpanelParametersSerializer()
+
+
 class CohortSyncMembersSerializer(serializers.Serializer[None]):
     # Child length mirrors CohortMembership.identifier.
     # TODO: this counts characters, but identity data is stored with a
diff --git a/api/cohorts/services.py b/api/cohorts/services.py
index 0b0688433f83..d81f03442b43 100644
--- a/api/cohorts/services.py
+++ b/api/cohorts/services.py
@@ -90,6 +90,7 @@ def create_cohort(
     name: str,
     description: str | None = None,
     source_type: CohortSourceType = CohortSourceType.CSV,
+    external_id: str | None = None,
 ) -> Cohort:
     with transaction.atomic():
         segment = Segment.objects.create(
@@ -100,7 +101,10 @@ def create_cohort(
         )
         rule = SegmentRule.objects.create(segment=segment, type=SegmentRule.ALL_RULE)
         cohort: Cohort = Cohort.objects.create(
-            environment=environment, segment=segment, source_type=source_type
+            environment=environment,
+            segment=segment,
+            source_type=source_type,
+            external_id=external_id,
         )
         Condition.objects.create(
             rule=rule,
@@ -124,10 +128,16 @@ def create_cohort_for_source(
     environment: "Environment",
     name: str,
     source_type: CohortSourceType,
+    external_id: str | None = None,
 ) -> Cohort:
     """Create a cohort on behalf of an external source, where no Flagsmith
     user is acting."""
-    cohort = create_cohort(environment=environment, name=name, source_type=source_type)
+    cohort = create_cohort(
+        environment=environment,
+        name=name,
+        source_type=source_type,
+        external_id=external_id,
+    )
     # Nothing records a user for these calls, so the audit log that Flagsmith
     # derives from historical records is skipped — and with it the environment
     # document rebuild that makes the new segment visible to SDKs. Write the
@@ -145,6 +155,21 @@ def create_cohort_for_source(
     return cohort
 
 
+def get_cohort_for_source(
+    *,
+    environment: "Environment",
+    source_type: CohortSourceType,
+    external_id: str,
+) -> Cohort | None:
+    cohort: Cohort | None = Cohort.objects.filter(
+        environment=environment,
+        source_type=source_type,
+        external_id=external_id,
+        deletion_requested_at__isnull=True,
+    ).first()
+    return cohort
+
+
 def add_cohort_members(cohort: Cohort, identifiers: "typing.Iterable[str]") -> None:
     from cohorts.tasks import apply_cohort_membership_deltas
 
diff --git a/api/cohorts/sync_urls.py b/api/cohorts/sync_urls.py
index 6afc517a0b6c..3b599f178e1f 100644
--- a/api/cohorts/sync_urls.py
+++ b/api/cohorts/sync_urls.py
@@ -1,6 +1,7 @@
+from django.urls import path
 from rest_framework.routers import SimpleRouter
 
-from cohorts.sync_views import AmplitudeCohortSyncViewSet
+from cohorts.sync_views import AmplitudeCohortSyncViewSet, MixpanelCohortSyncView
 
 app_name = "cohort-sync"
 
@@ -8,4 +9,7 @@
 router = SimpleRouter()
 router.register(r"amplitude/lists", AmplitudeCohortSyncViewSet, basename="amplitude")
 
-urlpatterns = router.urls
+urlpatterns = [
+    path("mixpanel/webhook/", MixpanelCohortSyncView.as_view(), name="mixpanel"),
+    *router.urls,
+]
diff --git a/api/cohorts/sync_views.py b/api/cohorts/sync_views.py
index eae69581f2ab..d4d764096c7e 100644
--- a/api/cohorts/sync_views.py
+++ b/api/cohorts/sync_views.py
@@ -1,12 +1,15 @@
+import json
 import typing
 import uuid as uuid_module
 
+import structlog
 from drf_spectacular.utils import extend_schema, extend_schema_view, inline_serializer
 from rest_framework import serializers, viewsets
 from rest_framework.decorators import action
-from rest_framework.exceptions import NotFound
+from rest_framework.exceptions import NotFound, ParseError
 from rest_framework.request import Request
 from rest_framework.response import Response
+from rest_framework.views import APIView
 
 from cohorts import services
 from cohorts.authentication import CohortSyncKeyAuthentication
@@ -15,12 +18,35 @@
 from cohorts.serializers import (
     AmplitudeListSerializer,
     CohortSyncMembersSerializer,
+    MixpanelWebhookSerializer,
 )
 
 _LIST_RESPONSE = inline_serializer(
     "AmplitudeListResponse", {"list_id": serializers.UUIDField()}
 )
 
+_MIXPANEL_RESPONSE = inline_serializer(
+    "MixpanelWebhookResponse",
+    {"action": serializers.CharField(), "status": serializers.CharField()},
+)
+
+_MIXPANEL_FAILURE_RESPONSE = inline_serializer(
+    "MixpanelWebhookFailureResponse",
+    {
+        "action": serializers.CharField(allow_null=True),
+        "status": serializers.CharField(),
+        "error": inline_serializer(
+            "MixpanelWebhookError",
+            {
+                "message": serializers.CharField(),
+                "code": serializers.IntegerField(),
+            },
+        ),
+    },
+)
+
+logger = structlog.get_logger("cohorts")
+
 
 @extend_schema_view(
     create=extend_schema(
@@ -83,3 +109,116 @@ def _get_cohort(self, request: Request, pk: str) -> Cohort:
         if cohort is None:
             raise NotFound("List not found.")
         return cohort
+
+
+class MixpanelCohortSyncView(APIView):
+    """
+    The receiving end of Mixpanel's Custom Webhook cohort destination:
+    https://docs.mixpanel.com/docs/cohort-sync/webhooks
+
+    Mixpanel POSTs every message to this one URL and reads the outcome from
+    the response body, which must repeat the action alongside a
+    success/failure status.
+    """
+
+    authentication_classes = [CohortSyncKeyAuthentication]
+    permission_classes = [HasCohortSyncKey]
+
+    @extend_schema(
+        description=(
+            "Called by Mixpanel every sync cycle with the cohort's full "
+            "membership (`members`) or the changes since the last sync "
+            "(`add_members`/`remove_members`)."
+        ),
+        request=MixpanelWebhookSerializer,
+        responses={
+            200: _MIXPANEL_RESPONSE,
+            400: _MIXPANEL_FAILURE_RESPONSE,
+            404: _MIXPANEL_FAILURE_RESPONSE,
+        },
+    )
+    def post(self, request: Request) -> Response:
+        serializer = MixpanelWebhookSerializer(data=request.data)
+        if not serializer.is_valid():
+            return self._failure(
+                request,
+                message=(
+                    f"Invalid payload: {json.dumps(serializer.errors, default=str)}"
+                ),
+                code=400,
+            )
+
+        data = serializer.validated_data
+        webhook_action: str = data["action"]
+        parameters = data["parameters"]
+        identifiers = [
+            member["mixpanel_distinct_id"] for member in parameters["members"]
+        ]
+        environment = typing.cast(CohortSyncKey, request.auth).environment
+
+        if webhook_action == "members":
+            # A large first sync arrives as several requests, each one page
+            # of members. Every page only adds; removals can't be detected
+            # without seeing all pages at once.
+            cohort = services.get_cohort_for_source(
+                environment=environment,
+                source_type=CohortSourceType.MIXPANEL,
+                external_id=parameters["mixpanel_cohort_id"],
+            ) or services.create_cohort_for_source(
+                environment=environment,
+                name=parameters["mixpanel_cohort_name"],
+                source_type=CohortSourceType.MIXPANEL,
+                external_id=parameters["mixpanel_cohort_id"],
+            )
+            services.add_cohort_members(cohort, identifiers)
+        else:
+            cohort_or_none = services.get_cohort_for_source(
+                environment=environment,
+                source_type=CohortSourceType.MIXPANEL,
+                external_id=parameters["mixpanel_cohort_id"],
+            )
+            if cohort_or_none is None:
+                # A 404 makes Mixpanel pause the sync and email the customer,
+                # which is what should happen when the cohort was deleted in
+                # Flagsmith but Mixpanel is still syncing it.
+                return self._failure(request, message="Cohort not found.", code=404)
+            if webhook_action == "add_members":
+                services.add_cohort_members(cohort_or_none, identifiers)
+            else:
+                services.remove_cohort_members(cohort_or_none, identifiers)
+
+        return Response({"action": webhook_action, "status": "success"})
+
+    def handle_exception(self, exc: Exception) -> Response:
+        if isinstance(exc, ParseError):
+            # A body that isn't valid JSON raises before post() runs, so the
+            # response is shaped here to keep the envelope Mixpanel expects.
+            return self._failure(self.request, message="Invalid payload.", code=400)
+        return super().handle_exception(exc)
+
+    def _failure(self, request: Request, *, message: str, code: int) -> Response:
+        logger.warning(
+            "sync_webhook.rejected",
+            source="mixpanel",
+            action=self._echo_action(request),
+            environment__id=typing.cast(CohortSyncKey, request.auth).environment_id,
+            error__message=message,
+            error__code=code,
+        )
+        return Response(
+            {
+                "action": self._echo_action(request),
+                "status": "failure",
+                "error": {"message": message, "code": code},
+            },
+            status=code,
+        )
+
+    def _echo_action(self, request: Request) -> str | None:
+        # Mixpanel expects the response to name the action it sent, even on
+        # failure; None when the request was too malformed to carry one.
+        if isinstance(request.data, dict) and isinstance(
+            action_value := request.data.get("action"), str
+        ):
+            return action_value
+        return None
diff --git a/api/tests/unit/cohorts/conftest.py b/api/tests/unit/cohorts/conftest.py
index 96467c9204e8..42c02715e832 100644
--- a/api/tests/unit/cohorts/conftest.py
+++ b/api/tests/unit/cohorts/conftest.py
@@ -42,6 +42,30 @@ def amplitude_cohort(
     return cohort
 
 
+@pytest.fixture()
+def postgres_cohort_sync_key(
+    environment: Environment,
+) -> typing.Tuple[CohortSyncKey, str]:
+    return typing.cast(
+        typing.Tuple[CohortSyncKey, str],
+        CohortSyncKey.objects.create_key(name="postgres key", environment=environment),
+    )
+
+
+@pytest.fixture()
+def mixpanel_cohort(environment: Environment) -> Cohort:
+    segment = Segment.objects.create(
+        name="mixpanel segment", project=environment.project
+    )
+    cohort: Cohort = Cohort.objects.create(
+        environment=environment,
+        segment=segment,
+        source_type=CohortSourceType.MIXPANEL,
+        external_id="mp-42",
+    )
+    return cohort
+
+
 @pytest.fixture()
 def edge_cohort(
     dynamo_enabled_project: Project,
diff --git a/api/tests/unit/cohorts/test_services.py b/api/tests/unit/cohorts/test_services.py
index 098902ca8f42..77b78bd14aef 100644
--- a/api/tests/unit/cohorts/test_services.py
+++ b/api/tests/unit/cohorts/test_services.py
@@ -2,7 +2,11 @@
 from pytest_mock import MockerFixture
 from pytest_structlog import StructuredLogCapture
 
-from cohorts.models import Cohort, CohortMembership, CohortMembershipState
+from cohorts.models import (
+    Cohort,
+    CohortMembership,
+    CohortMembershipState,
+)
 from cohorts.services import (
     apply_pending_memberships,
     create_cohort,
diff --git a/api/tests/unit/cohorts/test_sync_views.py b/api/tests/unit/cohorts/test_sync_views.py
index f8ddd2a0de68..992862d659a9 100644
--- a/api/tests/unit/cohorts/test_sync_views.py
+++ b/api/tests/unit/cohorts/test_sync_views.py
@@ -1,9 +1,11 @@
+import base64
 import typing
 
 from django.urls import reverse
 from django.utils import timezone
 from flag_engine.segments.constants import IS_SET
 from pytest_django.fixtures import SettingsWrapper
+from pytest_structlog import StructuredLogCapture
 from rest_framework import status
 from rest_framework.test import APIClient
 
@@ -17,6 +19,7 @@
     CohortSyncKey,
 )
 from environments.dynamodb import DynamoIdentityWrapper
+from environments.identities.models import Identity
 from environments.models import Environment
 
 _KeyAndPlaintext = typing.Tuple[CohortSyncKey, str]
@@ -400,3 +403,414 @@ def test_amplitude_add_members__master_api_key_throttle_enabled__succeeds(
 
     # Then
     assert response.status_code == status.HTTP_200_OK
+
+
+def _basic_auth_client(plaintext_key: str) -> APIClient:
+    credentials = base64.b64encode(f"flagsmith:{plaintext_key}".encode()).decode()
+    client = APIClient()
+    client.credentials(HTTP_AUTHORIZATION=f"Basic {credentials}")
+    return client
+
+
+def test_mixpanel_webhook__members_action_unknown_cohort__creates_cohort_and_memberships(
+    postgres_cohort_sync_key: _KeyAndPlaintext,
+) -> None:
+    # Given
+    key, plaintext = postgres_cohort_sync_key
+    client = _basic_auth_client(plaintext)
+    url = reverse("api-v1:cohort-sync:mixpanel")
+
+    # When
+    response = client.post(
+        url,
+        data={
+            "action": "members",
+            "parameters": {
+                "mixpanel_cohort_id": "mp-42",
+                "mixpanel_cohort_name": "Power users",
+                "members": [
+                    {"mixpanel_distinct_id": "user-1"},
+                    {"mixpanel_distinct_id": "user-2"},
+                ],
+            },
+        },
+        format="json",
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json() == {"action": "members", "status": "success"}
+    cohort = Cohort.objects.get(
+        environment=key.environment,
+        source_type=CohortSourceType.MIXPANEL,
+        external_id="mp-42",
+    )
+    assert cohort.segment.name == "Power users"
+    assert sorted(
+        CohortMembership.objects.filter(cohort=cohort).values_list(
+            "identifier", "state"
+        )
+    ) == [
+        ("user-1", CohortMembershipState.APPLIED),
+        ("user-2", CohortMembershipState.APPLIED),
+    ]
+
+
+def test_mixpanel_webhook__members_action_existing_cohort__adds_to_it(
+    postgres_cohort_sync_key: _KeyAndPlaintext,
+    mixpanel_cohort: Cohort,
+) -> None:
+    # Given
+    _, plaintext = postgres_cohort_sync_key
+    client = _basic_auth_client(plaintext)
+    url = reverse("api-v1:cohort-sync:mixpanel")
+
+    # When
+    response = client.post(
+        url,
+        data={
+            "action": "members",
+            "parameters": {
+                "mixpanel_cohort_id": "mp-42",
+                "mixpanel_cohort_name": "Power users",
+                "members": [{"mixpanel_distinct_id": "user-1"}],
+            },
+        },
+        format="json",
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_200_OK
+    assert Cohort.objects.filter(source_type=CohortSourceType.MIXPANEL).count() == 1
+    membership = CohortMembership.objects.get(cohort=mixpanel_cohort)
+    assert (membership.identifier, membership.state) == (
+        "user-1",
+        CohortMembershipState.APPLIED,
+    )
+
+
+def test_mixpanel_webhook__add_members_action__sets_system_trait(
+    postgres_cohort_sync_key: _KeyAndPlaintext,
+    mixpanel_cohort: Cohort,
+) -> None:
+    # Given
+    _, plaintext = postgres_cohort_sync_key
+    client = _basic_auth_client(plaintext)
+    url = reverse("api-v1:cohort-sync:mixpanel")
+
+    # When
+    response = client.post(
+        url,
+        data={
+            "action": "add_members",
+            "parameters": {
+                "mixpanel_cohort_id": "mp-42",
+                "mixpanel_cohort_name": "Power users",
+                "members": [{"mixpanel_distinct_id": "user-1"}],
+            },
+        },
+        format="json",
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json() == {"action": "add_members", "status": "success"}
+    identity = Identity.objects.get(
+        environment=mixpanel_cohort.environment, identifier="user-1"
+    )
+    assert identity.system_traits == {mixpanel_cohort.system_trait_key: True}
+
+
+def test_mixpanel_webhook__remove_members_action__unsets_system_trait(
+    postgres_cohort_sync_key: _KeyAndPlaintext,
+    mixpanel_cohort: Cohort,
+) -> None:
+    # Given
+    Identity.objects.create(
+        environment=mixpanel_cohort.environment,
+        identifier="member",
+        system_traits={mixpanel_cohort.system_trait_key: True},
+    )
+    CohortMembership.objects.create(
+        cohort=mixpanel_cohort,
+        identifier="member",
+        state=CohortMembershipState.APPLIED,
+    )
+    _, plaintext = postgres_cohort_sync_key
+    client = _basic_auth_client(plaintext)
+    url = reverse("api-v1:cohort-sync:mixpanel")
+
+    # When
+    response = client.post(
+        url,
+        data={
+            "action": "remove_members",
+            "parameters": {
+                "mixpanel_cohort_id": "mp-42",
+                "mixpanel_cohort_name": "Power users",
+                "members": [{"mixpanel_distinct_id": "member"}],
+            },
+        },
+        format="json",
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json() == {"action": "remove_members", "status": "success"}
+    assert not CohortMembership.objects.filter(cohort=mixpanel_cohort).exists()
+    identity = Identity.objects.get(
+        environment=mixpanel_cohort.environment, identifier="member"
+    )
+    assert identity.system_traits == {}
+
+
+def test_mixpanel_webhook__add_members_unknown_cohort__returns_404_failure(
+    postgres_cohort_sync_key: _KeyAndPlaintext,
+    log: StructuredLogCapture,
+) -> None:
+    # Given
+    key, plaintext = postgres_cohort_sync_key
+    client = _basic_auth_client(plaintext)
+    url = reverse("api-v1:cohort-sync:mixpanel")
+
+    # When
+    response = client.post(
+        url,
+        data={
+            "action": "add_members",
+            "parameters": {
+                "mixpanel_cohort_id": "unknown",
+                "mixpanel_cohort_name": "Power users",
+                "members": [{"mixpanel_distinct_id": "user-1"}],
+            },
+        },
+        format="json",
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_404_NOT_FOUND
+    assert response.json() == {
+        "action": "add_members",
+        "status": "failure",
+        "error": {"message": "Cohort not found.", "code": 404},
+    }
+    assert log.events == [
+        {
+            "level": "warning",
+            "event": "sync_webhook.rejected",
+            "source": "mixpanel",
+            "action": "add_members",
+            "environment__id": key.environment_id,
+            "error__message": "Cohort not found.",
+            "error__code": 404,
+        }
+    ]
+
+
+def test_mixpanel_webhook__deletion_requested_cohort__returns_404_failure(
+    postgres_cohort_sync_key: _KeyAndPlaintext,
+    mixpanel_cohort: Cohort,
+) -> None:
+    # Given
+    mixpanel_cohort.deletion_requested_at = timezone.now()
+    mixpanel_cohort.save()
+    _, plaintext = postgres_cohort_sync_key
+    client = _basic_auth_client(plaintext)
+    url = reverse("api-v1:cohort-sync:mixpanel")
+
+    # When
+    response = client.post(
+        url,
+        data={
+            "action": "remove_members",
+            "parameters": {
+                "mixpanel_cohort_id": "mp-42",
+                "mixpanel_cohort_name": "Power users",
+                "members": [{"mixpanel_distinct_id": "user-1"}],
+            },
+        },
+        format="json",
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_404_NOT_FOUND
+    assert response.json()["status"] == "failure"
+
+
+def test_mixpanel_webhook__missing_parameters__returns_400_failure(
+    postgres_cohort_sync_key: _KeyAndPlaintext,
+) -> None:
+    # Given
+    _, plaintext = postgres_cohort_sync_key
+    client = _basic_auth_client(plaintext)
+    url = reverse("api-v1:cohort-sync:mixpanel")
+
+    # When
+    response = client.post(url, data={"action": "members"}, format="json")
+
+    # Then
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    body = response.json()
+    assert body["action"] == "members"
+    assert body["status"] == "failure"
+    assert body["error"]["code"] == 400
+    assert "parameters" in body["error"]["message"]
+
+
+def test_mixpanel_webhook__non_object_payload__returns_400_failure(
+    postgres_cohort_sync_key: _KeyAndPlaintext,
+) -> None:
+    # Given
+    _, plaintext = postgres_cohort_sync_key
+    client = _basic_auth_client(plaintext)
+    url = reverse("api-v1:cohort-sync:mixpanel")
+
+    # When
+    response = client.post(url, data=["not", "an", "object"], format="json")
+
+    # Then
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    body = response.json()
+    assert body["action"] is None
+    assert body["status"] == "failure"
+    assert body["error"]["code"] == 400
+
+
+def test_mixpanel_webhook__unparseable_body__returns_400_failure(
+    postgres_cohort_sync_key: _KeyAndPlaintext,
+) -> None:
+    # Given
+    _, plaintext = postgres_cohort_sync_key
+    client = _basic_auth_client(plaintext)
+    url = reverse("api-v1:cohort-sync:mixpanel")
+
+    # When
+    response = client.post(url, data="{not json", content_type="application/json")
+
+    # Then
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.json() == {
+        "action": None,
+        "status": "failure",
+        "error": {"message": "Invalid payload.", "code": 400},
+    }
+
+
+def test_mixpanel_webhook__other_environment_key__returns_404_failure(
+    mixpanel_cohort: Cohort,
+) -> None:
+    # Given - a key scoped to a different environment than the cohort's
+    other_environment = Environment.objects.create(
+        name="Other environment", project=mixpanel_cohort.environment.project
+    )
+    _, plaintext = CohortSyncKey.objects.create_key(
+        name="other key", environment=other_environment
+    )
+    client = _basic_auth_client(plaintext)
+    url = reverse("api-v1:cohort-sync:mixpanel")
+
+    # When
+    response = client.post(
+        url,
+        data={
+            "action": "add_members",
+            "parameters": {
+                "mixpanel_cohort_id": "mp-42",
+                "mixpanel_cohort_name": "Power users",
+                "members": [{"mixpanel_distinct_id": "user-1"}],
+            },
+        },
+        format="json",
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_404_NOT_FOUND
+    assert not CohortMembership.objects.filter(cohort=mixpanel_cohort).exists()
+
+
+def test_mixpanel_webhook__empty_members_page__returns_success(
+    postgres_cohort_sync_key: _KeyAndPlaintext,
+) -> None:
+    # Given
+    key, plaintext = postgres_cohort_sync_key
+    client = _basic_auth_client(plaintext)
+    url = reverse("api-v1:cohort-sync:mixpanel")
+
+    # When
+    response = client.post(
+        url,
+        data={
+            "action": "members",
+            "parameters": {
+                "mixpanel_cohort_id": "mp-42",
+                "mixpanel_cohort_name": "Empty cohort",
+                "members": [],
+            },
+        },
+        format="json",
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_200_OK
+    cohort = Cohort.objects.get(environment=key.environment, external_id="mp-42")
+    assert not CohortMembership.objects.filter(cohort=cohort).exists()
+
+
+def test_mixpanel_webhook__non_ascii_basic_credentials__returns_401(
+    db: None,
+) -> None:
+    # Given - a header byte outside ASCII, which base64 decoding rejects
+    client = APIClient()
+    client.credentials(HTTP_AUTHORIZATION="Basic dXNlcjprÿZXk=")
+    url = reverse("api-v1:cohort-sync:mixpanel")
+
+    # When
+    response = client.post(url, data={"action": "members"}, format="json")
+
+    # Then
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_mixpanel_webhook__nul_byte_in_basic_password__returns_401(
+    db: None,
+) -> None:
+    # Given - valid base64 whose decoded password contains a NUL character
+    credentials = base64.b64encode(b"user:\x00key").decode()
+    client = APIClient()
+    client.credentials(HTTP_AUTHORIZATION=f"Basic {credentials}")
+    url = reverse("api-v1:cohort-sync:mixpanel")
+
+    # When
+    response = client.post(url, data={"action": "members"}, format="json")
+
+    # Then
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_mixpanel_webhook__malformed_basic_credentials__returns_401(
+    db: None,
+) -> None:
+    # Given
+    client = APIClient()
+    client.credentials(HTTP_AUTHORIZATION="Basic not-base64!!")
+    url = reverse("api-v1:cohort-sync:mixpanel")
+
+    # When
+    response = client.post(url, data={"action": "members"}, format="json")
+
+    # Then
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_mixpanel_webhook__unknown_key_in_basic_password__returns_401(
+    db: None,
+) -> None:
+    # Given
+    client = _basic_auth_client("not-a-key")
+    url = reverse("api-v1:cohort-sync:mixpanel")
+
+    # When
+    response = client.post(url, data={"action": "members"}, format="json")
+
+    # Then
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md
index 742e613bf0b3..398ccc3c5874 100644
--- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md
+++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md
@@ -74,7 +74,7 @@ Attributes:
 ### `cohorts.cohort.created`
 
 Logged at `info` from:
- - `api/cohorts/services.py:111`
+ - `api/cohorts/services.py:115`
 
 Attributes:
  - `cohort.id`
@@ -86,7 +86,7 @@ Attributes:
 ### `cohorts.cohort.deleted`
 
 Logged at `info` from:
- - `api/cohorts/services.py:219`
+ - `api/cohorts/services.py:244`
 
 Attributes:
  - `cohort.id`
@@ -95,7 +95,7 @@ Attributes:
 ### `cohorts.cohort.deletion_requested`
 
 Logged at `info` from:
- - `api/cohorts/services.py:203`
+ - `api/cohorts/services.py:228`
 
 Attributes:
  - `cohort.id`
@@ -132,8 +132,8 @@ Attributes:
 ### `cohorts.membership.deltas_received`
 
 Logged at `info` from:
- - `api/cohorts/services.py:168`
- - `api/cohorts/services.py:187`
+ - `api/cohorts/services.py:193`
+ - `api/cohorts/services.py:212`
 
 Attributes:
  - `action`
@@ -142,6 +142,18 @@ Attributes:
  - `environment.id`
  - `members.matched`
 
+### `cohorts.sync_webhook.rejected`
+
+Logged at `warning` from:
+ - `api/cohorts/sync_views.py:200`
+
+Attributes:
+ - `action`
+ - `environment.id`
+ - `error.code`
+ - `error.message`
+ - `source`
+
 ### `core.encrypted_field.decrypt_failed`
 
 Logged at `warning` from:
diff --git a/mcp/src/flagsmith_mcp/openapi.json b/mcp/src/flagsmith_mcp/openapi.json
index ef2dbc10e145..545e240db8e7 100644
--- a/mcp/src/flagsmith_mcp/openapi.json
+++ b/mcp/src/flagsmith_mcp/openapi.json
@@ -7761,6 +7761,11 @@
         "scheme": "bearer",
         "description": "For cohort sync endpoints called by an external cohort source, such as Amplitude."
       },
+      "Cohort Sync Key (Basic)": {
+        "type": "http",
+        "scheme": "basic",
+        "description": "For cohort sync endpoints called by an external cohort source that can only send Basic credentials, such as Mixpanel. The key is the password; the username is ignored."
+      },
       "Environment API Key": {
         "type": "apiKey",
         "in": "header",
diff --git a/openapi.yaml b/openapi.yaml
index 8b03b8f842d7..ac809f434dce 100644
--- a/openapi.yaml
+++ b/openapi.yaml
@@ -1683,6 +1683,7 @@ paths:
                 $ref: '#/components/schemas/AmplitudeListResponse'
       security:
         - Cohort Sync Key: []
+        - Cohort Sync Key (Basic): []
       tags:
         - Other
   '/api/v1/cohort-sync/amplitude/lists/{id}/add/':
@@ -1711,6 +1712,7 @@ paths:
           description: No response body
       security:
         - Cohort Sync Key: []
+        - Cohort Sync Key (Basic): []
       tags:
         - Other
   '/api/v1/cohort-sync/amplitude/lists/{id}/remove/':
@@ -1739,8 +1741,49 @@ paths:
           description: No response body
       security:
         - Cohort Sync Key: []
+        - Cohort Sync Key (Basic): []
       tags:
         - Other
+  /api/v1/cohort-sync/mixpanel/webhook/:
+    post:
+      operationId: api_v1_cohort_sync_mixpanel_webhook_create
+      description: Called by Mixpanel every sync cycle with the cohort's full membership (`members`) or the changes since the last sync (`add_members`/`remove_members`).
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/MixpanelWebhook'
+          application/x-www-form-urlencoded:
+            schema:
+              $ref: '#/components/schemas/MixpanelWebhook'
+          multipart/form-data:
+            schema:
+              $ref: '#/components/schemas/MixpanelWebhook'
+      responses:
+        '200':
+          description: ''
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/MixpanelWebhookResponse'
+        '400':
+          description: ''
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/MixpanelWebhookFailureResponse'
+        '404':
+          description: ''
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/MixpanelWebhookFailureResponse'
+      security:
+        - Cohort Sync Key: []
+        - Cohort Sync Key (Basic): []
+      tags:
+        - Webhooks
   /api/v1/environment-document/:
     get:
       operationId: sdk_v1_environment_document
@@ -18664,6 +18707,16 @@ paths:
         - processor
 components:
   schemas:
+    ActionEnum:
+      description: |-
+        * `members` - members
+        * `add_members` - add_members
+        * `remove_members` - remove_members
+      type: string
+      enum:
+        - members
+        - add_members
+        - remove_members
     ActionTypeEnum:
       description: |-
         * `TOGGLE_FEATURE` - Enable/Disable Feature for the environment
@@ -22700,6 +22753,76 @@ components:
           maxLength: 200
       required:
         - api_key
+    MixpanelMember:
+      type: object
+      properties:
+        mixpanel_distinct_id:
+          type: string
+          maxLength: 2000
+      required:
+        - mixpanel_distinct_id
+    MixpanelParameters:
+      type: object
+      properties:
+        mixpanel_cohort_id:
+          type: string
+          maxLength: 255
+        mixpanel_cohort_name:
+          type: string
+          maxLength: 2000
+        members:
+          type: array
+          items:
+            $ref: '#/components/schemas/MixpanelMember'
+      required:
+        - members
+        - mixpanel_cohort_id
+        - mixpanel_cohort_name
+    MixpanelWebhook:
+      type: object
+      properties:
+        action:
+          $ref: '#/components/schemas/ActionEnum'
+        parameters:
+          $ref: '#/components/schemas/MixpanelParameters'
+      required:
+        - action
+        - parameters
+    MixpanelWebhookError:
+      type: object
+      properties:
+        message:
+          type: string
+        code:
+          type: integer
+      required:
+        - code
+        - message
+    MixpanelWebhookFailureResponse:
+      type: object
+      properties:
+        action:
+          type:
+            - string
+            - 'null'
+        status:
+          type: string
+        error:
+          $ref: '#/components/schemas/MixpanelWebhookError'
+      required:
+        - action
+        - error
+        - status
+    MixpanelWebhookResponse:
+      type: object
+      properties:
+        action:
+          type: string
+        status:
+          type: string
+      required:
+        - action
+        - status
     Monitoring:
       type: object
       properties:
@@ -27346,10 +27469,12 @@ components:
       description: |-
         * `csv` - CSV
         * `amplitude` - Amplitude
+        * `mixpanel` - Mixpanel
       type: string
       enum:
         - csv
         - amplitude
+        - mixpanel
     StageAction:
       type: object
       properties:
@@ -29362,6 +29487,10 @@ components:
       type: http
       scheme: bearer
       description: 'For cohort sync endpoints called by an external cohort source, such as Amplitude.'
+    Cohort Sync Key (Basic):
+      type: http
+      scheme: basic
+      description: 'For cohort sync endpoints called by an external cohort source that can only send Basic credentials, such as Mixpanel. The key is the password; the username is ignored.'
     Environment API Key:
       type: apiKey
       in: header
diff --git a/sdk/openapi.yaml b/sdk/openapi.yaml
index e534853228a9..5b8787a1056d 100644
--- a/sdk/openapi.yaml
+++ b/sdk/openapi.yaml
@@ -549,6 +549,10 @@ components:
       type: http
       scheme: bearer
       description: 'For cohort sync endpoints called by an external cohort source, such as Amplitude.'
+    Cohort Sync Key (Basic):
+      type: http
+      scheme: basic
+      description: 'For cohort sync endpoints called by an external cohort source that can only send Basic credentials, such as Mixpanel. The key is the password; the username is ignored.'
     Environment API Key:
       type: apiKey
       in: header