TOS MCP Server

Instructions

Implementation Reference

• src/tos_mcp_server/handlers.py:102-141 (handler)

  The main handler function that executes the tool logic: retrieves the object from TOS bucket using tos_client.get_object, reads content, encodes as base64 or UTF-8 text based on parameters, and returns structured JSON result.

  async def get_object(args: Dict[str, Any]) -> List[TextContent]:
      """下载对象"""
      bucket_name = args["bucket_name"]
      object_key = args["object_key"]
      return_as_base64 = args.get("return_as_base64", False)
      
      try:
          resp = tos_client.get_object(bucket_name, object_key)
          content = resp.read()
          
          if return_as_base64:
              content_str = base64.b64encode(content).decode('utf-8')
              result = {
                  "content": content_str,
                  "content_type": resp.content_type,
                  "content_length": resp.content_length,
                  "encoding": "base64"
              }
          else:
              try:
                  content_str = content.decode('utf-8')
                  result = {
                      "content": content_str,
                      "content_type": resp.content_type,
                      "content_length": resp.content_length,
                      "encoding": "utf-8"
                  }
              except UnicodeDecodeError:
                  content_str = base64.b64encode(content).decode('utf-8')
                  result = {
                      "content": content_str,
                      "content_type": resp.content_type,
                      "content_length": resp.content_length,
                      "encoding": "base64"
                  }
          
          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:126-144 (schema)

  Input schema definition for the tos_get_object tool, specifying parameters: bucket_name (required), object_key (required), return_as_base64 (optional boolean).

  inputSchema={
      "type": "object",
      "properties": {
          "bucket_name": {
              "type": "string",
              "description": "存储桶名称"
          },
          "object_key": {
              "type": "string",
              "description": "对象键名"
          },
          "return_as_base64": {
              "type": "boolean",
              "description": "是否以base64格式返回内容",
              "default": False
          }
      },
      "required": ["bucket_name", "object_key"]
  }

• src/tos_mcp_server/server.py:123-145 (registration)

  Tool registration in list_tools(): defines name, description, and input schema for tos_get_object.

  Tool(
      name="tos_get_object",
      description="从 TOS 下载对象",
      inputSchema={
          "type": "object",
          "properties": {
              "bucket_name": {
                  "type": "string",
                  "description": "存储桶名称"
              },
              "object_key": {
                  "type": "string",
                  "description": "对象键名"
              },
              "return_as_base64": {
                  "type": "boolean",
                  "description": "是否以base64格式返回内容",
                  "default": False
              }
          },
          "required": ["bucket_name", "object_key"]
      }
  ),

• src/tos_mcp_server/server.py:347-348 (registration)

  Dispatch registration in call_tool(): maps tool name 'tos_get_object' to the get_object handler function.

  elif name == "tos_get_object":
      return await get_object(arguments)