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

src/mcp_server/core/live_streaming/tools.py:221-239 (handler)

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))]

src/mcp_server/core/live_streaming/tools.py:225-235 (schema)

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"],
    },
)

src/mcp_server/core/live_streaming/live_streaming.py:526-573 (helper)

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)
```

Qiniu MCP Server