WIP

Restores the process_exception method that was removed in v6.7.5 (PR #328),
which broke exception capture from Django views and downstream middleware.

Django converts view exceptions into responses before they propagate through
the middleware stack's __call__ method, so the context manager's exception
handler never sees them. Django provides these exceptions via the
process_exception hook instead.

Changes:
- Add process_exception method to capture exceptions from views and downstream
  middleware with proper request context and tags
- Add tests verifying process_exception behavior and settings (capture_exceptions,
  request_filter)

CHANGELOG.md

@@ -1,3 +1,19 @@
+# Unreleased
+
+- fix(django): Restore process_exception method to capture view and downstream middleware exceptions (fixes #329)
+
+# 6.7.11 - 2025-10-28
+
+- feat(ai): Add `$ai_framework` property for framework integrations (e.g. LangChain)
+
+# 6.7.10 - 2025-10-24
+
+- fix(django): Make middleware truly hybrid - compatible with both sync (WSGI) and async (ASGI) Django stacks without breaking sync-only deployments
+
+# 6.7.9 - 2025-10-22
+
+- fix(flags): multi-condition flags with static cohorts returning wrong variants
+
 # 6.7.8 - 2025-10-16
 
 - fix(llma): missing async for OpenAI's streaming implementation

posthog/__init__.py

@@ -2,7 +2,13 @@ import datetime  # noqa: F401
 from typing import Callable, Dict, Optional, Any  # noqa: F401
 from typing_extensions import Unpack
 
-from posthog.args import OptionalCaptureArgs, OptionalSetArgs, ExceptionArg
+from posthog.args import (
+    OptionalCaptureArgs,
+    OptionalSetArgs,
+    OptionalCaptureAIArgs,
+    AI_EVENT_TYPE,
+    ExceptionArg,
+)
 from posthog.client import Client
 from posthog.contexts import (
     new_context as inner_new_context,
@@ -11,7 +17,15 @@ from posthog.contexts import (
     set_context_session as inner_set_context_session,
     identify_context as inner_identify_context,
 )
-from posthog.types import FeatureFlag, FlagsAndPayloads, FeatureFlagResult
+from posthog.feature_flags import (
+    InconclusiveMatchError as InconclusiveMatchError,
+    RequiresServerEvaluation as RequiresServerEvaluation,
+)
+from posthog.types import (
+    FeatureFlag,
+    FlagsAndPayloads,
+    FeatureFlagResult as FeatureFlagResult,
+)
 from posthog.version import VERSION
 
 __version__ = VERSION
@@ -385,6 +399,81 @@ def capture_exception(
     return _proxy("capture_exception", exception=exception, **kwargs)
 
 
+def capture_ai(
+    event: AI_EVENT_TYPE, **kwargs: Unpack[OptionalCaptureAIArgs]
+) -> Optional[str]:
+    """
+    Capture an AI event to the dedicated AI endpoint with support for large payloads.
+
+    This method sends AI events (like $ai_generation, $ai_trace, etc.) to PostHog's
+    specialized AI endpoint (/i/v0/ai) which supports large payloads through multipart/form-data
+    and blob storage in S3.
+
+    Args:
+        event: The AI event type. Must be one of: "$ai_generation", "$ai_trace", "$ai_span",
+               "$ai_embedding", "$ai_metric", "$ai_feedback"
+        distinct_id: The distinct ID of the user.
+        properties: A dictionary of AI event properties. Must include required properties based on event type:
+            - All events: "$ai_model" (required)
+            - $ai_generation: "$ai_provider", "$ai_trace_id" (required)
+            - $ai_trace: "$ai_trace_id" (required)
+            - $ai_span: "$ai_trace_id", "$ai_span_id" (required)
+            - $ai_embedding: "$ai_provider", "$ai_trace_id" (required)
+        blob_properties: List of property names to send as blobs (large data stored in S3).
+            Common blob properties: "$ai_input", "$ai_output_choices", "$ai_input_state", "$ai_output_state"
+            If not provided, defaults to common blob properties based on event type.
+        timestamp: The timestamp of the event.
+        uuid: A unique identifier for the event.
+        groups: A dictionary of group information.
+        disable_geoip: Whether to disable GeoIP for this event.
+
+    Examples:
+        ```python
+        # $ai_generation event with blobs
+        from posthog import capture_ai
+        capture_ai(
+            "$ai_generation",
+            distinct_id="user_123",
+            properties={
+                "$ai_model": "gpt-4",
+                "$ai_provider": "openai",
+                "$ai_trace_id": "trace_abc123",
+                "$ai_input": {
+                    "messages": [
+                        {"role": "user", "content": "Hello!"}
+                    ]
+                },
+                "$ai_output_choices": {
+                    "choices": [{"message": {"role": "assistant", "content": "Hi there!"}}]
+                },
+                "$ai_completion_tokens": 150,
+                "$ai_prompt_tokens": 50
+            },
+            blob_properties=["$ai_input", "$ai_output_choices"]
+        )
+        ```
+
+        ```python
+        # $ai_trace event
+        from posthog import capture_ai
+        capture_ai(
+            "$ai_trace",
+            distinct_id="user_123",
+            properties={
+                "$ai_model": "gpt-4",
+                "$ai_trace_id": "trace_abc123"
+            }
+        )
+        ```
+
+    Category:
+        AI Events
+
+    Note: This method sends events synchronously to the AI endpoint, bypassing the queue system.
+    """
+    return _proxy("capture_ai", event, **kwargs)
+
+
 def feature_enabled(
     key,  # type: str
     distinct_id,  # type: str

posthog/ai/langchain/callbacks.py

@@ -486,6 +486,7 @@ class CallbackHandler(BaseCallbackHandler):
             "$ai_latency": run.latency,
             "$ai_span_name": run.name,
             "$ai_span_id": run_id,
+            "$ai_framework": "langchain",
         }
         if parent_run_id is not None:
             event_properties["$ai_parent_id"] = parent_run_id
@@ -556,6 +557,7 @@ class CallbackHandler(BaseCallbackHandler):
             "$ai_http_status": 200,
             "$ai_latency": run.latency,
             "$ai_base_url": run.base_url,
+            "$ai_framework": "langchain",
         }
 
         if run.tools:

posthog/args.py

@@ -1,4 +1,4 @@
-from typing import TypedDict, Optional, Any, Dict, Union, Tuple, Type
+from typing import TypedDict, Optional, Any, Dict, List, Union, Tuple, Type
 from types import TracebackType
 from typing_extensions import NotRequired  # For Python < 3.11 compatibility
 from datetime import datetime
@@ -69,3 +69,33 @@ ExcInfo = Union[
 ]
 
 ExceptionArg = Union[BaseException, ExcInfo]
+
+
+# AI Event Types (literal strings to enforce valid event types)
+AI_EVENT_TYPE = Union[
+    str,  # Allow str for flexibility but document the expected values
+]  # "$ai_generation", "$ai_trace", "$ai_span", "$ai_embedding", "$ai_metric", "$ai_feedback"
+
+
+class OptionalCaptureAIArgs(TypedDict):
+    """Optional arguments for the capture_ai method.
+
+    Args:
+        distinct_id: Unique identifier for the person associated with this AI event. If not set, the context
+            distinct_id is used, if available, otherwise a UUID is generated.
+        properties: Dictionary of AI event properties to track. Must include required properties for the event type.
+        blob_properties: List of property names that should be sent as blobs (e.g., '$ai_input', '$ai_output_choices').
+            These properties will be extracted from `properties` and sent as multipart blobs.
+        timestamp: When the event occurred (defaults to current time)
+        uuid: Unique identifier for this specific event. If not provided, one is generated.
+        groups: Group identifiers to associate with this event (format: {group_type: group_key})
+        disable_geoip: Whether to disable GeoIP lookup for this event.
+    """
+
+    distinct_id: NotRequired[Optional[ID_TYPES]]
+    properties: NotRequired[Optional[Dict[str, Any]]]
+    blob_properties: NotRequired[Optional[List[str]]]
+    timestamp: NotRequired[Optional[Union[datetime, str]]]
+    uuid: NotRequired[Optional[str]]
+    groups: NotRequired[Optional[Dict[str, str]]]
+    disable_geoip: NotRequired[Optional[bool]]

posthog/client.py

@@ -1,16 +1,25 @@
 import atexit
+import json
 import logging
 import os
 import sys
 from datetime import datetime, timedelta
-from typing import Any, Callable, Dict, Optional, Union
+from io import BytesIO
+from typing import Any, Dict, List, Optional, Tuple, Union
 from typing_extensions import Unpack
 from uuid import uuid4
 
 from dateutil.tz import tzutc
 from six import string_types
 
-from posthog.args import OptionalCaptureArgs, OptionalSetArgs, ID_TYPES, ExceptionArg
+from posthog.args import (
+    OptionalCaptureArgs,
+    OptionalSetArgs,
+    OptionalCaptureAIArgs,
+    AI_EVENT_TYPE,
+    ID_TYPES,
+    ExceptionArg,
+)
 from posthog.consumer import Consumer
 from posthog.exception_capture import ExceptionCapture
 from posthog.exception_utils import (
@@ -20,7 +29,11 @@ from posthog.exception_utils import (
     exception_is_already_captured,
     mark_exception_as_captured,
 )
-from posthog.feature_flags import InconclusiveMatchError, match_feature_flag_properties
+from posthog.feature_flags import (
+    InconclusiveMatchError,
+    RequiresServerEvaluation,
+    match_feature_flag_properties,
+)
 from posthog.poller import Poller
 from posthog.request import (
     DEFAULT_HOST,
@@ -99,6 +112,106 @@ def add_context_tags(properties):
     return properties
 
 
+def _generate_multipart_boundary() -> str:
+    """Generate a random boundary string for multipart requests."""
+    return f"----WebKitFormBoundary{uuid4().hex[:16]}"
+
+
+def _encode_multipart_part(
+    boundary: str,
+    name: str,
+    content: bytes,
+    content_type: str,
+    filename: Optional[str] = None,
+) -> bytes:
+    """Encode a single part of a multipart/form-data request."""
+    part = f"--{boundary}\r\n".encode("utf-8")
+
+    if filename:
+        part += f'Content-Disposition: form-data; name="{name}"; filename="{filename}"\r\n'.encode(
+            "utf-8"
+        )
+    else:
+        part += f'Content-Disposition: form-data; name="{name}"\r\n'.encode("utf-8")
+
+    part += f"Content-Type: {content_type}\r\n\r\n".encode("utf-8")
+    part += content
+    part += b"\r\n"
+
+    return part
+
+
+def _build_multipart_body(
+    event_data: Dict[str, Any],
+    properties: Optional[Dict[str, Any]],
+    blob_properties: Optional[List[str]],
+) -> Tuple[bytes, str]:
+    """
+    Build a multipart/form-data body for AI capture endpoint.
+
+    Args:
+        event_data: The event data (uuid, event, distinct_id, timestamp)
+        properties: Event properties (small properties)
+        blob_properties: List of property names to send as blobs
+
+    Returns:
+        Tuple of (body_bytes, content_type_header)
+    """
+    boundary = _generate_multipart_boundary()
+    body = BytesIO()
+
+    # Part 1: Event (required, must be first)
+    event_json = json.dumps(event_data).encode("utf-8")
+    body.write(
+        _encode_multipart_part(boundary, "event", event_json, "application/json")
+    )
+
+    # Separate properties into small properties and blob properties
+    small_properties = {}
+    blob_data = {}
+
+    if properties:
+        blob_property_names = set(blob_properties or [])
+        for key, value in properties.items():
+            if key in blob_property_names:
+                blob_data[key] = value
+            else:
+                small_properties[key] = value
+
+    # Part 2: Event properties (if there are any small properties)
+    if small_properties:
+        properties_json = json.dumps(small_properties).encode("utf-8")
+        body.write(
+            _encode_multipart_part(
+                boundary, "event.properties", properties_json, "application/json"
+            )
+        )
+
+    # Part 3+: Blob parts
+    for property_name, property_value in blob_data.items():
+        # Serialize the blob data as JSON
+        blob_json = json.dumps(property_value).encode("utf-8")
+        blob_filename = f"blob_{uuid4().hex[:8]}"
+
+        body.write(
+            _encode_multipart_part(
+                boundary,
+                f"event.properties.{property_name}",
+                blob_json,
+                "application/json",
+                filename=blob_filename,
+            )
+        )
+
+    # Final boundary
+    body.write(f"--{boundary}--\r\n".encode("utf-8"))
+
+    body_bytes = body.getvalue()
+    content_type = f"multipart/form-data; boundary={boundary}"
+
+    return body_bytes, content_type
+
+
 def no_throw(default_return=None):
     """
     Decorator to prevent raising exceptions from public API methods.
@@ -1001,6 +1114,196 @@ class Client(object):
         except Exception as e:
             self.log.exception(f"Failed to capture exception: {e}")
 
+    @no_throw()
+    def capture_ai(
+        self, event: AI_EVENT_TYPE, **kwargs: Unpack[OptionalCaptureAIArgs]
+    ) -> Optional[str]:
+        """
+        Capture an AI event to the dedicated AI endpoint with support for large payloads.
+
+        This method sends AI events (like $ai_generation, $ai_trace, etc.) to PostHog's
+        specialized AI endpoint (/i/v0/ai) which supports large payloads through multipart/form-data
+        and blob storage in S3.
+
+        Args:
+            event: The AI event type. Must be one of: "$ai_generation", "$ai_trace", "$ai_span",
+                   "$ai_embedding", "$ai_metric", "$ai_feedback"
+            distinct_id: The distinct ID of the user.
+            properties: A dictionary of AI event properties. Must include required properties based on event type:
+                - All events: "$ai_model" (required)
+                - $ai_generation: "$ai_provider", "$ai_trace_id" (required)
+                - $ai_trace: "$ai_trace_id" (required)
+                - $ai_span: "$ai_trace_id", "$ai_span_id" (required)
+                - $ai_embedding: "$ai_provider", "$ai_trace_id" (required)
+            blob_properties: List of property names to send as blobs (large data stored in S3).
+                Common blob properties: "$ai_input", "$ai_output_choices", "$ai_input_state", "$ai_output_state"
+                If not provided, defaults to common blob properties based on event type.
+            timestamp: The timestamp of the event.
+            uuid: A unique identifier for the event.
+            groups: A dictionary of group information.
+            disable_geoip: Whether to disable GeoIP for this event.
+
+        Examples:
+            ```python
+            # $ai_generation event with blobs
+            posthog.capture_ai(
+                "$ai_generation",
+                distinct_id="user_123",
+                properties={
+                    "$ai_model": "gpt-4",
+                    "$ai_provider": "openai",
+                    "$ai_trace_id": "trace_abc123",
+                    "$ai_input": {
+                        "messages": [
+                            {"role": "user", "content": "Hello!"}
+                        ]
+                    },
+                    "$ai_output_choices": {
+                        "choices": [{"message": {"role": "assistant", "content": "Hi there!"}}]
+                    },
+                    "$ai_completion_tokens": 150,
+                    "$ai_prompt_tokens": 50
+                },
+                blob_properties=["$ai_input", "$ai_output_choices"]
+            )
+            ```
+
+            ```python
+            # $ai_trace event
+            posthog.capture_ai(
+                "$ai_trace",
+                distinct_id="user_123",
+                properties={
+                    "$ai_model": "gpt-4",
+                    "$ai_trace_id": "trace_abc123"
+                }
+            )
+            ```
+
+            ```python
+            # $ai_metric event
+            posthog.capture_ai(
+                "$ai_metric",
+                distinct_id="user_123",
+                properties={
+                    "$ai_model": "gpt-4",
+                    "$ai_trace_id": "trace_abc123",
+                    "$ai_metric_name": "accuracy",
+                    "$ai_metric_value": "0.95"
+                }
+            )
+            ```
+
+        Category:
+            AI Events
+
+        Note: This method sends events synchronously to the AI endpoint, bypassing the queue system.
+        """
+        import requests
+
+        if self.disabled:
+            return None
+
+        distinct_id = kwargs.get("distinct_id", None)
+        properties = kwargs.get("properties", None)
+        timestamp = kwargs.get("timestamp", None)
+        uuid = kwargs.get("uuid", None)
+        groups = kwargs.get("groups", None)
+        blob_properties = kwargs.get("blob_properties", None)
+
+        # Default blob properties based on event type if not provided
+        if blob_properties is None:
+            default_blob_properties = {
+                "$ai_generation": ["$ai_input", "$ai_output_choices"],
+                "$ai_trace": ["$ai_input_state", "$ai_output_state"],
+                "$ai_span": ["$ai_input_state", "$ai_output_state"],
+                "$ai_embedding": ["$ai_input"],
+            }
+            blob_properties = default_blob_properties.get(event, [])
+
+        properties = properties or {}
+
+        # Get distinct_id
+        (distinct_id, personless) = get_identity_state(distinct_id)
+
+        if personless and "$process_person_profile" not in properties:
+            properties["$process_person_profile"] = False
+
+        # Prepare timestamp
+        if timestamp is None:
+            timestamp = datetime.now(tz=tzutc())
+
+        timestamp = guess_timezone(timestamp)
+        timestamp_str = timestamp.isoformat()
+
+        # Generate UUID if not provided
+        if uuid:
+            uuid_str = stringify_id(uuid)
+        else:
+            uuid_str = stringify_id(uuid4())
+
+        # Add groups to properties
+        if groups:
+            properties["$groups"] = groups
+
+        # Prepare event data (the main event part)
+        event_data = {
+            "uuid": uuid_str,
+            "event": event,
+            "distinct_id": stringify_id(distinct_id),
+            "timestamp": timestamp_str,
+        }
+
+        # Build multipart body
+        body_bytes, content_type = _build_multipart_body(
+            event_data, properties, blob_properties
+        )
+
+        # Send request to AI endpoint
+        url = remove_trailing_slash(self.host) + "/i/v0/ai"
+
+        headers = {
+            "Content-Type": content_type,
+            "Authorization": f"Bearer {self.api_key}",
+            "User-Agent": f"posthog-python/{VERSION}",
+        }
+
+        self.log.debug(f"Sending AI event to {url}: {event} (uuid: {uuid_str})")
+
+        try:
+            response = requests.post(
+                url,
+                data=body_bytes,
+                headers=headers,
+                timeout=self.timeout,
+            )
+
+            if response.status_code == 200:
+                self.log.debug(f"AI event captured successfully: {event}")
+                return uuid_str
+            else:
+                error_message = f"AI capture failed with status {response.status_code}"
+                try:
+                    error_detail = response.json()
+                    error_message = f"{error_message}: {error_detail}"
+                except Exception:
+                    error_message = f"{error_message}: {response.text}"
+
+                self.log.error(error_message)
+
+                if self.debug:
+                    raise Exception(error_message)
+
+                return None
+
+        except Exception as e:
+            self.log.exception(f"Error sending AI event: {e}")
+
+            if self.debug:
+                raise e
+
+            return None
+
     def _enqueue(self, msg, disable_geoip):
         # type: (...) -> Optional[str]
         """Push a new `msg` onto the queue, return `(success, msg)`"""
@@ -1583,7 +1886,7 @@ class Client(object):
                     self.log.debug(
                         f"Successfully computed flag locally: {key} -> {response}"
                     )
-                except InconclusiveMatchError as e:
+                except (RequiresServerEvaluation, InconclusiveMatchError) as e:
                     self.log.debug(f"Failed to compute flag {key} locally: {e}")
                 except Exception as e:
                     self.log.exception(

posthog/feature_flags.py

@@ -22,6 +22,18 @@ class InconclusiveMatchError(Exception):
     pass
 
 
+class RequiresServerEvaluation(Exception):
+    """
+    Raised when feature flag evaluation requires server-side data that is not
+    available locally (e.g., static cohorts, experience continuity).
+
+    This error should propagate immediately to trigger API fallback, unlike
+    InconclusiveMatchError which allows trying other conditions.
+    """
+
+    pass
+
+
 # This function takes a distinct_id and a feature flag key and returns a float between 0 and 1.
 # Given the same distinct_id and key, it'll always return the same float. These floats are
 # uniformly distributed between 0 and 1, so if we want to show this feature to 20% of traffic
@@ -239,7 +251,12 @@ def match_feature_flag_properties(
                 else:
                     variant = get_matching_variant(flag, distinct_id)
                 return variant or True
+        except RequiresServerEvaluation:
+            # Static cohort or other missing server-side data - must fallback to API
+            raise
         except InconclusiveMatchError:
+            # Evaluation error (bad regex, invalid date, missing property, etc.)
+            # Track that we had an inconclusive match, but try other conditions
             is_inconclusive = True
 
     if is_inconclusive:
@@ -449,8 +466,8 @@ def match_cohort(
     # }
     cohort_id = str(property.get("value"))
     if cohort_id not in cohort_properties:
-        raise InconclusiveMatchError(
-            "can't match cohort without a given cohort property value"
+        raise RequiresServerEvaluation(
+            f"cohort {cohort_id} not found in local cohorts - likely a static cohort that requires server evaluation"
         )
 
     property_group = cohort_properties[cohort_id]
@@ -503,6 +520,9 @@ def match_property_group(
                     # OR group
                     if matches:
                         return True
+            except RequiresServerEvaluation:
+                # Immediately propagate - this condition requires server-side data
+                raise
             except InconclusiveMatchError as e:
                 log.debug(f"Failed to compute property {prop} locally: {e}")
                 error_matching_locally = True
@@ -552,6 +572,9 @@ def match_property_group(
                         return True
                     if not matches and negation:
                         return True
+            except RequiresServerEvaluation:
+                # Immediately propagate - this condition requires server-side data
+                raise
             except InconclusiveMatchError as e:
                 log.debug(f"Failed to compute property {prop} locally: {e}")
                 error_matching_locally = True

posthog/integrations/django.py

@@ -3,13 +3,19 @@ from posthog import contexts
 from posthog.client import Client
 
 try:
-    from asgiref.sync import iscoroutinefunction
+    from asgiref.sync import iscoroutinefunction, markcoroutinefunction
 except ImportError:
-    # Fallback for older Django versions
+    # Fallback for older Django versions without asgiref
     import asyncio
 
     iscoroutinefunction = asyncio.iscoroutinefunction
 
+    # No-op fallback for markcoroutinefunction
+    # Older Django versions without asgiref typically don't support async middleware anyway
+    def markcoroutinefunction(func):
+        return func
+
+
 if TYPE_CHECKING:
     from django.http import HttpRequest, HttpResponse  # noqa: F401
     from typing import Callable, Dict, Any, Optional, Union, Awaitable  # noqa: F401
@@ -39,26 +45,24 @@ class PosthogContextMiddleware:
     See the context documentation for more information. The extracted distinct ID and session ID, if found, are used to
     associate all events captured in the middleware context with the same distinct ID and session as currently active on the
     frontend. See the documentation for `set_context_session` and `identify_context` for more details.
+
+    This middleware is hybrid-capable: it supports both WSGI (sync) and ASGI (async) Django applications. The middleware
+    detects at initialization whether the next middleware in the chain is async or sync, and adapts its behavior accordingly.
+    This ensures compatibility with both pure sync and pure async middleware chains, as well as mixed chains in ASGI mode.
     """
 
-    # Django middleware capability flags
     sync_capable = True
     async_capable = True
 
     def __init__(self, get_response):
         # type: (Union[Callable[[HttpRequest], HttpResponse], Callable[[HttpRequest], Awaitable[HttpResponse]]]) -> None
+        self.get_response = get_response
         self._is_coroutine = iscoroutinefunction(get_response)
-        self._async_get_response = None  # type: Optional[Callable[[HttpRequest], Awaitable[HttpResponse]]]
-        self._sync_get_response = None  # type: Optional[Callable[[HttpRequest], HttpResponse]]
 
+        # Mark this instance as a coroutine function if get_response is async
+        # This is required for Django to correctly detect async middleware
         if self._is_coroutine:
-            self._async_get_response = cast(
-                "Callable[[HttpRequest], Awaitable[HttpResponse]]", get_response
-            )
-        else:
-            self._sync_get_response = cast(
-                "Callable[[HttpRequest], HttpResponse]", get_response
-            )
+            markcoroutinefunction(self)
 
         from django.conf import settings
 
@@ -181,40 +185,67 @@ class PosthogContextMiddleware:
         return user_id, email
 
     def __call__(self, request):
-        # type: (HttpRequest) -> HttpResponse
-        # Purely defensive around django's internal sync/async handling - this should be unreachable, but if it's reached, we may
-        # as well return something semi-meaningful
+        # type: (HttpRequest) -> Union[HttpResponse, Awaitable[HttpResponse]]
+        """
+        Unified entry point for both sync and async request handling.
+
+        When sync_capable and async_capable are both True, Django passes requests
+        without conversion. This method detects the mode and routes accordingly.
+        """
         if self._is_coroutine:
-            raise RuntimeError(
-                "PosthogContextMiddleware received sync call but get_response is async"
-            )
+            return self.__acall__(request)
+        else:
+            # Synchronous path
+            if self.request_filter and not self.request_filter(request):
+                return self.get_response(request)
 
-        if self.request_filter and not self.request_filter(request):
-            assert self._sync_get_response is not None
-            return self._sync_get_response(request)
+            with contexts.new_context(self.capture_exceptions, client=self.client):
+                for k, v in self.extract_tags(request).items():
+                    contexts.tag(k, v)
 
-        with contexts.new_context(self.capture_exceptions, client=self.client):
-            for k, v in self.extract_tags(request).items():
-                contexts.tag(k, v)
-
-            assert self._sync_get_response is not None
-            return self._sync_get_response(request)
+                return self.get_response(request)
 
     async def __acall__(self, request):
-        # type: (HttpRequest) -> HttpResponse
+        # type: (HttpRequest) -> Awaitable[HttpResponse]
+        """
+        Asynchronous entry point for async request handling.
+
+        This method is called when the middleware chain is async.
+        """
         if self.request_filter and not self.request_filter(request):
-            if self._async_get_response is not None:
-                return await self._async_get_response(request)
-            else:
-                assert self._sync_get_response is not None
-                return self._sync_get_response(request)
+            return await self.get_response(request)
 
         with contexts.new_context(self.capture_exceptions, client=self.client):
             for k, v in self.extract_tags(request).items():
                 contexts.tag(k, v)
 
-            if self._async_get_response is not None:
-                return await self._async_get_response(request)
-            else:
-                assert self._sync_get_response is not None
-                return self._sync_get_response(request)
+            return await self.get_response(request)
+
+    def process_exception(self, request, exception):
+        # type: (HttpRequest, Exception) -> None
+        """
+        Process exceptions from views and downstream middleware.
+
+        Django calls this WHILE still inside the context created by __call__,
+        so request tags have already been extracted and set. This method just
+        needs to capture the exception directly.
+
+        Django converts view exceptions into responses before they propagate through
+        the middleware stack, so the context manager in __call__/__acall__ never sees them.
+
+        Note: Django's process_exception is always synchronous, even for async views.
+        """
+        if self.request_filter and not self.request_filter(request):
+            return
+
+        if not self.capture_exceptions:
+            return
+
+        # Context and tags already set by __call__ or __acall__
+        # Just capture the exception
+        if self.client:
+            self.client.capture_exception(exception)
+        else:
+            from posthog import capture_exception
+
+            capture_exception(exception)

posthog/test/ai/langchain/test_callbacks.py

@@ -204,6 +204,7 @@ def test_basic_chat_chain(mock_client, stream):
     # Generation is second
     assert generation_args["event"] == "$ai_generation"
     assert "distinct_id" in generation_args
+    assert generation_props["$ai_framework"] == "langchain"
     assert "$ai_model" in generation_props
     assert "$ai_provider" in generation_props
     assert generation_props["$ai_input"] == [

posthog/test/integrations/test_middleware.py

@@ -4,7 +4,21 @@ from posthog.contexts import (
     get_context_distinct_id,
 )
 import unittest
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
+import asyncio
+
+# Configure Django settings before importing middleware
+import django
+from django.conf import settings
+
+if not settings.configured:
+    settings.configure(
+        DEBUG=True,
+        SECRET_KEY="test-secret-key",
+        INSTALLED_APPS=[],
+        MIDDLEWARE=[],
+    )
+    django.setup()
 
 from posthog.integrations.django import PosthogContextMiddleware
 
@@ -38,14 +52,33 @@ class TestPosthogContextMiddleware(unittest.TestCase):
         request_filter=None,
         tag_map=None,
         capture_exceptions=True,
+        get_response=None,
     ):
-        """Helper to create middleware instance without calling __init__"""
-        middleware = PosthogContextMiddleware.__new__(PosthogContextMiddleware)
-        middleware.get_response = Mock()
-        middleware.extra_tags = extra_tags
-        middleware.request_filter = request_filter
-        middleware.tag_map = tag_map
-        middleware.capture_exceptions = capture_exceptions
+        """Helper to create middleware instance with mock Django settings"""
+        if get_response is None:
+            get_response = Mock()
+
+        with patch("django.conf.settings") as mock_settings:
+            # Configure mock settings
+            mock_settings.POSTHOG_MW_EXTRA_TAGS = extra_tags
+            mock_settings.POSTHOG_MW_REQUEST_FILTER = request_filter
+            mock_settings.POSTHOG_MW_TAG_MAP = tag_map
+            mock_settings.POSTHOG_MW_CAPTURE_EXCEPTIONS = capture_exceptions
+            mock_settings.POSTHOG_MW_CLIENT = None
+
+            # Make hasattr work correctly
+            def mock_hasattr(obj, name):
+                return name in [
+                    "POSTHOG_MW_EXTRA_TAGS",
+                    "POSTHOG_MW_REQUEST_FILTER",
+                    "POSTHOG_MW_TAG_MAP",
+                    "POSTHOG_MW_CAPTURE_EXCEPTIONS",
+                    "POSTHOG_MW_CLIENT",
+                ]
+
+            with patch("builtins.hasattr", side_effect=mock_hasattr):
+                middleware = PosthogContextMiddleware(get_response)
+
         return middleware
 
     def test_extract_tags_basic(self):
@@ -168,6 +201,348 @@ class TestPosthogContextMiddleware(unittest.TestCase):
 
         self.assertEqual(tags["$request_method"], "PATCH")
 
+    def test_process_exception_called_during_view_exception(self):
+        """
+        Unit test verifying process_exception captures exceptions per Django's contract.
+
+        Since this is a library test (no Django runtime), we simulate how Django
+        would invoke our middleware in production:
+        1. Middleware.__call__ creates context with request tags
+        2. View raises exception inside get_response
+        3. Django's BaseHandler catches it, calls process_exception, returns error response
+        4. Exception never propagates to middleware's context manager
+
+        We manually call process_exception to simulate Django's behavior - this is
+        the only way to test the hook without a full Django integration test.
+        """
+        mock_client = Mock()
+        view_exception = ValueError("View raised this error")
+        error_response = Mock(status_code=500)
+
+        def mock_get_response(request):
+            # Simulate Django's exception handling: catches view exception,
+            # calls process_exception hook if it exists, returns error response
+            if hasattr(middleware, "process_exception"):
+                middleware.process_exception(request, view_exception)
+            return error_response
+
+        middleware = self.create_middleware(get_response=mock_get_response)
+        middleware.client = mock_client
+
+        request = MockRequest(
+            headers={"X-POSTHOG-DISTINCT-ID": "test-user"},
+            method="POST",
+            path="/api/endpoint",
+        )
+        response = middleware(request)
+
+        self.assertEqual(response.status_code, 500)
+        mock_client.capture_exception.assert_called_once_with(view_exception)
+
+    def test_process_exception_respects_capture_exceptions_false(self):
+        """Verify process_exception respects capture_exceptions=False setting"""
+        mock_client = Mock()
+        view_exception = ValueError("Should not be captured")
+
+        def mock_get_response(request):
+            if hasattr(middleware, "process_exception"):
+                middleware.process_exception(request, view_exception)
+            return Mock(status_code=500)
+
+        middleware = self.create_middleware(
+            capture_exceptions=False, get_response=mock_get_response
+        )
+        middleware.client = mock_client
+
+        request = MockRequest()
+        middleware(request)
+
+        mock_client.capture_exception.assert_not_called()
+
+    def test_process_exception_respects_request_filter(self):
+        """Verify process_exception respects request_filter setting"""
+        mock_client = Mock()
+        view_exception = ValueError("Should be filtered")
+
+        def mock_get_response(request):
+            if hasattr(middleware, "process_exception"):
+                middleware.process_exception(request, view_exception)
+            return Mock(status_code=500)
+
+        middleware = self.create_middleware(
+            request_filter=lambda req: False,
+            capture_exceptions=True,
+            get_response=mock_get_response,
+        )
+        middleware.client = mock_client
+
+        request = MockRequest()
+        middleware(request)
+
+        mock_client.capture_exception.assert_not_called()
+
+
+class TestPosthogContextMiddlewareSync(unittest.TestCase):
+    """Test synchronous middleware behavior"""
+
+    def test_sync_middleware_call(self):
+        """Test that sync middleware correctly processes requests"""
+        mock_response = Mock()
+        get_response = Mock(return_value=mock_response)
+
+        # Create middleware with sync get_response
+        middleware = PosthogContextMiddleware(get_response)
+
+        # Verify sync mode detected
+        self.assertFalse(middleware._is_coroutine)
+
+        request = MockRequest(
+            headers={"X-POSTHOG-SESSION-ID": "test-session"},
+            method="GET",
+            path="/test",
+        )
+
+        with new_context():
+            response = middleware(request)
+
+            # Verify response returned
+            self.assertEqual(response, mock_response)
+            get_response.assert_called_once_with(request)
+
+    def test_sync_middleware_with_filter(self):
+        """Test sync middleware respects request filter"""
+        mock_response = Mock()
+        get_response = Mock(return_value=mock_response)
+
+        # Create middleware with request filter that filters all requests
+        request_filter = lambda req: False
+        middleware = PosthogContextMiddleware.__new__(PosthogContextMiddleware)
+        middleware.get_response = get_response
+        middleware._is_coroutine = False
+        middleware.request_filter = request_filter
+        middleware.capture_exceptions = True
+        middleware.client = None
+
+        request = MockRequest()
+
+        # Should skip context creation and return response directly
+        response = middleware(request)
+        self.assertEqual(response, mock_response)
+        get_response.assert_called_once_with(request)
+
+    def test_view_exceptions_only_captured_via_process_exception(self):
+        """
+        Demonstrates that process_exception is required to capture view exceptions.
+
+        In production Django, view exceptions don't propagate to middleware's context
+        manager because Django's BaseHandler catches them first and converts them to
+        error responses. Django provides the exception via process_exception hook instead.
+
+        This unit test proves:
+        1. Context manager in __call__ never sees view exceptions (Django intercepts)
+        2. Only process_exception can capture them
+        3. Without process_exception, exceptions are silently lost (v6.7.5 regression)
+
+        We manually call process_exception to verify the hook works - in production,
+        Django's BaseHandler would call it when a view raises.
+        """
+        mock_client = Mock()
+        get_response = Mock(return_value=Mock(status_code=500))
+
+        middleware = PosthogContextMiddleware(get_response)
+        middleware.client = mock_client
+
+        def get_response_simulating_django(request):
+            # Simulates Django behavior: view exception converted to error response,
+            # never propagates to middleware's context manager
+            return Mock(status_code=500)
+
+        middleware._sync_get_response = get_response_simulating_django
+
+        request = MockRequest()
+
+        response = middleware(request)
+        self.assertEqual(response.status_code, 500)
+
+        # Context manager didn't capture anything - exception was intercepted by Django
+        mock_client.capture_exception.assert_not_called()
+
+        # Verify process_exception hook exists and captures exceptions when called
+        if hasattr(middleware, "process_exception"):
+            exception = ValueError("View error")
+            middleware.process_exception(request, exception)
+            mock_client.capture_exception.assert_called_once_with(exception)
+        else:
+            self.fail(
+                "process_exception missing - view exceptions will not be captured!"
+            )
+
+
+class TestPosthogContextMiddlewareAsync(unittest.TestCase):
+    """Test asynchronous middleware behavior"""
+
+    def test_async_middleware_detection(self):
+        """Test that async get_response is correctly detected"""
+
+        async def async_get_response(request):
+            return Mock()
+
+        middleware = PosthogContextMiddleware(async_get_response)
+
+        # Verify async mode detected
+        self.assertTrue(middleware._is_coroutine)
+
+    def test_async_middleware_call(self):
+        """Test that async middleware correctly processes requests"""
+
+        async def run_test():
+            mock_response = Mock()
+
+            async def async_get_response(request):
+                return mock_response
+
+            middleware = PosthogContextMiddleware(async_get_response)
+
+            request = MockRequest(
+                headers={"X-POSTHOG-SESSION-ID": "async-session"},
+                method="POST",
+                path="/async-test",
+            )
+
+            with new_context():
+                # Call should return the coroutine from __acall__
+                result = middleware(request)
+
+                # Verify it's a coroutine
+                self.assertTrue(asyncio.iscoroutine(result))
+
+                # Await the result
+                response = await result
+                self.assertEqual(response, mock_response)
+
+        asyncio.run(run_test())
+
+    def test_async_middleware_with_filter(self):
+        """Test async middleware respects request filter"""
+
+        async def run_test():
+            mock_response = Mock()
+
+            async def async_get_response(request):
+                return mock_response
+
+            # Properly initialize middleware
+            middleware = PosthogContextMiddleware(async_get_response)
+            # Override request filter after initialization
+            middleware.request_filter = lambda req: False
+
+            request = MockRequest()
+
+            # Should skip context creation and return response directly
+            result = middleware(request)
+            response = await result
+            self.assertEqual(response, mock_response)
+
+        asyncio.run(run_test())
+
+    def test_async_middleware_context_propagation(self):
+        """Test that async middleware properly propagates context"""
+
+        async def run_test():
+            mock_response = Mock()
+
+            async def async_get_response(request):
+                # Verify context is available during async processing
+                session_id = get_context_session_id()
+                self.assertEqual(session_id, "async-session-123")
+                return mock_response
+
+            middleware = PosthogContextMiddleware(async_get_response)
+
+            request = MockRequest(
+                headers={"X-POSTHOG-SESSION-ID": "async-session-123"},
+                method="GET",
+            )
+
+            with new_context():
+                result = middleware(request)
+                await result
+
+        asyncio.run(run_test())
+
+    def test_async_middleware_exception_capture(self):
+        """Test that async middleware captures exceptions during request processing"""
+
+        async def run_test():
+            mock_client = Mock()
+
+            # Make async_get_response raise an exception
+            async def raise_exception(request):
+                raise ValueError("Async test exception")
+
+            # Properly initialize middleware
+            middleware = PosthogContextMiddleware(raise_exception)
+            middleware.client = mock_client  # Override with mock client
+
+            request = MockRequest()
+
+            # Should capture exception and re-raise
+            with self.assertRaises(ValueError):
+                result = middleware(request)
+                await result
+
+            # Verify exception was captured by middleware
+            mock_client.capture_exception.assert_called_once()
+            captured_exception = mock_client.capture_exception.call_args[0][0]
+            self.assertIsInstance(captured_exception, ValueError)
+            self.assertEqual(str(captured_exception), "Async test exception")
+
+        asyncio.run(run_test())
+
+
+class TestPosthogContextMiddlewareHybrid(unittest.TestCase):
+    """Test hybrid middleware behavior with mixed sync/async chains"""
+
+    def test_hybrid_flags_set(self):
+        """Test that both capability flags are set"""
+        self.assertTrue(PosthogContextMiddleware.sync_capable)
+        self.assertTrue(PosthogContextMiddleware.async_capable)
+
+    def test_sync_to_async_routing(self):
+        """Test that __call__ routes to __acall__ when async"""
+
+        async def run_test():
+            async def async_get_response(request):
+                return Mock()
+
+            middleware = PosthogContextMiddleware(async_get_response)
+
+            # Verify routing happens
+            request = MockRequest()
+            result = middleware(request)
+
+            # Should be a coroutine from __acall__
+            self.assertTrue(asyncio.iscoroutine(result))
+            await result  # Clean up
+
+        asyncio.run(run_test())
+
+    def test_sync_path_direct_return(self):
+        """Test that sync path returns directly without coroutine"""
+        mock_response = Mock()
+
+        def sync_get_response(request):
+            return mock_response
+
+        middleware = PosthogContextMiddleware(sync_get_response)
+
+        request = MockRequest()
+        result = middleware(request)
+
+        # Should NOT be a coroutine
+        self.assertFalse(asyncio.iscoroutine(result))
+        self.assertEqual(result, mock_response)
+
 
 if __name__ == "__main__":
     unittest.main()

posthog/test/test_feature_flags.py

@@ -3013,6 +3013,75 @@ class TestLocalEvaluation(unittest.TestCase):
         )
         self.assertEqual(patch_flags.call_count, 0)
 
+    @mock.patch("posthog.client.flags")
+    @mock.patch("posthog.client.get")
+    def test_fallback_to_api_when_flag_has_static_cohort_in_multi_condition(
+        self, patch_get, patch_flags
+    ):
+        """
+        When a flag has multiple conditions and one contains a static cohort,
+        the SDK should fallback to API for the entire flag, not just skip that
+        condition and evaluate the next one locally.
+
+        This prevents returning wrong variants when later conditions could match
+        locally but the user is actually in the static cohort.
+        """
+        client = Client(FAKE_TEST_API_KEY, personal_api_key=FAKE_TEST_API_KEY)
+
+        # Mock the local flags response - cohort 999 is NOT in cohorts map (static cohort)
+        client.feature_flags = [
+            {
+                "id": 1,
+                "key": "multi-condition-flag",
+                "active": True,
+                "filters": {
+                    "groups": [
+                        {
+                            "properties": [
+                                {"key": "id", "value": 999, "type": "cohort"}
+                            ],
+                            "rollout_percentage": 100,
+                            "variant": "set-1",
+                        },
+                        {
+                            "properties": [
+                                {
+                                    "key": "$geoip_country_code",
+                                    "operator": "exact",
+                                    "value": ["DE"],
+                                    "type": "person",
+                                }
+                            ],
+                            "rollout_percentage": 100,
+                            "variant": "set-8",
+                        },
+                    ],
+                    "multivariate": {
+                        "variants": [
+                            {"key": "set-1", "rollout_percentage": 50},
+                            {"key": "set-8", "rollout_percentage": 50},
+                        ]
+                    },
+                },
+            }
+        ]
+        client.cohorts = {}  # Note: cohort 999 is NOT here - it's a static cohort
+
+        # Mock the API response - user is in the static cohort
+        patch_flags.return_value = {"featureFlags": {"multi-condition-flag": "set-1"}}
+
+        result = client.get_feature_flag(
+            "multi-condition-flag",
+            "test-distinct-id",
+            person_properties={"$geoip_country_code": "DE"},
+        )
+
+        # Should return the API result (set-1), not local evaluation (set-8)
+        self.assertEqual(result, "set-1")
+
+        # Verify API was called (fallback occurred)
+        self.assertEqual(patch_flags.call_count, 1)
+
 
 class TestMatchProperties(unittest.TestCase):
     def property(self, key, value, operator=None):
@@ -4006,6 +4075,60 @@ class TestCaptureCalls(unittest.TestCase):
 
         patch_capture.reset_mock()
 
+    @mock.patch("posthog.client.flags")
+    def test_fallback_to_api_in_get_feature_flag_payload_when_flag_has_static_cohort(
+        self, patch_flags
+    ):
+        """
+        Test that get_feature_flag_payload falls back to API when evaluating
+        a flag with static cohorts, similar to get_feature_flag behavior.
+        """
+        client = Client(FAKE_TEST_API_KEY, personal_api_key=FAKE_TEST_API_KEY)
+
+        # Mock the local flags response - cohort 999 is NOT in cohorts map (static cohort)
+        client.feature_flags = [
+            {
+                "id": 1,
+                "name": "Multi-condition Flag",
+                "key": "multi-condition-flag",
+                "active": True,
+                "filters": {
+                    "groups": [
+                        {
+                            "properties": [
+                                {"key": "id", "value": 999, "type": "cohort"}
+                            ],
+                            "rollout_percentage": 100,
+                            "variant": "variant-1",
+                        }
+                    ],
+                    "multivariate": {
+                        "variants": [{"key": "variant-1", "rollout_percentage": 100}]
+                    },
+                    "payloads": {"variant-1": '{"message": "local-payload"}'},
+                },
+            }
+        ]
+        client.cohorts = {}  # Note: cohort 999 is NOT here - it's a static cohort
+
+        # Mock the API response - user is in the static cohort
+        patch_flags.return_value = {
+            "featureFlags": {"multi-condition-flag": "variant-1"},
+            "featureFlagPayloads": {"multi-condition-flag": '{"message": "from-api"}'},
+        }
+
+        # Call get_feature_flag_payload without match_value to trigger evaluation
+        result = client.get_feature_flag_payload(
+            "multi-condition-flag",
+            "test-distinct-id",
+        )
+
+        # Should return the API payload, not local payload
+        self.assertEqual(result, {"message": "from-api"})
+
+        # Verify API was called (fallback occurred)
+        self.assertEqual(patch_flags.call_count, 1)
+
     @mock.patch.object(Client, "capture")
     @mock.patch("posthog.client.flags")
     def test_disable_geoip_get_flag_capture_call(self, patch_flags, patch_capture):

posthog/version.py

@@ -1,4 +1,4 @@
-VERSION = "6.7.8"
+VERSION = "6.7.11"
 
 if __name__ == "__main__":
     print(VERSION, end="")  # noqa: T201

references/posthog-python-references-latest.json

@@ -2,7 +2,7 @@
   "id": "posthog-python",
   "hogRef": "0.3",
   "info": {
-    "version": "6.7.7",
+    "version": "6.7.11",
     "id": "posthog-python",
     "title": "PostHog Python SDK",
     "description": "Integrate PostHog into any python application.",

test_capture_ai_local.py

@@ -0,0 +1,326 @@
+#!/usr/bin/env python3
+"""
+Test script to send capture_ai events to localhost:8010.
+This script tests the actual network request to a local PostHog instance.
+"""
+
+from posthog import Posthog
+from uuid import uuid4
+
+# Create a client pointing to localhost:8010
+posthog = Posthog(
+    "test-api-key",  # Use your actual project API key if needed
+    host="http://localhost:8010",
+    debug=True,  # Enable debug mode to see detailed logs
+)
+
+print("Testing capture_ai with localhost:8010")
+print("=" * 60)
+
+# Test 1: $ai_generation event with blobs
+print("\n1. Testing $ai_generation event with blobs...")
+print("-" * 60)
+
+trace_id = f"trace_{uuid4().hex[:8]}"
+
+try:
+    event_uuid = posthog.capture_ai(
+        "$ai_generation",
+        distinct_id="test_user_123",
+        properties={
+            "$ai_model": "gpt-4",
+            "$ai_provider": "openai",
+            "$ai_trace_id": trace_id,
+            "$ai_input": {
+                "messages": [
+                    {
+                        "role": "system",
+                        "content": "You are a helpful assistant that answers questions about Python.",
+                    },
+                    {
+                        "role": "user",
+                        "content": "What is the difference between a list and a tuple?",
+                    },
+                ],
+                "temperature": 0.7,
+                "max_tokens": 500,
+            },
+            "$ai_output_choices": {
+                "choices": [
+                    {
+                        "index": 0,
+                        "message": {
+                            "role": "assistant",
+                            "content": "A list is mutable (can be changed) while a tuple is immutable (cannot be changed after creation). Lists use square brackets [] and tuples use parentheses ().",
+                        },
+                        "finish_reason": "stop",
+                    }
+                ],
+                "model": "gpt-4",
+                "usage": {
+                    "prompt_tokens": 45,
+                    "completion_tokens": 32,
+                    "total_tokens": 77,
+                },
+            },
+            "$ai_completion_tokens": 32,
+            "$ai_prompt_tokens": 45,
+            "$ai_total_tokens": 77,
+            "$ai_latency": 1.234,
+        },
+        blob_properties=["$ai_input", "$ai_output_choices"],
+    )
+
+    if event_uuid:
+        print("✓ SUCCESS: $ai_generation event sent")
+        print(f"  UUID: {event_uuid}")
+        print(f"  Trace ID: {trace_id}")
+    else:
+        print("✗ FAILED: No UUID returned")
+
+except Exception as e:
+    print(f"✗ ERROR: {e}")
+    import traceback
+
+    traceback.print_exc()
+
+# Test 2: $ai_trace event
+print("\n2. Testing $ai_trace event...")
+print("-" * 60)
+
+try:
+    event_uuid = posthog.capture_ai(
+        "$ai_trace",
+        distinct_id="test_user_123",
+        properties={
+            "$ai_model": "gpt-4",
+            "$ai_trace_id": trace_id,
+            "$ai_trace_name": "python_qa_session",
+            "$ai_input_state": {
+                "session_id": "session_123",
+                "user_context": "learning Python",
+            },
+            "$ai_output_state": {"questions_answered": 1, "satisfaction_score": 5},
+        },
+        blob_properties=["$ai_input_state", "$ai_output_state"],
+    )
+
+    if event_uuid:
+        print("✓ SUCCESS: $ai_trace event sent")
+        print(f"  UUID: {event_uuid}")
+        print(f"  Trace ID: {trace_id}")
+    else:
+        print("✗ FAILED: No UUID returned")
+
+except Exception as e:
+    print(f"✗ ERROR: {e}")
+    import traceback
+
+    traceback.print_exc()
+
+# Test 3: $ai_span event
+print("\n3. Testing $ai_span event...")
+print("-" * 60)
+
+span_id = f"span_{uuid4().hex[:8]}"
+
+try:
+    event_uuid = posthog.capture_ai(
+        "$ai_span",
+        distinct_id="test_user_123",
+        properties={
+            "$ai_model": "gpt-4",
+            "$ai_trace_id": trace_id,
+            "$ai_span_id": span_id,
+            "$ai_span_name": "answer_generation",
+            "$ai_parent_id": trace_id,
+            "$ai_span_kind": "llm",
+            "$ai_latency": 0.8,
+        },
+    )
+
+    if event_uuid:
+        print("✓ SUCCESS: $ai_span event sent")
+        print(f"  UUID: {event_uuid}")
+        print(f"  Span ID: {span_id}")
+    else:
+        print("✗ FAILED: No UUID returned")
+
+except Exception as e:
+    print(f"✗ ERROR: {e}")
+    import traceback
+
+    traceback.print_exc()
+
+# Test 4: $ai_embedding event
+print("\n4. Testing $ai_embedding event...")
+print("-" * 60)
+
+try:
+    event_uuid = posthog.capture_ai(
+        "$ai_embedding",
+        distinct_id="test_user_123",
+        properties={
+            "$ai_model": "text-embedding-ada-002",
+            "$ai_provider": "openai",
+            "$ai_trace_id": trace_id,
+            "$ai_input": {
+                "text": "What is the difference between a list and a tuple in Python?"
+            },
+            "$ai_embedding_dimension": 1536,
+            "$ai_latency": 0.123,
+        },
+        blob_properties=["$ai_input"],
+    )
+
+    if event_uuid:
+        print("✓ SUCCESS: $ai_embedding event sent")
+        print(f"  UUID: {event_uuid}")
+    else:
+        print("✗ FAILED: No UUID returned")
+
+except Exception as e:
+    print(f"✗ ERROR: {e}")
+    import traceback
+
+    traceback.print_exc()
+
+# Test 5: $ai_metric event
+print("\n5. Testing $ai_metric event...")
+print("-" * 60)
+
+try:
+    event_uuid = posthog.capture_ai(
+        "$ai_metric",
+        distinct_id="test_user_123",
+        properties={
+            "$ai_model": "gpt-4",
+            "$ai_trace_id": trace_id,
+            "$ai_metric_name": "response_quality",
+            "$ai_metric_value": "0.95",
+        },
+    )
+
+    if event_uuid:
+        print("✓ SUCCESS: $ai_metric event sent")
+        print(f"  UUID: {event_uuid}")
+    else:
+        print("✗ FAILED: No UUID returned")
+
+except Exception as e:
+    print(f"✗ ERROR: {e}")
+    import traceback
+
+    traceback.print_exc()
+
+# Test 6: $ai_feedback event
+print("\n6. Testing $ai_feedback event...")
+print("-" * 60)
+
+try:
+    event_uuid = posthog.capture_ai(
+        "$ai_feedback",
+        distinct_id="test_user_123",
+        properties={
+            "$ai_model": "gpt-4",
+            "$ai_trace_id": trace_id,
+            "$ai_feedback_text": "Great explanation! Very clear and helpful.",
+            "$ai_feedback_rating": 5,
+        },
+    )
+
+    if event_uuid:
+        print("✓ SUCCESS: $ai_feedback event sent")
+        print(f"  UUID: {event_uuid}")
+    else:
+        print("✗ FAILED: No UUID returned")
+
+except Exception as e:
+    print(f"✗ ERROR: {e}")
+    import traceback
+
+    traceback.print_exc()
+
+# Test 7: Test with custom blob properties
+print("\n7. Testing with custom blob properties...")
+print("-" * 60)
+
+try:
+    event_uuid = posthog.capture_ai(
+        "$ai_generation",
+        distinct_id="test_user_123",
+        properties={
+            "$ai_model": "claude-3-opus",
+            "$ai_provider": "anthropic",
+            "$ai_trace_id": trace_id,
+            "$ai_input": {
+                "messages": [{"role": "user", "content": "Write a haiku about Python"}]
+            },
+            "$ai_output_choices": {
+                "choices": [
+                    {
+                        "message": {
+                            "role": "assistant",
+                            "content": "Snake glides through code\nSimple syntax, powerful tools\nDevelopers smile",
+                        }
+                    }
+                ]
+            },
+            "$ai_custom_data": {
+                "large_context": "This is some large custom data that should be sent as a blob"
+            },
+            "$ai_completion_tokens": 20,
+            "$ai_prompt_tokens": 10,
+        },
+        # Custom blob properties - including the default ones plus a custom one
+        blob_properties=["$ai_input", "$ai_output_choices", "$ai_custom_data"],
+    )
+
+    if event_uuid:
+        print("✓ SUCCESS: Event with custom blob properties sent")
+        print(f"  UUID: {event_uuid}")
+    else:
+        print("✗ FAILED: No UUID returned")
+
+except Exception as e:
+    print(f"✗ ERROR: {e}")
+    import traceback
+
+    traceback.print_exc()
+
+# Test 8: Test with groups
+print("\n8. Testing with groups...")
+print("-" * 60)
+
+try:
+    event_uuid = posthog.capture_ai(
+        "$ai_generation",
+        distinct_id="test_user_123",
+        groups={"company": "posthog_inc", "team": "engineering"},
+        properties={
+            "$ai_model": "gpt-4",
+            "$ai_provider": "openai",
+            "$ai_trace_id": trace_id,
+            "$ai_input": {"messages": [{"role": "user", "content": "test"}]},
+            "$ai_output_choices": {
+                "choices": [{"message": {"role": "assistant", "content": "response"}}]
+            },
+        },
+    )
+
+    if event_uuid:
+        print("✓ SUCCESS: Event with groups sent")
+        print(f"  UUID: {event_uuid}")
+    else:
+        print("✗ FAILED: No UUID returned")
+
+except Exception as e:
+    print(f"✗ ERROR: {e}")
+    import traceback
+
+    traceback.print_exc()
+
+print("\n" + "=" * 60)
+print("All tests completed!")
+print("\nMake sure your local PostHog instance is running on http://localhost:8010")
+print("and that the /i/v0/ai endpoint is available.")