Image Processing MCP Server

Instructions

Implementation Reference

• tools/basic.py:106-153 (handler)

  Main execution logic for the load_image tool: validates source, loads image using processor, retrieves info, outputs image reference, and returns JSON response.

  async def load_image(source: str) -> list[TextContent]:
      """
      加载图片
      
      Args:
          source: 图片源（文件路径或base64编码）
          
      Returns:
          包含图片信息和文件引用的响应
      """
      try:
          # 验证图片源
          ensure_valid_image_source(source)
          
          # 加载图片
          image = processor.load_image(source)
          
          # 获取图片信息
          info = processor.get_image_info(image)
          
          # 输出图片（文件引用模式）
          output_info = processor.output_image(image, "loaded")
          
          result = {
              "success": True,
              "message": "图片加载成功",
              "data": {
                  **output_info,
                  "info": info
              }
          }
          
          return [TextContent(type="text", text=json.dumps(result, ensure_ascii=False))]
          
      except ValidationError as e:
          error_result = {
              "success": False,
              "error": f"参数验证失败: {str(e)}"
          }
          return [TextContent(type="text", text=json.dumps(error_result, ensure_ascii=False))]
          
      except Exception as e:
          error_result = {
              "success": False,
              "error": f"图片加载失败: {str(e)}"
          }
          return [TextContent(type="text", text=json.dumps(error_result, ensure_ascii=False))]

• main.py:107-120 (registration)

  MCP server tool registration using @mcp.tool() decorator. Defines input schema via Annotated Field and delegates execution to basic_load_image handler.

  @mcp.tool()
  def load_image(
      source: Annotated[str, Field(description="图片文件路径或base64编码的图片数据。支持本地文件路径（如 'image.jpg'）或base64编码字符串")]
  ) -> str:
      """加载图片文件或base64编码的图片"""
      try:
          result = safe_run_async(basic_load_image(source))
          return result[0].text
      except Exception as e:
          return json.dumps({
              "success": False,
              "error": f"加载图片失败: {str(e)}"
          }, ensure_ascii=False, indent=2)

• main.py:109-109 (schema)

  Input schema definition for the load_image tool using Pydantic Annotated and Field for source parameter validation and description.

  source: Annotated[str, Field(description="图片文件路径或base64编码的图片数据。支持本地文件路径（如 'image.jpg'）或base64编码字符串")]

• utils/image_processor.py:25-64 (helper)

  Core utility function in ImageProcessor class that loads image from file path or base64 string into PIL Image object, with size validation.

  def load_image(self, source: Union[str, bytes]) -> Image.Image:
      """
      加载图片，支持文件路径、base64编码
      
      Args:
          source: 图片源，可以是文件路径或base64编码字符串
          
      Returns:
          PIL Image对象
          
      Raises:
          ValueError: 当图片源无效时
          IOError: 当图片无法加载时
      """
      try:
          if isinstance(source, str):
              if source.startswith('data:image'):
                  # base64编码的图片
                  header, data = source.split(',', 1)
                  image_data = base64.b64decode(data)
                  image = Image.open(io.BytesIO(image_data))
              else:
                  # 文件路径
                  if not os.path.exists(source):
                      raise ValueError(f"图片文件不存在: {source}")
                  image = Image.open(source)
          elif isinstance(source, bytes):
              image = Image.open(io.BytesIO(source))
          else:
              raise ValueError("不支持的图片源类型")
          
          # 检查图片尺寸
          if image.size[0] > self.max_image_size[0] or image.size[1] > self.max_image_size[1]:
              raise ValueError(f"图片尺寸过大，最大支持: {self.max_image_size}")
          
          return image
          
      except Exception as e:
          raise IOError(f"图片加载失败: {str(e)}")