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

src/mcp_server/core/live_streaming/live_streaming.py:184-234 (handler)

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
                }

src/mcp_server/core/live_streaming/tools.py:64-86 (schema)

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

src/mcp_server/core/live_streaming/tools.py:88-90 (handler)

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

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

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}"

Qiniu MCP Server