Skip to content

fix: 修复图片序列化错误，解决 FastMCP Image 类型兼容性问题 #171

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hexonal wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix: 修复图片序列化错误，解决 FastMCP Image 类型兼容性问题 #171

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 34 additions & 29 deletions src/mcp_feedback_enhanced/server.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,7 @@
from typing import Annotated, Any

from fastmcp import FastMCP
from fastmcp.utilities.types import Image as MCPImage
from mcp.types import TextContent
from mcp.types import TextContent, ImageContent
from pydantic import Field

# 導入統一的調試功能
Expand Down Expand Up @@ -360,57 +359,63 @@ def create_feedback_text(feedback_data: dict) -> str:
return "\n\n".join(text_parts) if text_parts else "用戶未提供任何回饋內容。"


def process_images(images_data: list[dict]) -> list[MCPImage]:
def process_images(images_data: list[dict]) -> list[ImageContent]:
"""
處理圖片資料，轉換為 MCP 圖片對象
處理圖片資料，轉換為標準 MCP ImageContent 對象

Args:
images_data: 圖片資料列表

Returns:
List[MCPImage]: MCP 圖片對象列表
List[ImageContent]: 標準 MCP ImageContent 對象列表
"""
mcp_images = []
image_contents = []

for i, img in enumerate(images_data, 1):
try:
if not img.get("data"):
debug_log(f"圖片 {i} 沒有資料，跳過")
continue

# 檢查數據類型並相應處理
# 處理圖片數據，確保轉換為 base64 字符串
if isinstance(img["data"], bytes):
# 如果是原始 bytes 數據，直接使用
image_bytes = img["data"]
# 如果是原始 bytes 數據，編碼為 base64 字符串
image_base64 = base64.b64encode(img["data"]).decode("utf-8")
debug_log(
f"圖片 {i} 使用原始 bytes 數據，大小: {len(image_bytes)} bytes"
f"圖片 {i} 從 bytes 編碼為 base64，原始大小: {len(img['data'])} bytes"
)
elif isinstance(img["data"], str):
# 如果是 base64 字符串，進行解碼
image_bytes = base64.b64decode(img["data"])
debug_log(f"圖片 {i} 從 base64 解碼，大小: {len(image_bytes)} bytes")
# 如果已經是 base64 字符串，直接使用
image_base64 = img["data"]
debug_log(f"圖片 {i} 使用現有 base64 字符串，長度: {len(image_base64)}")
else:
debug_log(f"圖片 {i} 數據類型不支援: {type(img['data'])}")
continue

if len(image_bytes) == 0:
if len(image_base64) == 0:
debug_log(f"圖片 {i} 數據為空，跳過")
continue

# 根據文件名推斷格式
# 根據文件名推斷 MIME 類型
file_name = img.get("name", "image.png")
if file_name.lower().endswith((".jpg", ".jpeg")):
image_format = "jpeg"
mime_type = "image/jpeg"
elif file_name.lower().endswith(".gif"):
image_format = "gif"
mime_type = "image/gif"
elif file_name.lower().endswith(".webp"):
mime_type = "image/webp"
else:
image_format = "png" # 默認使用 PNG
mime_type = "image/png" # 默認使用 PNG
Comment on lines +399 to +408
Copy link

@gemini-code-assist gemini-code-assist bot Aug 3, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

For better maintainability and readability, you could use a dictionary to map file extensions to MIME types instead of a long if/elif/else chain. This makes it easier to add support for new image formats in the future. Using os.path.splitext() is also a more robust way to extract file extensions.

Suggested change
# 根據文件名推斷 MIME 類型
file_name = img.get("name", "image.png")
if file_name.lower().endswith((".jpg", ".jpeg")):
image_format = "jpeg"
mime_type = "image/jpeg"
elif file_name.lower().endswith(".gif"):
image_format = "gif"
mime_type = "image/gif"
elif file_name.lower().endswith(".webp"):
mime_type = "image/webp"
else:
image_format = "png" # 默認使用 PNG
mime_type = "image/png" # 默認使用 PNG
# 根據文件名推斷 MIME 類型
file_name = img.get("name", "image.png")
file_ext = os.path.splitext(file_name)[1].lower()
mime_type = {
".jpeg": "image/jpeg",
".jpg": "image/jpeg",
".gif": "image/gif",
".webp": "image/webp",
}.get(file_ext, "image/png") # 默認使用 PNG


# 創建 MCPImage 對象
mcp_image = MCPImage(data=image_bytes, format=image_format)
mcp_images.append(mcp_image)
# 創建標準 MCP ImageContent 對象
image_content = ImageContent(
type="image",
data=image_base64,
mimeType=mime_type
)
image_contents.append(image_content)

debug_log(f"圖片 {i} ({file_name}) 處理成功，格式: {image_format}")
debug_log(f"圖片 {i} ({file_name}) 處理成功，MIME類型: {mime_type}")

except Exception as e:
# 使用統一錯誤處理（不影響 JSON RPC）
Expand All @@ -421,8 +426,8 @@ def process_images(images_data: list[dict]) -> list[MCPImage]:
)
debug_log(f"圖片 {i} 處理失敗 [錯誤ID: {error_id}]: {e}")

debug_log(f"共處理 {len(mcp_images)} 張圖片")
return mcp_images
debug_log(f"共處理 {len(image_contents)} 張圖片")
return image_contents


# ===== MCP 工具定義 =====
Expand All @@ -449,7 +454,7 @@ async def interactive_feedback(
timeout: Timeout in seconds for waiting user feedback (default: 600 seconds)

Returns:
list: List containing TextContent and MCPImage objects representing user feedback
list: List containing TextContent and ImageContent objects representing user feedback
"""
# 環境偵測
is_remote = is_remote_environment()
Expand Down Expand Up @@ -491,10 +496,10 @@ async def interactive_feedback(

# 添加圖片回饋
if result.get("images"):
mcp_images = process_images(result["images"])
# 修復 arg-type 錯誤 - 直接擴展列表
feedback_items.extend(mcp_images)
debug_log(f"已添加 {len(mcp_images)} 張圖片")
image_contents = process_images(result["images"])
# 擴展列表添加 ImageContent 對象
feedback_items.extend(image_contents)
debug_log(f"已添加 {len(image_contents)} 張圖片")

# 確保至少有一個回饋項目
if not feedback_items:
Expand Down