update miku mcp tools

Author: callmefisher

src/mcp_server/core/live_streaming/live_streaming.py

@@ -37,9 +37,13 @@ def _get_auth_header(self, method: str, url: str, content_type: Optional[str] =
             # Generate Qiniu token for the request
             # For live streaming API, we use a simple token format
             token = self._generate_qiniu_token(method, url, content_type, body)
+            # token = generate_signature(method, url, body,self.access_key, self.secret_key)
             return {
                 "Authorization": f"Qiniu {token}"
             }
+        return  {
+            "Authorization": f"Qiniu ak:sk"
+        }
 
     def _build_bucket_url(self, bucket: str) -> str:
         """Build S3-style bucket URL"""
@@ -82,15 +86,36 @@ async def create_bucket(self, bucket: str) -> Dict[str, Any]:
             Dict containing the response status and message
         """
         url = self._build_bucket_url(bucket)
-        headers = self._get_auth_header(method="PUT",url=url)
-
-        logger.info(f"Creating bucket: {bucket} at {url}")
+        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(f"方法: 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) as response:
+            async with session.put(url, headers=headers, data=bodyJson) as response:
                 status = response.status
                 text = await response.text()
 
+                print(f"=== HTTP 响应信息 ===")
+                print(f"状态码: {status}")
+                print(f"响应内容: {text}")
+                print("==================")
+
                 if status == 200 or status == 201:
                     logger.info(f"Successfully created bucket: {bucket}")
                     return {
@@ -122,12 +147,17 @@ async def create_stream(self, bucket: str, stream: str) -> Dict[str, Any]:
             Dict containing the response status and message
         """
         url = self._build_stream_url(bucket, stream)
-        headers = self._get_auth_header(method="PUT", url=url)
+        data = {}
+        bodyJson = json.dumps(data)
+        headers = {
+            **self._get_auth_header(method="PUT", url=url, content_type="application/json", body=bodyJson),
+            "Content-Type": "application/json"
+        }
 
         logger.info(f"Creating stream: {stream} in bucket: {bucket} at {url}")
 
         async with aiohttp.ClientSession() as session:
-            async with session.put(url, headers=headers) as response:
+            async with session.put(url, headers=headers, data=bodyJson) as response:
                 status = response.status
                 text = await response.text()
 
@@ -458,19 +488,18 @@ async def list_streams(self, bucket_id: str) -> Dict[str, Any]:
     def _generate_qiniu_token(self, method: str, url: str, content_type: Optional[str] = None, body: Optional[str] = None) -> str:
         if not self.access_key or not self.secret_key:
             raise ValueError("QINIU_ACCESS_KEY and QINIU_SECRET_KEY are required")
-
         # Parse the URL
         parsed = urlparse(url)
 
         # 1. Add Method and Path
-        data = f"{method} {parsed.path}"
+        data = f"{method} {parsed.path or '/'}"
 
         # 2. Add Query if exists
         if parsed.query:
             data += f"?{parsed.query}"
 
         # 3. Add Host
-        data += f"\nHost: {parsed.netloc}"
+        data += f"\nHost: {parsed.hostname}"
 
         # 4. Add Content-Type if exists and not empty
         if content_type:
@@ -495,9 +524,41 @@ def _generate_qiniu_token(self, method: str, url: str, content_type: Optional[st
         sign = hmac.new(secret_bytes, data_bytes, hashlib.sha1).digest()
 
         # 8. URL-safe Base64 encode
-        encoded_sign = base64.urlsafe_b64encode(sign).decode('utf-8')
+        encoded_pre = base64.b64encode(sign).decode('utf-8')
+        encoded_sign = encoded_pre.replace('+', '-').replace('/', '_')
 
         # 9. Construct and return the Qiniu token
         qiniu_token = f"{self.access_key}:{encoded_sign}"
+        return qiniu_token
+
+
+
+
+
+
+def generate_signature(method, url, body, ak, sk):
+    parsed_url = urlparse(url)
+
+    # 构建签名数据
+    data = method + " " + parsed_url.path
+
+    if parsed_url.query:
+        data += "?" + parsed_url.query
+
+    data += "\nHost: " + parsed_url.hostname
+    data += "\nContent-Type: application/json"
+
+    if body:
+        data += "\n\n" + body
+    print(data)
+    # 使用HMAC-SHA1进行签名
+    hmac_sha1 = hmac.new(sk.encode('utf-8'), data.encode('utf-8'), hashlib.sha1)
+    hmac_result = hmac_sha1.digest()
+
+    sign = ak + ":" + base64_url_safe_encode(hmac_result)
+    return sign
 
-        return qiniu_token
+def base64_url_safe_encode(data):
+    encoded = base64.b64encode(data).decode('utf-8')
+    encoded = encoded.replace('+', '-').replace('/', '_')
+    return encoded