live_streaming_create_stream

Create a new live streaming channel using S3-style API for broadcasting video content to audiences.

Instructions

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

Input Schema

TableJSON Schema

Name	Required	Description	Default
`bucket`	Yes	LiveStreaming bucket name
`stream`	Yes	LiveStreaming stream name

Implementation Reference

src/mcp_server/core/live_streaming/tools.py:39-61 (handler)

MCP tool handler for 'live_streaming_create_stream': decorated function that validates input via schema and delegates to LiveStreamingService.create_stream, returning result as TextContent.

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

src/mcp_server/core/live_streaming/tools.py:40-57 (schema)

Input schema definition for the live_streaming_create_stream tool, specifying bucket and stream names as required strings.

types.Tool(
    name="live_streaming_create_stream",
    description="Create a new stream in LiveStreaming using S3-style API. The stream will be created at https://<bucket>.<endpoint_url>/<stream>",
    inputSchema={
        "type": "object",
        "properties": {
            "bucket": {
                "type": "string",
                "description": _BUCKET_DESC,
            },
            "stream": {
                "type": "string",
                "description": _STREAM_DESC,
            },
        },
        "required": ["bucket", "stream"],
    },
)

src/mcp_server/core/live_streaming/__init__.py:6-9 (registration)
Top-level registration entrypoint: creates LiveStreamingService instance and calls register_tools to auto-register all live streaming tools including live_streaming_create_stream.
```
def load(cfg: config.Config):
    live = LiveStreamingService(cfg)
    register_tools(live)
```

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

Registers the tool handler instance by calling tools.auto_register_tools on the list including create_stream.

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:137-183 (helper)

Underlying service method that performs the actual HTTP PUT request to create the stream at the LiveStreaming S3-style endpoint with authentication.

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

    Args:
        bucket: The bucket name
        stream: The stream name to create

    Returns:
        Dict containing the response status and message
    """
    url = self._build_stream_url(bucket, stream)
    data = {}
    bodyJson = json.dumps(data)
    headers = {
        **self._get_auth_header(method="PUT", url=url, content_type="application/json", body=bodyJson),
        "Content-Type": "application/json"
    }

    logger.info(f"Creating stream: {stream} in 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()

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

Qiniu MCP Server