TOS MCP Server

Instructions

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)