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)