upload_local_file
Easily transfer local files to Qiniu Cloud Storage by specifying the bucket, unique key, file path, and overwrite option.
Instructions
Upload a local file to Qiniu bucket.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| bucket | Yes | Qiniu Cloud Storage bucket Name | |
| file_path | Yes | The file path of file to upload. | |
| 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. | |
| overwrite | No | Whether to overwrite the existing object if it already exists. |
Implementation Reference
- Tool metadata decorator defining the input schema, description, and registration for the upload_local_file MCP tool.@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"], } ) )
- The handler function for the MCP tool 'upload_local_file', which calls the storage service and returns formatted TextContent with upload URLs.def upload_local_file(self, **kwargs) -> list[types.TextContent]: urls = self.storage.upload_local_file(**kwargs) return [types.TextContent(type="text", text=str(urls))]
- Core implementation of the file upload logic using Qiniu SDK, generating upload token and performing the upload.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)
- src/mcp_server/core/storage/tools.py:236-248 (registration)Function that creates a tool instance and registers the upload_local_file tool (and others) with the MCP tools system.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, ] )