Skip to content

media processing tools #3

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Apr 9, 2025
Merged

media processing tools #3

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9e88e7c
processing part
Apr 9, 2025
a51e678
Merge branch 'main' of github.com:qiniu/mcp-server
Apr 9, 2025
3a8cf26
processing part
Apr 9, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 6 additions & 3 deletions mcp_server/core/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
211 changes: 166 additions & 45 deletions mcp_server/core/media_processing/tools.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,50 +8,55 @@

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 [
types.TextContent(type="text", text="percent must be between 1 and 999")
]

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",
Expand All @@ -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", "")
Expand All @@ -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",
Expand All @@ -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)
Loading
Loading