Qiniu MCP Server

Instructions

Implementation Reference

• src/mcp_server/core/live_streaming/tools.py:92-119 (handler)

  MCP tool handler for 'live_streaming_bind_play_domain': decorated function that delegates to LiveStreamingService.bind_play_domain
  @tools.tool_meta( types.Tool( name="live_streaming_bind_play_domain", description="Bind a playback domain to a LiveStreaming bucket for live streaming. This allows you to configure the domain for playing back streams via FLV/M3U8/WHEP.", inputSchema={ "type": "object", "properties": { "bucket": { "type": "string", "description": _BUCKET_DESC, }, "domain": { "type": "string", "description": "The playback domain name (e.g., mcp-play1.qiniu.com)", }, "domain_type": { "type": "string", "description": "The type of playback domain (default: live)", "default": "live", }, }, "required": ["bucket", "domain"], }, ) ) async def bind_play_domain(self, **kwargs) -> list[types.TextContent]: result = await self.live_streaming.bind_play_domain(**kwargs) return [types.TextContent(type="text", text=str(result))]
• src/mcp_server/core/live_streaming/tools.py:96-114 (schema)

  Input schema defining parameters: bucket (required), domain (required), domain_type (optional, default 'live')
  inputSchema={ "type": "object", "properties": { "bucket": { "type": "string", "description": _BUCKET_DESC, }, "domain": { "type": "string", "description": "The playback domain name (e.g., mcp-play1.qiniu.com)", }, "domain_type": { "type": "string", "description": "The type of playback domain (default: live)", "default": "live", }, }, "required": ["bucket", "domain"], },
• src/mcp_server/core/live_streaming/live_streaming.py:235-284 (handler)

  Core implementation of bind_play_domain in LiveStreamingService: makes authenticated POST request to bind playback domain to bucket
  async def bind_play_domain(self, bucket: str, domain: str, domain_type: str = "live") -> Dict[str, Any]: """ Bind a playback domain to the bucket Args: bucket: The bucket name domain: The playback domain name domain_type: The type of playback domain (default: live) Returns: Dict containing the response status and message """ url = f"{self._build_bucket_url(bucket)}/?domain" 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 playback 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 playback domain: {domain} to bucket: {bucket}") return { "status": "success", "bucket": bucket, "domain": domain, "type": domain_type, "message": f"Playback domain '{domain}' bound successfully to bucket '{bucket}'", "status_code": status } else: logger.error(f"Failed to bind playback domain: {domain}, status: {status}, response: {text}") return { "status": "error", "bucket": bucket, "domain": domain, "type": domain_type, "message": f"Failed to bind playback domain: {text}", "status_code": status }
• src/mcp_server/core/live_streaming/tools.py:242-257 (registration)

  register_tools function creates _ToolImpl instance and calls tools.auto_register_tools including bind_play_domain tool
  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-9 (registration)

  load function instantiates LiveStreamingService and calls register_tools to register all live streaming tools
  def load(cfg: config.Config): live = LiveStreamingService(cfg) register_tools(live)