Qiniu MCP Server

Instructions

Implementation Reference

• src/mcp_server/core/live_streaming/tools.py:19-37 (handler)

  The primary handler function for the 'live_streaming_create_bucket' MCP tool. It defines the tool metadata (including input schema) and executes the logic by delegating to LiveStreamingService.create_bucket.

  @tools.tool_meta(
      types.Tool(
          name="live_streaming_create_bucket",
          description="Create a new bucket in LiveStreaming using S3-style API. The bucket will be created at https://<bucket>.<endpoint_url>",
          inputSchema={
              "type": "object",
              "properties": {
                  "bucket": {
                      "type": "string",
                      "description": _BUCKET_DESC,
                  },
              },
              "required": ["bucket"],
          },
      )
  )
  async def create_bucket(self, **kwargs) -> list[types.TextContent]:
      result = await self.live_streaming.create_bucket(**kwargs)
      return [types.TextContent(type="text", text=str(result))]

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

  The register_tools function where the live_streaming_create_bucket handler is registered into the MCP tools registry via auto_register_tools.

  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:78-136 (helper)

  Core helper implementation in LiveStreamingService that performs the HTTP PUT request to create the bucket, handles authentication, and processes the response.

  async def create_bucket(self, bucket: str) -> Dict[str, Any]:
      """
      Create a bucket using S3-style API
  
      Args:
          bucket: The bucket name to create
  
      Returns:
          Dict containing the response status and message
      """
      url = self._build_bucket_url(bucket)
      data = {}
      bodyJson = json.dumps(data)
      auth_headers = self._get_auth_header(method="PUT", url=url, content_type="application/json", body=bodyJson)
      headers = {"Content-Type": "application/json"}
      # 如果有认证头，添加到headers中
      if auth_headers:
          headers.update(auth_headers)
  
      # 打印 HTTP 请求信息
      print("=== HTTP 请求信息 ===")
      print("方法: PUT")
      print(f"URL: {url}")
      print("请求头:")
      for key, value in headers.items():
          print(f"  {key}: {value}")
      print("请求体: {}")
      print("===================")
  
      print(f"Creating bucket: {bucket} at {url}")
  
      async with aiohttp.ClientSession() as session:
          async with session.put(url, headers=headers, data=bodyJson) as response:
              status = response.status
              text = await response.text()
  
              print(f"状态码: {status}")
              print(f"响应内容: {text}")
              print("==================")
  
              if status == 200 or status == 201:
                  logger.info(f"Successfully created bucket: {bucket}")
                  return {
                      "status": "success",
                      "bucket": bucket,
                      "url": url,
                      "message": f"Bucket '{bucket}' created successfully",
                      "status_code": status
                  }
              else:
                  logger.error(f"Failed to create bucket: {bucket}, status: {status}, response: {text}")
                  return {
                      "status": "error",
                      "bucket": bucket,
                      "url": url,
                      "message": f"Failed to create bucket: {text}",
                      "status_code": status
                  }

• src/mcp_server/core/live_streaming/live_streaming.py:48-62 (helper)

  Helper method to construct the S3-style bucket URL used in the create_bucket implementation.

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

• src/mcp_server/core/live_streaming/__init__.py:6-8 (registration)

  The load function that instantiates LiveStreamingService and calls register_tools to register all live streaming tools including live_streaming_create_bucket.

  def load(cfg: config.Config):
      live = LiveStreamingService(cfg)
      register_tools(live)