diff --git a/server/mcp_server_mediakit/README.md b/server/mcp_server_mediakit/README.md index d41f4480..21a918f4 100644 --- a/server/mcp_server_mediakit/README.md +++ b/server/mcp_server_mediakit/README.md @@ -2,6 +2,14 @@ MediaKit MCP Server is a standard AI capability plugin for Volcano Engine AI MediaKit. It is built on the MCP (Model Context Protocol) protocol and exposes cloud media capabilities such as video editing, audio processing, subtitle processing, and video enhancement as tools that can be called by AI agents. With MediaKit MCP, developers can use natural language to drive intelligent media production workflows. +<<<<<<< HEAD +| Field | Value | +| ----------- | ----------------------------------------------------------------- | +| Version | v1.0.0 | +| Description | MediaKit MCP intelligent media assistant | +| Categories | Media cloud, audio/video editing, video enhancement | +| Tags | MCP, MediaKit, video editing, audio processing, video enhancement | +======= | Field | Value | | --- | --- | | Version | v1.0.0 | @@ -9,6 +17,8 @@ MediaKit MCP Server is a standard AI capability plugin for Volcano Engine AI Med | Categories | Media cloud, audio/video editing, video enhancement | | Tags | MCP, MediaKit, video editing, audio processing, video enhancement | +> > > > > > > 62a9e293bc254026e04eaec63ca9d229aaa8e89e + ## Tool Overview MediaKit MCP provides tools that cover the full workflow from asynchronous task query to deep media editing and video enhancement. All tools support dynamic loading by group or by tool name to optimize agent reasoning efficiency. @@ -133,11 +143,19 @@ Trae is an AI-native IDE with strong agent collaboration capabilities. By connec Choose one of the following modes based on your usage scenario: +<<<<<<< HEAD +| Mode | Best for | Access method | +| --------------------------- | ------------------------------------------------------------------ | ---------------------------------------------------------------------------------------------- | +| **Local Mode (JSON Local)** | Personal debugging, quick trials, no self-hosted service required. | Use `uvx` to launch MediaKit MCP directly from the `mcp-server` repository subdirectory. | +| **Cloud Mode (JSON URL)** | Team sharing, long-term usage, centralized operations. | Deploy MediaKit MCP Server yourself, then connect using the deployed Streamable HTTP endpoint. | +======= | Mode | Best for | Access method | | --- | --- | --- | | **Local Mode (JSON Local)** | Personal debugging, quick trials, no self-hosted service required. | Use `uvx` to launch MediaKit MCP directly from the `mcp-server` repository subdirectory. | | **Cloud Mode (JSON URL)** | Team sharing, long-term usage, centralized operations. | Deploy MediaKit MCP Server yourself, then connect using the deployed Streamable HTTP endpoint. | +> > > > > > > 62a9e293bc254026e04eaec63ca9d229aaa8e89e + ### Step 2: Add MCP Configuration 1. Open Trae and click the settings button in the top-right corner. @@ -317,12 +335,21 @@ The table below lists the core MediaKit MCP configuration fields for cloud mode For self-hosted cloud mode, you can also configure the following startup parameters: +<<<<<<< HEAD +| Environment variable | Default value | Description | +| ---------------------- | ------------- | --------------------------- | +| `MCP_SERVER_HOST` | `0.0.0.0` | MCP service bind address. | +| `MCP_SERVER_PORT` | `8000` | MCP service listening port. | +| `STREAMABLE_HTTP_PATH` | `/mcp` | Streamable HTTP path. | +======= | Environment variable | Default value | Description | | --- | --- | --- | | `MCP_SERVER_HOST` | `0.0.0.0` | MCP service bind address. | | `MCP_SERVER_PORT` | `8000` | MCP service listening port. | | `STREAMABLE_HTTP_PATH` | `/mcp` | Streamable HTTP path. | +> > > > > > > 62a9e293bc254026e04eaec63ca9d229aaa8e89e + ## Tool Details ### query_task @@ -384,3 +411,9 @@ Enhance video quality for scenarios such as `common`, `ugc`, `short_series`, `ai ## License MIT + +This software calls MediaKit APIs at runtime. Use of these APIs is subject to the following terms and privacy policies: + +- [Video Cloud Service Special Terms](https://www.volcengine.com/docs/6448/79646?lang=zh) +- [Intelligent Processing Service Billing Rules](https://www.volcengine.com/docs/6448/104992?lang=zh) +- [Intelligent Processing Service Level Agreement](https://www.volcengine.com/docs/6448/79648?lang=zh) diff --git a/server/mcp_server_mediakit/README_zh.md b/server/mcp_server_mediakit/README_zh.md index af723844..62cd3466 100644 --- a/server/mcp_server_mediakit/README_zh.md +++ b/server/mcp_server_mediakit/README_zh.md @@ -2,12 +2,21 @@ MediaKit MCP 是火山引擎 AI MediaKit 面向 AI 时代推出的标准能力插件。它基于 MCP（Model Context Protocol）协议，将云端专业的视频剪辑、音频处理、字幕处理、画质增强等原子能力封装为智能体可直观调用的工具。通过 MediaKit MCP，开发者可直接以自然语言驱动 AI 智能体完成复杂的云端媒体处理任务。 +<<<<<<< HEAD +| 字段 | 取值 | +| ---- | ------------------------------------------- | +| 版本 | v1.0.0 | +| 描述 | MediaKit MCP 智能媒体助手 | +| 分类 | 视频云、音视频编辑、画质增强 | +======= | 字段 | 取值 | | --- | --- | | 版本 | v1.0.0 | | 描述 | MediaKit MCP 智能媒体助手 | | 分类 | 视频云、音视频编辑、画质增强 | -| 标签 | MCP、MediaKit、视频剪辑、音频处理、画质增强 | + +> > > > > > > 62a9e293bc254026e04eaec63ca9d229aaa8e89e +> > > > > > > | 标签 | MCP、MediaKit、视频剪辑、音频处理、画质增强 | ## 工具概览 @@ -133,11 +142,19 @@ Trae 是一款 AI 原生 IDE，提供了强大的智能体协作能力。通过 根据您的使用场景，选择以下两种接入模式之一： +<<<<<<< HEAD +| 模式 | 适用场景 | 接入方式 | +| -------------------------- | ---------------------------------- | ------------------------------------------------------------------------- | +| **本地模式（JSON Local）** | 个人调试、快速试用、无需自建服务。 | 通过 `uvx` 直接从 `mcp-server` 仓库子目录拉起 MediaKit MCP。 | +| **云端模式（JSON URL）** | 团队共享、长期稳定使用、统一运维。 | 先自行部署 MediaKit MCP Server，再使用部署后的 Streamable HTTP 地址接入。 | +======= | 模式 | 适用场景 | 接入方式 | | --- | --- | --- | | **本地模式（JSON Local）** | 个人调试、快速试用、无需自建服务。 | 通过 `uvx` 直接从 `mcp-server` 仓库子目录拉起 MediaKit MCP。 | | **云端模式（JSON URL）** | 团队共享、长期稳定使用、统一运维。 | 先自行部署 MediaKit MCP Server，再使用部署后的 Streamable HTTP 地址接入。 | +> > > > > > > 62a9e293bc254026e04eaec63ca9d229aaa8e89e + ### 步骤 2：添加 MCP 配置 1. 打开 Trae，单击窗口右上角“设置”按钮。 @@ -317,12 +334,21 @@ uvx --from "git+https://github.com/volcengine/mcp-server.git#subdirectory=server 云端自部署时，还可按需使用以下服务启动参数： +<<<<<<< HEAD +| 环境变量 | 默认值 | 说明 | +| ---------------------- | --------- | ---------------------- | +| `MCP_SERVER_HOST` | `0.0.0.0` | MCP 服务监听地址。 | +| `MCP_SERVER_PORT` | `8000` | MCP 服务监听端口。 | +| `STREAMABLE_HTTP_PATH` | `/mcp` | Streamable HTTP 路径。 | +======= | 环境变量 | 默认值 | 说明 | | --- | --- | --- | | `MCP_SERVER_HOST` | `0.0.0.0` | MCP 服务监听地址。 | | `MCP_SERVER_PORT` | `8000` | MCP 服务监听端口。 | | `STREAMABLE_HTTP_PATH` | `/mcp` | Streamable HTTP 路径。 | +> > > > > > > 62a9e293bc254026e04eaec63ca9d229aaa8e89e + ## 工具详情 ### query_task @@ -384,3 +410,9 @@ uvx --from "git+https://github.com/volcengine/mcp-server.git#subdirectory=server ## License MIT + +该软件运行时会调用 MediaKit 的 API，使用这些 API 需要遵守如下协议和隐私政策： + +- [视频云服务专用条款](https://www.volcengine.com/docs/6448/79646?lang=zh) +- [智能处理服务计费结算规则](https://www.volcengine.com/docs/6448/104992?lang=zh) +- [智能处理服务等级协议](https://www.volcengine.com/docs/6448/79648?lang=zh) 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, *,