live_streaming_create_bucket

Create a new bucket for live streaming storage using S3-style API to organize and manage streaming content on Qiniu Cloud.

Instructions

Create a new bucket in LiveStreaming using S3-style API. The bucket will be created at https://.<endpoint_url>

Input Schema

TableJSON Schema

Name	Required	Description	Default
`bucket`	Yes	LiveStreaming bucket name

Implementation Reference

src/mcp_server/core/live_streaming/tools.py:19-37 (handler)

The primary handler function for the 'live_streaming_create_bucket' MCP tool. It defines the tool metadata (including input schema) and executes the logic by delegating to LiveStreamingService.create_bucket.

@tools.tool_meta(
    types.Tool(
        name="live_streaming_create_bucket",
        description="Create a new bucket in LiveStreaming using S3-style API. The bucket will be created at https://<bucket>.<endpoint_url>",
        inputSchema={
            "type": "object",
            "properties": {
                "bucket": {
                    "type": "string",
                    "description": _BUCKET_DESC,
                },
            },
            "required": ["bucket"],
        },
    )
)
async def create_bucket(self, **kwargs) -> list[types.TextContent]:
    result = await self.live_streaming.create_bucket(**kwargs)
    return [types.TextContent(type="text", text=str(result))]

src/mcp_server/core/live_streaming/tools.py:242-256 (registration)

The register_tools function where the live_streaming_create_bucket handler is registered into the MCP tools registry via auto_register_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,
        ]
    )

src/mcp_server/core/live_streaming/live_streaming.py:78-136 (helper)

Core helper implementation in LiveStreamingService that performs the HTTP PUT request to create the bucket, handles authentication, and processes the response.

async def create_bucket(self, bucket: str) -> Dict[str, Any]:
    """
    Create a bucket using S3-style API

    Args:
        bucket: The bucket name to create

    Returns:
        Dict containing the response status and message
    """
    url = self._build_bucket_url(bucket)
    data = {}
    bodyJson = json.dumps(data)
    auth_headers = self._get_auth_header(method="PUT", url=url, content_type="application/json", body=bodyJson)
    headers = {"Content-Type": "application/json"}
    # 如果有认证头，添加到headers中
    if auth_headers:
        headers.update(auth_headers)

    # 打印 HTTP 请求信息
    print("=== HTTP 请求信息 ===")
    print("方法: PUT")
    print(f"URL: {url}")
    print("请求头:")
    for key, value in headers.items():
        print(f"  {key}: {value}")
    print("请求体: {}")
    print("===================")

    print(f"Creating bucket: {bucket} at {url}")

    async with aiohttp.ClientSession() as session:
        async with session.put(url, headers=headers, data=bodyJson) as response:
            status = response.status
            text = await response.text()

            print(f"状态码: {status}")
            print(f"响应内容: {text}")
            print("==================")

            if status == 200 or status == 201:
                logger.info(f"Successfully created bucket: {bucket}")
                return {
                    "status": "success",
                    "bucket": bucket,
                    "url": url,
                    "message": f"Bucket '{bucket}' created successfully",
                    "status_code": status
                }
            else:
                logger.error(f"Failed to create bucket: {bucket}, status: {status}, response: {text}")
                return {
                    "status": "error",
                    "bucket": bucket,
                    "url": url,
                    "message": f"Failed to create bucket: {text}",
                    "status_code": status
                }

src/mcp_server/core/live_streaming/live_streaming.py:48-62 (helper)

Helper method to construct the S3-style bucket URL used in the create_bucket implementation.

def _build_bucket_url(self, bucket: str) -> str:
    """Build S3-style bucket URL"""
    if not self.live_endpoint:
        self.live_endpoint = "mls.cn-east-1.qiniumiku.com"

    # Remove protocol if present in live_endpoint
    endpoint = self.live_endpoint
    if endpoint.startswith("http://"):
        endpoint = endpoint[7:]
    elif endpoint.startswith("https://"):
        endpoint = endpoint[8:]

    # Build URL in format: https://<bucket>.<endpoint>
    return f"https://{bucket}.{endpoint}"

src/mcp_server/core/live_streaming/__init__.py:6-8 (registration)
The load function that instantiates LiveStreamingService and calls register_tools to register all live streaming tools including live_streaming_create_bucket.
```
def load(cfg: config.Config):
    live = LiveStreamingService(cfg)
    register_tools(live)
```

Qiniu MCP Server