process_images
Process images in a directory with operations like enhancement, OCR text extraction, resizing, and deduplication to organize and extract information from visual content.
Instructions
Process images in a directory with various operations.
Args:
image_dir: Directory containing images to process
operations: List of operations (enhance, ocr, resize, deduplicate)
ocr_language: Language for OCR processing (default: eng)
Returns:
JSON string with processing results.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| image_dir | Yes | ||
| operations | No | ||
| ocr_language | No | eng |
Implementation Reference
- server.py:258-301 (handler)The primary handler for the 'process_images' MCP tool, decorated with @mcp.tool() for automatic registration and schema inference from type hints. Implements the tool logic by delegating to ImageProcessor.process_batch.@mcp.tool() async def process_images( image_dir: str, operations: List[str] = ["enhance"], ocr_language: str = "eng" ) -> str: """ Process images in a directory with various operations. Args: image_dir: Directory containing images to process operations: List of operations (enhance, ocr, resize, deduplicate) ocr_language: Language for OCR processing (default: eng) Returns: JSON string with processing results. """ try: ip = get_image_processor() results = ip.process_batch(image_dir, operations) # Add OCR language info to results if OCR was performed if "ocr" in operations: for ocr_result in results.get("ocr_results", []): ocr_result["language"] = ocr_language result = { "status": "success", "image_directory": image_dir, "operations": operations, "results": results } return json.dumps(result, indent=2) except Exception as e: logger.error(f"Failed to process images: {e}") return json.dumps({ "status": "error", "error": str(e), "image_directory": image_dir, "operations": operations })
- processing.py:176-241 (helper)Core implementation of batch image processing (enhance, OCR, resize, deduplicate) in ImageProcessor.process_batch, directly called by the tool handler to perform the specified operations on images in a directory.def process_batch(self, image_dir: str, operations: List[str] = None) -> Dict[str, List[str]]: """ Process a batch of images in a directory. Args: image_dir: Directory containing images operations: List of operations ('enhance', 'ocr', 'resize', 'deduplicate') Returns: Dictionary with results of operations """ if operations is None: operations = ['enhance'] image_paths = [] for ext in self.supported_formats: image_paths.extend(Path(image_dir).glob(f"*{ext}")) image_paths.extend(Path(image_dir).glob(f"*{ext.upper()}")) image_paths = [str(p) for p in image_paths] results = { 'processed_files': [], 'enhanced_files': [], 'ocr_results': [], 'resized_files': [], 'duplicates': {} } try: # Find duplicates first if 'deduplicate' in operations: results['duplicates'] = self.find_duplicates(image_paths) for image_path in image_paths: results['processed_files'].append(image_path) try: # Enhance image if 'enhance' in operations: enhanced_path = self.enhance_image(image_path) results['enhanced_files'].append(enhanced_path) # Extract text if 'ocr' in operations: text = self.extract_text(image_path) results['ocr_results'].append({ 'file': image_path, 'text': text }) # Resize image if 'resize' in operations: resized_path = self.resize_image(image_path) results['resized_files'].append(resized_path) except Exception as e: logger.error(f"Failed to process {image_path}: {e}") continue logger.info(f"Batch processed {len(results['processed_files'])} images") return results except Exception as e: logger.error(f"Failed to process batch in {image_dir}: {e}") raise