diff --git a/mcp_server/core/__init__.py b/mcp_server/core/__init__.py index 1369c2b..e294ce4 100644 --- a/mcp_server/core/__init__.py +++ b/mcp_server/core/__init__.py @@ -8,6 +8,9 @@ def load(): # 加载配置 cfg = config.load_config() - load_storage(cfg) # 存储业务 - load_cdn(cfg) # CDN - load_media_processing(cfg) # dora + # 存储业务 + load_storage(cfg) + # CDN + load_cdn(cfg) + # 智能多媒体 + load_media_processing(cfg) diff --git a/mcp_server/core/media_processing/tools.py b/mcp_server/core/media_processing/tools.py index 346456c..e42ecb5 100644 --- a/mcp_server/core/media_processing/tools.py +++ b/mcp_server/core/media_processing/tools.py @@ -8,42 +8,47 @@ logger = logging.getLogger(consts.LOGGER_NAME) -_OBJECT_URL_DESC = "图片的 URL，可以通过 GetObjectURL 工具获取的 URL，也可以是其他 Fop 工具生成的 url。 Length Constraints: Minimum length of 1." +_OBJECT_URL_DESC = "The URL of the image. This can be a URL obtained via the GetObjectURL tool or a URL generated by other Fop tools. Length Constraints: Minimum length of 1." -class _ToolImplement: +class _ToolImpl: def __init__(self, cli: Client): self.client = cli - @staticmethod - def image_scale_by_percent_fop() -> types.Tool: - return types.Tool( - name="ImageScaleByPercentFop", - description="图片缩放工具，根据缩放的百分比对图片进行缩放，返回缩放后图片的信息，信息中包含图片的缩放后的 object_url，图片必须存储在七牛云 Bucket 中。", + @tools.tool_meta( + types.Tool( + name="ImageScaleByPercent", + description="""Image scaling tool that resizes images based on a percentage and returns information about the scaled image. + The information includes the object_url of the scaled image, which users can directly use for HTTP GET requests to retrieve the image content or open in a browser to view the file. + The image must be stored in a Qiniu Cloud Bucket. + Supported original image formats: psd, jpeg, png, gif, webp, tiff, bmp, avif, heic. Image width and height cannot exceed 30,000 pixels, and total pixels cannot exceed 150 million. + """, inputSchema={ "type": "object", "properties": { - "object_url": {"type": "string", "description": _OBJECT_URL_DESC}, + "object_url": { + "type": "string", + "description": _OBJECT_URL_DESC + }, "percent": { "type": "integer", - "description": "缩放百分比，范围在[1,999]，比如：90 即为是图片的宽高均缩小至原来的 90%；200 即为是图片的宽高均扩大至原来的 200%", + "description": "Scaling percentage, range [1,999]. For example: 90 means the image width and height are reduced to 90% of the original; 200 means the width and height are enlarged to 200% of the original.", + "minimum": 1, + "maximum": 999 }, }, - "required": ["percent"], + "required": ["object_url", "percent"], }, ) - + ) def image_scale_by_percent( - self, **kwargs + self, **kwargs ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]: object_url = kwargs.get("object_url", "") percent = kwargs.get("percent", "") if object_url is None or len(object_url) == 0: return [types.TextContent(type="text", text="object_url is required")] - if percent is None or len(percent) == 0: - return [types.TextContent(type="text", text="percent is required")] - percent_int = int(percent) if percent_int < 1 or percent_int > 999: return [ @@ -51,7 +56,7 @@ def image_scale_by_percent( ] fop = f"imageMogr2/thumbnail/!{percent}p" - object_url = utils.url_add_processing_tool(object_url, fop) + object_url = utils.url_add_processing_func(object_url, fop) return [ types.TextContent( type="text", @@ -63,31 +68,43 @@ def image_scale_by_percent( ) ] - @staticmethod - def image_scale_by_size_tool() -> types.Tool: - return types.Tool( - name="ImageScaleBySizeFop", - description="图片缩放工具，可以根据新图片的宽或高对图片进行缩放，返回缩放后图片的信息，信息中包含图片的缩放后的 object_url，图片必须存储在七牛云 Bucket 中。原图格式支持： psd、jpeg、png、gif、webp、tiff、bmp、avif、heic。图片 width 和 height 不能超过3万像素，总像素不能超过1.5亿像素", + @tools.tool_meta( + types.Tool( + name="ImageScaleBySize", + description="""Image scaling tool that resizes images based on a specified width or height and returns information about the scaled image. + The information includes the object_url of the scaled image, which users can directly use for HTTP GET requests to retrieve the image content or open in a browser to view the file. + The image must be stored in a Qiniu Cloud Bucket. + Supported original image formats: psd, jpeg, png, gif, webp, tiff, bmp, avif, heic. Image width and height cannot exceed 30,000 pixels, and total pixels cannot exceed 150 million. + """, inputSchema={ "type": "object", "properties": { - "object_url": {"type": "string", "description": _OBJECT_URL_DESC}, + "object_url": { + "type": "string", + "description": _OBJECT_URL_DESC + }, "width": { "type": "integer", - "description": "指定图片宽度进行缩放，也即图片缩放到指定的宽度，图片高度按照宽度缩放比例进行适应。", + "description": "Specifies the width for image scaling. The image will be scaled to the specified width, and the height will be adjusted proportionally.", + "minimum": 1 }, "height": { "type": "integer", - "description": "指定图片高度进行缩放，也即图片缩放到指定的高度，图片宽度按照高度缩放比例进行适应。", + "description": "Specifies the height for image scaling. The image will be scaled to the specified height, and the width will be adjusted proportionally.", + "minimum": 1 }, }, - "required": [], + "required": ["object_url"], + "anyOf": [ + {"required": ["width"]}, + {"required": ["height"]} + ] }, ) - + ) def image_scale_by_size( - self, **kwargs - ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]: + self, **kwargs + ) -> list[types.TextContent]: object_url = kwargs.get("object_url", "") width = kwargs.get("width", "") height = kwargs.get("height", "") @@ -103,7 +120,7 @@ def image_scale_by_size( ] fop = f"imageMogr2/thumbnail/{fop}" - object_url = utils.url_add_processing_tool(object_url, fop) + object_url = utils.url_add_processing_func(object_url, fop) return [ types.TextContent( type="text", @@ -115,36 +132,140 @@ def image_scale_by_size( ) ] - @staticmethod - def get_fop_status_tool() -> types.Tool: - return types.Tool( + @tools.tool_meta( + types.Tool( + name="ImageRoundCorner", + description="""Image rounded corner tool that processes images based on width, height, and corner radius, returning information about the processed image. + If only radius_x or radius_y is set, the other parameter will be assigned the same value, meaning horizontal and vertical parameters will be identical. + The information includes the object_url of the processed image, which users can directly use for HTTP GET requests to retrieve the image content or open in a browser to view the file. + The image must be stored in a Qiniu Cloud Bucket. + Supported original image formats: psd, jpeg, png, gif, webp, tiff, bmp, avif, heic. Image width and height cannot exceed 30,000 pixels, and total pixels cannot exceed 150 million. + Corner radius supports pixels and percentages, but cannot be negative. Pixels are represented by numbers, e.g., 200 means 200px; percentages use !xp, e.g., !25p means 25%.""", + inputSchema={ + "type": "object", + "properties": { + "object_url": { + "type": "string", + "description": _OBJECT_URL_DESC + }, + "radius_x": { + "type": "string", + "description": "Parameter for horizontal corner size. Can use: pixel values (e.g., 200 for 200px) or percentages (e.g., !25p for 25%), all non-negative values." + }, + "radius_y": { + "type": "string", + "description": "Parameter for vertical corner size. Can use: pixel values (e.g., 200 for 200px) or percentages (e.g., !25p for 25%), all non-negative values." + }, + }, + "required": ["object_url"], + "anyOf": [ + {"required": ["radius_x"]}, + {"required": ["radius_y"]} + ] + } + ) + ) + def image_round_corner(self, **kwargs) -> list[types.TextContent]: + object_url = kwargs.get("object_url", "") + radius_x = kwargs.get("radius_x", "") + radius_y = kwargs.get("radius_y", "") + if object_url is None or len(object_url) == 0: + return [ + types.TextContent( + type="text", + text="object_url is required" + ) + ] + + if (radius_x is None or len(radius_x) == 0) and (radius_y is None or len(radius_y) == 0) is None: + return [ + types.TextContent( + type="text", + text="At least one of radius_x or radius_y must be set" + ) + ] + + if radius_x is None or len(radius_x) == 0: + radius_x = radius_y + elif radius_y is None or len(radius_y) == 0: + radius_y = radius_x + + func = f"roundPic/radiusx/{radius_x}/radiusy/{radius_y}" + object_url = utils.url_add_processing_func(object_url, func) + return [ + types.TextContent( + type="text", + text=str({ + "object_url": object_url, + }) + ) + ] + + @tools.tool_meta( + types.Tool( + name="ImageInfo", + description="Retrieves basic image information, including image format, size, and color model.", + inputSchema={ + "type": "object", + "properties": { + "object_url": { + "type": "string", + "description": _OBJECT_URL_DESC + }, + }, + "required": ["object_url"], + }, + ) + ) + def image_info(self, **kwargs) -> list[types.TextContent]: + object_url = kwargs.get("object_url", "") + if object_url is None or len(object_url) == 0: + return [ + types.TextContent( + type="text", + text="object_url is required" + ) + ] + + func = "imageInfo" + object_url = utils.url_add_processing_func(object_url, func) + return [ + types.TextContent( + type="text", + text=str({ + "object_url": object_url, + }) + ) + ] + + @tools.tool_meta( + types.Tool( name="GetFopStatus", - description="获取 Fop 执行状态", + description="Retrieves the execution status of a Fop operation.", inputSchema={ "type": "object", "properties": { "persistent_id": { "type": "string", - "description": "执行 Fop 返回的操作 ID", + "description": "Operation ID returned from executing a Fop operation", }, }, "required": ["persistent_id"], }, ) - - def get_fop_status( - self, **kwargs - ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]: + ) + def get_fop_status(self, **kwargs) -> list[types.TextContent]: status = self.client.get_fop_status(**kwargs) return [types.TextContent(type="text", text=str(status))] def register_tools(cli: Client): - tool = _ToolImplement(cli) - tools.register_tool( - _ToolImplement.image_scale_by_percent_fop(), tool.image_scale_by_percent - ) - tools.register_tool( - _ToolImplement.image_scale_by_size_tool(), tool.image_scale_by_size + tool_impl = _ToolImpl(cli) + tools.auto_register_tools( + [ + tool_impl.image_scale_by_percent, + tool_impl.image_scale_by_size, + tool_impl.image_round_corner, + tool_impl.image_info, + ] ) - # tools.register_tool(_ToolImplement.get_fop_status_tool(), tool.get_fop_status) diff --git a/mcp_server/core/media_processing/utils.py b/mcp_server/core/media_processing/utils.py index 8842cd4..9e3d972 100644 --- a/mcp_server/core/media_processing/utils.py +++ b/mcp_server/core/media_processing/utils.py @@ -1,62 +1,66 @@ from urllib import parse +FUNC_POSITION_NONE = "none" +FUNC_POSITION_PREFIX = "prefix" +FUNC_POSITION_SUFFIX = "suffix" -def url_add_processing_tool(url: str, tool: str) -> str: - tool_items = tool.split("/") - tool_prefix = tool_items[0] + +def url_add_processing_func(url: str, func: str) -> str: + func_items = func.split("/") + func_prefix = func_items[0] url_info = parse.urlparse(url) - new_query = _query_add_processing_tool(url_info.query, tool, tool_prefix) + new_query = _query_add_processing_func(url_info.query, func, func_prefix) new_query = parse.quote(new_query, safe="") url_info = url_info._replace(query=new_query) new_url = parse.urlunparse(url_info) return str(new_url) -def _query_add_processing_tool(query: str, tool: str, tool_prefix: str) -> str: +def _query_add_processing_func(query: str, func: str, func_prefix: str) -> str: queries = query.split("&") - if "" in queries: - queries.remove("") + if '' in queries: + queries.remove('') # query 中不包含任何数据 if len(queries) == 0: - return tool + return func - # tool 会放在第一个元素中 + # funcs 会放在第一个元素中 first_query = parse.unquote(queries[0]) queries.remove(queries[0]) - # tool 不存在 + # funcs 不存在 if len(first_query) == 0: - queries.insert(0, tool) + queries.insert(0, func) return "&".join(queries) - # 未找到当前类别的 tool，则直接拼接在后面 - if first_query.find(tool_prefix) < 0: - tool = first_query + "|" + tool - queries.insert(0, tool) + # 未找到当前类别的 func + if first_query.find(func_prefix) < 0: + func = first_query + "|" + func + queries.insert(0, func) return "&".join(queries) - query_tools = first_query.split("|") - if "" in query_tools: - query_tools.remove("") + query_funcs = first_query.split("|") + if '' in query_funcs: + query_funcs.remove('') - # 只有一个 tool，且和当前 tool 相同，拼接气候 - if len(query_tools) == 1: - tool = first_query + tool.removeprefix(tool_prefix) - queries.insert(0, tool) + # 只有一个 func，且和当前 func 相同，拼接其后 + if len(query_funcs) == 1: + func = first_query + func.removeprefix(func_prefix) + queries.insert(0, func) return "&".join(queries) - # 多个 tool，查看最后一个是否和当前 tool 匹配 - last_tool = query_tools[-1] + # 多个 func，查看最后一个是否和当前 func 匹配 + last_func = query_funcs[-1] # 最后一个不匹配，只用管道符拼接 - if last_tool.find(tool_prefix) < 0: - tool = first_query + "|" + tool - queries.insert(0, tool) + if last_func.find(func_prefix) < 0: + func = first_query + "|" + func + queries.insert(0, func) return "&".join(queries) # 最后一个匹配，则直接拼接在后面 - tool = first_query + tool.removeprefix(tool_prefix) - queries.insert(0, tool) + func = first_query + func.removeprefix(func_prefix) + queries.insert(0, func) return "&".join(queries) diff --git a/mcp_server/core/storage/tools.py b/mcp_server/core/storage/tools.py index fe62218..4ce88b3 100644 --- a/mcp_server/core/storage/tools.py +++ b/mcp_server/core/storage/tools.py @@ -2,6 +2,8 @@ import base64 from mcp import types +from mcp.types import ImageContent, TextContent + from .storage import StorageService from ...consts import consts from ...tools import tools @@ -85,7 +87,7 @@ async def list_objects(self, **kwargs) -> list[types.TextContent]: }, ) ) - async def get_object(self, **kwargs) -> list[types.TextContent]: + async def get_object(self, **kwargs) -> list[ImageContent] | list[TextContent]: response = await self.storage.get_object(**kwargs) file_content = response["Body"] content_type = response.get("ContentType", "application/octet-stream") diff --git a/mcp_server/tools/tools.py b/mcp_server/tools/tools.py index ff15027..81e912e 100644 --- a/mcp_server/tools/tools.py +++ b/mcp_server/tools/tools.py @@ -85,7 +85,7 @@ def sync_wrapper(*args, **kwargs): return _add_metadata(tool_meta=meta) -def auto_register_tools(func_list: Union[ToolFunc, AsyncToolFunc]): +def auto_register_tools(func_list: list[Union[ToolFunc, AsyncToolFunc]]): """尝试自动注册带有 tool_meta 的工具""" for func in func_list: if hasattr(func, "tool_meta"):