live_streaming_get_push_urls

Generate RTMP and WHIP push URLs for live streaming by providing domain, bucket, and stream name parameters.

Instructions

Get push URLs for a stream. Returns RTMP and WHIP push URLs that can be used to push live streams.

Input Schema

TableJSON Schema

Name	Required	Description
`push_domain`	Yes	The push domain name
`bucket`	Yes	LiveStreaming bucket name
`stream_name`	Yes	The stream name

Implementation Reference

src/mcp_server/core/live_streaming/tools.py:121-147 (handler)

The handler function for the 'live_streaming_get_push_urls' tool, decorated with @tools.tool_meta that defines the tool name, description, input schema, and executes by calling the service method.

@tools.tool_meta(
    types.Tool(
        name="live_streaming_get_push_urls",
        description="Get push URLs for a stream. Returns RTMP and WHIP push URLs that can be used to push live streams.",
        inputSchema={
            "type": "object",
            "properties": {
                "push_domain": {
                    "type": "string",
                    "description": "The push domain name",
                },
                "bucket": {
                    "type": "string",
                    "description": _BUCKET_DESC,
                },
                "stream_name": {
                    "type": "string",
                    "description": "The stream name",
                },
            },
            "required": ["push_domain", "bucket", "stream_name"],
        },
    )
)
async def get_push_urls(self, **kwargs) -> list[types.TextContent]:
    result = self.live_streaming.get_push_urls(**kwargs)
    return [types.TextContent(type="text", text=str(result))]

src/mcp_server/core/live_streaming/live_streaming.py:285-309 (helper)

Core helper method in LiveStreamingService that implements the logic to generate RTMP and WHIP push URLs for the given push_domain, bucket, and stream_name.

def get_push_urls(self, push_domain: str, bucket: str, stream_name: str) -> Dict[str, Any]:
    """
    Generate push URLs for RTMP and WHIP protocols

    Args:
        push_domain: The push domain
        bucket: The bucket name
        stream_name: The stream name

    Returns:
        Dict containing RTMP and WHIP push URLs
    """
    rtmp_url = f"rtmp://{push_domain}/{bucket}/{stream_name}"
    whip_url = f"https://{push_domain}/{bucket}/{stream_name}.whip"

    logger.info(f"Generated push URLs for stream: {stream_name}")
    return {
        "status": "success",
        "push_domain": push_domain,
        "bucket": bucket,
        "stream_name": stream_name,
        "rtmp_url": rtmp_url,
        "whip_url": whip_url,
        "message": "Push URLs generated successfully"
    }

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

register_tools function that instantiates _ToolImpl and calls tools.auto_register_tools with the list of tool methods, including get_push_urls.

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/__init__.py:6-8 (registration)
Top-level load function that creates LiveStreamingService and calls register_tools to register the live streaming tools to the MCP server.
```
def load(cfg: config.Config):
    live = LiveStreamingService(cfg)
    register_tools(live)
```

Qiniu MCP Server