Qiniu MCP Server

Instructions

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