Python: feat(anthropic): support ImageContent multimodal vision blocks in user messages

python/semantic_kernel/connectors/ai/anthropic/services/utils.py

@@ -1,5 +1,6 @@
 # Copyright (c) Microsoft. All rights reserved.
 
+import base64
 import json
 import logging
 from collections.abc import Callable, Mapping
@@ -9,8 +10,10 @@
 from semantic_kernel.contents.chat_message_content import ChatMessageContent
 from semantic_kernel.contents.function_call_content import FunctionCallContent
 from semantic_kernel.contents.function_result_content import FunctionResultContent
+from semantic_kernel.contents.image_content import ImageContent
 from semantic_kernel.contents.text_content import TextContent
 from semantic_kernel.contents.utils.author_role import AuthorRole
+from semantic_kernel.exceptions.service_exceptions import ServiceInvalidRequestError
 from semantic_kernel.functions.kernel_function_metadata import KernelFunctionMetadata
 
 logger: logging.Logger = logging.getLogger(__name__)
@@ -21,6 +24,102 @@
     from semantic_kernel.connectors.ai.prompt_execution_settings import PromptExecutionSettings
 
 
+ANTHROPIC_SUPPORTED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp"}
+
+
+def _infer_image_mime_type(raw_bytes: bytes | bytearray | None, mime_type: str | None) -> str:
+    """Infer or validate a supported image MIME type for Anthropic vision."""
+    if mime_type and mime_type in ANTHROPIC_SUPPORTED_IMAGE_TYPES:
+        return mime_type
+    if raw_bytes:
+        prefix = bytes(raw_bytes[:16])
+        if prefix.startswith(b"\x89PNG\r\n\x1a\n"):
+            return "image/png"
+        if prefix.startswith(b"\xff\xd8\xff"):
+            return "image/jpeg"
+        if prefix.startswith((b"GIF87a", b"GIF89a")):
+            return "image/gif"
+        if prefix.startswith(b"RIFF") and b"WEBP" in prefix:
+            return "image/webp"
+    if mime_type and mime_type in ANTHROPIC_SUPPORTED_IMAGE_TYPES:
+        return mime_type
+
+    if mime_type == "text/plain" or not mime_type:
+        return "image/jpeg"
+
+    raise ServiceInvalidRequestError(
+        f"Unsupported image format: '{mime_type}'. Anthropic Claude vision only supports "
+        f"{', '.join(sorted(ANTHROPIC_SUPPORTED_IMAGE_TYPES))}."
+    )
+
+
+def _has_inline_image_data(image_content: ImageContent) -> bool:
+    """Check whether the ImageContent carries inline image data (not just a remote reference)."""
+    data = image_content.data
+    if isinstance(data, (bytes, bytearray)):
+        # BinaryContent defaults data to b"" when only a URI is given.
+        return len(data) > 0
+    return bool(image_content.data_string) or data is not None
+
+
+def _create_base64_image_content(image_content: ImageContent) -> dict[str, Any]:
+    """Create an Anthropic base64 image content block from inline image data."""
+    raw_bytes = image_content.data if isinstance(image_content.data, (bytes, bytearray)) else None
+    mime_type = _infer_image_mime_type(raw_bytes, image_content.mime_type)
+
+    if raw_bytes is not None:
+        data = base64.b64encode(raw_bytes).decode("utf-8")
+    elif image_content.data_string:
+        data = image_content.data_string
+    elif image_content.data is not None:
+        data = str(image_content.data)
+    else:
+        raise ServiceInvalidRequestError(
+            "ImageContent without data, data_uri, or valid http(s) uri while formatting message for Anthropic."
+        )
+
+    if not data:
+        raise ServiceInvalidRequestError(
+            "ImageContent without data, data_uri, or valid http(s) uri while formatting message for Anthropic."
+        )
+
+    return {
+        "type": "image",
+        "source": {
+            "type": "base64",
+            "media_type": mime_type,
+            "data": data,
+        },
+    }
+
+
+def _create_image_content(image_content: ImageContent) -> dict[str, Any]:
+    """Create an Anthropic image content block from an ImageContent object.
+
+    Inline image data takes precedence over a remote URL: per ImageContent's
+    contract, its URI may refer to different content than its data, so sending
+    the URL when bytes are available can send the wrong image (or fail for
+    URLs requiring authentication).
+    """
+    if _has_inline_image_data(image_content):
+        return _create_base64_image_content(image_content)
+
+    if image_content.uri:
+        uri_str = str(image_content.uri)
+        if uri_str.startswith(("http://", "https://")):
+            return {
+                "type": "image",
+                "source": {
+                    "type": "url",
+                    "url": uri_str,
+                },
+            }
+
+    raise ServiceInvalidRequestError(
+        "ImageContent without data, data_uri, or valid http(s) uri while formatting message for Anthropic."
+    )
+
+
 def _format_user_message(message: ChatMessageContent) -> dict[str, Any]:
     """Format a user message to the expected object for the Anthropic client.
 
@@ -30,9 +129,30 @@ def _format_user_message(message: ChatMessageContent) -> dict[str, Any]:
     Returns:
         The formatted user message.
     """
+    if not message.items or (len(message.items) == 1 and isinstance(message.items[0], TextContent)):
+        return {
+            "role": "user",
+            "content": message.content,
+        }
+
+    contents: list[dict[str, Any]] = []
+    for item in message.items:
+        if isinstance(item, TextContent):
+            if item.text:
+                contents.append({
+                    "type": "text",
+                    "text": item.text,
+                })
+        elif isinstance(item, ImageContent):
+            contents.append(_create_image_content(item))
+        else:
+            logger.warning(
+                f"Unsupported item type in User message while formatting chat history for Anthropic: {type(item)}"
+            )
+
     return {
         "role": "user",
-        "content": message.content,
+        "content": contents,
     }
 
 

python/tests/unit/connectors/ai/anthropic/services/test_anthropic_chat_completion.py

@@ -16,8 +16,10 @@
 )
 from semantic_kernel.contents.chat_history import ChatHistory
 from semantic_kernel.contents.chat_message_content import ChatMessageContent, FunctionCallContent, TextContent
+from semantic_kernel.contents.image_content import ImageContent
 from semantic_kernel.contents.streaming_chat_message_content import StreamingChatMessageContent
 from semantic_kernel.contents.utils.author_role import AuthorRole
+from semantic_kernel.connectors.ai.anthropic.services.utils import _format_user_message
 from semantic_kernel.exceptions import (
     ServiceInitializationError,
     ServiceInvalidExecutionSettingsError,
@@ -293,6 +295,181 @@ def test_prompt_execution_settings_class(anthropic_unit_test_env):
     assert prompt_execution_settings == AnthropicChatPromptExecutionSettings
 
 
+def test_format_user_message_text_only():
+    message = ChatMessageContent(role=AuthorRole.USER, content="Hello Anthropic")
+    formatted = _format_user_message(message)
+    assert formatted == {"role": "user", "content": "Hello Anthropic"}
+
+
+def test_format_user_message_with_image():
+    dummy_b64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg=="
+    message = ChatMessageContent(
+        role=AuthorRole.USER,
+        items=[
+            TextContent(text="Describe this image"),
+            ImageContent(data_uri=f"data:image/png;base64,{dummy_b64}"),
+        ],
+    )
+    formatted = _format_user_message(message)
+    assert formatted["role"] == "user"
+    assert len(formatted["content"]) == 2
+    assert formatted["content"][0] == {"type": "text", "text": "Describe this image"}
+    assert formatted["content"][1] == {
+        "type": "image",
+        "source": {
+            "type": "base64",
+            "media_type": "image/png",
+            "data": dummy_b64,
+        },
+    }
+
+
+def test_format_user_message_image_only():
+    dummy_b64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg=="
+    message = ChatMessageContent(
+        role=AuthorRole.USER,
+        items=[
+            ImageContent(data_uri=f"data:image/png;base64,{dummy_b64}"),
+        ],
+    )
+    formatted = _format_user_message(message)
+    assert formatted["role"] == "user"
+    assert len(formatted["content"]) == 1
+    assert formatted["content"][0]["type"] == "image"
+    assert formatted["content"][0]["source"]["media_type"] == "image/png"
+    assert formatted["content"][0]["source"]["data"] == dummy_b64
+
+
+def test_format_user_message_raw_bytes():
+    raw_png = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x06\x00\x00\x00\x1f\x15c4"
+    message = ChatMessageContent(
+        role=AuthorRole.USER,
+        items=[
+            ImageContent(data=raw_png, mime_type="image/png"),
+        ],
+    )
+    formatted = _format_user_message(message)
+    assert formatted["role"] == "user"
+    assert len(formatted["content"]) == 1
+    assert formatted["content"][0]["type"] == "image"
+    assert formatted["content"][0]["source"]["media_type"] == "image/png"
+    import base64
+    assert formatted["content"][0]["source"]["data"] == base64.b64encode(raw_png).decode("utf-8")
+
+
+def test_format_user_message_image_url_only():
+    url = "https://example.com/sample_image.png"
+    message = ChatMessageContent(
+        role=AuthorRole.USER,
+        items=[
+            ImageContent(uri=url),
+        ],
+    )
+    formatted = _format_user_message(message)
+    assert formatted["role"] == "user"
+    assert len(formatted["content"]) == 1
+    assert formatted["content"][0]["type"] == "image"
+    assert formatted["content"][0]["source"]["type"] == "url"
+    assert formatted["content"][0]["source"]["url"] == url
+
+
+def test_format_user_message_dual_source_prefers_inline_bytes():
+    # ImageContent documents that its URI may refer to different content than
+    # its data, so inline bytes must win over the remote URL.
+    raw_png = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x06\x00\x00\x00\x1f\x15c4"
+    message = ChatMessageContent(
+        role=AuthorRole.USER,
+        items=[
+            ImageContent(
+                uri="https://example.com/remote.png",
+                data=raw_png,
+                mime_type="image/png",
+            ),
+        ],
+    )
+    formatted = _format_user_message(message)
+    assert formatted["role"] == "user"
+    assert len(formatted["content"]) == 1
+    assert formatted["content"][0]["type"] == "image"
+    assert formatted["content"][0]["source"]["type"] == "base64"
+    assert formatted["content"][0]["source"]["media_type"] == "image/png"
+    import base64
+    assert formatted["content"][0]["source"]["data"] == base64.b64encode(raw_png).decode("utf-8")
+
+
+def test_format_user_message_dual_source_prefers_data_uri():
+    dummy_b64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg=="
+    message = ChatMessageContent(
+        role=AuthorRole.USER,
+        items=[
+            ImageContent(
+                uri="https://example.com/remote.png",
+                data_uri=f"data:image/png;base64,{dummy_b64}",
+            ),
+        ],
+    )
+    formatted = _format_user_message(message)
+    assert formatted["role"] == "user"
+    assert len(formatted["content"]) == 1
+    assert formatted["content"][0]["type"] == "image"
+    assert formatted["content"][0]["source"]["type"] == "base64"
+    assert formatted["content"][0]["source"]["media_type"] == "image/png"
+    assert formatted["content"][0]["source"]["data"] == dummy_b64
+
+
+def test_format_user_message_raw_bytes_default_mime_type():
+    raw_png = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x06\x00\x00\x00\x1f\x15c4"
+    # When mime_type is omitted, BinaryContent defaults to text/plain
+    message = ChatMessageContent(
+        role=AuthorRole.USER,
+        items=[
+            ImageContent(data=raw_png),
+        ],
+    )
+    formatted = _format_user_message(message)
+    assert formatted["role"] == "user"
+    assert len(formatted["content"]) == 1
+    assert formatted["content"][0]["type"] == "image"
+    assert formatted["content"][0]["source"]["media_type"] == "image/png"
+
+
+def test_format_user_message_jpeg_inferred_mime_type():
+    raw_jpeg = b"\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x01\x00`\x00`\x00\x00"
+    message = ChatMessageContent(
+        role=AuthorRole.USER,
+        items=[
+            ImageContent(data=raw_jpeg),
+        ],
+    )
+    formatted = _format_user_message(message)
+    assert formatted["role"] == "user"
+    assert len(formatted["content"]) == 1
+    assert formatted["content"][0]["type"] == "image"
+    assert formatted["content"][0]["source"]["media_type"] == "image/jpeg"
+
+
+def test_format_user_message_unsupported_svg_image_raises():
+    message = ChatMessageContent(
+        role=AuthorRole.USER,
+        items=[
+            ImageContent(data=b"<svg></svg>", mime_type="image/svg+xml"),
+        ],
+    )
+    with pytest.raises(ServiceInvalidRequestError, match="Unsupported image format: 'image/svg\\+xml'"):
+        _format_user_message(message)
+
+
+def test_format_user_message_unsupported_bmp_data_uri_raises():
+    message = ChatMessageContent(
+        role=AuthorRole.USER,
+        items=[
+            ImageContent(data_uri="data:image/bmp;base64,Qk0="),
+        ],
+    )
+    with pytest.raises(ServiceInvalidRequestError, match="Unsupported image format: 'image/bmp'"):
+        _format_user_message(message)
+
+
 async def test_with_different_execution_settings(kernel: Kernel, mock_anthropic_client_completion: MagicMock):
     chat_history = MagicMock()
     settings = OpenAIChatPromptExecutionSettings(temperature=0.2)