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 | Default |
|---|---|---|---|
| push_domain | Yes | The push domain name | |
| bucket | Yes | LiveStreaming bucket name | |
| stream_name | Yes | The stream name |
Implementation Reference
- 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))] - 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)