live_streaming_bind_push_domain
Configure a push domain for RTMP/WHIP live streams by binding it to a LiveStreaming bucket in Qiniu Cloud.
Instructions
Bind a push domain to a LiveStreaming bucket for live streaming. This allows you to configure the domain for pushing RTMP/WHIP streams.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| bucket | Yes | LiveStreaming bucket name | |
| domain | Yes | The push domain name (e.g., mcp-push1.qiniu.com) | |
| domain_type | No | The type of push domain (default: pushRtmp) | pushRtmp |
Implementation Reference
- Core handler implementation in LiveStreamingService that performs the HTTP POST request to bind a push domain to a bucket.async def bind_push_domain(self, bucket: str, domain: str, domain_type: str = "pushRtmp") -> Dict[str, Any]: """ Bind a push domain to the bucket Args: bucket: The bucket name domain: The push domain name domain_type: The type of push domain (default: pushRtmp) Returns: Dict containing the response status and message """ url = f"{self._build_bucket_url(bucket)}/?pushDomain" data = { "domain": domain, "type": domain_type } body_str = json.dumps(data) headers = { **self._get_auth_header(method="POST", url=url, content_type="application/json", body=body_str), "Content-Type": "application/json" } logger.info(f"Binding push domain: {domain} (type: {domain_type}) to bucket: {bucket}") async with aiohttp.ClientSession() as session: async with session.post(url, headers=headers, json=data) as response: status = response.status text = await response.text() if status == 200 or status == 201: logger.info(f"Successfully bound push domain: {domain} to bucket: {bucket}") return { "status": "success", "bucket": bucket, "domain": domain, "type": domain_type, "message": f"Push domain '{domain}' bound successfully to bucket '{bucket}'", "status_code": status } else: logger.error(f"Failed to bind push domain: {domain}, status: {status}, response: {text}") return { "status": "error", "bucket": bucket, "domain": domain, "type": domain_type, "message": f"Failed to bind push domain: {text}", "status_code": status }
- Tool schema defining input parameters, description, and name for the MCP tool.types.Tool( name="live_streaming_bind_push_domain", description="Bind a push domain to a LiveStreaming bucket for live streaming. This allows you to configure the domain for pushing RTMP/WHIP streams.", inputSchema={ "type": "object", "properties": { "bucket": { "type": "string", "description": _BUCKET_DESC, }, "domain": { "type": "string", "description": "The push domain name (e.g., mcp-push1.qiniu.com)", }, "domain_type": { "type": "string", "description": "The type of push domain (default: pushRtmp)", "default": "pushRtmp", }, }, "required": ["bucket", "domain"], }, )
- MCP tool handler wrapper in _ToolImpl that delegates to LiveStreamingService and formats response.async def bind_push_domain(self, **kwargs) -> list[types.TextContent]: result = await self.live_streaming.bind_push_domain(**kwargs) return [types.TextContent(type="text", text=str(result))]
- src/mcp_server/core/live_streaming/tools.py:242-257 (registration)Function to register all LiveStreaming MCP tools, including live_streaming_bind_push_domain.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, ] )
- Helper method to construct the bucket URL used in bind_push_domain.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}"