app/mcp/tools/image.py

from __future__ import annotations

from typing import Any

from mcp.server.fastmcp import FastMCP

from app.mcp.registry import MCPRegistry, MediaInput
from app.operations.concat import image_slideshow
from app.operations.concat import image_to_video as image_to_video_operation
from app.operations.resize import resize_image as resize_image_operation
from app.operations.watermark import watermark_image as watermark_image_operation


def register_image_tools(server: FastMCP[Any], registry: MCPRegistry) -> None:
    """Register image MCP tools backed by existing operations."""

    @server.tool(description="Turn one still image into an H.264 video.")
    async def image_to_video(
        input: MediaInput, duration: float = 5, fps: int = 30
    ) -> dict[str, Any]:
        """Create a video from one image."""
        return await registry.run_operation(
            "image_to_video",
            [input],
            {"duration": duration, "fps": fps},
            image_to_video_operation,
        )

    @server.tool(description="Create a video slideshow from multiple images.")
    async def slideshow(
        inputs: list[MediaInput],
        duration_per_image: float = 3,
        width: int = 1280,
        height: int = 720,
        fps: int = 30,
    ) -> dict[str, Any]:
        """Create a slideshow through the shared image operation."""
        return await registry.run_operation(
            "slideshow",
            inputs,
            {
                "duration_per_image": duration_per_image,
                "width": width,
                "height": height,
                "fps": fps,
            },
            image_slideshow,
        )

    @server.tool(description="Apply an image watermark to another image.")
    async def watermark_image(
        image: MediaInput,
        watermark: MediaInput,
        position: str = "bottom-right",
        opacity: float = 1.0,
        watermark_scale: float = 1.0,
        format: str = "png",
    ) -> dict[str, Any]:
        """Apply a watermark through the existing image operation."""
        return await registry.run_operation(
            "watermark_image",
            [image, watermark],
            {
                "position": position,
                "opacity": opacity,
                "watermark_scale": watermark_scale,
                "format": format,
            },
            watermark_image_operation,
        )

    @server.tool(description="Resize an image with contain, cover, or fill behavior.")
    async def resize_image(
        input: MediaInput,
        width: int = 1280,
        height: int = 720,
        fit: str = "contain",
        format: str = "png",
    ) -> dict[str, Any]:
        """Resize one image through the existing image operation."""
        return await registry.run_operation(
            "resize_image",
            [input],
            {"width": width, "height": height, "fit": fit, "format": format},
            resize_image_operation,
        )