list_objects
List objects in Qiniu Cloud Storage buckets with pagination support. Use start_after parameter to continue listing from a specific key and prefix filtering to narrow results.
Instructions
List objects in Qiniu Cloud, list a part each time, you can set start_after to continue listing, when the number of listed objects is less than max_keys, it means that all files are listed. start_after can be the key of the last file in the previous listing.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| bucket | Yes | Qiniu Cloud Storage bucket Name | |
| max_keys | No | Sets the max number of keys returned, default: 20 | |
| prefix | No | Specify the prefix of the operation response key. Only keys that meet this prefix will be listed. | |
| start_after | No | start_after is where you want Qiniu Cloud to start listing from. Qiniu Cloud starts listing after this specified key. start_after can be any key in the bucket. |
Implementation Reference
- The handler function that executes the list_objects tool logic, delegating to the storage service and formatting the response as TextContent.
async def list_objects(self, **kwargs) -> list[types.TextContent]: objects = await self.storage.list_objects(**kwargs) return [types.TextContent(type="text", text=str(objects))] - The input schema and metadata definition for the list_objects tool, including parameters like bucket, max_keys, prefix, start_after.
types.Tool( name="list_objects", description="List objects in Qiniu Cloud, list a part each time, you can set start_after to continue listing, when the number of listed objects is less than max_keys, it means that all files are listed. start_after can be the key of the last file in the previous listing.", inputSchema={ "type": "object", "properties": { "bucket": { "type": "string", "description": _BUCKET_DESC, }, "max_keys": { "type": "integer", "description": "Sets the max number of keys returned, default: 20", }, "prefix": { "type": "string", "description": "Specify the prefix of the operation response key. Only keys that meet this prefix will be listed.", }, "start_after": { "type": "string", "description": "start_after is where you want Qiniu Cloud to start listing from. Qiniu Cloud starts listing after this specified key. start_after can be any key in the bucket.", }, }, "required": ["bucket"], }, ) ) - src/mcp_server/core/storage/tools.py:236-247 (registration)Registers the list_objects tool (via tool_impl.list_objects) 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, ] ) - src/mcp_server/core/storage/__init__.py:7-10 (registration)Invokes register_tools to register storage tools including list_objects during service initialization.
def load(cfg: config.Config): storage = StorageService(cfg) register_tools(storage) register_resource_provider(storage) - Core implementation of list_objects in StorageService using aioboto3 S3 client to list objects in the specified bucket.
async def list_objects( self, bucket: str, prefix: str = "", max_keys: int = 20, start_after: str = "" ) -> List[dict]: if self.config.buckets and bucket not in self.config.buckets: logger.warning(f"Bucket {bucket} not in configured bucket list") return [] if isinstance(max_keys, str): max_keys = int(max_keys) if max_keys > 100: max_keys = 100 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: response = await s3.list_objects_v2( Bucket=bucket, Prefix=prefix, MaxKeys=max_keys, StartAfter=start_after, ) return response.get("Contents", [])