live_streaming_list_buckets
Retrieve a list of all available live streaming spaces or buckets to manage and organize your streaming content on Qiniu Cloud.
Instructions
List all live streaming spaces/buckets. Returns information about all available live streaming buckets.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- MCP tool handler function for 'live_streaming_list_buckets' that delegates to the LiveStreamingService.list_buckets method.
async def list_buckets(self, **kwargs) -> list[types.TextContent]: result = await self.live_streaming.list_buckets(**kwargs) return [types.TextContent(type="text", text=str(result))] - Input schema definition for the 'live_streaming_list_buckets' tool, which requires no parameters.
types.Tool( name="live_streaming_list_buckets", description="List all live streaming spaces/buckets. Returns information about all available live streaming buckets.", inputSchema={ "type": "object", "properties": {}, "required": [], }, ) - src/mcp_server/core/live_streaming/tools.py:242-257 (registration)Tool registration function that registers the 'list_buckets' handler along with other live streaming tools.
def register_tools(live_streaming: LiveStreamingService): tool_impl = _ToolImpl(live_streaming) tools.auto_register_tools( [ tool_impl.create_bucket, tool_impl.create_stream, tool_impl.bind_push_domain, tool_impl.bind_play_domain, tool_impl.get_push_urls, tool_impl.get_play_urls, tool_impl.query_live_traffic_stats, tool_impl.list_buckets, tool_impl.list_streams, ] ) - Core helper method in LiveStreamingService that implements the bucket listing logic by making an authenticated GET request to the live streaming API endpoint.
async def list_buckets(self) -> Dict[str, Any]: """ List all live streaming spaces/buckets Returns: Dict containing the list of buckets """ if not self.live_endpoint: self.live_endpoint = "mls.cn-east-1.qiniumiku.com" # Remove protocol to get base endpoint endpoint = self.live_endpoint if endpoint.startswith("http://"): endpoint = endpoint[7:] elif endpoint.startswith("https://"): endpoint = endpoint[8:] url = f"https://{endpoint}/" headers = self._get_auth_header(method="GET", url=url) logger.info(f"Listing all live streaming buckets from {url}") async with aiohttp.ClientSession() as session: async with session.get(url, headers=headers) as response: status = response.status text = await response.text() if status == 200: logger.info("Successfully listed all buckets") return { "status": "success", "data": text, "message": "Buckets listed successfully", "status_code": status } else: logger.error(f"Failed to list buckets, status: {status}, response: {text}") return { "status": "error", "message": f"Failed to list buckets: {text}", "status_code": status }