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