MCP Image Server

Instructions

Implementation Reference

• mcp_image.py:392-449 (handler)

  The main handler function for the 'fetch_images' tool. It validates input, calls process_images_async to handle URLs and local files concurrently, extracts Image objects, and returns a list of processed Images or None on failure. Includes decorator for registration and schema docstring.
  @mcp.tool() async def fetch_images(image_sources: List[str], ctx: Context) -> List[Image | None]: """ Fetch and process images from URLs or local file paths, returning them in a format suitable for LLMs. This tool accepts a list of image sources which can be either: 1. URLs pointing to web-hosted images (http:// or https://) 2. Local file paths pointing to images stored on the local filesystem (e.g., "C:/images/photo1.jpg") For a single image, provide a one-element list. The function will process images in parallel when multiple sources are provided. Images that exceed the size limit (1MB) will be automatically compressed while maintaining aspect ratio and reasonable quality. Args: image_sources: A list of image URLs or local file paths. For a single image, provide a one-element list. Returns: A list of Image objects or None values (if processing failed) in the same order as the input sources. """ try: start_time = asyncio.get_event_loop().time() # Validate input if not image_sources: ctx.error("No image sources provided") logger.error("fetch_images called with empty source list") return [] # Log the types of sources we're processing url_count = sum(1 for src in image_sources if is_url(src)) local_count = len(image_sources) - url_count logger.debug(f"Processing {len(image_sources)} image sources: {url_count} URLs and {local_count} local files") # Process all images results = await process_images_async(image_sources, ctx) # Extract just the Image objects or None values image_results = [] for result in results: if "image" in result: image_results.append(result["image"]) else: image_results.append(None) elapsed = asyncio.get_event_loop().time() - start_time success_count = sum(1 for r in image_results if r is not None) logger.debug( f"Processed {len(image_sources)} images in {elapsed:.2f} seconds. " f"Success: {success_count}, Failed: {len(image_sources) - success_count}" ) return image_results except Exception as e: logger.exception("Error in fetch_images") ctx.error(f"Failed to process images: {str(e)}") return [None] * len(image_sources)
• mcp_image.py:394-409 (schema)

  Docstring defining the tool schema, including description, parameters (image_sources: List[str]), and return type (List[Image | None]).
  """ Fetch and process images from URLs or local file paths, returning them in a format suitable for LLMs. This tool accepts a list of image sources which can be either: 1. URLs pointing to web-hosted images (http:// or https://) 2. Local file paths pointing to images stored on the local filesystem (e.g., "C:/images/photo1.jpg") For a single image, provide a one-element list. The function will process images in parallel when multiple sources are provided. Images that exceed the size limit (1MB) will be automatically compressed while maintaining aspect ratio and reasonable quality. Args: image_sources: A list of image URLs or local file paths. For a single image, provide a one-element list. Returns: A list of Image objects or None values (if processing failed) in the same order as the input sources.
• mcp_image.py:356-390 (helper)

  Key helper function that separates URLs from local paths, processes them concurrently using asyncio.gather (fetch_single_image for URLs, process_local_image for locals), and preserves input order in results.
  async def process_images_async(image_sources: List[str], ctx: Context) -> List[Dict[str, Any]]: """Process multiple images (URLs or local files) concurrently.""" if not image_sources: raise ValueError("No image sources provided") # Separate URLs from local file paths urls = [src for src in image_sources if is_url(src)] local_paths = [src for src in image_sources if not is_url(src)] results = [] # Process URLs if any if urls: logger.debug(f"Processing {len(urls)} URLs") async with httpx.AsyncClient() as client: url_tasks = [fetch_single_image(url, client, ctx) for url in urls] url_results = await asyncio.gather(*url_tasks) results.extend(url_results) # Process local files if any if local_paths: logger.debug(f"Processing {len(local_paths)} local files") local_tasks = [process_local_image(path, ctx) for path in local_paths] local_results = await asyncio.gather(*local_tasks) results.extend(local_results) # Ensure results are in the same order as input sources ordered_results = [] for src in image_sources: for result in results: if (src == result.get("url", None)) or (src == result.get("path", None)): ordered_results.append(result) break return ordered_results
• mcp_image.py:50-160 (helper)

  Core helper for processing image bytes: logs dimensions, handles small images directly, for large ones compresses iteratively by reducing quality or scaling to fit under 800KB, converts to JPEG.
  async def process_image_data(data: bytes, content_type: str, image_source: str, ctx: Context) -> Image | None: """Process image data and return an MCP Image object.""" try: # If image is not large, try to log dimensions without processing if len(data) <= 1048576: try: with PILImage.open(BytesIO(data)) as img: width, height = img.size logger.debug(f"Original image dimensions from {image_source}: {width}x{height}") logger.debug(f"Image format from PIL: {img.format}, mode: {img.mode}") except Exception as e: logger.debug(f"Could not determine dimensions for {image_source}: {e}") # Ensure content_type is valid and doesn't include 'image/' if content_type.startswith('image/'): content_type = content_type.split('/')[-1] logger.debug(f"Creating Image object with format: {content_type}") return Image(data=data, format=content_type) # For large images, save to temp file and process temp_path = os.path.join(TEMP_DIR, f"temp_image_{hash(image_source)}." + content_type.split('/')[-1]) with open(temp_path, "wb") as f: f.write(data) try: # First pass: get dimensions and basic info with PILImage.open(temp_path) as img: orig_width, orig_height = img.size orig_format = img.format orig_mode = img.mode logger.debug(f"Original image dimensions from {image_source}: {orig_width}x{orig_height}") logger.debug(f"Large image format from PIL: {orig_format}, mode: {orig_mode}") # Calculate optimal resize factor if image is very large max_dimension = max(orig_width, orig_height) initial_scale = 1.0 if max_dimension > 3000: initial_scale = 3000 / max_dimension logger.debug(f"Very large image detected ({max_dimension}px), will start with scale factor: {initial_scale}") # Second pass: process the image with PILImage.open(temp_path) as img: if img.mode in ('RGBA', 'P'): img = img.convert('RGB') # Apply initial scale if needed if initial_scale < 1.0: width = int(orig_width * initial_scale) height = int(orig_height * initial_scale) img = img.resize((width, height), PILImage.LANCZOS) else: width, height = img.size quality = 85 scale_factor = 1.0 while True: img_byte_arr = BytesIO() # Create a copy for this iteration to avoid accumulating transforms if scale_factor < 1.0: current_width = int(width * scale_factor) current_height = int(height * scale_factor) current_img = img.resize((current_width, current_height), PILImage.LANCZOS) else: current_img = img current_width, current_height = width, height current_img.save(img_byte_arr, format='JPEG', quality=quality, optimize=True) processed_data = img_byte_arr.getvalue() # Clean up the temporary image if we created one if scale_factor < 1.0 and hasattr(current_img, 'close'): current_img.close() # Target 800KB to leave buffer for any MCP overhead if len(processed_data) <= 819200: # 800KB logger.debug(f"Processed image dimensions from {image_source}: {current_width}x{current_height} (quality={quality})") logger.debug(f"Returning processed image with format: jpeg, size: {len(processed_data)} bytes") return Image(data=processed_data, format='jpeg') # Try reducing quality first if quality > 20: quality -= 10 logger.debug(f"Reducing quality to {quality} for {image_source}, current size: {len(processed_data)} bytes") else: # Then try scaling down scale_factor *= 0.8 if current_width * scale_factor < 200 or current_height * scale_factor < 200: ctx.error("Unable to compress image to acceptable size while maintaining quality") logger.error(f"Failed processing image from {image_source}: dimensions too small") return None logger.debug(f"Applying scale factor {scale_factor} to image from {image_source}") quality = 85 # Reset quality when changing size except MemoryError as e: ctx.error(f"Out of memory processing large image: {str(e)}") logger.error(f"MemoryError processing image from {image_source}: {str(e)}") return None except Exception as e: ctx.error(f"Image processing error: {str(e)}") logger.exception(f"Exception processing image from {image_source}") return None finally: if os.path.exists(temp_path): os.remove(temp_path) except Exception as e: ctx.error(f"Error processing image: {str(e)}") logger.exception(f"Unexpected error processing {image_source}") return None