tos_video_snapshot
Extract frames from videos stored in TOS at specific timestamps and save them as images in designated buckets for analysis or processing.
Instructions
视频截帧（支持持久化）
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| bucket_name | Yes | 存储桶名称 | |
| format | No | 输出格式 | jpg |
| object_key | Yes | 视频对象键名 | |
| save_bucket | Yes | 保存截帧图片的存储桶名称 | |
| save_key | Yes | 保存截帧图片的对象键名 | |
| time | No | 截帧时间点（毫秒），如300表示第300毫秒 |
Implementation Reference
- src/tos_mcp_server/handlers.py:297-344 (handler)The core handler function `video_snapshot` that implements the `tos_video_snapshot` tool logic. It constructs a video processing parameter, performs the snapshot using TOS SDK's get_object with process and save parameters, waits for completion, generates a presigned URL for the resulting image, and returns the result.async def video_snapshot(args: Dict[str, Any]) -> List[TextContent]: """视频截帧（支持持久化）""" bucket_name = args["bucket_name"] object_key = args["object_key"] time = args.get("time", 300) format = args.get("format", "jpg") save_bucket = args["save_bucket"] save_key = args["save_key"] try: # 构建视频截帧处理参数，时间单位为毫秒 process = f"video/snapshot,t_{int(time)},f_{format}" # 使用官方SDK写法，通过save_bucket和save_object参数执行视频截帧和持久化 resp = tos_client.get_object( bucket=bucket_name, key=object_key, process=process, save_bucket=base64.b64encode(save_bucket.encode("utf-8")).decode("utf-8"), save_object=base64.b64encode(save_key.encode("utf-8")).decode("utf-8") ) # 读取处理结果以确保截帧完成 processed_data = resp.read() # 等待一下确保回写完成 import time as time_module time_module.sleep(1.0) # 生成截帧图片的预签名 URL download_url = tos_client.pre_signed_url(tos.HttpMethodType.Http_Method_Get, save_bucket, save_key, 3600) result = { "presigned_url": download_url.signed_url, "source_bucket": bucket_name, "source_key": object_key, "save_bucket": save_bucket, "save_key": save_key, "time": time, "format": format, "processed_size": len(processed_data), "expires_in": 3600, "status": "processed" } return [TextContent(type="text", text=json.dumps(result, indent=2, ensure_ascii=False))] except Exception as e: return [TextContent(type="text", text=f"视频截帧失败: {str(e)}")]
- src/tos_mcp_server/server.py:276-312 (schema)The input schema definition for the `tos_video_snapshot` tool, registered in the `list_tools()` handler. Defines parameters like bucket_name, object_key, time, format, save_bucket, save_key.Tool( name="tos_video_snapshot", description="视频截帧（支持持久化）", inputSchema={ "type": "object", "properties": { "bucket_name": { "type": "string", "description": "存储桶名称" }, "object_key": { "type": "string", "description": "视频对象键名" }, "time": { "type": "number", "description": "截帧时间点（毫秒），如300表示第300毫秒", "default": 300 }, "format": { "type": "string", "description": "输出格式", "enum": ["jpg", "png"], "default": "jpg" }, "save_bucket": { "type": "string", "description": "保存截帧图片的存储桶名称" }, "save_key": { "type": "string", "description": "保存截帧图片的对象键名" } }, "required": ["bucket_name", "object_key", "save_bucket", "save_key"] } ),
- src/tos_mcp_server/server.py:359-360 (registration)Tool dispatch/registration in the `call_tool()` function, which routes calls to `tos_video_snapshot` to the `video_snapshot` handler.elif name == "tos_video_snapshot": return await video_snapshot(arguments)