list_buckets
Retrieve configured Buckets filtered by a specified prefix using the Qiniu MCP Server, enabling efficient storage management and access within AI large model contexts.
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
- The decorated handler function implementing the logic for the 'list_buckets' MCP tool. Includes tool metadata with name, description, and input schema. Calls the underlying 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/__init__.py:7-11 (registration)Initialization function that creates the StorageService instance and invokes register_tools to register all storage tools, including 'list_buckets'.def load(cfg: config.Config): storage = StorageService(cfg) register_tools(storage) register_resource_provider(storage)
- src/mcp_server/core/storage/tools.py:236-248 (registration)The register_tools function that instantiates the tool implementation class and registers the list_buckets tool (and others) via 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, ] )
- Helper method in StorageService that performs the actual bucket listing using the S3-compatible API, filtering by configured buckets and optional prefix.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]