upload_text_data
Store text content in Qiniu Cloud Storage buckets by specifying bucket name, file key, and text data, with optional overwrite control.
Instructions
Upload text data 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. | |
| data | Yes | The data to upload. | |
| overwrite | No | Whether to overwrite the existing object if it already exists. |
Implementation Reference
- The MCP tool handler function for 'upload_text_data' that invokes the storage service method and formats the response as TextContent.
def upload_text_data(self, **kwargs) -> list[types.TextContent]: urls = self.storage.upload_text_data(**kwargs) return [types.TextContent(type="text", text=str(urls))] - Input schema defining parameters for the upload_text_data tool: bucket, key, data (required), and optional overwrite.
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"], } - src/mcp_server/core/storage/tools.py:236-248 (registration)Tool registration function that registers upload_text_data among other storage tools 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 implementation in StorageService that handles the actual upload to Qiniu using upload_token and put_data, then returns object URLs.
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)