ref(pydantic-ai): Extract content serialization helpers and add ImageUrl support

Move shared binary/image serialization logic from ai_client.py and invoke_agent.py
into a shared _serialize_binary_content_item and _serialize_image_url_item helpers
in spans/utils.py. Add support for pydantic-ai's ImageUrl type, with base64 data
URLs redacted and regular HTTP URLs preserved. Remove unused tool_definition
parameter from execute_tool_span.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: dingsdax

sentry_sdk/integrations/pydantic_ai/consts.py

@@ -1 +1,9 @@
+import re
+
 SPAN_ORIGIN = "auto.ai.pydantic_ai"
+
+# Matches data URLs with base64-encoded content, e.g. "data:image/png;base64,iVBORw0K..."
+# Group 1: MIME type (e.g. "image/png"), Group 2: base64 data
+DATA_URL_BASE64_REGEX = re.compile(
+    r"^data:([a-zA-Z]+/[a-zA-Z]+);base64,([A-Za-z0-9+/\-_]+={0,2})$"
+)

sentry_sdk/integrations/pydantic_ai/patches/tools.py

@@ -50,7 +50,6 @@ async def wrapped_execute_tool_call(
         call = validated.call
         name = call.tool_name
         tool = self.tools.get(name) if self.tools else None
-        selected_tool_definition = getattr(tool, "tool_def", None)
 
         # Determine tool type by checking tool.toolset
         tool_type = "function"
@@ -74,7 +73,6 @@ async def wrapped_execute_tool_call(
                     args_dict,
                     agent,
                     tool_type=tool_type,
-                    tool_definition=selected_tool_definition,
                 ) as span:
                     try:
                         result = await original_execute_tool_call(
@@ -129,7 +127,6 @@ async def wrapped_call_tool(
         # Extract tool info before calling original
         name = call.tool_name
         tool = self.tools.get(name) if self.tools else None
-        selected_tool_definition = getattr(tool, "tool_def", None)
 
         # Determine tool type by checking tool.toolset
         tool_type = "function"  # default
@@ -153,7 +150,6 @@ async def wrapped_call_tool(
                     args_dict,
                     agent,
                     tool_type=tool_type,
-                    tool_definition=selected_tool_definition,
                 ) as span:
                     try:
                         result = await original_call_tool(

sentry_sdk/integrations/pydantic_ai/spans/ai_client.py

@@ -1,12 +1,10 @@
 import json
 
 import sentry_sdk
-from sentry_sdk._types import BLOB_DATA_SUBSTITUTE
 from sentry_sdk.ai.utils import (
     normalize_message_roles,
     set_data_normalized,
     truncate_and_annotate_messages,
-    get_modality_from_mime_type,
 )
 from sentry_sdk.consts import OP, SPANDATA
 from sentry_sdk.utils import safe_serialize
@@ -21,7 +19,11 @@
     get_current_agent,
     get_is_streaming,
 )
-from .utils import _set_usage_data
+from .utils import (
+    _serialize_binary_content_item,
+    _serialize_image_url_item,
+    _set_usage_data,
+)
 
 from typing import TYPE_CHECKING
 
@@ -40,6 +42,7 @@
         TextPart,
         ThinkingPart,
         BinaryContent,
+        ImageUrl,
     )
 except ImportError:
     # Fallback if these classes are not available
@@ -50,6 +53,7 @@
     TextPart = None
     ThinkingPart = None
     BinaryContent = None
+    ImageUrl = None
 
 
 def _transform_system_instructions(
@@ -158,22 +162,14 @@ def _set_input_messages(span: "sentry_sdk.tracing.Span", messages: "Any") -> Non
                             for item in part.content:
                                 if isinstance(item, str):
                                     content.append({"type": "text", "text": item})
+                                elif ImageUrl and isinstance(item, ImageUrl):
+                                    content.append(_serialize_image_url_item(item))
                                 elif BinaryContent and isinstance(item, BinaryContent):
-                                    content.append(
-                                        {
-                                            "type": "blob",
-                                            "modality": get_modality_from_mime_type(
-                                                item.media_type
-                                            ),
-                                            "mime_type": item.media_type,
-                                            "content": BLOB_DATA_SUBSTITUTE,
-                                        }
-                                    )
+                                    content.append(_serialize_binary_content_item(item))
                                 else:
                                     content.append(safe_serialize(item))
                         else:
                             content.append({"type": "text", "text": str(part.content)})
-
                     # Add message if we have content or tool calls
                     if content or tool_calls:
                         message: "Dict[str, Any]" = {"role": role}

sentry_sdk/integrations/pydantic_ai/spans/execute_tool.py

@@ -9,15 +9,10 @@
 
 if TYPE_CHECKING:
     from typing import Any, Optional
-    from pydantic_ai._tool_manager import ToolDefinition  # type: ignore
 
 
 def execute_tool_span(
-    tool_name: str,
-    tool_args: "Any",
-    agent: "Any",
-    tool_type: str = "function",
-    tool_definition: "Optional[ToolDefinition]" = None,
+    tool_name: str, tool_args: "Any", agent: "Any", tool_type: str = "function"
 ) -> "sentry_sdk.tracing.Span":
     """Create a span for tool execution.
 
@@ -26,7 +21,6 @@ def execute_tool_span(
         tool_args: The arguments passed to the tool
         agent: The agent executing the tool
         tool_type: The type of tool ("function" for regular tools, "mcp" for MCP services)
-        tool_definition: The definition of the tool, if available
     """
     span = sentry_sdk.start_span(
         op=OP.GEN_AI_EXECUTE_TOOL,
@@ -38,12 +32,6 @@ def execute_tool_span(
     span.set_data(SPANDATA.GEN_AI_TOOL_TYPE, tool_type)
     span.set_data(SPANDATA.GEN_AI_TOOL_NAME, tool_name)
 
-    if tool_definition is not None and hasattr(tool_definition, "description"):
-        span.set_data(
-            SPANDATA.GEN_AI_TOOL_DESCRIPTION,
-            tool_definition.description,
-        )
-
     _set_agent_data(span, agent)
 
     if _should_send_prompts() and tool_args is not None:

sentry_sdk/integrations/pydantic_ai/spans/invoke_agent.py

@@ -1,7 +1,5 @@
 import sentry_sdk
-from sentry_sdk._types import BLOB_DATA_SUBSTITUTE
 from sentry_sdk.ai.utils import (
-    get_modality_from_mime_type,
     get_start_span_function,
     normalize_message_roles,
     set_data_normalized,
@@ -16,17 +14,22 @@
     _set_model_data,
     _should_send_prompts,
 )
-from .utils import _set_usage_data
+from .utils import (
+    _serialize_binary_content_item,
+    _serialize_image_url_item,
+    _set_usage_data,
+)
 
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
     from typing import Any
 
 try:
-    from pydantic_ai.messages import BinaryContent  # type: ignore
+    from pydantic_ai.messages import BinaryContent, ImageUrl  # type: ignore
 except ImportError:
     BinaryContent = None
+    ImageUrl = None
 
 
 def invoke_agent_span(
@@ -105,17 +108,10 @@ def invoke_agent_span(
                 for item in user_prompt:
                     if isinstance(item, str):
                         content.append({"text": item, "type": "text"})
+                    elif ImageUrl and isinstance(item, ImageUrl):
+                        content.append(_serialize_image_url_item(item))
                     elif BinaryContent and isinstance(item, BinaryContent):
-                        content.append(
-                            {
-                                "type": "blob",
-                                "modality": get_modality_from_mime_type(
-                                    item.media_type
-                                ),
-                                "mime_type": item.media_type,
-                                "content": BLOB_DATA_SUBSTITUTE,
-                            }
-                        )
+                        content.append(_serialize_binary_content_item(item))
                 if content:
                     messages.append(
                         {

sentry_sdk/integrations/pydantic_ai/spans/utils.py

@@ -1,14 +1,56 @@
 """Utility functions for PydanticAI span instrumentation."""
 
 import sentry_sdk
+from sentry_sdk._types import BLOB_DATA_SUBSTITUTE
+from sentry_sdk.ai.utils import get_modality_from_mime_type
 from sentry_sdk.consts import SPANDATA
 
+from ..consts import DATA_URL_BASE64_REGEX
+
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
-    from typing import Union, Dict, Any, List
+    from typing import Union, Dict, Any, List, Optional
     from pydantic_ai.usage import RequestUsage, RunUsage  # type: ignore
 
+try:
+    from pydantic_ai.messages import BinaryContent, ImageUrl  # type: ignore
+except ImportError:
+    BinaryContent = None
+    ImageUrl = None
+
+
+def _serialize_image_url_item(item: "Any") -> "Dict[str, Any]":
+    """Serialize an ImageUrl content item for span data.
+
+    For data URLs containing base64-encoded images, the content is redacted.
+    For regular HTTP URLs, the URL string is preserved.
+    """
+    data_url_matches = DATA_URL_BASE64_REGEX.match(item.url)
+
+    if data_url_matches:
+        mime_type = data_url_matches[1] or "image"
+        return {
+            "type": "image",
+            "mime_type": mime_type,
+            "content": BLOB_DATA_SUBSTITUTE,
+        }
+
+    return {
+        "type": "image",
+        "content": str(item.url),
+    }
+
+
+def _serialize_binary_content_item(item: "Any") -> "Dict[str, Any]":
+    """Serialize a BinaryContent item for span data, redacting the blob data."""
+    return {
+        "type": "blob",
+        "modality": get_modality_from_mime_type(item.media_type),
+        "mime_type": item.media_type,
+        "content": BLOB_DATA_SUBSTITUTE,
+    }
+
 
 def _set_usage_data(
     span: "sentry_sdk.tracing.Span", usage: "Union[RequestUsage, RunUsage]"