list_objects
Retrieve object keys from a Qiniu Cloud Storage bucket with pagination support. Specify a prefix for filtered results and use start_after to continue listing from a specific key location. Ideal for managing large datasets.
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 for the "list_objects" MCP tool. It invokes the storage service's list_objects method and formats the result 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 schema definition for the list_objects tool, including input parameters like bucket, max_keys, prefix, start_after.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-248 (registration)The register_tools function that creates a tool implementation instance and registers the list_objects tool (and others) 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:9-9 (registration)Invocation of register_tools during the storage module load, which registers the list_objects tool with the MCP server.register_tools(storage)
- The core helper method in StorageService that performs the actual S3-compatible list_objects_v2 call to retrieve objects from the 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", [])