live_streaming_list_streams
Retrieve all live streams from a specified Qiniu Cloud bucket to manage and monitor streaming content.
Instructions
List all streams in a specific live streaming bucket. Returns the list of streams for the given bucket ID.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| bucket_id | Yes | The bucket ID/name to list streams from |
Implementation Reference
- Handler for the 'live_streaming_list_streams' MCP tool, decorated with metadata including schema. It acts as a thin wrapper invoking LiveStreamingService.list_streams and formatting the MCP response.
@tools.tool_meta( types.Tool( name="live_streaming_list_streams", description="List all streams in a specific live streaming bucket. Returns the list of streams for the given bucket ID.", inputSchema={ "type": "object", "properties": { "bucket_id": { "type": "string", "description": "The bucket ID/name to list streams from", }, }, "required": ["bucket_id"], }, ) ) async def list_streams(self, **kwargs) -> list[types.TextContent]: result = await self.live_streaming.list_streams(**kwargs) return [types.TextContent(type="text", text=str(result))] - JSON Schema defining the input parameters for the live_streaming_list_streams tool (requires 'bucket_id').
inputSchema={ "type": "object", "properties": { "bucket_id": { "type": "string", "description": "The bucket ID/name to list streams from", }, }, "required": ["bucket_id"], }, ) - Core helper method in LiveStreamingService implementing the logic to list streams via HTTP GET request to the live streaming API endpoint.
async def list_streams(self, bucket_id: str) -> Dict[str, Any]: """ List all streams in a specific live streaming bucket Args: bucket_id: The bucket ID/name Returns: Dict containing the list of streams in the bucket """ 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}/?streamlist&bucketId={bucket_id}" headers = self._get_auth_header(method="GET", url=url) logger.info(f"Listing streams in bucket: {bucket_id}") 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(f"Successfully listed streams in bucket: {bucket_id}") return { "status": "success", "bucket_id": bucket_id, "data": text, "message": f"Streams in bucket '{bucket_id}' listed successfully", "status_code": status } else: logger.error(f"Failed to list streams in bucket: {bucket_id}, status: {status}, response: {text}") return { "status": "error", "bucket_id": bucket_id, "message": f"Failed to list streams: {text}", "status_code": status } - src/mcp_server/core/live_streaming/tools.py:242-257 (registration)Module-level registration function that creates a tool implementation instance and auto-registers all live streaming tools, including list_streams.
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, ] ) - src/mcp_server/core/live_streaming/__init__.py:6-9 (registration)Top-level load function that instantiates LiveStreamingService and calls register_tools to register the live streaming MCP tools.
def load(cfg: config.Config): live = LiveStreamingService(cfg) register_tools(live)