Spaces:
Running
Running
File size: 2,981 Bytes
fba6023 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | 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,
)
|