feat(ai): add OpenTelemetry integration for AI span export (#494)

## Problem

Users instrumenting their AI/LLM applications with OpenTelemetry have no way to route AI spans to PostHog without writing custom glue code. This is the Python equivalent of PostHog/posthog-js#3358.

## Changes

Adds a `posthog.ai.otel` module with two integration patterns:

- **`PostHogSpanProcessor`** (recommended) — self-contained `SpanProcessor` that wraps `BatchSpanProcessor` + `OTLPSpanExporter`, filtering to only forward AI spans to PostHog's OTLP endpoint at `/i/v0/ai/otel`
- **`PostHogTraceExporter`** — `SpanExporter` that filters AI spans before forwarding, for setups that only accept an exporter (e.g. passed to `SimpleSpanProcessor`)
- **`is_ai_span()`** — shared predicate matching `gen_ai.*`, `llm.*`, `ai.*`, `traceloop.*` prefixes on span names and attribute keys

Also adds `posthog[otel]` optional dependency group for `opentelemetry-sdk` and `opentelemetry-exporter-otlp-proto-http`.

### Usage

```python
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.resources import Resource, SERVICE_NAME
from posthog.ai.otel import PostHogSpanProcessor

resource = Resource(
    attributes={
        SERVICE_NAME: "my-app",
        "posthog.distinct_id": "user-123",
    }
)
provider = TracerProvider(resource=resource)
provider.add_span_processor(
    PostHogSpanProcessor(api_key="phc_...")
)
trace.set_tracer_provider(provider)
```

### Consistency with #482

Uses the same OTLP endpoint (`/i/v0/ai/otel`) and auth header format (`Authorization: Bearer {api_key}`) as the migrated examples in #482.

## How did you test this code?

31 unit tests covering span filtering, processor forwarding/dropping, exporter filtering, endpoint configuration, and lifecycle delegation.

Author: richardsolomou

posthog/ai/otel/__init__.py

@@ -0,0 +1,33 @@
+"""PostHog OpenTelemetry integration for AI tracing.
+
+Provides components to route AI-related OpenTelemetry spans to PostHog's
+OTLP endpoint. Only spans matching known AI semantic convention prefixes
+(gen_ai, llm, ai, traceloop) are forwarded; all other spans are silently
+dropped.
+
+Two integration patterns are supported:
+
+1. **PostHogSpanProcessor** (recommended) - Self-contained processor that
+   handles batching and export internally::
+
+       provider = TracerProvider()
+       provider.add_span_processor(
+           PostHogSpanProcessor(api_key="phc_...")
+       )
+
+2. **PostHogTraceExporter** - Exporter for use with your own
+   BatchSpanProcessor or frameworks that only accept a SpanExporter::
+
+       provider = TracerProvider()
+       provider.add_span_processor(
+           BatchSpanProcessor(
+               PostHogTraceExporter(api_key="phc_...")
+           )
+       )
+"""
+
+from posthog.ai.otel.exporter import PostHogTraceExporter
+from posthog.ai.otel.processor import PostHogSpanProcessor
+from posthog.ai.otel.spans import is_ai_span
+
+__all__ = ["PostHogSpanProcessor", "PostHogTraceExporter", "is_ai_span"]

posthog/ai/otel/exporter.py

@@ -0,0 +1,67 @@
+"""PostHog trace exporter for OpenTelemetry.
+
+Provides a SpanExporter that filters AI-related spans before forwarding them
+to PostHog's OTLP endpoint. Use this when your setup only accepts a
+SpanExporter (e.g. as an argument to BatchSpanProcessor).
+"""
+
+from typing import Optional, Sequence
+
+from opentelemetry.sdk.trace import ReadableSpan
+from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
+
+from .spans import DEFAULT_HOST, is_ai_span
+
+
+class PostHogTraceExporter(SpanExporter):
+    """Span exporter that filters AI spans and forwards them to PostHog.
+
+    Wraps an OTLPSpanExporter configured for PostHog's OTLP endpoint. Spans
+    that are not AI-related are silently dropped, returning SUCCESS immediately.
+
+    Usage::
+
+        from opentelemetry.sdk.trace import TracerProvider
+        from opentelemetry.sdk.trace.export import BatchSpanProcessor
+        from posthog.ai.otel import PostHogTraceExporter
+
+        provider = TracerProvider()
+        provider.add_span_processor(
+            BatchSpanProcessor(
+                PostHogTraceExporter(api_key="phc_...")
+            )
+        )
+    """
+
+    def __init__(
+        self,
+        api_key: str,
+        host: str = DEFAULT_HOST,
+    ):
+        """
+        Args:
+            api_key: PostHog project API key.
+            host: PostHog host URL. Defaults to US cloud.
+        """
+        self._api_key = api_key
+        self._host = host.rstrip("/")
+
+        self._exporter = OTLPSpanExporter(
+            endpoint=f"{self._host}/i/v0/ai/otel",
+            headers={"Authorization": f"Bearer {self._api_key}"},
+        )
+
+    def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
+        ai_spans = [span for span in spans if is_ai_span(span)]
+        if not ai_spans:
+            return SpanExportResult.SUCCESS
+        return self._exporter.export(ai_spans)
+
+    def shutdown(self) -> None:
+        self._exporter.shutdown()
+
+    def force_flush(self, timeout_millis: Optional[int] = None) -> bool:
+        if timeout_millis is not None:
+            return self._exporter.force_flush(timeout_millis)
+        return self._exporter.force_flush()

posthog/ai/otel/processor.py

@@ -0,0 +1,69 @@
+"""PostHog span processor for OpenTelemetry.
+
+Provides a self-contained SpanProcessor that filters AI-related spans and
+exports them to PostHog's OTLP endpoint. This is the recommended integration
+for setups using TracerProvider.add_span_processor().
+"""
+
+from typing import Optional
+
+from opentelemetry.context import Context
+from opentelemetry.sdk.trace import ReadableSpan, Span, SpanProcessor
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
+
+from .spans import DEFAULT_HOST, is_ai_span
+
+
+class PostHogSpanProcessor(SpanProcessor):
+    """Span processor that filters AI spans and exports them to PostHog.
+
+    Wraps a BatchSpanProcessor and OTLPSpanExporter internally, configured
+    to send to PostHog's OTLP traces endpoint. Only spans identified as
+    AI-related (by name or attribute prefix) are forwarded for export.
+
+    Usage::
+
+        from opentelemetry.sdk.trace import TracerProvider
+        from posthog.ai.otel import PostHogSpanProcessor
+
+        provider = TracerProvider()
+        provider.add_span_processor(
+            PostHogSpanProcessor(api_key="phc_...")
+        )
+    """
+
+    def __init__(
+        self,
+        api_key: str,
+        host: str = DEFAULT_HOST,
+    ):
+        """
+        Args:
+            api_key: PostHog project API key.
+            host: PostHog host URL. Defaults to US cloud.
+        """
+        self._api_key = api_key
+        self._host = host.rstrip("/")
+
+        exporter = OTLPSpanExporter(
+            endpoint=f"{self._host}/i/v0/ai/otel",
+            headers={"Authorization": f"Bearer {self._api_key}"},
+        )
+        self._processor = BatchSpanProcessor(exporter)
+
+    def on_start(self, span: Span, parent_context: Optional[Context] = None) -> None:
+        pass
+
+    def on_end(self, span: ReadableSpan) -> None:
+        if not is_ai_span(span):
+            return
+        self._processor.on_end(span)
+
+    def shutdown(self) -> None:
+        self._processor.shutdown()
+
+    def force_flush(self, timeout_millis: Optional[int] = None) -> bool:
+        if timeout_millis is not None:
+            return self._processor.force_flush(timeout_millis)
+        return self._processor.force_flush()

posthog/ai/otel/spans.py

@@ -0,0 +1,28 @@
+"""Shared AI span filtering logic and constants for OpenTelemetry integration."""
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from opentelemetry.sdk.trace import ReadableSpan
+
+DEFAULT_HOST = "https://us.i.posthog.com"
+
+AI_SPAN_PREFIXES = ("gen_ai.", "llm.", "ai.", "traceloop.")
+
+
+def is_ai_span(span: "ReadableSpan") -> bool:
+    """Check if a span is AI-related by examining its name and attribute keys.
+
+    Matches spans whose name or any attribute key starts with one of the
+    known AI semantic convention prefixes (gen_ai, llm, ai, traceloop).
+    """
+    name = span.name
+    if any(name.startswith(prefix) for prefix in AI_SPAN_PREFIXES):
+        return True
+
+    attributes = span.attributes or {}
+    for key in attributes:
+        if any(key.startswith(prefix) for prefix in AI_SPAN_PREFIXES):
+            return True
+
+    return False

posthog/test/ai/otel/__init__.py

@@ -0,0 +1,8 @@
+from unittest.mock import MagicMock
+
+
+def make_span(name: str = "test", attributes: dict | None = None) -> MagicMock:
+    span = MagicMock()
+    span.name = name
+    span.attributes = attributes or {}
+    return span

posthog/test/ai/otel/conftest.py

@@ -0,0 +1,98 @@
+import unittest
+from unittest.mock import patch
+
+from opentelemetry.sdk.trace.export import SpanExportResult
+
+from posthog.ai.otel.exporter import PostHogTraceExporter
+from posthog.test.ai.otel.conftest import make_span
+
+
+class TestPostHogTraceExporter(unittest.TestCase):
+    @patch("posthog.ai.otel.exporter.OTLPSpanExporter")
+    def test_configures_exporter_with_correct_endpoint(self, mock_otlp_cls):
+        PostHogTraceExporter(api_key="phc_test123")
+        mock_otlp_cls.assert_called_once_with(
+            endpoint="https://us.i.posthog.com/i/v0/ai/otel",
+            headers={"Authorization": "Bearer phc_test123"},
+        )
+
+    @patch("posthog.ai.otel.exporter.OTLPSpanExporter")
+    def test_configures_custom_host(self, mock_otlp_cls):
+        PostHogTraceExporter(api_key="phc_test", host="https://eu.i.posthog.com")
+        mock_otlp_cls.assert_called_once_with(
+            endpoint="https://eu.i.posthog.com/i/v0/ai/otel",
+            headers={"Authorization": "Bearer phc_test"},
+        )
+
+    @patch("posthog.ai.otel.exporter.OTLPSpanExporter")
+    def test_exports_ai_spans(self, mock_otlp_cls):
+        exporter = PostHogTraceExporter(api_key="phc_test")
+        inner = mock_otlp_cls.return_value
+        inner.export.return_value = SpanExportResult.SUCCESS
+
+        spans = [make_span("gen_ai.chat"), make_span("llm.call")]
+        result = exporter.export(spans)
+
+        self.assertEqual(result, SpanExportResult.SUCCESS)
+        inner.export.assert_called_once_with(spans)
+
+    @patch("posthog.ai.otel.exporter.OTLPSpanExporter")
+    def test_filters_out_non_ai_spans(self, mock_otlp_cls):
+        exporter = PostHogTraceExporter(api_key="phc_test")
+        inner = mock_otlp_cls.return_value
+        inner.export.return_value = SpanExportResult.SUCCESS
+
+        ai_span = make_span("gen_ai.chat")
+        http_span = make_span("http.request")
+        result = exporter.export([ai_span, http_span])
+
+        self.assertEqual(result, SpanExportResult.SUCCESS)
+        inner.export.assert_called_once_with([ai_span])
+
+    @patch("posthog.ai.otel.exporter.OTLPSpanExporter")
+    def test_returns_success_when_no_ai_spans(self, mock_otlp_cls):
+        exporter = PostHogTraceExporter(api_key="phc_test")
+        inner = mock_otlp_cls.return_value
+
+        result = exporter.export([make_span("http.request"), make_span("db.query")])
+
+        self.assertEqual(result, SpanExportResult.SUCCESS)
+        inner.export.assert_not_called()
+
+    @patch("posthog.ai.otel.exporter.OTLPSpanExporter")
+    def test_returns_success_for_empty_batch(self, mock_otlp_cls):
+        exporter = PostHogTraceExporter(api_key="phc_test")
+        inner = mock_otlp_cls.return_value
+
+        result = exporter.export([])
+
+        self.assertEqual(result, SpanExportResult.SUCCESS)
+        inner.export.assert_not_called()
+
+    @patch("posthog.ai.otel.exporter.OTLPSpanExporter")
+    def test_exports_spans_with_ai_attributes(self, mock_otlp_cls):
+        exporter = PostHogTraceExporter(api_key="phc_test")
+        inner = mock_otlp_cls.return_value
+        inner.export.return_value = SpanExportResult.SUCCESS
+
+        span = make_span("http.request", {"gen_ai.system": "openai"})
+        result = exporter.export([span])
+
+        self.assertEqual(result, SpanExportResult.SUCCESS)
+        inner.export.assert_called_once_with([span])
+
+    @patch("posthog.ai.otel.exporter.OTLPSpanExporter")
+    def test_shutdown_delegates(self, mock_otlp_cls):
+        exporter = PostHogTraceExporter(api_key="phc_test")
+        inner = mock_otlp_cls.return_value
+
+        exporter.shutdown()
+        inner.shutdown.assert_called_once()
+
+    @patch("posthog.ai.otel.exporter.OTLPSpanExporter")
+    def test_force_flush_delegates(self, mock_otlp_cls):
+        exporter = PostHogTraceExporter(api_key="phc_test")
+        inner = mock_otlp_cls.return_value
+
+        exporter.force_flush(timeout_millis=5000)
+        inner.force_flush.assert_called_once_with(5000)