some tools, and openai framework to generate image

mcp_server.py

@@ -2,6 +2,9 @@ from mcp.server.fastmcp import FastMCP
 from src.utils.change_format import change_format
 from src.utils.image_helpers import remove_background_from_url
 from src.utils.visualize_image import visualize_base64_image
+from src.utils.generate_image import generate_image
+from src.utils.apply_filter import apply_filter
+from src.utils.add_text import add_text_to_image
 
 mcp = FastMCP("Youtube Service")
 
@@ -21,6 +24,9 @@ def say_hello(name: str) -> str:
 mcp.add_tool(remove_background_from_url)
 mcp.add_tool(change_format)
 mcp.add_tool(visualize_base64_image)
+mcp.add_tool(generate_image)
+mcp.add_tool(apply_filter)
+mcp.add_tool(add_text_to_image)
 
 if __name__ == "__main__":
     mcp.run()

requirements.txt

@@ -2,4 +2,5 @@ fastmcp
 requests
 Pillow
 rembg
-onnxruntime
+onnxruntime
+openai

src/utils/add_text.py

@@ -0,0 +1,68 @@
+from PIL import Image, ImageDraw, ImageFont
+import os
+from typing import Optional, Tuple, Dict, Any
+
+def add_text_to_image(
+    image_path: str,
+    text: str,
+    color: Optional[Tuple[int, int, int]] = None,
+    output_name: Optional[str] = None
+) -> Dict[str, Any]:
+    """
+    Adds centered text to an image and saves the result in the same folder.
+    If no output_name is provided, '_edited' is appended to the original filename.
+    If no color is provided, black is used by default.
+
+    Args:
+        image_path: Path to the original image.
+        text: Text to write on the image.
+        color: Optional RGB color of the text. Defaults to black.
+        output_name: Optional output filename (without extension).
+
+    Returns:
+        Dictionary with success status and info.
+    """
+    try:
+        if color is None:
+            color = (0, 0, 0)
+
+        image = Image.open(image_path)
+        draw = ImageDraw.Draw(image)
+        font = ImageFont.load_default()
+
+        text_width = draw.textlength(text, font=font)
+        x = (image.width - text_width) / 2
+        y = image.height / 2
+
+        draw.text((x, y), text, fill=color, font=font)
+
+        base_dir = os.path.dirname(image_path)
+        base_name, ext = os.path.splitext(os.path.basename(image_path))
+
+        if output_name:
+            new_filename = f"{output_name}{ext}"
+        else:
+            new_filename = f"{base_name}_edited{ext}"
+
+        new_path = os.path.join(base_dir, new_filename)
+
+        image.save(new_path)
+        output_size = os.path.getsize(new_path)
+
+        return {
+            "success": True,
+            "message": "Text added successfully to the image",
+            "input_path": image_path,
+            "output_path": new_path,
+            "output_size_bytes": output_size,
+            "text": text,
+            "color": color
+        }
+
+    except Exception as e:
+        return {
+            "success": False,
+            "error": str(e),
+            "input_path": image_path,
+            "output_path": None
+        }

src/utils/add_watermark.py

@@ -0,0 +1,56 @@
+from PIL import Image, ImageDraw, ImageFont
+import os
+from typing import Dict, Any
+
+def add_watermark(image_path: str, watermark_text: str) -> Dict[str, Any]:
+    try:
+        image = Image.open(image_path)
+        
+        overlay = Image.new('RGBA', image.size, (255, 255, 255, 0))
+        draw = ImageDraw.Draw(overlay)
+        
+        try:
+            font_size = min(image.width, image.height) // 20
+            font = ImageFont.truetype("arial.ttf", font_size)
+        except:
+            font = ImageFont.load_default()
+        
+        bbox = draw.textbbox((0, 0), watermark_text, font=font)
+        text_width = bbox[2] - bbox[0]
+        text_height = bbox[3] - bbox[1]
+        
+        x = (image.width - text_width) // 2
+        y = (image.height - text_height) // 2
+        
+        text_color = (255, 255, 255, 128)
+        
+        draw.text((x-2, y-2), watermark_text, fill=(0, 0, 0, 64), font=font)
+        draw.text((x, y), watermark_text, fill=text_color, font=font)
+        
+        watermarked = Image.alpha_composite(image.convert('RGBA'), overlay)
+        final_image = watermarked.convert('RGB')
+        
+        base_dir = os.path.dirname(image_path)
+        base_name, ext = os.path.splitext(os.path.basename(image_path))
+        new_filename = f"{base_name}_watermarked{ext}"
+        new_path = os.path.join(base_dir, new_filename)
+        
+        final_image.save(new_path, quality=95)
+        output_size = os.path.getsize(new_path)
+        
+        return {
+            "success": True,
+            "message": "Watermark added successfully",
+            "input_path": image_path,
+            "output_path": new_path,
+            "output_size_bytes": output_size,
+            "watermark_text": watermark_text
+        }
+        
+    except Exception as e:
+        return {
+            "success": False,
+            "error": str(e),
+            "input_path": image_path,
+            "output_path": None
+        }

src/utils/apply_filter.py

@@ -0,0 +1,123 @@
+from PIL import Image, ImageFilter, ImageEnhance
+from io import BytesIO
+import requests
+import base64
+
+def apply_filter(image_url: str, filter_type: str, intensity: float = 1.0, output_format: str = 'JPEG') -> str:
+    """
+    Apply various filters to an image from a URL.
+    
+    Args:
+        image_url: The URL of the input image.
+        filter_type: Type of filter to apply ('blur', 'sharpen', 'emboss', 'edge', 'smooth', 
+                    'brightness', 'contrast', 'saturation', 'sepia', 'grayscale').
+        intensity: Filter intensity (0.1 to 3.0, default 1.0).
+        output_format: The desired output format (e.g., 'JPEG', 'PNG').
+        
+    Returns:
+        The filtered image as a base64-encoded string.
+    """
+    
+    response = requests.get(image_url, timeout=30)
+    response.raise_for_status()
+    
+    img = Image.open(BytesIO(response.content))
+    
+    if img.mode != 'RGB':
+        img = img.convert('RGB')
+    
+    if filter_type == 'blur':
+        img = img.filter(ImageFilter.GaussianBlur(radius=intensity))
+    elif filter_type == 'sharpen':
+        img = img.filter(ImageFilter.UnsharpMask(radius=2, percent=int(intensity * 150), threshold=3))
+    elif filter_type == 'emboss':
+        img = img.filter(ImageFilter.EMBOSS)
+    elif filter_type == 'edge':
+        img = img.filter(ImageFilter.FIND_EDGES)
+    elif filter_type == 'smooth':
+        img = img.filter(ImageFilter.SMOOTH_MORE)
+    elif filter_type == 'brightness':
+        enhancer = ImageEnhance.Brightness(img)
+        img = enhancer.enhance(intensity)
+    elif filter_type == 'contrast':
+        enhancer = ImageEnhance.Contrast(img)
+        img = enhancer.enhance(intensity)
+    elif filter_type == 'saturation':
+        enhancer = ImageEnhance.Color(img)
+        img = enhancer.enhance(intensity)
+    elif filter_type == 'sepia':
+        img = apply_sepia_filter(img)
+    elif filter_type == 'grayscale':
+        img = img.convert('L').convert('RGB')
+    
+    output = BytesIO()
+    img.save(output, format=output_format, quality=95)
+    output.seek(0)
+    
+    encoded_image = base64.b64encode(output.getvalue()).decode('utf-8')
+    
+    return encoded_image
+
+def apply_sepia_filter(img: Image.Image) -> Image.Image:
+    """
+    Apply sepia tone effect to an image.
+    
+    Args:
+        img: PIL Image object.
+        
+    Returns:
+        Image with sepia effect applied.
+    """
+    
+    pixels = img.load()
+    width, height = img.size
+    
+    for y in range(height):
+        for x in range(width):
+            r, g, b = pixels[x, y]
+            
+            tr = int(0.393 * r + 0.769 * g + 0.189 * b)
+            tg = int(0.349 * r + 0.686 * g + 0.168 * b)
+            tb = int(0.272 * r + 0.534 * g + 0.131 * b)
+            
+            pixels[x, y] = (min(255, tr), min(255, tg), min(255, tb))
+    
+    return img
+
+def apply_vintage_filter(image_url: str, output_format: str = 'JPEG') -> str:
+    """
+    Apply a vintage effect combining multiple filters.
+    
+    Args:
+        image_url: The URL of the input image.
+        output_format: The desired output format.
+        
+    Returns:
+        The vintage-filtered image as a base64-encoded string.
+    """
+    
+    response = requests.get(image_url, timeout=30)
+    response.raise_for_status()
+    
+    img = Image.open(BytesIO(response.content))
+    
+    if img.mode != 'RGB':
+        img = img.convert('RGB')
+    
+    contrast_enhancer = ImageEnhance.Contrast(img)
+    img = contrast_enhancer.enhance(0.8)
+    
+    brightness_enhancer = ImageEnhance.Brightness(img)
+    img = brightness_enhancer.enhance(1.1)
+    
+    img = apply_sepia_filter(img)
+    
+    img = img.filter(ImageFilter.GaussianBlur(radius=0.5))
+    
+    output = BytesIO()
+    img.save(output, format=output_format, quality=90)
+    output.seek(0)
+    
+    encoded_image = base64.b64encode(output.getvalue()).decode('utf-8')
+    
+    return encoded_image

src/utils/generate_image.py

@@ -0,0 +1,85 @@
+import os
+import base64
+from typing import Dict, Any
+from openai import OpenAI
+
+async def generate_image(
+    prompt: str,
+    output_path: str = "generated_image.png",
+    width: int = 1024,
+    height: int = 1024,
+    num_inference_steps: int = 28,
+    negative_prompt: str = "",
+    seed: int = -1
+) -> Dict[str, Any]:
+    """
+    Generate an image using Nebius API.
+    
+    Args:
+        prompt: Text prompt for image generation
+        output_path: Path where to save the generated image
+        width: Image width
+        height: Image height
+        num_inference_steps: Number of inference steps
+        negative_prompt: Negative prompt for generation
+        seed: Random seed (-1 for random)
+    
+    Returns:
+        Dictionary with result information
+    """
+
+    try:
+        client = OpenAI(
+            base_url="https://api.studio.nebius.com/v1/",
+            api_key=os.environ.get("NEBIUS_API_KEY")
+        )
+        
+        response = client.images.generate(
+            model="black-forest-labs/flux-dev",
+            response_format="b64_json",
+            extra_body={
+                "response_extension": "png",
+                "width": width,
+                "height": height,
+                "num_inference_steps": num_inference_steps,
+                "negative_prompt": negative_prompt,
+                "seed": seed
+            },
+            prompt=prompt
+        )
+        
+        image_data = base64.b64decode(response.data[0].b64_json)
+        
+        with open(output_path, 'wb') as output_file:
+            output_file.write(image_data)
+        
+        output_size = os.path.getsize(output_path)
+        
+        return {
+            "success": True,
+            "message": "Image generated successfully",
+            "prompt": prompt,
+            "output_path": output_path,
+            "output_size_bytes": output_size,
+            "generation_params": {
+                "width": width,
+                "height": height,
+                "num_inference_steps": num_inference_steps,
+                "negative_prompt": negative_prompt,
+                "seed": seed
+            }
+        }
+        
+    except Exception as e:
+        if "NEBIUS_API_KEY" in str(e) or not os.environ.get("NEBIUS_API_KEY"):
+            return {
+                "success": False,
+                "error": "NEBIUS_API_KEY environment variable not set",
+                "output_path": None,
+                'user': os.environ.get("USER")
+            }
+        return {
+            "success": False,
+            "error": f"Failed to generate image: {str(e)}",
+            "output_path": None
+        }