upload_local_file
Upload local files to Qiniu Cloud Storage buckets for cloud storage and management. Specify bucket, file path, and key to transfer files from your local system to Qiniu's cloud infrastructure.
Instructions
Upload a local file to Qiniu bucket.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| bucket | Yes | Qiniu Cloud Storage bucket Name | |
| key | Yes | 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 | Yes | The file path of file to upload. | |
| overwrite | No | Whether to overwrite the existing object if it already exists. |
Implementation Reference
- MCP tool handler that receives tool call parameters, invokes the storage service's upload_local_file method, and returns the result as TextContent.
def upload_local_file(self, **kwargs) -> list[types.TextContent]: urls = self.storage.upload_local_file(**kwargs) return [types.TextContent(type="text", text=str(urls))] - Input schema defining parameters for the upload_local_file tool: bucket, key, file_path (required), overwrite (optional).
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"], } - src/mcp_server/core/storage/tools.py:236-248 (registration)Registration function that instantiates the tool implementation and registers all storage-related tools, including upload_local_file, using tools.auto_register_tools.
def register_tools(storage: StorageService): tool_impl = _ToolImpl(storage) tools.auto_register_tools( [ 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, ] ) - Core helper function in StorageService that implements the file upload logic using Qiniu SDK: generates upload token with policy, uploads file with qiniu.put_file, handles overwrite, and returns download URLs.
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)