Image Processing MCP Server

Instructions

Implementation Reference

• tools/effects.py:634-782 (handler)

  Core handler function implementing the watermark addition logic: validates parameters, loads image with ImageProcessor, creates RGBA watermark layer (text using ImageFont or resized image), positions it (top-left etc.), applies opacity, alpha composites with original, outputs base64 with metadata.
  async def add_watermark(arguments: Dict[str, Any]) -> List[TextContent]: """ 为图片添加水印 Args: arguments: 包含图片源和水印参数的字典 Returns: List[TextContent]: 处理结果 """ try: # 参数验证 image_source = arguments.get("image_source") ensure_valid_image_source(image_source) watermark_text = arguments.get("watermark_text") watermark_image = arguments.get("watermark_image") position = arguments.get("position", "bottom-right") opacity = arguments.get("opacity", 0.5) scale = arguments.get("scale", 0.2) output_format = arguments.get("output_format", DEFAULT_IMAGE_FORMAT) # 验证参数 if not watermark_text and not watermark_image: raise ValidationError("必须提供watermark_text或watermark_image之一") validate_numeric_range(opacity, 0.0, 1.0, "opacity") validate_numeric_range(scale, 0.1, 2.0, "scale") # 加载图片 processor = ImageProcessor() image = processor.load_image(image_source) # 转换为RGBA模式 if image.mode != "RGBA": image = image.convert("RGBA") # 创建水印图层 watermark_layer = Image.new("RGBA", image.size, (0, 0, 0, 0)) if watermark_text: # 文字水印 from PIL import ImageFont try: # 尝试使用系统字体 font_size = int(min(image.width, image.height) * scale * 0.1) font = ImageFont.truetype("arial.ttf", font_size) except: # 使用默认字体 font = ImageFont.load_default() draw = ImageDraw.Draw(watermark_layer) # 获取文字尺寸 bbox = draw.textbbox((0, 0), watermark_text, font=font) text_width = bbox[2] - bbox[0] text_height = bbox[3] - bbox[1] # 计算位置 if position == "top-left": x, y = 10, 10 elif position == "top-right": x, y = image.width - text_width - 10, 10 elif position == "bottom-left": x, y = 10, image.height - text_height - 10 elif position == "bottom-right": x, y = image.width - text_width - 10, image.height - text_height - 10 else: # center x, y = (image.width - text_width) // 2, (image.height - text_height) // 2 # 绘制文字 alpha = int(255 * opacity) draw.text((x, y), watermark_text, font=font, fill=(255, 255, 255, alpha)) else: # 图片水印 watermark_img = processor.load_image(watermark_image) # 调整水印大小 watermark_width = int(image.width * scale) watermark_height = int(watermark_img.height * watermark_width / watermark_img.width) watermark_img = watermark_img.resize((watermark_width, watermark_height), Image.Resampling.LANCZOS) # 转换为RGBA if watermark_img.mode != "RGBA": watermark_img = watermark_img.convert("RGBA") # 调整透明度 alpha_channel = watermark_img.split()[-1] alpha_channel = alpha_channel.point(lambda p: int(p * opacity)) watermark_img.putalpha(alpha_channel) # 计算位置 if position == "top-left": x, y = 10, 10 elif position == "top-right": x, y = image.width - watermark_width - 10, 10 elif position == "bottom-left": x, y = 10, image.height - watermark_height - 10 elif position == "bottom-right": x, y = image.width - watermark_width - 10, image.height - watermark_height - 10 else: # center x, y = (image.width - watermark_width) // 2, (image.height - watermark_height) // 2 # 粘贴水印 watermark_layer.paste(watermark_img, (x, y), watermark_img) # 合成最终图片 result_image = Image.alpha_composite(image, watermark_layer) # 转换为base64 output_info = processor.output_image(result_image, "border", output_format) return [TextContent( type="text", text=json.dumps({ "success": True, "message": "成功添加水印", "data": { **output_info, "metadata": { "size": f"{image.width}x{image.height}", "watermark_type": "text" if watermark_text else "image", "position": position, "opacity": opacity, "scale": scale, "format": output_format } } }, ensure_ascii=False) )] except ValidationError as e: return [TextContent( type="text", text=json.dumps({ "success": False, "error": f"参数验证失败: {str(e)}" }, ensure_ascii=False) )] except Exception as e: return [TextContent( type="text", text=json.dumps({ "success": False, "error": f"添加水印失败: {str(e)}" }, ensure_ascii=False) )]
• main.py:571-602 (registration)

  FastMCP tool registration for 'add_watermark' using @mcp.tool() decorator. Defines input parameters with descriptions, validation (ge, le, etc.), defaults; constructs arguments dict and calls the core handler from effects.py.
  @mcp.tool() def add_watermark( image_source: Annotated[str, Field(description="图片源，可以是文件路径或base64编码的图片数据")], watermark_text: Annotated[Optional[str], Field(description="水印文字内容，与watermark_image二选一", default=None)], watermark_image: Annotated[Optional[str], Field(description="水印图片路径或base64数据，与watermark_text二选一", default=None)], position: Annotated[str, Field(description="水印位置：top-left、top-right、bottom-left、bottom-right、center", default="bottom-right")], opacity: Annotated[float, Field(description="水印不透明度，范围 0.0-1.0", ge=0.0, le=1.0, default=0.5)], scale: Annotated[float, Field(description="水印缩放比例，1.0为原始大小", gt=0, default=1.0)], output_format: Annotated[str, Field(description="输出格式：PNG、JPEG、WEBP 等", default="PNG")] ) -> str: """为图片添加水印""" try: arguments = { "image_source": image_source, "position": position, "opacity": opacity, "scale": scale, "output_format": output_format } if watermark_text: arguments["watermark_text"] = watermark_text if watermark_image: arguments["watermark_image"] = watermark_image result = safe_run_async(effects_add_watermark(arguments)) return result[0].text except Exception as e: return json.dumps({ "success": False, "error": f"添加水印失败: {str(e)}" }, ensure_ascii=False, indent=2)
• tools/effects.py:167-214 (schema)

  JSON input schema definition for add_watermark tool in the get_effect_tools() function, specifying properties, types, enums, defaults, and required fields for validation.
  Tool( name="add_watermark", description="为图片添加水印", inputSchema={ "type": "object", "properties": { "image_source": { "type": "string", "description": "图片源（文件路径或base64编码）" }, "watermark_text": { "type": "string", "description": "水印文字（与watermark_image二选一）" }, "watermark_image": { "type": "string", "description": "水印图片源（与watermark_text二选一）" }, "position": { "type": "string", "description": "水印位置", "enum": ["top-left", "top-right", "bottom-left", "bottom-right", "center"], "default": "bottom-right" }, "opacity": { "type": "number", "description": "水印透明度（0.0-1.0）", "minimum": 0.0, "maximum": 1.0, "default": 0.5 }, "scale": { "type": "number", "description": "水印缩放比例", "minimum": 0.1, "maximum": 2.0, "default": 0.2 }, "output_format": { "type": "string", "description": "输出格式", "enum": ["PNG", "JPEG", "WEBP"], "default": "PNG" } }, "required": ["image_source"] } ),