get_object_url
Generate a file download URL from Qiniu Cloud Storage. Specify the bucket and object key to retrieve the link, with options to disable SSL and set 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 | |
| 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. | |
| key | Yes | Key of the object to get. |
Implementation Reference
- MCP tool handler for 'get_object_url'. Delegates to StorageService.get_object_url and formats the result 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))]
- Input schema and metadata definition for the 'get_object_url' tool, including parameters like bucket, key, disable_ssl, and expires.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-248 (registration)Registration of the 'get_object_url' tool (along with others) via tools.auto_register_tools in the register_tools function.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 method in StorageService that generates download URLs (public or signed private) for Qiniu Cloud objects using bucket domains.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