Image Processing MCP Server

Instructions

Implementation Reference

• tools/transform.py:229-284 (handler)

  Core handler function that loads the image, validates crop coordinates, performs the crop using PIL, and returns JSON result with cropped image data.
  async def crop_image(image_data: str, left: int, top: int, right: int, bottom: int) -> list[TextContent]: """ 裁剪图片 Args: image_data: 图片数据（base64编码） left, top, right, bottom: 裁剪坐标 Returns: 裁剪后的图片数据 """ try: # 验证参数 if not image_data: raise ValidationError("图片数据不能为空") # 加载图片 image = processor.load_image(image_data) image_width, image_height = image.size # 验证裁剪坐标 if not validate_crop_coordinates(left, top, right, bottom, image_width, image_height): raise ValidationError(f"无效的裁剪坐标: ({left}, {top}, {right}, {bottom}), 图片尺寸: {image_width}x{image_height}") # 裁剪图片 cropped_image = image.crop((left, top, right, bottom)) # 输出裁剪后的图片 output_info = processor.output_image(cropped_image, f"crop_{left}_{top}_{right}_{bottom}") result = { "success": True, "message": f"图片裁剪成功: ({left}, {top}, {right}, {bottom})", "data": { **output_info, "original_size": (image_width, image_height), "crop_box": (left, top, right, bottom), "cropped_size": cropped_image.size } } 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:188-204 (registration)

  MCP tool registration decorator @mcp.tool() defining the synchronous wrapper function for crop_image, which delegates to the core async handler.
  @mcp.tool() def crop_image( image_source: Annotated[str, Field(description="图片源，可以是文件路径或base64编码的图片数据")], left: Annotated[int, Field(description="裁剪区域左边界坐标（像素）", ge=0)], top: Annotated[int, Field(description="裁剪区域上边界坐标（像素）", ge=0)], right: Annotated[int, Field(description="裁剪区域右边界坐标（像素）", gt=0)], bottom: Annotated[int, Field(description="裁剪区域下边界坐标（像素）", gt=0)] ) -> str: """裁剪图片""" try: result = safe_run_async(transform_crop_image(image_source, left, top, right, bottom)) return result[0].text except Exception as e: return json.dumps({ "success": False, "error": f"裁剪图片失败: {str(e)}" }, ensure_ascii=False, indent=2)
• tools/transform.py:48-79 (schema)

  JSON schema definition for the crop_image tool input parameters.
  Tool( name="crop_image", description="裁剪图片", inputSchema={ "type": "object", "properties": { "image_data": { "type": "string", "description": "图片数据（base64编码）" }, "left": { "type": "integer", "description": "左边界坐标", "minimum": 0 }, "top": { "type": "integer", "description": "上边界坐标", "minimum": 0 }, "right": { "type": "integer", "description": "右边界坐标" }, "bottom": { "type": "integer", "description": "下边界坐标" } }, "required": ["image_data", "left", "top", "right", "bottom"] } ),
• main.py:189-195 (schema)

  Pydantic schema inferred from Annotated Field descriptions in the registered tool function signature.
  def crop_image( image_source: Annotated[str, Field(description="图片源，可以是文件路径或base64编码的图片数据")], left: Annotated[int, Field(description="裁剪区域左边界坐标（像素）", ge=0)], top: Annotated[int, Field(description="裁剪区域上边界坐标（像素）", ge=0)], right: Annotated[int, Field(description="裁剪区域右边界坐标（像素）", gt=0)], bottom: Annotated[int, Field(description="裁剪区域下边界坐标（像素）", gt=0)] ) -> str:
• main.py:88-101 (helper)

  Utility function used by the wrapper to safely execute the async core handler from sync context.
  def safe_run_async(coro): """安全地运行异步函数，处理事件循环问题""" try: # 尝试获取当前事件循环 loop = asyncio.get_running_loop() # 如果已经在事件循环中，创建一个任务 import concurrent.futures with concurrent.futures.ThreadPoolExecutor() as executor: future = executor.submit(asyncio.run, coro) return future.result() except RuntimeError: # 没有运行的事件循环，可以直接使用asyncio.run return asyncio.run(coro)