fix(aio): never drop or leak multimodal content in ai capture

.sampo/changesets/multimodal-capture-correctness.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: minor
+---
+
+AI capture now records multimodal and structured content (thinking blocks, tool calls, media, and Responses API output items) faithfully across all providers and streaming paths, and redacts base64 media structurally without leaking raw bytes or over-redacting legitimate values.

posthog/ai/anthropic/anthropic.py

@@ -22,7 +22,6 @@
     handle_anthropic_tool_delta,
     finalize_anthropic_tool_input,
 )
-from posthog.ai.sanitization import sanitize_anthropic
 from posthog.client import Client as PostHogClient
 from posthog import setup
 
@@ -169,7 +168,7 @@ def generator():
                         if block:
                             content_blocks.append(block)
 
-                            if block.get("type") == "text":
+                            if block.get("type") in ("text", "thinking"):
                                 current_text_block = block
                             else:
                                 current_text_block = None
@@ -248,16 +247,14 @@ def _capture_streaming_event(
         )
         from posthog.ai.utils import capture_streaming_event
 
-        # Prepare standardized event data
         formatted_input = format_anthropic_streaming_input(kwargs)
-        sanitized_input = sanitize_anthropic(formatted_input, self._client._ph_client)
 
         event_data = StreamingEventData(
             provider="anthropic",
             model=kwargs.get("model", "unknown"),
             base_url=str(self._client.base_url),
             kwargs=kwargs,
-            formatted_input=sanitized_input,
+            formatted_input=formatted_input,
             formatted_output=format_anthropic_streaming_output_complete(
                 content_blocks, accumulated_content
             ),

posthog/ai/anthropic/anthropic_async.py

@@ -24,7 +24,6 @@
     handle_anthropic_tool_delta,
     finalize_anthropic_tool_input,
 )
-from posthog.ai.sanitization import sanitize_anthropic
 from posthog.client import Client as PostHogClient
 
 
@@ -170,7 +169,7 @@ async def generator():
                         if block:
                             content_blocks.append(block)
 
-                            if block.get("type") == "text":
+                            if block.get("type") in ("text", "thinking"):
                                 current_text_block = block
                             else:
                                 current_text_block = None
@@ -249,16 +248,14 @@ async def _capture_streaming_event(
         )
         from posthog.ai.utils import capture_streaming_event
 
-        # Prepare standardized event data
         formatted_input = format_anthropic_streaming_input(kwargs)
-        sanitized_input = sanitize_anthropic(formatted_input, self._client._ph_client)
 
         event_data = StreamingEventData(
             provider="anthropic",
             model=kwargs.get("model", "unknown"),
             base_url=str(self._client.base_url),
             kwargs=kwargs,
-            formatted_input=sanitized_input,
+            formatted_input=formatted_input,
             formatted_output=format_anthropic_streaming_output_complete(
                 content_blocks, accumulated_content
             ),

posthog/ai/anthropic/anthropic_converter.py

@@ -8,6 +8,7 @@
 import json
 from typing import Any, Dict, List, Optional, Tuple
 
+from posthog.ai.media import to_plain
 from posthog.ai.types import (
     FormattedContentItem,
     FormattedFunctionCall,
@@ -69,6 +70,30 @@ def format_anthropic_response(response: Any) -> List[FormattedMessage]:
                 }
                 content.append(function_call)
 
+            elif getattr(choice, "type", None) == "thinking":
+                content.append(
+                    {
+                        "type": "thinking",
+                        "thinking": getattr(choice, "thinking", None),
+                        "signature": getattr(choice, "signature", None),
+                    }
+                )
+
+            elif getattr(choice, "type", None) == "redacted_thinking":
+                content.append(
+                    {
+                        "type": "redacted_thinking",
+                        "data": getattr(choice, "data", None),
+                    }
+                )
+
+            else:
+                # Catches blocks the branches above skip on falsy values (e.g. TextBlock(text=""))
+                # so empty-but-valid content survives instead of being silently dropped.
+                plain = to_plain(choice)
+                if isinstance(plain, dict) and plain.get("type"):
+                    content.append(plain)
+
     if content:
         message: FormattedMessage = {
             "role": "assistant",
@@ -102,10 +127,15 @@ def format_anthropic_input(
     # Add user messages
     if messages:
         for msg in messages:
-            # Messages are already in the correct format, just ensure type safety
+            raw_content = msg.get("content", "")
+            content: Any = (
+                [to_plain(item) for item in raw_content]
+                if isinstance(raw_content, list)
+                else raw_content
+            )
             formatted_msg: FormattedMessage = {
                 "role": msg.get("role", "user"),
-                "content": msg.get("content", ""),
+                "content": content,
             }
             formatted_messages.append(formatted_msg)
 
@@ -161,6 +191,23 @@ def format_anthropic_streaming_content(
                 }
             )
 
+        elif block.get("type") == "thinking":
+            formatted.append(
+                {
+                    "type": "thinking",
+                    "thinking": block.get("thinking") or "",
+                    "signature": block.get("signature"),
+                }
+            )
+
+        elif block.get("type") == "redacted_thinking":
+            formatted.append(
+                {
+                    "type": "redacted_thinking",
+                    "data": block.get("data"),
+                }
+            )
+
     return formatted
 
 
@@ -325,18 +372,39 @@ def handle_anthropic_content_block_start(
         tool_in_progress: ToolInProgress = {"block": tool_block, "input_string": ""}
         return tool_block, tool_in_progress
 
+    elif block.type == "thinking":
+        thinking_block: StreamingContentBlock = {
+            "type": "thinking",
+            "thinking": getattr(block, "thinking", "") or "",
+        }
+        signature = getattr(block, "signature", None)
+        if signature:
+            thinking_block["signature"] = signature
+        return thinking_block, None
+
+    elif block.type == "redacted_thinking":
+        redacted_block: StreamingContentBlock = {
+            "type": "redacted_thinking",
+            "data": getattr(block, "data", None),
+        }
+        return redacted_block, None
+
     return None, None
 
 
 def handle_anthropic_text_delta(
     event: Any, current_block: Optional[StreamingContentBlock]
 ) -> Optional[str]:
     """
-    Handle text delta event from Anthropic streaming.
+    Handle text, thinking, and signature delta events from Anthropic streaming.
+
+    Thinking and signature deltas are accumulated into current_block in place but,
+    unlike text deltas, are not returned — the caller's accumulated_content is a
+    plain-text fallback and thinking output must not leak into it.
 
     Args:
         event: Delta event
-        current_block: Current text block being accumulated
+        current_block: Current block being accumulated
 
     Returns:
         Text delta if present
@@ -354,6 +422,24 @@ def handle_anthropic_text_delta(
 
         return delta_text
 
+    if hasattr(event, "delta") and hasattr(event.delta, "thinking"):
+        delta_thinking = event.delta.thinking or ""
+
+        if current_block is not None and current_block.get("type") == "thinking":
+            thinking_val = current_block.get("thinking")
+            current_block["thinking"] = (thinking_val or "") + delta_thinking
+
+        return None
+
+    if hasattr(event, "delta") and hasattr(event.delta, "signature"):
+        delta_signature = event.delta.signature or ""
+
+        if current_block is not None and current_block.get("type") == "thinking":
+            signature_val = current_block.get("signature")
+            current_block["signature"] = (signature_val or "") + delta_signature
+
+        return None
+
     return None
 
 

posthog/ai/claude_agent_sdk/client.py

@@ -23,6 +23,10 @@
         "Please install the Claude Agent SDK to use this feature: 'pip install claude-agent-sdk'"
     )
 
+from posthog.ai.claude_agent_sdk.formatting import (
+    format_assistant_blocks,
+    format_tool_result_content,
+)
 from posthog.ai.claude_agent_sdk.processor import (
     PostHogClaudeAgentProcessor,
     _GenerationTracker,
@@ -159,7 +163,6 @@ async def receive_response(self):
                         self._tracker.current_span_id
                         or self._current_generation_span_id
                     )
-                    output_content: List[Dict[str, Any]] = []
                     for block in message.content:
                         if isinstance(block, ToolUseBlock):
                             self._processor._emit_tool_span(
@@ -171,17 +174,7 @@ async def receive_response(self):
                                 self._privacy,
                                 self._groups,
                             )
-                            output_content.append(
-                                {
-                                    "type": "function",
-                                    "function": {
-                                        "name": block.name,
-                                        "arguments": block.input,
-                                    },
-                                }
-                            )
-                        elif hasattr(block, "text"):
-                            output_content.append({"type": "text", "text": block.text})
+                    output_content = format_assistant_blocks(message.content)
                     if output_content:
                         self._pending_output = [
                             {"role": "assistant", "content": output_content}
@@ -199,9 +192,9 @@ async def receive_response(self):
                                     {
                                         "type": "tool_result",
                                         "tool_use_id": block.tool_use_id,
-                                        "content": str(block.content)[:500]
-                                        if block.content
-                                        else None,
+                                        "content": format_tool_result_content(
+                                            block, self._processor._client
+                                        ),
                                     }
                                 )
                             elif hasattr(block, "text"):

posthog/ai/claude_agent_sdk/formatting.py

@@ -0,0 +1,55 @@
+from typing import Any, Dict, List
+
+try:
+    from claude_agent_sdk import ToolUseBlock
+except ImportError:
+    raise ModuleNotFoundError(
+        "Please install the Claude Agent SDK to use this feature: 'pip install claude-agent-sdk'"
+    )
+
+from posthog.ai.media import to_plain
+from posthog.ai.sanitization import redact_media
+
+
+def format_tool_result_content(block: Any, ph_client: Any = None) -> Any:
+    """Structured, redacted tool-result content (replaces str(block.content)[:500]).
+
+    Calls redact_media directly (not finalize_ai_content) for the
+    max_string_len truncation, which finalize_ai_content doesn't support.
+    The processor.py/client.py emit sites re-run finalize_ai_content on the
+    already-redacted result when building $ai_input; that's a no-op since
+    placeholders don't re-match.
+    """
+    content = block.content
+    if isinstance(content, list):
+        content = [to_plain(c) for c in content]
+    return redact_media(content, max_string_len=5000, ph_client=ph_client)
+
+
+def format_assistant_blocks(blocks: Any) -> List[Dict[str, Any]]:
+    out: List[Dict[str, Any]] = []
+    for block in blocks:
+        if getattr(block, "thinking", None) is not None:
+            thinking_block: Dict[str, Any] = {
+                "type": "thinking",
+                "thinking": block.thinking,
+            }
+            signature = getattr(block, "signature", None)
+            if signature is not None:
+                thinking_block["signature"] = signature
+            out.append(thinking_block)
+        elif isinstance(block, ToolUseBlock):
+            out.append(
+                {
+                    "type": "function",
+                    "function": {
+                        "name": block.name,
+                        "arguments": block.input,
+                    },
+                }
+            )
+        elif getattr(block, "text", None) is not None:
+            out.append({"type": "text", "text": block.text})
+        else:
+            out.append(to_plain(block))
+    return out