list_buckets
Retrieve configured storage buckets from Qiniu Cloud with optional prefix filtering to manage and organize cloud storage resources.
Instructions
Return the Bucket you configured based on the conditions.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| prefix | No | Bucket prefix. The listed Buckets will be filtered based on this prefix, and only those matching the prefix will be output. |
Implementation Reference
- MCP tool handler for 'list_buckets', including schema definition and execution logic that delegates to StorageService.list_buckets()
@tools.tool_meta( types.Tool( name="list_buckets", description="Return the Bucket you configured based on the conditions.", inputSchema={ "type": "object", "properties": { "prefix": { "type": "string", "description": "Bucket prefix. The listed Buckets will be filtered based on this prefix, and only those matching the prefix will be output.", }, }, "required": [], }, ) ) async def list_buckets(self, **kwargs) -> list[types.TextContent]: buckets = await self.storage.list_buckets(**kwargs) return [types.TextContent(type="text", text=str(buckets))] - src/mcp_server/core/storage/tools.py:236-247 (registration)Registration of storage tools including list_buckets 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 of list_buckets in StorageService using aioboto3 S3 client to list configured buckets, with prefix filtering.
async def list_buckets(self, prefix: Optional[str] = None) -> List[dict]: if not self.config.buckets or len(self.config.buckets) == 0: return [] max_buckets = 50 async with self.s3_session.client( "s3", aws_access_key_id=self.config.access_key, aws_secret_access_key=self.config.secret_key, endpoint_url=self.config.endpoint_url, region_name=self.config.region_name, ) as s3: # If buckets are configured, only return those response = await s3.list_buckets() all_buckets = response.get("Buckets", []) configured_bucket_list = [ bucket for bucket in all_buckets if bucket["Name"] in self.config.buckets ] if prefix: configured_bucket_list = [ b for b in configured_bucket_list if b["Name"] > prefix ] return configured_bucket_list[:max_buckets]