get_object_url
Generate download URLs for files stored in Qiniu Cloud Storage buckets, with options for SSL configuration and token expiration for private buckets.
Instructions
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.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| bucket | Yes | Qiniu Cloud Storage bucket Name | |
| key | Yes | Key of the object to get. | |
| disable_ssl | No | Whether to disable SSL. By default, it is not disabled (HTTP protocol is used). If disabled, the HTTP protocol will be used. | |
| expires | No | Token expiration time (in seconds) for download links. When the bucket is private, a signed Token is required to access file objects. Public buckets do not require Token signing. |
Implementation Reference
- MCP tool handler implementation for 'get_object_url'. It delegates to StorageService.get_object_url and formats the response as TextContent.def get_object_url(self, **kwargs) -> list[types.TextContent]: urls = self.storage.get_object_url(**kwargs) return [types.TextContent(type="text", text=str(urls))]
- Tool schema definition including name, description, and inputSchema for 'get_object_url'.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", "properties": { "bucket": { "type": "string", "description": _BUCKET_DESC, }, "key": { "type": "string", "description": "Key of the object to get.", }, "disable_ssl": { "type": "boolean", "description": "Whether to disable SSL. By default, it is not disabled (HTTP protocol is used). If disabled, the HTTP protocol will be used.", }, "expires": { "type": "integer", "description": "Token expiration time (in seconds) for download links. When the bucket is private, a signed Token is required to access file objects. Public buckets do not require Token signing.", }, }, "required": ["bucket", "key"], }, )
- src/mcp_server/core/storage/tools.py:236-247 (registration)Registration of the 'get_object_url' tool handler 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 generates public and private download URLs for objects using Qiniu SDK.def get_object_url( self, bucket: str, key: str, disable_ssl: bool = False, expires: int = 3600 ) -> list[dict[str:Any]]: # 获取下载域名 domains_getter = getattr(self.bucket_manager, "_BucketManager__uc_do_with_retrier") domains_list, domain_response = domains_getter('/v3/domains?tbl={0}'.format(bucket)) if domain_response.status_code != 200: raise Exception( f"get bucket domain error：{domain_response.exception} reqId:{domain_response.req_id}" ) if not domains_list or len(domains_list) == 0: raise Exception( f"get bucket domain error：domains_list is empty reqId:{domain_response.req_id}" ) http_schema = "https" if not disable_ssl else "http" object_public_urls = [] for domain in domains_list: # 被冻结 freeze_types = domain.get("freeze_types") if freeze_types is not None: continue domain_url = domain.get("domain") if domain_url is None: continue object_public_urls.append({ "object_url": f"{http_schema}://{domain_url}/{key}", "domain_type": "cdn" if domain.get("domaintype") is None or domain.get("domaintype") == 0 else "origin" }) object_urls = [] bucket_info, bucket_info_response = self.bucket_manager.bucket_info(bucket) if domain_response.status_code != 200: raise Exception( f"get bucket domain error：{bucket_info_response.exception} reqId:{bucket_info_response.req_id}" ) if bucket_info["private"] != 0: for url_info in object_public_urls: public_url = url_info.get("object_url") if public_url is None: continue url_info["object_url"] = self.auth.private_download_url(public_url, expires=expires) object_urls.append(url_info) else: for url_info in object_public_urls: object_urls.append(url_info) return object_urls
- src/mcp_server/core/storage/__init__.py:7-10 (registration)Higher-level registration trigger: storage/__init__.py load() calls register_tools to register the tools including get_object_url.def load(cfg: config.Config): storage = StorageService(cfg) register_tools(storage) register_resource_provider(storage)