support upload

CHANGELOG.md

@@ -1,3 +1,6 @@
+# v1.2.0
+- 支持文件上传至七牛 Bucket
+
 # v1.1.1
 - 支持 AK、SK 为空字符串
 

README.md

@@ -7,6 +7,20 @@ Server 来访问七牛云存储、智能多媒体服务等。
 
 关于访问七牛云存储详细情况请参考 [基于 MCP 使用大模型访问七牛云存储](https://developer.qiniu.com/kodo/12914/mcp-aimodel-kodo)。
 
+能力集：
+- 存储
+  - 获取 Bucket 列表
+  - 获取 Bucket 中的文件列表
+  - 上传本地文件，以及给出文件内容进行上传
+  - 读取文件内容
+  - 获取文件下载链接
+- 智能多媒体
+  - 图片缩放
+  - 图片切圆角
+- CDN
+  - 根据链接刷新文件
+  - 根据链接预取文件
+
 ## 环境要求
 
 - Python 3.12 或更高版本

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "qiniu-mcp-server"
-version = "1.1.1"
+version = "1.2.0"
 description = "A MCP server project of Qiniu."
 requires-python = ">=3.12"
 authors = [

src/mcp_server/core/cdn/tools.py

@@ -29,7 +29,7 @@ def __init__(self, cdn: CDNService):
 
     @tools.tool_meta(
         types.Tool(
-            name="CDNPrefetchUrls",
+            name="cdn_prefetch_urls",
             description="Newly added resources are proactively retrieved by the CDN and stored on its cache nodes in advance. Users simply submit the resource URLs, and the CDN automatically triggers the prefetch process.",
             inputSchema={
                 "type": "object",
@@ -76,7 +76,7 @@ def prefetch_urls(self, **kwargs) -> list[types.TextContent]:
 
     @tools.tool_meta(
         types.Tool(
-            name="CDNRefresh",
+            name="cdn_refresh",
             description="This function marks resources cached on CDN nodes as expired. When users access these resources again, the CDN nodes will fetch the latest version from the origin server and store them anew.",
             inputSchema={
                 "type": "object",

src/mcp_server/core/media_processing/tools.py

@@ -17,7 +17,7 @@ def __init__(self, cli: MediaProcessingService):
 
     @tools.tool_meta(
         types.Tool(
-            name="ImageScaleByPercent",
+            name="image_scale_by_percent",
             description="""Image scaling tool that resizes images based on a percentage and returns information about the scaled image.
             The information includes the object_url of the scaled image, which users can directly use for HTTP GET requests to retrieve the image content or open in a browser to view the file.
             The image must be stored in a Qiniu Cloud Bucket.
@@ -70,7 +70,7 @@ def image_scale_by_percent(
 
     @tools.tool_meta(
         types.Tool(
-            name="ImageScaleBySize",
+            name="image_scale_by_size",
             description="""Image scaling tool that resizes images based on a specified width or height and returns information about the scaled image.
             The information includes the object_url of the scaled image, which users can directly use for HTTP GET requests to retrieve the image content or open in a browser to view the file.
             The image must be stored in a Qiniu Cloud Bucket.
@@ -134,7 +134,7 @@ def image_scale_by_size(
 
     @tools.tool_meta(
         types.Tool(
-            name="ImageRoundCorner",
+            name="image_round_corner",
             description="""Image rounded corner tool that processes images based on width, height, and corner radius, returning information about the processed image.
             If only radius_x or radius_y is set, the other parameter will be assigned the same value, meaning horizontal and vertical parameters will be identical.
             The information includes the object_url of the processed image, which users can directly use for HTTP GET requests to retrieve the image content or open in a browser to view the file.
@@ -203,7 +203,7 @@ def image_round_corner(self, **kwargs) -> list[types.TextContent]:
 
     @tools.tool_meta(
         types.Tool(
-            name="ImageInfo",
+            name="image_info",
             description="Retrieves basic image information, including image format, size, and color model.",
             inputSchema={
                 "type": "object",
@@ -240,7 +240,7 @@ def image_info(self, **kwargs) -> list[types.TextContent]:
 
     @tools.tool_meta(
         types.Tool(
-            name="GetFopStatus",
+            name="get_fop_status",
             description="Retrieves the execution status of a Fop operation.",
             inputSchema={
                 "type": "object",

src/mcp_server/core/storage/resource.py

@@ -5,9 +5,12 @@
 from mcp import types
 from urllib.parse import unquote
 
+from mcp.server.lowlevel.helper_types import ReadResourceContents
+
 from .storage import StorageService
 from ...consts import consts
 from ...resource import resource
+from ...resource.resource import ResourceContents
 
 logger = logging.getLogger(consts.LOGGER_NAME)
 
@@ -88,7 +91,7 @@ async def process_bucket_with_semaphore(bucket):
         logger.info(f"Returning {len(resources)} resources")
         return resources
 
-    async def read_resource(self, uri: types.AnyUrl, **kwargs) -> str:
+    async def read_resource(self, uri: types.AnyUrl, **kwargs) -> ResourceContents:
         """
         Read content from an S3 resource and return structured response
 
@@ -120,7 +123,7 @@ async def read_resource(self, uri: types.AnyUrl, **kwargs) -> str:
         if content_type.startswith("image/"):
             file_content = base64.b64encode(file_content).decode("utf-8")
 
-        return file_content
+        return [ReadResourceContents(mime_type=content_type, content=file_content)]
 
 
 def register_resource_provider(storage: StorageService):

src/mcp_server/core/storage/storage.py

@@ -160,6 +160,44 @@ async def get_object(self, bucket: str, key: str) -> Dict[str, Any]:
             response["Body"] = b"".join(chunks)
             return response
 
+    def upload_text_data(self, bucket: str, key: str, data: str, overwrite: bool = False) -> list[dict[str:Any]]:
+        policy = {
+            "insertOnly": 1,
+        }
+
+        if overwrite:
+            policy["insertOnly"] = 0
+            policy["scope"] = f"{bucket}:{key}"
+
+        token = self.auth.upload_token(bucket=bucket, key=key, policy=policy)
+        ret, info = qiniu.put_data(up_token=token, key=key, data=bytes(data, encoding="utf-8"))
+        if info.status_code != 200:
+            raise Exception(f"Failed to upload object: {info}")
+
+        return self.get_object_url(bucket, key)
+
+    def upload_local_file(self, bucket: str, key: str, file_path: str, overwrite: bool = False) -> list[dict[str:Any]]:
+        policy = {
+            "insertOnly": 1,
+        }
+
+        if overwrite:
+            policy["insertOnly"] = 0
+            policy["scope"] = f"{bucket}:{key}"
+
+        token = self.auth.upload_token(bucket=bucket, key=key, policy=policy)
+        ret, info = qiniu.put_file(up_token=token, key=key, file_path=file_path)
+        if info.status_code != 200:
+            raise Exception(f"Failed to upload object: {info}")
+
+        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}")
+
+        return self.get_object_url(bucket, key)
 
     def is_text_file(self, key: str) -> bool:
         text_extensions = {

src/mcp_server/core/storage/tools.py

@@ -18,7 +18,7 @@ def __init__(self, storage: StorageService):
 
     @tools.tool_meta(
         types.Tool(
-            name="ListBuckets",
+            name="list_buckets",
             description="Return the Bucket you configured based on the conditions.",
             inputSchema={
                 "type": "object",
@@ -38,7 +38,7 @@ async def list_buckets(self, **kwargs) -> list[types.TextContent]:
 
     @tools.tool_meta(
         types.Tool(
-            name="ListObjects",
+            name="list_objects",
             description="List objects in Qiniu Cloud, list a part each time, you can set start_after to continue listing, when the number of listed objects is less than max_keys, it means that all files are listed. start_after can be the key of the last file in the previous listing.",
             inputSchema={
                 "type": "object",
@@ -70,7 +70,7 @@ async def list_objects(self, **kwargs) -> list[types.TextContent]:
 
     @tools.tool_meta(
         types.Tool(
-            name="GetObject",
+            name="get_object",
             description="Get an object contents from Qiniu Cloud bucket. In the GetObject request, specify the full key name for the object.",
             inputSchema={
                 "type": "object",
@@ -110,7 +110,99 @@ async def get_object(self, **kwargs) -> list[ImageContent] | list[TextContent]:
 
     @tools.tool_meta(
         types.Tool(
-            name="GetObjectURL",
+            name="upload_text_data",
+            description="Upload text data to Qiniu bucket.",
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "bucket": {
+                        "type": "string",
+                        "description": _BUCKET_DESC,
+                    },
+                    "key": {
+                        "type": "string",
+                        "description": "The key under which a file is saved in Qiniu Cloud Storage serves as the unique identifier for the file within that space, typically using the filename.",
+                    },
+                    "data": {
+                        "type": "string",
+                        "description": "The data to upload.",
+                    },
+                    "overwrite": {
+                        "type": "boolean",
+                        "description": "Whether to overwrite the existing object if it already exists.",
+                    },
+                },
+                "required": ["bucket", "key", "data"],
+            }
+        )
+    )
+    def upload_text_data(self, **kwargs) -> list[types.TextContent]:
+        urls = self.storage.upload_text_data(**kwargs)
+        return [types.TextContent(type="text", text=str(urls))]
+
+    @tools.tool_meta(
+        types.Tool(
+            name="upload_local_file",
+            description="Upload a local file to Qiniu bucket.",
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "bucket": {
+                        "type": "string",
+                        "description": _BUCKET_DESC,
+                    },
+                    "key": {
+                        "type": "string",
+                        "description": "The key under which a file is saved in Qiniu Cloud Storage serves as the unique identifier for the file within that space, typically using the filename.",
+                    },
+                    "file_path": {
+                        "type": "string",
+                        "description": "The file path of file to upload.",
+                    },
+                    "overwrite": {
+                        "type": "boolean",
+                        "description": "Whether to overwrite the existing object if it already exists.",
+                    },
+                },
+                "required": ["bucket", "key", "file_path"],
+            }
+        )
+    )
+    def upload_local_file(self, **kwargs) -> list[types.TextContent]:
+        urls = self.storage.upload_local_file(**kwargs)
+        return [types.TextContent(type="text", text=str(urls))]
+
+    @tools.tool_meta(
+        types.Tool(
+            name="fetch_object",
+            description="Fetch a http object to Qiniu bucket.",
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "bucket": {
+                        "type": "string",
+                        "description": _BUCKET_DESC,
+                    },
+                    "key": {
+                        "type": "string",
+                        "description": "The key under which a file is saved in Qiniu Cloud Storage serves as the unique identifier for the file within that space, typically using the filename.",
+                    },
+                    "url": {
+                        "type": "string",
+                        "description": "The URL of the object to fetch.",
+                    },
+                },
+                "required": ["bucket", "key", "url"],
+            }
+        )
+    )
+    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="get_object_url",
             description="Get the file download URL, and note that the Bucket where the file is located must be bound to a domain name. If using Qiniu Cloud test domain, HTTPS access will not be available, and users need to make adjustments for this themselves.",
             inputSchema={
                 "type": "object",
@@ -121,7 +213,7 @@ async def get_object(self, **kwargs) -> list[ImageContent] | list[TextContent]:
                     },
                     "key": {
                         "type": "string",
-                        "description": "Key of the object to get. Length Constraints: Minimum length of 1.",
+                        "description": "Key of the object to get.",
                     },
                     "disable_ssl": {
                         "type": "boolean",
@@ -148,6 +240,8 @@ def register_tools(storage: StorageService):
             tool_impl.list_buckets,
             tool_impl.list_objects,
             tool_impl.get_object,
+            tool_impl.upload_text_data,
+            tool_impl.upload_local_file,
             tool_impl.get_object_url,
         ]
     )

src/mcp_server/core/version/tools.py

@@ -11,7 +11,7 @@ def __init__(self):
 
     @tools.tool_meta(
         types.Tool(
-            name="Version",
+            name="version",
             description="qiniu mcp server version info.",
             inputSchema={
                 "type": "object",

src/mcp_server/core/version/version.py

@@ -1,2 +1,2 @@
 
-VERSION = '1.1.1'
+VERSION = '1.2.0'

src/mcp_server/resource/resource.py

@@ -1,12 +1,15 @@
 import logging
 from abc import abstractmethod
-from typing import Dict, AsyncGenerator
+from typing import Dict, AsyncGenerator, Iterable
 
 from mcp import types
+from mcp.server.lowlevel.helper_types import ReadResourceContents
+
 from ..consts import consts
 
 logger = logging.getLogger(consts.LOGGER_NAME)
 
+ResourceContents = str | bytes | Iterable[ReadResourceContents]
 
 class ResourceProvider:
     def __init__(self, scheme: str):
@@ -17,7 +20,7 @@ async def list_resources(self, **kwargs) -> list[types.Resource]:
         pass
 
     @abstractmethod
-    async def read_resource(self, uri: types.AnyUrl, **kwargs) -> str:
+    async def read_resource(self, uri: types.AnyUrl, **kwargs) -> ResourceContents:
         pass
 
 
@@ -35,7 +38,7 @@ async def list_resources(**kwargs) -> AsyncGenerator[types.Resource, None]:
     return
 
 
-async def read_resource(uri: types.AnyUrl, **kwargs) -> str:
+async def read_resource(uri: types.AnyUrl, **kwargs) -> ResourceContents:
     if len(_all_resource_providers) == 0:
         return ""
 
@@ -52,6 +55,7 @@ def register_resource_provider(provider: ResourceProvider):
 
 
 __all__ = [
+    "ResourceContents",
     "ResourceProvider",
     "list_resources",
     "read_resource",