fix: sending OpenAI-style image_url causes Anthropic 400 invalid tag error

[KBVsent]

When using Anthropic as the LLM provider, processing quoted images in messages fails with error:

Error code: 400 - {'type': 'error', 'error': {'type': 'invalid_request_error', 'message': "messages.0.content.2: Input tag 'image_url' found using 'type' does not match any of the expected tags: 'document', 'image', 'redacted_thinking', 'search_result', 'server_tool_use', 'text', 'thinking', 'tool_result', 'tool_use', 'web_search_tool_result'"}, 'request_id': 'req_011CX2qbvsswzamt2xxxxq2D'}

Historical context messages from entities.py use OpenAI format (type: "image_url") which Anthropic API doesn't recognize

Modifications / 改动点

Modified: anthropic_source.py

• Updated _prepare_payload() method to handle user messages containing image_url type content
• Convert OpenAI image_url format to Anthropic image format with proper structure
• Decode base64 image data and use existing _detect_image_mime_type() method to detect actual MIME type from magic bytes

• This is NOT a breaking change. / 这不是一个破坏性变更。

Screenshots or Test Results / 运行截图或测试结果

---

Checklist / 检查清单

• 😊 如果 PR 中有新加入的功能，已经通过 Issue / 邮件等方式和作者讨论过。/ If there are new features added in the PR, I have discussed it with the authors through issues/emails, etc.
• 👀 我的更改经过了良好的测试，并已在上方提供了“验证步骤”和“运行截图”。/ My changes have been well-tested, and "Verification Steps" and "Screenshots" have been provided above.
• 🤓 我确保没有引入新依赖库，或者引入了新依赖库的同时将其添加到了 requirements.txt 和 pyproject.toml 文件相应位置。/ I have ensured that no new dependencies are introduced, OR if new dependencies are introduced, they have been added to the appropriate locations in requirements.txt and pyproject.toml.
• 😮 我的更改没有引入恶意代码。/ My changes do not introduce malicious code.

Summary by Sourcery

在为 Anthropic 准备请求负载时，处理包含 OpenAI 风格 image_url 内容的用户消息，以避免出现无效标签错误。

Bug Fixes:

• 将用户消息中 OpenAI 风格的 image_url 内容转换为与 Anthropic 兼容的图像内容，防止在引用图像时出现 400 invalid_request_error 响应。

Enhancements:

• 对不受支持或格式错误的图像数据 URI 进行优雅地跳过或发出警告，同时在 Anthropic 请求中保留用户消息的其他内容。

Original summary in English

Summary by Sourcery

Handle user messages with OpenAI-style image_url content when preparing Anthropic payloads to avoid invalid tag errors.

Bug Fixes:

• Convert OpenAI-style image_url content in user messages into Anthropic-compatible image content to prevent 400 invalid_request_error responses when images are quoted.

Enhancements:

• Gracefully skip or warn on unsupported or malformed image data URIs while preserving other user message content in Anthropic requests.

[sourcery-ai]

Hey - 我发现了两个问题，并给出了一些整体性的反馈：

• 当一个 data: URL 解析失败或处于不支持的格式时，对应的 image part 会在仅记录 warning 的情况下被静默丢弃；建议要么保留原始的 image_url part，要么增加一个清晰的回退逻辑，以避免消息语义被无意间改变。
• 针对 user 消息的处理现在有了自己的循环和内容转换逻辑；如果在其他地方也有对消息 part 做归一化处理的逻辑，建议抽取一个共享的 helper，避免对类似内容结构产生行为差异。

给 AI Agents 的提示词

Please address the comments from this code review:

## Overall Comments
- When a `data:` URL fails to parse or is in an unsupported format, the corresponding image part is silently dropped while only logging a warning; consider either preserving the original `image_url` part or adding a clear fallback so message semantics are not unintentionally changed.
- The `user`-message handling now has its own loop and content transformation logic; if there are other places that normalize message parts, consider extracting a shared helper to avoid diverging behavior for similar content structures.

## Individual Comments

### Comment 1
<location> `astrbot/core/provider/sources/anthropic_source.py:139-143` </location>
<code_context>
+                            image_url_data = part.get("image_url", {})
+                            url = image_url_data.get("url", "")
+                            if url.startswith("data:"):
+                                try:
+                                    _, base64_data = url.split(",", 1)
+                                    # Detect actual image format from binary data
+                                    image_bytes = base64.b64decode(base64_data)
+                                    media_type = self._detect_image_mime_type(
+                                        image_bytes
+                                    )
</code_context>

<issue_to_address>
**suggestion (bug_risk):** Exception handling around base64 decoding is too narrow and may miss decoding errors.

The current `try`/`except ValueError` also wraps `base64.b64decode`, which raises `binascii.Error` for invalid input, so malformed data will currently escape this handler. Please either catch `binascii.Error` explicitly (in addition to `ValueError`) or broaden the except clause (e.g., to `Exception`) and log the error so bad image data doesn’t cause `_prepare_payload` to fail.

Suggested implementation:

```python
import base64
import binascii

```

```python
                                except (ValueError, binascii.Error):

```
</issue_to_address>

### Comment 2
<location> `astrbot/core/provider/sources/anthropic_source.py:130` </location>
<code_context>
                         ],
                     },
                 )
+            elif message["role"] == "user":
+                if isinstance(message.get("content"), list):
+                    converted_content = []
</code_context>

<issue_to_address>
**issue (complexity):** Consider extracting the user message image-conversion logic into dedicated helper methods so `_prepare_payload` stays focused on simple message shaping and control flow.

You can reduce the added complexity by extracting the user-content/image conversion into small helpers and keeping `_prepare_payload` focused on message shaping.

For example, instead of inlining all of this logic:

```python
elif message["role"] == "user":
    if isinstance(message.get("content"), list):
        converted_content = []
        for part in message["content"]:
            if part.get("type") == "image_url":
                # Convert OpenAI image_url format to Anthropic image format
                image_url_data = part.get("image_url", {})
                url = image_url_data.get("url", "")
                if url.startswith("data:"):
                    try:
                        _, base64_data = url.split(",", 1)
                        # Detect actual image format from binary data
                        image_bytes = base64.b64decode(base64_data)
                        media_type = self._detect_image_mime_type(image_bytes)
                        converted_content.append(
                            {
                                "type": "image",
                                "source": {
                                    "type": "base64",
                                    "media_type": media_type,
                                    "data": base64_data,
                                },
                            }
                        )
                    except ValueError:
                        logger.warning(
                            f"Failed to parse image data URI: {url[:50]}..."
                        )
                else:
                    logger.warning(
                        f"Unsupported image URL format for Anthropic: {url[:50]}..."
                    )
            else:
                converted_content.append(part)
        new_messages.append(
            {
                "role": "user",
                "content": converted_content,
            }
        )
    else:
        new_messages.append(message)
```

you can move the conversion into helpers and keep the control flow flatter:

```python
def _convert_user_content(self, content):
    if not isinstance(content, list):
        return content

    converted = []
    for part in content:
        if part.get("type") == "image_url":
            converted_part = self._convert_image_part(part)
            if converted_part is not None:
                converted.append(converted_part)
            # if conversion failed, we just drop the part or could append original,
            # depending on current behavior you want to preserve
        else:
            converted.append(part)
    return converted

def _convert_image_part(self, part):
    image_url_data = part.get("image_url", {})
    url = image_url_data.get("url", "")
    if not url.startswith("data:"):
        logger.warning(
            f"Unsupported image URL format for Anthropic: {url[:50]}..."
        )
        return None

    try:
        _, base64_data = url.split(",", 1)
        image_bytes = base64.b64decode(base64_data)
        media_type = self._detect_image_mime_type(image_bytes)
        return {
            "type": "image",
            "source": {
                "type": "base64",
                "media_type": media_type,
                "data": base64_data,
            },
        }
    except ValueError:
        logger.warning(
            f"Failed to parse image data URI: {url[:50]}..."
        )
        return None
```

Then `_prepare_payload` only needs:

```python
elif message["role"] == "user":
    content = self._convert_user_content(message.get("content"))
    new_messages.append(
        {
            "role": "user",
            "content": content,
        }
        if isinstance(content, list)
        else {**message, "content": content}
    )
```

Benefits:

- Flattens nesting in `_prepare_payload` to a single `elif role == 'user'` branch.
- Localizes base64/MIME/logging details in `_convert_image_part`, making it easier to test and reason about.
- Separates “how to convert content” from “how to append the message,” which makes control flow clearer and avoids duplicated `new_messages.append` shapes.
</issue_to_address>

---

Sourcery 对开源项目是免费的——如果你觉得我们的代码评审有帮助，欢迎分享 ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{帮我变得更有用！请在每条评论上点 👍 或 👎，我会根据你的反馈改进之后的评审。}

Original comment in English

Hey - I've found 2 issues, and left some high level feedback:

• When a data: URL fails to parse or is in an unsupported format, the corresponding image part is silently dropped while only logging a warning; consider either preserving the original image_url part or adding a clear fallback so message semantics are not unintentionally changed.
• The user-message handling now has its own loop and content transformation logic; if there are other places that normalize message parts, consider extracting a shared helper to avoid diverging behavior for similar content structures.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- When a `data:` URL fails to parse or is in an unsupported format, the corresponding image part is silently dropped while only logging a warning; consider either preserving the original `image_url` part or adding a clear fallback so message semantics are not unintentionally changed.
- The `user`-message handling now has its own loop and content transformation logic; if there are other places that normalize message parts, consider extracting a shared helper to avoid diverging behavior for similar content structures.

## Individual Comments

### Comment 1
<location> `astrbot/core/provider/sources/anthropic_source.py:139-143` </location>
<code_context>
+                            image_url_data = part.get("image_url", {})
+                            url = image_url_data.get("url", "")
+                            if url.startswith("data:"):
+                                try:
+                                    _, base64_data = url.split(",", 1)
+                                    # Detect actual image format from binary data
+                                    image_bytes = base64.b64decode(base64_data)
+                                    media_type = self._detect_image_mime_type(
+                                        image_bytes
+                                    )
</code_context>

<issue_to_address>
**suggestion (bug_risk):** Exception handling around base64 decoding is too narrow and may miss decoding errors.

The current `try`/`except ValueError` also wraps `base64.b64decode`, which raises `binascii.Error` for invalid input, so malformed data will currently escape this handler. Please either catch `binascii.Error` explicitly (in addition to `ValueError`) or broaden the except clause (e.g., to `Exception`) and log the error so bad image data doesn’t cause `_prepare_payload` to fail.

Suggested implementation:

```python
import base64
import binascii

```

```python
                                except (ValueError, binascii.Error):

```
</issue_to_address>

### Comment 2
<location> `astrbot/core/provider/sources/anthropic_source.py:130` </location>
<code_context>
                         ],
                     },
                 )
+            elif message["role"] == "user":
+                if isinstance(message.get("content"), list):
+                    converted_content = []
</code_context>

<issue_to_address>
**issue (complexity):** Consider extracting the user message image-conversion logic into dedicated helper methods so `_prepare_payload` stays focused on simple message shaping and control flow.

You can reduce the added complexity by extracting the user-content/image conversion into small helpers and keeping `_prepare_payload` focused on message shaping.

For example, instead of inlining all of this logic:

```python
elif message["role"] == "user":
    if isinstance(message.get("content"), list):
        converted_content = []
        for part in message["content"]:
            if part.get("type") == "image_url":
                # Convert OpenAI image_url format to Anthropic image format
                image_url_data = part.get("image_url", {})
                url = image_url_data.get("url", "")
                if url.startswith("data:"):
                    try:
                        _, base64_data = url.split(",", 1)
                        # Detect actual image format from binary data
                        image_bytes = base64.b64decode(base64_data)
                        media_type = self._detect_image_mime_type(image_bytes)
                        converted_content.append(
                            {
                                "type": "image",
                                "source": {
                                    "type": "base64",
                                    "media_type": media_type,
                                    "data": base64_data,
                                },
                            }
                        )
                    except ValueError:
                        logger.warning(
                            f"Failed to parse image data URI: {url[:50]}..."
                        )
                else:
                    logger.warning(
                        f"Unsupported image URL format for Anthropic: {url[:50]}..."
                    )
            else:
                converted_content.append(part)
        new_messages.append(
            {
                "role": "user",
                "content": converted_content,
            }
        )
    else:
        new_messages.append(message)
```

you can move the conversion into helpers and keep the control flow flatter:

```python
def _convert_user_content(self, content):
    if not isinstance(content, list):
        return content

    converted = []
    for part in content:
        if part.get("type") == "image_url":
            converted_part = self._convert_image_part(part)
            if converted_part is not None:
                converted.append(converted_part)
            # if conversion failed, we just drop the part or could append original,
            # depending on current behavior you want to preserve
        else:
            converted.append(part)
    return converted

def _convert_image_part(self, part):
    image_url_data = part.get("image_url", {})
    url = image_url_data.get("url", "")
    if not url.startswith("data:"):
        logger.warning(
            f"Unsupported image URL format for Anthropic: {url[:50]}..."
        )
        return None

    try:
        _, base64_data = url.split(",", 1)
        image_bytes = base64.b64decode(base64_data)
        media_type = self._detect_image_mime_type(image_bytes)
        return {
            "type": "image",
            "source": {
                "type": "base64",
                "media_type": media_type,
                "data": base64_data,
            },
        }
    except ValueError:
        logger.warning(
            f"Failed to parse image data URI: {url[:50]}..."
        )
        return None
```

Then `_prepare_payload` only needs:

```python
elif message["role"] == "user":
    content = self._convert_user_content(message.get("content"))
    new_messages.append(
        {
            "role": "user",
            "content": content,
        }
        if isinstance(content, list)
        else {**message, "content": content}
    )
```

Benefits:

- Flattens nesting in `_prepare_payload` to a single `elif role == 'user'` branch.
- Localizes base64/MIME/logging details in `_convert_image_part`, making it easier to test and reason about.
- Separates “how to convert content” from “how to append the message,” which makes control flow clearer and avoids duplicated `new_messages.append` shapes.
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}