add upload

Author: YangSen-qn

src/mcp_server/core/storage/storage.py

@@ -237,6 +237,13 @@ def upload_file(self, bucket: str, key: str, file_path: str, overwrite: bool = F
 
         return self.get_object_url(bucket, key)
 
+    def fetch_object(self, bucket: str, key: str, url: str):
+        ret, info = self.bucket_manager.fetch(url, bucket, key=key)
+        if info.status_code != 200:
+            raise Exception(f"Failed to fetch object: {info.text}")
+
+        return self.get_object_url(bucket, key)
+
     def is_text_file(self, key: str) -> bool:
         """Determine if a file is text-based by its extension"""
         text_extensions = {

src/mcp_server/core/storage/tools.py

@@ -112,7 +112,7 @@ async def get_object(self, **kwargs) -> list[ImageContent] | list[TextContent]:
     @tools.tool_meta(
         types.Tool(
             name="UploadTextData",
-            description="Upload text data to Qiniu bucket. In the UploadData request, specify the full key name for the object. Path-style requests are not supported.",
+            description="Upload text data to Qiniu bucket.",
             inputSchema={
                 "type": "object",
                 "properties": {
@@ -144,7 +144,7 @@ def upload_text_data(self, **kwargs) -> list[types.TextContent]:
     @tools.tool_meta(
         types.Tool(
             name="UploadFile",
-            description="Upload file to Qiniu bucket. In the UploadFile request, specify the full key name for the object. Path-style requests are not supported.",
+            description="Upload a local file to Qiniu bucket.",
             inputSchema={
                 "type": "object",
                 "properties": {
@@ -173,6 +173,28 @@ def upload_file(self, **kwargs) -> list[types.TextContent]:
         urls = self.storage.upload_file(**kwargs)
         return [types.TextContent(type="text", text=str(urls))]
 
+    @tools.tool_meta(
+        types.Tool(
+            name="FetchObject",
+            description="Fetch a http object to Qiniu bucket.",
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "bucket": {
+                        "type": "string",
+                        "description": _BUCKET_DESC,
+                    },
+                    "key": {
+                        "type": "string",
+                    }
+                }
+            }
+        )
+    )
+    def fetch_object(self, **kwargs) -> list[types.TextContent]:
+        urls = self.storage.fetch_object(**kwargs)
+        return [types.TextContent(type="text", text=str(urls))]
+
     @tools.tool_meta(
         types.Tool(
             name="GetObjectURL",
@@ -215,6 +237,7 @@ def register_tools(storage: StorageService):
             tool_impl.get_object,
             tool_impl.upload_text_data,
             tool_impl.upload_file,
+            tool_impl.fetch_object,
             tool_impl.get_object_url,
         ]
     )