diff --git a/server/mcp_server_mediakit/src/base/api_info.py b/server/mcp_server_mediakit/src/base/api_info.py index d588388d..3d41341a 100644 --- a/server/mcp_server_mediakit/src/base/api_info.py +++ b/server/mcp_server_mediakit/src/base/api_info.py @@ -1,16 +1,41 @@ api_info = { "add_image_to_video": {"path": "/api/v1/tools/add-image-to-video", "method": "POST"}, "add_subtitle_to_video": {"path": "/api/v1/tools/add-subtitle-to-video", "method": "POST"}, + "adjust_audio_speed": {"path": "/api/v1/tools/adjust-audio-speed", "method": "POST"}, "adjust_video_speed": {"path": "/api/v1/tools/adjust-video-speed", "method": "POST"}, + "adjust_video_volume": {"path": "/api/v1/tools/adjust-video-volume", "method": "POST"}, + "analyze_video_highlights": {"path": "/api/v1/tools/analyze-video-highlights", "method": "POST"}, + "analyze_video_storyline": {"path": "/api/v1/tools/analyze-video-storyline", "method": "POST"}, + "apply_video_filter": {"path": "/api/v1/tools/apply-video-filter", "method": "POST"}, + "asr_subtitles": {"path": "/api/v1/tools/asr-subtitles", "method": "POST"}, "concat_audio": {"path": "/api/v1/tools/concat-audio", "method": "POST"}, "concat_video": {"path": "/api/v1/tools/concat-video", "method": "POST"}, + "enhance_image": {"path": "/api/v1/tools-sync/enhance-image", "method": "POST"}, "enhance_video": {"path": "/api/v1/tools/enhance-video", "method": "POST"}, + "enhance_video_generative": {"path": "/api/v1/tools/enhance-video-generative", "method": "POST"}, + "erase_image": {"path": "/api/v1/tools-sync/erase-image", "method": "POST"}, + "erase_video_subtitle": {"path": "/api/v1/tools/erase-video-subtitle", "method": "POST"}, "erase_video_subtitle_pro": {"path": "/api/v1/tools/erase-video-subtitle-pro", "method": "POST"}, + "evaluate_image_quality": {"path": "/api/v1/tools-sync/evaluate-image-quality", "method": "POST"}, "extract_audio": {"path": "/api/v1/tools/extract-audio", "method": "POST"}, + "fade_audio": {"path": "/api/v1/tools/fade-audio", "method": "POST"}, + "fade_video_audio": {"path": "/api/v1/tools/fade-video-audio", "method": "POST"}, "flip_video": {"path": "/api/v1/tools/flip-video", "method": "POST"}, + "generate_highlights_microdrama": {"path": "/api/v1/tools/generate-highlights-microdrama", "method": "POST"}, + "generate_highlights_minigame": {"path": "/api/v1/tools/generate-highlights-minigame", "method": "POST"}, + "image_ocr": {"path": "/api/v1/tools-sync/image-ocr", "method": "POST"}, "image_to_video": {"path": "/api/v1/tools/image-to-video", "method": "POST"}, + "matte_greenscreen_video": {"path": "/api/v1/tools/matte-greenscreen-video", "method": "POST"}, + "matte_portrait_video": {"path": "/api/v1/tools/matte-portrait-video", "method": "POST"}, + "mix_audio": {"path": "/api/v1/tools/mix-audio", "method": "POST"}, "mux_audio_video": {"path": "/api/v1/tools/mux-audio-video", "method": "POST"}, + "probe_audio_metadata": {"path": "/api/v1/tools/probe-audio-metadata", "method": "POST"}, + "probe_video_metadata": {"path": "/api/v1/tools/probe-video-metadata", "method": "POST"}, "query_task": {"path": "/api/v1/tasks/{task_id}", "method": "GET"}, + "remove_image_background": {"path": "/api/v1/tools-sync/remove-image-background", "method": "POST"}, + "segment_scenes": {"path": "/api/v1/tools/segment-scenes", "method": "POST"}, + "separate_voice": {"path": "/api/v1/tools/separate-voice", "method": "POST"}, "trim_audio": {"path": "/api/v1/tools/trim-audio", "method": "POST"}, "trim_video": {"path": "/api/v1/tools/trim-video", "method": "POST"}, + "video_ocr": {"path": "/api/v1/tools/video-ocr", "method": "POST"}, } diff --git a/server/mcp_server_mediakit/src/mediakit/mcp_tools/audio.py b/server/mcp_server_mediakit/src/mediakit/mcp_tools/audio.py new file mode 100644 index 00000000..bd933551 --- /dev/null +++ b/server/mcp_server_mediakit/src/mediakit/mcp_tools/audio.py @@ -0,0 +1,68 @@ +from __future__ import annotations + +from typing import Any, Dict, List, Optional +from typing_extensions import NotRequired, Required, TypedDict + +try: + from pydantic import Field +except Exception: # pragma: no cover + def Field(*args, **kwargs): + if args: + return args[0] + return kwargs.get("default", None) + +try: + from mcp.server.fastmcp.server import Context + from mcp.server.session import ServerSession +except Exception: # pragma: no cover + class Context: # type: ignore + pass + + class ServerSession: # type: ignore + pass + +from base.client import MediKitClient +from ..utils.response import async_task_response, error_response + +TOOL_NAMES = ['probe_audio_metadata', 'separate_voice'] + + +def register_tools(mcp, client: MediKitClient) -> None: + @mcp.tool(name="separate_voice", description="将音频中的人声与背景音精准分离，输出为两个独立的音轨文件。\n支持格式：主流音视频格式（如mp4、mov、mp3、m4a、wav等）。\n输入：video_url和audio_url二选一。\n输出格式：AAC。 使用 task_id, 调用 query_task 方法获取结果") + async def separate_voice( + video_url: Optional[str] = Field(None, description="输入视频 Url（需公网可访问），与audio_url二选一，都存在时优先取video_url"), + audio_url: Optional[str] = Field(None, description="输入音频 Url（需公网可访问），与video_url二选一，不能都为空"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """将音频中的人声与背景音精准分离，输出为两个独立的音轨文件。 +支持格式：主流音视频格式（如mp4、mov、mp3、m4a、wav等）。 +输入：video_url和audio_url二选一。 +输出格式：AAC。""" + try: + result = client.call(api_name="separate_voice", video_url=video_url, audio_url=audio_url, callback_args=callback_args, client_token=client_token) + return async_task_response(result) + except Exception as exc: + return error_response(str(exc)) + + @mcp.tool(name="probe_audio_metadata", description="获取指定音频的详细元信息，输出容器层信息（format_meta）与音频流元信息（audio_stream_meta）。\n字段分类参考 ffprobe，并对 VOD 原始返回做精简与统一。\n使用限制：支持公网 HTTP/HTTPS URL。 使用 task_id, 调用 query_task 方法获取结果") + async def probe_audio_metadata( + audio_url: str = Field(..., description="待探测的音频公网 HTTP/HTTPS URL。"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """获取指定音频的详细元信息，输出容器层信息（format_meta）与音频流元信息（audio_stream_meta）。 +字段分类参考 ffprobe，并对 VOD 原始返回做精简与统一。 +使用限制：支持公网 HTTP/HTTPS URL。""" + try: + result = client.call(api_name="probe_audio_metadata", audio_url=audio_url, callback_args=callback_args, client_token=client_token) + return async_task_response(result) + except Exception as exc: + return error_response(str(exc)) + + if hasattr(mcp, "register_domain_tools"): + mcp.register_domain_tools("audio", TOOL_NAMES) diff --git a/server/mcp_server_mediakit/src/mediakit/mcp_tools/editing.py b/server/mcp_server_mediakit/src/mediakit/mcp_tools/editing.py index 9f3be50e..81bc5788 100644 --- a/server/mcp_server_mediakit/src/mediakit/mcp_tools/editing.py +++ b/server/mcp_server_mediakit/src/mediakit/mcp_tools/editing.py @@ -36,7 +36,7 @@ class ImageToVideoImagesItem(TypedDict): animation_in: NotRequired[float] animation_out: NotRequired[float] -TOOL_NAMES = ['add_image_to_video', 'add_subtitle_to_video', 'adjust_video_speed', 'concat_audio', 'concat_video', 'extract_audio', 'flip_video', 'image_to_video', 'mux_audio_video', 'trim_audio', 'trim_video'] +TOOL_NAMES = ['add_image_to_video', 'add_subtitle_to_video', 'adjust_audio_speed', 'adjust_video_speed', 'adjust_video_volume', 'apply_video_filter', 'concat_audio', 'concat_video', 'extract_audio', 'fade_audio', 'fade_video_audio', 'flip_video', 'image_to_video', 'mix_audio', 'mux_audio_video', 'trim_audio', 'trim_video'] def register_tools(mcp, client: MediKitClient) -> None: @@ -233,5 +233,106 @@ async def trim_video( except Exception as exc: return error_response(str(exc)) + @mcp.tool(name="adjust_audio_speed", description="调整音频的播放倍速，实现快放或慢放效果。 使用 task_id, 调用 query_task 方法获取结果") + async def adjust_audio_speed( + audio_url: str = Field(..., description="输入音频。支持http://xxx或https://xxx格式 Url，支持 mp3、m4a、wav 等格式"), + speed: Optional[float] = Field(1, description="调整速度的倍数，Float类型，取值范围为0.1～4。0.1=放慢至原速的 0.1 倍，1=原速，4=加速至原速的 4 倍。"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """调整音频的播放倍速，实现快放或慢放效果。""" + try: + result = client.call(api_name="adjust_audio_speed", audio_url=audio_url, speed=speed, callback_args=callback_args, client_token=client_token) + return async_task_response(result) + except Exception as exc: + return error_response(str(exc)) + + @mcp.tool(name="adjust_video_volume", description="调整视频音量大小，支持静音；输出 mp4，分辨率与原片一致。 使用 task_id, 调用 query_task 方法获取结果") + async def adjust_video_volume( + video_url: str = Field(..., description="输入视频。支持http://xxx或https://xxx格式 URL，支持 mp4、mov、flv、ts、avi、wmv、mkv 等格式，最高 4K"), + volume: Optional[float] = Field(1, description="音量倍数。Float 类型，取值范围 0~4。0=静音，1=原音量，4=放大 4 倍。"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """调整视频音量大小，支持静音；输出 mp4，分辨率与原片一致。""" + try: + result = client.call(api_name="adjust_video_volume", video_url=video_url, volume=volume, callback_args=callback_args, client_token=client_token) + return async_task_response(result) + except Exception as exc: + return error_response(str(exc)) + + @mcp.tool(name="apply_video_filter", description="为视频添加指定滤镜效果，输出mp4，分辨率与原片一致。 使用 task_id, 调用 query_task 方法获取结果") + async def apply_video_filter( + video_url: str = Field(..., description="输入视频。支持http://xxx或https://xxx格式 URL，支持 mp4、mov、flv、ts、avi、wmv、mkv 等格式，最高 4K"), + filter_style: Optional[str] = Field('spring', description="滤镜风格。根据用户想要的视频画面效果选择：\n- spring：春日滤镜\n- sunset：晚霞滤镜\n- vivid：鲜亮滤镜\n- fair_skin：白皙滤镜\n- food：食物滤镜\n"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """为视频添加指定滤镜效果，输出mp4，分辨率与原片一致。""" + try: + result = client.call(api_name="apply_video_filter", video_url=video_url, filter_style=filter_style, callback_args=callback_args, client_token=client_token) + return async_task_response(result) + except Exception as exc: + return error_response(str(exc)) + + @mcp.tool(name="fade_audio", description="对输入音频实现淡入淡出效果，输出 mp3。 使用 task_id, 调用 query_task 方法获取结果") + async def fade_audio( + audio_url: str = Field(..., description="输入音频。支持http://xxx或https://xxx格式 URL，支持 mp3、m4a、wav、flac 等格式"), + fade_in_duration: Optional[float] = Field(1, description="声音淡入时长。单位：秒，可传小数（最多3位小数）。0 表示不淡入。"), + fade_out_duration: Optional[float] = Field(1, description="声音淡出时长。单位：秒，可传小数（最多3位小数）。0 表示不淡出。"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """对输入音频实现淡入淡出效果，输出 mp3。""" + try: + result = client.call(api_name="fade_audio", audio_url=audio_url, fade_in_duration=fade_in_duration, fade_out_duration=fade_out_duration, callback_args=callback_args, client_token=client_token) + return async_task_response(result) + except Exception as exc: + return error_response(str(exc)) + + @mcp.tool(name="fade_video_audio", description="对输入视频的声轨实现淡入淡出效果。\n输出 mp4，分辨率与原片一致。 使用 task_id, 调用 query_task 方法获取结果") + async def fade_video_audio( + video_url: str = Field(..., description="输入视频。支持http://xxx或https://xxx格式 URL，支持 mp4、mov、flv、ts、avi、wmv、mkv 等格式，最高 4K"), + fade_in_duration: Optional[float] = Field(1, description="声音淡入时长。单位：秒，可传小数（最多3位小数）。0 表示不淡入。"), + fade_out_duration: Optional[float] = Field(1, description="声音淡出时长。单位：秒，可传小数（最多3位小数）。0 表示不淡出。"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """对输入视频的声轨实现淡入淡出效果。 +输出 mp4，分辨率与原片一致。""" + try: + result = client.call(api_name="fade_video_audio", video_url=video_url, fade_in_duration=fade_in_duration, fade_out_duration=fade_out_duration, callback_args=callback_args, client_token=client_token) + return async_task_response(result) + except Exception as exc: + return error_response(str(exc)) + + @mcp.tool(name="mix_audio", description="将多个音频文件（如背景音乐、音效、人声）进行混音，生成一个新的音频文件。\n处理耗时：处理耗时与视频时长正相关。视频时长越长，处理耗时越长。平均 RTF（处理耗时/原片时长）为 1。\n输出音频的时长以最长的音频为准。\n输出视频格式：mp3 使用 task_id, 调用 query_task 方法获取结果") + async def mix_audio( + audio_urls: List[str] = Field(..., description="待混合的音频列表，Array类型。最少传入1个，最多传入100个。\n子项说明：待混合的输入音频。支持http://xxx或https://xxx格式 URL，支持 mp3、wav、flac 等格式"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """将多个音频文件（如背景音乐、音效、人声）进行混音，生成一个新的音频文件。 +处理耗时：处理耗时与视频时长正相关。视频时长越长，处理耗时越长。平均 RTF（处理耗时/原片时长）为 1。 +输出音频的时长以最长的音频为准。 +输出视频格式：mp3""" + try: + result = client.call(api_name="mix_audio", audio_urls=audio_urls, callback_args=callback_args, client_token=client_token) + return async_task_response(result) + except Exception as exc: + return error_response(str(exc)) + if hasattr(mcp, "register_domain_tools"): mcp.register_domain_tools("editing", TOOL_NAMES) diff --git a/server/mcp_server_mediakit/src/mediakit/mcp_tools/image.py b/server/mcp_server_mediakit/src/mediakit/mcp_tools/image.py new file mode 100644 index 00000000..9d552a63 --- /dev/null +++ b/server/mcp_server_mediakit/src/mediakit/mcp_tools/image.py @@ -0,0 +1,126 @@ +from __future__ import annotations + +from typing import Any, Dict, List, Optional +from typing_extensions import NotRequired, Required, TypedDict + +try: + from pydantic import Field +except Exception: # pragma: no cover + def Field(*args, **kwargs): + if args: + return args[0] + return kwargs.get("default", None) + +try: + from mcp.server.fastmcp.server import Context + from mcp.server.session import ServerSession +except Exception: # pragma: no cover + class Context: # type: ignore + pass + + class ServerSession: # type: ignore + pass + +from base.client import MediKitClient +from ..utils.response import sync_result_response, error_response + +TOOL_NAMES = ['enhance_image', 'erase_image', 'evaluate_image_quality', 'image_ocr', 'remove_image_background'] + + +def register_tools(mcp, client: MediKitClient) -> None: + @mcp.tool(name="image_ocr", description="识别图片中的通用印刷体文字，返回可编辑文本、文字框坐标和置信度。\n本期支持简体中文和英文通用场景识别。") + async def image_ocr( + image_url: str = Field(..., description="输入图片 URL，需为公网可访问的 png/jpg/jpeg/webp/heic/avif 图片，单图不超过 10MB。"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """识别图片中的通用印刷体文字，返回可编辑文本、文字框坐标和置信度。 +本期支持简体中文和英文通用场景识别。""" + try: + result = client.call(api_name="image_ocr", image_url=image_url, callback_args=callback_args, client_token=client_token) + return sync_result_response(result) + except Exception as exc: + return error_response(str(exc)) + + @mcp.tool(name="erase_image", description="自动检测并擦除图片中的常见图标、文字或指定区域内容，并对擦除区域进行背景智能填充。") + async def erase_image( + image_url: str = Field(..., description="输入图片 URL，需为公网可访问的 png/jpg/jpeg/webp/tiff/bmp/heic 图片，单图不超过 10MB。"), + tool_version: Optional[str] = Field('standard', description="图像擦除修复选用的模型版本。- standard：标准版。基于明确的规则（如文本匹配、矩形框坐标）擦除指定内容。适用于简单、明确的擦除任务。默认 standard。"), + standard_scene: Optional[str] = Field('full_screen_text_erase', description="标准版擦除场景，仅 standard 版本生效。full_screen_text_erase：全屏文字擦除，可通过standard_erase_text字段指定要擦除的文字，不指定则默认擦除所有文字内容。full_screen_icon_erase：全屏图标擦除。"), + standard_erase_text: Optional[str] = Field(None, description="标准版文字擦除，指定要擦除的文字，不指定则默认擦除所有文字内容。"), + output_format: Optional[str] = Field('webp', description="输出图片格式；默认 webp。"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """自动检测并擦除图片中的常见图标、文字或指定区域内容，并对擦除区域进行背景智能填充。""" + try: + result = client.call(api_name="erase_image", image_url=image_url, tool_version=tool_version, standard_scene=standard_scene, standard_erase_text=standard_erase_text, output_format=output_format, callback_args=callback_args, client_token=client_token) + return sync_result_response(result) + except Exception as exc: + return error_response(str(exc)) + + @mcp.tool(name="remove_image_background", description="自动识别并保留图像主体，移除背景并生成透明背景图片。\n支持通用、人像、商品场景，可在人像/商品场景中生成主体描边或裁剪透明背景。") + async def remove_image_background( + image_url: str = Field(..., description="输入图片 URL，需为公网可访问的 png/jpg/jpeg/webp/tiff/bmp/ico 图片，单图不超过 10MB。"), + scene: str = Field(..., description="背景移除场景：general 为通用场景，适用于期望抠出图像主体但不确定该主体所属分类的场景。human 为人像抠图场景，适用于仅需抠出图像中的人像主体的场景，product 为商品抠图场景，适用于仅需抠出图像中的商品主体的场景。"), + need_contour: Optional[bool] = Field(False, description="是否为主体生成描边；默认 false，仅 human/product 场景生效，general 场景忽略。"), + contour_color: Optional[str] = Field('#FFFFFF', description="主体描边颜色，十六进制 RGB；默认 #FFFFFF，仅 need_contour=true 且 human/product 场景生效。"), + contour_size: Optional[int] = Field(10, description="主体描边宽度，单位 px；默认 10，仅 need_contour=true 且 human/product 场景生效。"), + need_crop_background: Optional[bool] = Field(False, description="是否裁剪透明背景到刚好包住主体；默认 false，仅 human/product 场景生效，general 场景忽略。"), + output_format: Optional[str] = Field('png', description="输出图片格式；默认 png。"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """自动识别并保留图像主体，移除背景并生成透明背景图片。 +支持通用、人像、商品场景，可在人像/商品场景中生成主体描边或裁剪透明背景。""" + try: + result = client.call(api_name="remove_image_background", image_url=image_url, scene=scene, need_contour=need_contour, contour_color=contour_color, contour_size=contour_size, need_crop_background=need_crop_background, output_format=output_format, callback_args=callback_args, client_token=client_token) + return sync_result_response(result) + except Exception as exc: + return error_response(str(exc)) + + @mcp.tool(name="enhance_image", description="基于图像内容理解智能决策，全方位提升图片分辨率、清晰度与色彩表现。") + async def enhance_image( + image_url: str = Field(..., description="输入图片。String 类型，支持http://xxx或https://xxx格式 URL"), + tool_version: Optional[str] = Field('standard', description="画质增强选用的模型版本，标准版:standard；专业版：professional。默认为标准版"), + multiple: Optional[float] = Field(None, description="图像处理后较原图的分辨率倍数，支持 2 位小数。取值最大不超过 30，取值范围[1,30]。注意：图像处理后的宽度和高度不能超过target_width、target_height的上限值。standard模式下，取值最大不超过 8。"), + target_width: Optional[int] = Field(None, description="图像处理后的宽度，单位为 px，取值不能超过 10240。注意：standard模式下，取值最大不超过 6144，且图像处理后较原图的分辨率倍数不能超过 8。"), + target_height: Optional[int] = Field(None, description="图像处理后的高度，单位为 px，取值不能超过 10240。注意：standard模式下，取值最大不超过 6144，且图像处理后较原图的分辨率倍数不能超过 8。"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """基于图像内容理解智能决策，全方位提升图片分辨率、清晰度与色彩表现。""" + try: + result = client.call(api_name="enhance_image", image_url=image_url, tool_version=tool_version, multiple=multiple, target_width=target_width, target_height=target_height, callback_args=callback_args, client_token=client_token) + return sync_result_response(result) + except Exception as exc: + return error_response(str(exc)) + + @mcp.tool(name="evaluate_image_quality", description="对输入图片进行主客观画质和美学评分，适用于质量监控、低质图筛查、内容审核、推荐排序和训练数据清洗等场景。\n支持标准版多维评分与专业版大模型评分。") + async def evaluate_image_quality( + image_url: str = Field(..., description="输入图片 URL，需为公网可访问的 png/jpeg/webp/heic 图片，单图不超过 10MB。"), + tool_version: Optional[str] = Field('standard', description="画质评估模型版本，standard 为标准版，professional 为专业版；默认 standard。"), + standard_evaluate_items: Optional[List[str]] = Field(['vqscore', 'noise', 'aesthetic', 'blur'], description="标准版选用的评估工具\n子项说明：评估工具。"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """对输入图片进行主客观画质和美学评分，适用于质量监控、低质图筛查、内容审核、推荐排序和训练数据清洗等场景。 +支持标准版多维评分与专业版大模型评分。""" + try: + result = client.call(api_name="evaluate_image_quality", image_url=image_url, tool_version=tool_version, standard_evaluate_items=standard_evaluate_items, callback_args=callback_args, client_token=client_token) + return sync_result_response(result) + except Exception as exc: + return error_response(str(exc)) + + if hasattr(mcp, "register_domain_tools"): + mcp.register_domain_tools("image", TOOL_NAMES) diff --git a/server/mcp_server_mediakit/src/mediakit/mcp_tools/video.py b/server/mcp_server_mediakit/src/mediakit/mcp_tools/video.py index b56213f9..13ce9c83 100644 --- a/server/mcp_server_mediakit/src/mediakit/mcp_tools/video.py +++ b/server/mcp_server_mediakit/src/mediakit/mcp_tools/video.py @@ -30,18 +30,80 @@ class EraseVideoSubtitleProEraseRatioLocationItem(TypedDict): bottom_right_x: Required[float] bottom_right_y: Required[float] -TOOL_NAMES = ['enhance_video', 'erase_video_subtitle_pro'] +TOOL_NAMES = ['analyze_video_highlights', 'analyze_video_storyline', 'asr_subtitles', 'enhance_video', 'enhance_video_generative', 'erase_video_subtitle', 'erase_video_subtitle_pro', 'generate_highlights_microdrama', 'generate_highlights_minigame', 'matte_greenscreen_video', 'matte_portrait_video', 'probe_video_metadata', 'segment_scenes', 'video_ocr'] def register_tools(mcp, client: MediKitClient) -> None: + @mcp.tool(name="analyze_video_highlights", description="智能捕捉视频\"情绪波峰\"与\"关键动作\"，输出精准时间戳、高光打分、OCR 文本和画面描述等元数据，供下游进行更灵活的二次开发。\n支持短剧（Miniseries）和小游戏（Game）两种分析模型。\n使用限制：单次最多 100 个视频，累计时长不超过 300 分钟。 使用 task_id, 调用 query_task 方法获取结果") + async def analyze_video_highlights( + video_urls: List[str] = Field(..., description="待处理的视频 URL 列表，支持 1-100 个视频\n子项说明：视频 URL，支持 http:// 或 https:// 格式"), + model: str = Field(..., description="分析场景模型，Miniseries（短剧）或 Game（小游戏）"), + mode: str = Field(..., description="高光提取模式。固定组合为：model=Miniseries 时 mode 只能传 StorylineCuts；model=Game 时 mode 只能传 HighlightExtract"), + minigame_info: Optional[Dict[str, Any]] = Field(None, description="小游戏描述信息，当 model=Game 时可选填，可辅助模型更精准识别高光内容"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """智能捕捉视频"情绪波峰"与"关键动作"，输出精准时间戳、高光打分、OCR 文本和画面描述等元数据，供下游进行更灵活的二次开发。 +支持短剧（Miniseries）和小游戏（Game）两种分析模型。 +使用限制：单次最多 100 个视频，累计时长不超过 300 分钟。""" + try: + result = client.call(api_name="analyze_video_highlights", video_urls=video_urls, model=model, mode=mode, minigame_info=minigame_info, callback_args=callback_args, client_token=client_token) + return async_task_response(result) + except Exception as exc: + return error_response(str(exc)) + + @mcp.tool(name="analyze_video_storyline", description="智能解析影视剧内容，生成结构化剧情线，供智能剪辑、内容检索与互动播放等场景使用。\n基于大模型视频理解能力，对输入的单个或多个长视频（如电影、电视剧）进行分析，提取并组织成一份完整的故事线。\n该故事线由一系列按时间顺序排列的剧情片段（Clips）和基于片段聚合的高光故事线（Highlights）组成。\n使用限制：单次最多 30 个视频，单个视频时长不超过 2.5 小时。 使用 task_id, 调用 query_task 方法获取结果") + async def analyze_video_storyline( + video_urls: List[str] = Field(..., description="待处理的视频 URL 列表，支持 HTTP/HTTPS 公网可访问链接，最多 30 个视频\n子项说明：视频 URL，支持 http:// 或 https:// 格式"), + enable_snapshot: Optional[bool] = Field(False, description="是否为每个剧情片段生成关键帧快照。默认为 false。开启后，结果中将包含 clip_snapshot_url 字段"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """智能解析影视剧内容，生成结构化剧情线，供智能剪辑、内容检索与互动播放等场景使用。 +基于大模型视频理解能力，对输入的单个或多个长视频（如电影、电视剧）进行分析，提取并组织成一份完整的故事线。 +该故事线由一系列按时间顺序排列的剧情片段（Clips）和基于片段聚合的高光故事线（Highlights）组成。 +使用限制：单次最多 30 个视频，单个视频时长不超过 2.5 小时。""" + try: + result = client.call(api_name="analyze_video_storyline", video_urls=video_urls, enable_snapshot=enable_snapshot, callback_args=callback_args, client_token=client_token) + return async_task_response(result) + except Exception as exc: + return error_response(str(exc)) + + @mcp.tool(name="asr_subtitles", description="对输入视频或音频进行语音识别，输出带时间戳的字幕片段。\n支持格式：主流音视频格式（如mp4、mov、mp3、m4a、wav等）。\n输入：video_url和audio_url二选一。 使用 task_id, 调用 query_task 方法获取结果") + async def asr_subtitles( + video_url: Optional[str] = Field(None, description="输入视频 Url（需公网可访问），与audio_url二选一，都存在时优先取video_url"), + audio_url: Optional[str] = Field(None, description="输入音频 Url（需公网可访问），与video_url二选一，不能都为空"), + content_type: Optional[str] = Field(None, description="识别类型，默认值为空，算法会自动探测类型，speech: 对话，singing: 歌唱"), + language: Optional[str] = Field(None, description="识别提示语言 ID (默认值为空，算法会自动探测语种）\n分类：简体中文，ID：cmn-Hans-CN\n分类：英语，ID：eng-US\n"), + enable_speaker_info: Optional[bool] = Field(False, description="是否开启说话人识别"), + enable_confidence: Optional[bool] = Field(False, description="是否返回置信度"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """对输入视频或音频进行语音识别，输出带时间戳的字幕片段。 +支持格式：主流音视频格式（如mp4、mov、mp3、m4a、wav等）。 +输入：video_url和audio_url二选一。""" + try: + result = client.call(api_name="asr_subtitles", video_url=video_url, audio_url=audio_url, content_type=content_type, language=language, enable_speaker_info=enable_speaker_info, enable_confidence=enable_confidence, callback_args=callback_args, client_token=client_token) + return async_task_response(result) + except Exception as exc: + return error_response(str(exc)) + @mcp.tool(name="enhance_video", description="画质增强：针对 AIGC / UGC / 短剧 / 教育 / 游戏 / 老片修复等场景，提供画质提升 + 超分增强一站式解决方案。依托 AI MediaKit 智能媒体处理引擎，融合视频内容理解、画质指标智能决策、多维度增强原子算法，实现画质的全面优化。\n支持格式：主流视频格式如mp4、flv、ts、avi、mov、wmv、mkv。\n使用限制：单文件大小不超过100G。 使用 task_id, 调用 query_task 方法获取结果") async def enhance_video( video_url: str = Field(..., description="输入视频。String 类型，支持http://xxx或https://xxx格式 URL"), scene: Optional[str] = Field('common', description="场景化模板类型。用于选择一个针对特定业务场景的预设画质增强模板。支持的取值如下：common（默认值）: 通用模板；ugc: UGC 短视频；short_series: 短剧；aigc: AIGC 内容；old_film: 老片修复"), tool_version: Optional[str] = Field('standard', description="工具版本，标准版:standard，专业版：professional，默认为标准版"), resolution: Optional[str] = Field(None, description="目标分辨率。支持的取值如下所示。配置此参数后，不可同时配置resolution_limit字段"), - resolution_limit: Optional[int] = Field(None, description="目标长宽限制，用于指定输出视频的长边或短边的最大像素值，取值范围为 [64, 2160]。配置此参数后，不可同时配置resolution字段"), - fps: Optional[float] = Field(None, description="目标帧率，单位为 fps。取值范围为 (0, 120]。"), + resolution_limit: Optional[int] = Field(None, description="指定输出视频的短边像素值，取值范围为 [128, 2160]。设置后，系统将锁定视频的短边像素值为设定值，并在保持原视频宽高比的前提下，等比缩放至该限制值。示例：若原视频为 640x480 (4:3)，设置 resolution_limit 为 720，则输出视频的短边将提升至 720，宽度等比缩放至 960。配置此参数后，不可同时配置resolution字段"), + bitrate_level: Optional[str] = Field('medium', description="码率档位。输出视频的目标平均码率。该参数将决定视频的视觉质量和最终的文件体积。参数取值：高码率、中码率（推荐码率）、低码率。非必填，默认为中码率。"), + fps: Optional[float] = Field(None, description="目标帧率，单位为 fps。取值范围为 [15, 120]。"), callback_args: Optional[str] = Field(None, description="可选，回调参数"), client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), *, @@ -51,7 +113,41 @@ async def enhance_video( 支持格式：主流视频格式如mp4、flv、ts、avi、mov、wmv、mkv。 使用限制：单文件大小不超过100G。""" try: - result = client.call(api_name="enhance_video", video_url=video_url, scene=scene, tool_version=tool_version, resolution=resolution, resolution_limit=resolution_limit, fps=fps, callback_args=callback_args, client_token=client_token) + result = client.call(api_name="enhance_video", video_url=video_url, scene=scene, tool_version=tool_version, resolution=resolution, resolution_limit=resolution_limit, bitrate_level=bitrate_level, fps=fps, callback_args=callback_args, client_token=client_token) + return async_task_response(result) + except Exception as exc: + return error_response(str(exc)) + + @mcp.tool(name="enhance_video_generative", description="生成式视频增强修复（generative_video_restoration）是基于扩散大模型（Diffusion-based Large Model）的生成式视频修复技术。不仅可以还原被破坏的像素，更借助大规模预训练积累的丰富视觉先验，主动补全细节、理解语义，生成真实、自然、高保真的视频内容。 使用 task_id, 调用 query_task 方法获取结果") + async def enhance_video_generative( + video_url: str = Field(..., description="输入视频。String 类型，支持http://xxx或https://xxx格式 URL"), + resolution: Optional[str] = Field('720p', description="目标分辨率。支持的取值如下所示。"), + bitrate_level: Optional[str] = Field('medium', description="码率档位。输出视频的目标平均码率。该参数将决定视频的视觉质量和最终的文件体积。参数取值：高码率、中码率（推荐码率）、低码率。非必填，默认为中码率。"), + fps: Optional[float] = Field(None, description="目标帧率，单位为 fps。若未指定 fps 参数，输出视频将保持与原始片源一致的帧率。取值范围为 [15, 120]。建议不超过原片的 4 倍。"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """生成式视频增强修复（generative_video_restoration）是基于扩散大模型（Diffusion-based Large Model）的生成式视频修复技术。不仅可以还原被破坏的像素，更借助大规模预训练积累的丰富视觉先验，主动补全细节、理解语义，生成真实、自然、高保真的视频内容。""" + try: + result = client.call(api_name="enhance_video_generative", video_url=video_url, resolution=resolution, bitrate_level=bitrate_level, fps=fps, callback_args=callback_args, client_token=client_token) + return async_task_response(result) + except Exception as exc: + return error_response(str(exc)) + + @mcp.tool(name="erase_video_subtitle", description="智能检测并擦除视频画面中已有的硬字幕，保留原始背景。\n支持格式：主流视频格式如mp4、flv、ts、avi、mov、wmv、mkv。 使用 task_id, 调用 query_task 方法获取结果") + async def erase_video_subtitle( + video_url: str = Field(..., description="输入视频。String 类型，支持http://xxx或https://xxx格式 Url"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """智能检测并擦除视频画面中已有的硬字幕，保留原始背景。 +支持格式：主流视频格式如mp4、flv、ts、avi、mov、wmv、mkv。""" + try: + result = client.call(api_name="erase_video_subtitle", video_url=video_url, callback_args=callback_args, client_token=client_token) return async_task_response(result) except Exception as exc: return error_response(str(exc)) @@ -75,5 +171,140 @@ async def erase_video_subtitle_pro( except Exception as exc: return error_response(str(exc)) + @mcp.tool(name="generate_highlights_microdrama", description="深度理解短剧角色、剧情与故事线，自动提取高光片段并混剪成投流视频。\n支持故事线混剪模式（StorylineCuts），可选\"短剧三要素\"视觉模板，输出高光集锦、单集预告等。\n支持输出详细分镜信息（storyboard）。\n使用限制：单次最多 100 个视频，累计时长不超过 300 分钟。 使用 task_id, 调用 query_task 方法获取结果") + async def generate_highlights_microdrama( + video_urls: List[str] = Field(..., description="待处理的短剧原片视频 URL 列表，支持 1-100 个视频\n子项说明：视频 URL，支持 http:// 或 https:// 格式"), + mode: Optional[str] = Field('StorylineCuts', description="短剧高光智剪模式，本期固定为 StorylineCuts（故事线混剪模式）"), + enable_generate_video: Optional[bool] = Field(True, description="是否生成混剪成片视频。true（默认）= 同时输出混剪视频与分镜信息；false = 仅输出高光分镜信息（clips/storyboard），不生成混剪视频，此时底层请求不会携带 Edit 字段，且传入的 edit_param 将被忽略。"), + enable_return_poster: Optional[bool] = Field(False, description="是否在结果中返回混剪视频封面图 URL。false（默认）= 不返回封面图；true = 若底层存在封面则返回 poster_url。"), + edit_param: Optional[Dict[str, Any]] = Field(None, description="成片剪辑参数配置"), + highlight_cuts_param: Optional[Dict[str, Any]] = Field(None, description="高光混剪参数配置"), + opening_hook_param: Optional[Dict[str, Any]] = Field(None, description="精彩前置功能参数配置（可选）"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """深度理解短剧角色、剧情与故事线，自动提取高光片段并混剪成投流视频。 +支持故事线混剪模式（StorylineCuts），可选"短剧三要素"视觉模板，输出高光集锦、单集预告等。 +支持输出详细分镜信息（storyboard）。 +使用限制：单次最多 100 个视频，累计时长不超过 300 分钟。""" + try: + result = client.call(api_name="generate_highlights_microdrama", video_urls=video_urls, mode=mode, enable_generate_video=enable_generate_video, enable_return_poster=enable_return_poster, edit_param=edit_param, highlight_cuts_param=highlight_cuts_param, opening_hook_param=opening_hook_param, callback_args=callback_args, client_token=client_token) + return async_task_response(result) + except Exception as exc: + return error_response(str(exc)) + + @mcp.tool(name="generate_highlights_minigame", description="识别小游戏录屏视频中的核心玩法与高光事件（如连击、通关、极限操作等），\n快速生成用于买量的视频素材。支持提供游戏名称、玩法描述、高光定义以辅助模型更精准识别。\n使用限制：本期仅支持单视频输入。 使用 task_id, 调用 query_task 方法获取结果") + async def generate_highlights_minigame( + video_urls: List[str] = Field(..., description="待处理的小游戏视频 URL 列表，本期仅支持单视频输入\n子项说明：视频 URL，支持 http:// 或 https:// 格式"), + mode: Optional[str] = Field('HighlightExtract', description="高光提取模式，本期支持 HighlightExtract"), + enable_generate_video: Optional[bool] = Field(True, description="是否生成混剪成片视频。true（默认）= 同时输出混剪视频（Edit.Mode=HighlightClips）与高光片段信息；false = 仅输出高光片段信息（clips），底层请求不携带 Edit 字段，也不会生成任何混剪视频。"), + minigame_info: Optional[Dict[str, Any]] = Field(None, description="小游戏描述信息，建议填写以辅助模型更精准识别高光内容"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """识别小游戏录屏视频中的核心玩法与高光事件（如连击、通关、极限操作等）， +快速生成用于买量的视频素材。支持提供游戏名称、玩法描述、高光定义以辅助模型更精准识别。 +使用限制：本期仅支持单视频输入。""" + try: + result = client.call(api_name="generate_highlights_minigame", video_urls=video_urls, mode=mode, enable_generate_video=enable_generate_video, minigame_info=minigame_info, callback_args=callback_args, client_token=client_token) + return async_task_response(result) + except Exception as exc: + return error_response(str(exc)) + + @mcp.tool(name="matte_greenscreen_video", description="对以绿幕或纯色为背景的视频进行抠图，自动识别主体（人物、物品、动物等），同时移除背景，生成背景透明的视频。\n输出视频格式为 WEBM（默认）或 MOV，分辨率与原片对齐。\n支持的格式：主流视频格式如 mp4、flv、ts、avi、mov、mkv、wmv。 使用 task_id, 调用 query_task 方法获取结果") + async def matte_greenscreen_video( + video_url: str = Field(..., description="输入视频 Url。支持 http://xxx 或 https://xxx 格式。"), + format: Optional[str] = Field('WEBM', description="输出视频格式：MOV / WEBM（默认）"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """对以绿幕或纯色为背景的视频进行抠图，自动识别主体（人物、物品、动物等），同时移除背景，生成背景透明的视频。 +输出视频格式为 WEBM（默认）或 MOV，分辨率与原片对齐。 +支持的格式：主流视频格式如 mp4、flv、ts、avi、mov、mkv、wmv。""" + try: + result = client.call(api_name="matte_greenscreen_video", video_url=video_url, format=format, callback_args=callback_args, client_token=client_token) + return async_task_response(result) + except Exception as exc: + return error_response(str(exc)) + + @mcp.tool(name="matte_portrait_video", description="自动识别人物主体，同时移除背景，生成背景透明的视频，适用于背景替换等场景。\n输出格式为 WEBM（默认）或 MOV，分辨率与原片对齐。\n支持的格式：主流视频格式如 mp4、flv、ts、avi、mov、mkv、wmv。 使用 task_id, 调用 query_task 方法获取结果") + async def matte_portrait_video( + video_url: str = Field(..., description="输入视频 Url。支持 http://xxx 或 https://xxx 格式。"), + format: Optional[str] = Field('WEBM', description="输出视频格式：MOV / WEBM（默认）"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """自动识别人物主体，同时移除背景，生成背景透明的视频，适用于背景替换等场景。 +输出格式为 WEBM（默认）或 MOV，分辨率与原片对齐。 +支持的格式：主流视频格式如 mp4、flv、ts、avi、mov、mkv、wmv。""" + try: + result = client.call(api_name="matte_portrait_video", video_url=video_url, format=format, callback_args=callback_args, client_token=client_token) + return async_task_response(result) + except Exception as exc: + return error_response(str(exc)) + + @mcp.tool(name="probe_video_metadata", description="对输入视频 URL 进行探测，输出标准化媒资元信息，覆盖容器层（format_meta）、视频流层（video_stream_meta）与音频流层（audio_stream_meta）。\n字段分类参考 ffprobe，并对 VOD 原始返回做精简与统一，便于上层做分辨率/帧率/码率/编码等策略判断。\n使用限制：仅支持公网 HTTP/HTTPS URL；输入视频分辨率最高支持 4K。 使用 task_id, 调用 query_task 方法获取结果") + async def probe_video_metadata( + video_url: str = Field(..., description="待探测的视频公网 HTTP/HTTPS URL。"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """对输入视频 URL 进行探测，输出标准化媒资元信息，覆盖容器层（format_meta）、视频流层（video_stream_meta）与音频流层（audio_stream_meta）。 +字段分类参考 ffprobe，并对 VOD 原始返回做精简与统一，便于上层做分辨率/帧率/码率/编码等策略判断。 +使用限制：仅支持公网 HTTP/HTTPS URL；输入视频分辨率最高支持 4K。""" + try: + result = client.call(api_name="probe_video_metadata", video_url=video_url, callback_args=callback_args, client_token=client_token) + return async_task_response(result) + except Exception as exc: + return error_response(str(exc)) + + @mcp.tool(name="segment_scenes", description="依据视频转场与画面变化自动切分场景，输出切片时间轴和（可选）切片文件。\n支持格式：MP4、FLV、ASF、RM、RMVB、MPEG、MOV、AVI、MPEGTS、M4S、WMV、3GP、TS、MPG、WEBM、MKV、WM、MPE、VOB、DAT、MP4V、M4V、F4V、MXF、QT 等主流视频格式。\n使用限制：单个视频时长不超过 2 小时。 使用 task_id, 调用 query_task 方法获取结果") + async def segment_scenes( + video_url: str = Field(..., description="待处理视频 Url，必须是公网可直接访问的 HTTP/HTTPS 链接"), + enable_clip_fade: Optional[bool] = Field(False, description="是否将检测到的淡入/淡出片段作为独立切片输出"), + segment_threshold: Optional[float] = Field(None, description="场景切分敏感度阈值，范围 [0, 100)，100 不可取。数值越低切得越细，参考经验值10"), + min_duration: Optional[float] = Field(None, description="单个切片最小时长（秒），参考经验值3，应小于等于max_duration"), + max_duration: Optional[float] = Field(None, description="单个切片最大时长（秒），参考经验值30，应大于等于min_duration"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """依据视频转场与画面变化自动切分场景，输出切片时间轴和（可选）切片文件。 +支持格式：MP4、FLV、ASF、RM、RMVB、MPEG、MOV、AVI、MPEGTS、M4S、WMV、3GP、TS、MPG、WEBM、MKV、WM、MPE、VOB、DAT、MP4V、M4V、F4V、MXF、QT 等主流视频格式。 +使用限制：单个视频时长不超过 2 小时。""" + try: + result = client.call(api_name="segment_scenes", video_url=video_url, enable_clip_fade=enable_clip_fade, segment_threshold=segment_threshold, min_duration=min_duration, max_duration=max_duration, callback_args=callback_args, client_token=client_token) + return async_task_response(result) + except Exception as exc: + return error_response(str(exc)) + + @mcp.tool(name="video_ocr", description="识别视频画面中的字幕/文字内容，输出带时间戳的字幕片段。\n支持格式：主流视频格式如 mp4、flv、ts、avi、mov、wmv、mkv。 使用 task_id, 调用 query_task 方法获取结果") + async def video_ocr( + video_url: str = Field(..., description="输入视频 Url（需公网可访问）"), + mode: Optional[str] = Field('Subtitle', description="工作模式（Subtitle: 识别字幕文本；Detailed: 识别更详细文本信息）"), + callback_args: Optional[str] = Field(None, description="可选，回调参数"), + client_token: Optional[str] = Field(None, description="可选，用于幂等，默认幂等，用户可根据需求进行调整"), + *, + ctx: Context, + ) -> dict: + """识别视频画面中的字幕/文字内容，输出带时间戳的字幕片段。 +支持格式：主流视频格式如 mp4、flv、ts、avi、mov、wmv、mkv。""" + try: + result = client.call(api_name="video_ocr", video_url=video_url, mode=mode, callback_args=callback_args, client_token=client_token) + return async_task_response(result) + except Exception as exc: + return error_response(str(exc)) + if hasattr(mcp, "register_domain_tools"): mcp.register_domain_tools("video", TOOL_NAMES) diff --git a/server/mcp_server_mediakit/src/mediakit/utils/response.py b/server/mcp_server_mediakit/src/mediakit/utils/response.py index 19f34875..4aa95bae 100644 --- a/server/mcp_server_mediakit/src/mediakit/utils/response.py +++ b/server/mcp_server_mediakit/src/mediakit/utils/response.py @@ -63,6 +63,43 @@ def query_task_response(result: dict[str, Any]) -> dict[str, Any]: return output +def sync_result_response(result: dict[str, Any]) -> dict[str, Any]: + """同步处理任务 — API 网关返回 {success, task_id, request_id, result: {...}}。 + + 与 query_task_response 处理方式一致：保留 task_id/request_id/status，将 result 中的字段解构平铺到外层。 + + 业务级失败（HTTP 2xx 但 success=false）会被识别并转为 error 字段输出。 + + Returns: + 正向: { task_id, request_id, status?, ...业务字段 } + 失败: { task_id, request_id, error: "" } + """ + if not isinstance(result, dict): + return {} + + if result.get("success") is False: + return error_response( + result.get("error"), + task_id=result.get("task_id"), + request_id=result.get("request_id"), + ) + + output: dict[str, Any] = {} + task_id = result.get("task_id") + request_id = result.get("request_id") + status = result.get("status") + if task_id is not None: + output["task_id"] = task_id + if request_id is not None: + output["request_id"] = request_id + if status is not None: + output["status"] = status + inner = result.get("result") + if isinstance(inner, dict): + output.update(inner) + return output + + def error_response( error: dict[str, Any] | str | None = None, *,