feat: 支持鉴权token

skills/beauty/scripts/beauty.py

@@ -17,6 +17,10 @@ FETCH_API_URL = "https://api.pearapi.ai/api/today_wife"
 FALLBACK_TEXT = "今天的美女图片暂时没拿到，等我再找找。"
 
 
+def _client_private_token() -> str:
+    return os.environ.get("ROBOT_CLIENT_PRIVATE_TOKEN", "").strip()
+
+
 def fetch_image_url() -> str | None:
     try:
         with urllib.request.urlopen(FETCH_API_URL, timeout=10) as response:
@@ -52,7 +56,10 @@ def send_image(image_url: str) -> bool:
     request = urllib.request.Request(
         api_url,
         data=body,
-        headers={"Content-Type": "application/json"},
+        headers={
+            "Content-Type": "application/json",
+            "X-Private-Token": _client_private_token(),
+        },
         method="POST",
     )
 

skills/create-memory/scripts/create_memory.py

@@ -31,6 +31,10 @@ def _require_env(name: str) -> str:
     return value
 
 
+def _client_private_token() -> str:
+    return os.environ.get("ROBOT_CLIENT_PRIVATE_TOKEN", "").strip()
+
+
 def _positive_int_env(name: str) -> int:
     raw = _require_env(name)
     try:
@@ -80,7 +84,10 @@ def _post_json(url: str, body: dict[str, Any]) -> dict[str, Any]:
     request = urllib.request.Request(
         url,
         data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
-        headers={"Content-Type": "application/json"},
+        headers={
+            "Content-Type": "application/json",
+            "X-Private-Token": _client_private_token(),
+        },
         method="POST",
     )
     try:

skills/create-scheduled-task/scripts/create_scheduled_task.py

@@ -43,6 +43,10 @@ class AmbiguousCreateError(RuntimeError):
     """The POST may have reached the server, so retrying could create a duplicate."""
 
 
+def _client_private_token() -> str:
+    return os.environ.get("ROBOT_CLIENT_PRIVATE_TOKEN", "").strip()
+
+
 class SkillArgumentParser(argparse.ArgumentParser):
     def error(self, message: str) -> NoReturn:
         raise ValueError(f"参数错误：{message}")
@@ -310,7 +314,11 @@ def _pick_unique_member(
 
 
 def _get_json(url: str, timeout: int = 15) -> dict[str, Any]:
-    request = urllib.request.Request(url, method="GET")
+    request = urllib.request.Request(
+        url,
+        headers={"X-Private-Token": _client_private_token()},
+        method="GET",
+    )
     try:
         with urllib.request.urlopen(request, timeout=timeout) as response:
             response_text = response.read().decode("utf-8", errors="replace")
@@ -435,7 +443,10 @@ def _post_json(url: str, payload: dict[str, Any], timeout: int = 30) -> dict[str
     request = urllib.request.Request(
         url,
         data=body,
-        headers={"Content-Type": "application/json"},
+        headers={
+            "Content-Type": "application/json",
+            "X-Private-Token": _client_private_token(),
+        },
         method="POST",
     )
 

skills/export-chat-history/scripts/export_chat_history.py

@@ -525,6 +525,10 @@ def _raise_for_client_error(payload: dict[str, Any]) -> None:
         raise RuntimeError(message)
 
 
+def _client_private_token() -> str:
+    return os.environ.get("ROBOT_CLIENT_PRIVATE_TOKEN", "").strip()
+
+
 def _send_file(client_port: str, chat_room_id: str, path: Path) -> None:
     url = (
         f"http://127.0.0.1:{client_port}"
@@ -536,7 +540,10 @@ def _send_file(client_port: str, chat_room_id: str, path: Path) -> None:
     request = urllib.request.Request(
         url,
         data=body,
-        headers={"Content-Type": "application/json"},
+        headers={
+            "Content-Type": "application/json",
+            "X-Private-Token": _client_private_token(),
+        },
         method="POST",
     )
     try:

skills/find-recent-chat-media/scripts/find_recent_chat_media.py

@@ -190,6 +190,10 @@ def _client_base_url(client_port: str) -> str:
     return f"http://127.0.0.1:{client_port}"
 
 
+def _client_private_token() -> str:
+    return os.environ.get("ROBOT_CLIENT_PRIVATE_TOKEN", "").strip()
+
+
 def _build_url(base_url: str, path: str, params: dict[str, object] | None = None) -> str:
     url = f"{base_url}{path}"
     if not params:
@@ -199,8 +203,13 @@ def _build_url(base_url: str, path: str, params: dict[str, object] | None = None
 
 
 def _http_get_bytes(url: str, timeout: int = 300) -> tuple[bytes, dict[str, str]]:
+    request = urllib.request.Request(
+        url,
+        headers={"X-Private-Token": _client_private_token()},
+        method="GET",
+    )
     try:
-        with urllib.request.urlopen(url, timeout=timeout) as resp:
+        with urllib.request.urlopen(request, timeout=timeout) as resp:
             headers = {key.lower(): value for key, value in resp.headers.items()}
             return resp.read(), headers
     except urllib.error.HTTPError as exc:
@@ -240,7 +249,10 @@ def _http_post_multipart(
     req = urllib.request.Request(
         url,
         data=body,
-        headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
+        headers={
+            "Content-Type": f"multipart/form-data; boundary={boundary}",
+            "X-Private-Token": _client_private_token(),
+        },
         method="POST",
     )
     try:

skills/image-to-image/scripts/image_to_image.py

@@ -159,6 +159,10 @@ def load_drawing_settings(conn, from_wx_id: str) -> tuple[bool, dict]:
 # API callers
 # ---------------------------------------------------------------------------
 
+def _client_private_token() -> str:
+    return os.environ.get("ROBOT_CLIENT_PRIVATE_TOKEN", "").strip()
+
+
 def _http_post_json(url: str, body: dict, headers: dict, timeout: int = 300) -> dict:
     data = json.dumps(body).encode("utf-8")
     req = urllib.request.Request(url, data=data, headers=headers, method="POST")
@@ -349,7 +353,15 @@ def _send_image_outputs(client_port: str, from_wx_id: str, image_outputs: list[s
             "to_wxid": from_wx_id,
             "image_urls": remote_urls,
         }
-        response = _http_post_json(send_url, send_body, {"Content-Type": "application/json"}, timeout=300)
+        response = _http_post_json(
+            send_url,
+            send_body,
+            {
+                "Content-Type": "application/json",
+                "X-Private-Token": _client_private_token(),
+            },
+            timeout=300,
+        )
         _debug_response("send image url response", response)
 
     for file_path in local_paths:
@@ -358,7 +370,15 @@ def _send_image_outputs(client_port: str, from_wx_id: str, image_outputs: list[s
             "to_wxid": from_wx_id,
             "file_path": file_path,
         }
-        response = _http_post_json(send_url, send_body, {"Content-Type": "application/json"}, timeout=300)
+        response = _http_post_json(
+            send_url,
+            send_body,
+            {
+                "Content-Type": "application/json",
+                "X-Private-Token": _client_private_token(),
+            },
+            timeout=300,
+        )
         _debug_response("send image local response", response)
 
 

skills/send-emoji/scripts/send_emoji.py

@@ -62,12 +62,19 @@ EMOJI_MAP: dict[str, dict[str, object]] = {
 }
 
 
+def _client_private_token() -> str:
+    return os.environ.get("ROBOT_CLIENT_PRIVATE_TOKEN", "").strip()
+
+
 def _http_post_json(url: str, body: dict, timeout: int = 300) -> dict:
     data = json.dumps(body).encode("utf-8")
     req = urllib.request.Request(
         url,
         data=data,
-        headers={"Content-Type": "application/json"},
+        headers={
+            "Content-Type": "application/json",
+            "X-Private-Token": _client_private_token(),
+        },
         method="POST",
     )
     with urllib.request.urlopen(req, timeout=timeout) as resp:

skills/send-file/scripts/send_file.py

@@ -15,6 +15,10 @@ sys.stderr = sys.stdout
 MAX_FILE_SIZE = 25 * 1024 * 1024
 
 
+def _client_private_token() -> str:
+    return os.environ.get("ROBOT_CLIENT_PRIVATE_TOKEN", "").strip()
+
+
 def _raise_for_client_error(response: dict) -> None:
     if not response:
         return
@@ -40,7 +44,10 @@ def _http_post_json(url: str, body: dict, timeout: int = 300) -> dict:
     req = urllib.request.Request(
         url,
         data=data,
-        headers={"Content-Type": "application/json"},
+        headers={
+            "Content-Type": "application/json",
+            "X-Private-Token": _client_private_token(),
+        },
         method="POST",
     )
     with urllib.request.urlopen(req, timeout=timeout) as resp:

skills/send-image/scripts/send_image.py

@@ -13,6 +13,10 @@ from pathlib import Path
 sys.stderr = sys.stdout
 
 
+def _client_private_token() -> str:
+    return os.environ.get("ROBOT_CLIENT_PRIVATE_TOKEN", "").strip()
+
+
 def _raise_for_client_error(response: dict) -> None:
     if not response:
         return
@@ -38,7 +42,10 @@ def _http_post_json(url: str, body: dict, timeout: int = 300) -> dict:
     req = urllib.request.Request(
         url,
         data=data,
-        headers={"Content-Type": "application/json"},
+        headers={
+            "Content-Type": "application/json",
+            "X-Private-Token": _client_private_token(),
+        },
         method="POST",
     )
     with urllib.request.urlopen(req, timeout=timeout) as resp:

skills/send-mention-message/scripts/send_mention_message.py

@@ -14,6 +14,10 @@ from pathlib import Path
 sys.stderr = sys.stdout
 
 
+def _client_private_token() -> str:
+    return os.environ.get("ROBOT_CLIENT_PRIVATE_TOKEN", "").strip()
+
+
 def _skill_root() -> Path:
     return Path(__file__).resolve().parent.parent
 
@@ -94,7 +98,10 @@ def _http_post_json(url: str, body: dict, timeout: int = 300) -> dict:
     req = urllib.request.Request(
         url,
         data=data,
-        headers={"Content-Type": "application/json"},
+        headers={
+            "Content-Type": "application/json",
+            "X-Private-Token": _client_private_token(),
+        },
         method="POST",
     )
     with urllib.request.urlopen(req, timeout=timeout) as resp:

skills/text-to-image/scripts/text_to_image.py

@@ -161,6 +161,10 @@ def load_drawing_settings(conn, from_wx_id: str) -> tuple[bool, dict]:
 # API callers
 # ---------------------------------------------------------------------------
 
+def _client_private_token() -> str:
+    return os.environ.get("ROBOT_CLIENT_PRIVATE_TOKEN", "").strip()
+
+
 def _http_post_json(url: str, body: dict, headers: dict, timeout: int = 300) -> dict:
     data = json.dumps(body).encode("utf-8")
     req = urllib.request.Request(url, data=data, headers=headers, method="POST")
@@ -360,7 +364,15 @@ def _send_image_outputs(client_port: str, from_wx_id: str, image_outputs: list[s
             "to_wxid": from_wx_id,
             "image_urls": remote_urls,
         }
-        response = _http_post_json(send_url, send_body, {"Content-Type": "application/json"}, timeout=300)
+        response = _http_post_json(
+            send_url,
+            send_body,
+            {
+                "Content-Type": "application/json",
+                "X-Private-Token": _client_private_token(),
+            },
+            timeout=300,
+        )
         _debug_response("send image url response", response)
 
     for file_path in local_paths:
@@ -369,7 +381,15 @@ def _send_image_outputs(client_port: str, from_wx_id: str, image_outputs: list[s
             "to_wxid": from_wx_id,
             "file_path": file_path,
         }
-        response = _http_post_json(send_url, send_body, {"Content-Type": "application/json"}, timeout=300)
+        response = _http_post_json(
+            send_url,
+            send_body,
+            {
+                "Content-Type": "application/json",
+                "X-Private-Token": _client_private_token(),
+            },
+            timeout=300,
+        )
         _debug_response("send image local response", response)
 
 

skills/video-generation/scripts/video_generation.py

@@ -30,6 +30,10 @@ DEFAULT_RESOLUTION = "720p"
 DEFAULT_DURATION = 5
 
 
+def _client_private_token() -> str:
+    return os.environ.get("ROBOT_CLIENT_PRIVATE_TOKEN", "").strip()
+
+
 def _skill_root() -> Path:
     script_dir = Path(__file__).resolve().parent
     return script_dir.parent
@@ -193,7 +197,15 @@ def send_videos(from_wx_id: str, video_urls: list[str]) -> None:
         "to_wxid": from_wx_id,
         "video_urls": [url for url in video_urls if url],
     }
-    _http_post_json(send_url, send_body, {"Content-Type": "application/json"}, timeout=60)
+    _http_post_json(
+        send_url,
+        send_body,
+        {
+            "Content-Type": "application/json",
+            "X-Private-Token": _client_private_token(),
+        },
+        timeout=60,
+    )
 
 
 def call_jimeng_video(

skills/voice-message/scripts/voice_message.py

@@ -59,6 +59,10 @@ MAX_CONTENT_LENGTH = 260
 STREAM_END_CODE = 20000000
 
 
+def _client_private_token() -> str:
+    return os.environ.get("ROBOT_CLIENT_PRIVATE_TOKEN", "").strip()
+
+
 def _skill_root() -> Path:
     return Path(__file__).resolve().parent.parent
 
@@ -252,7 +256,11 @@ def _download_referenced_voice_clone(message_id: str) -> str:
         f"http://127.0.0.1:{client_port}/api/v1/robot/chat/voice/download"
         f"?message_id={encoded_message_id}"
     )
-    req = urllib.request.Request(download_url, method="GET")
+    req = urllib.request.Request(
+        download_url,
+        headers={"X-Private-Token": _client_private_token()},
+        method="GET",
+    )
     try:
         with urllib.request.urlopen(req, timeout=60) as response:
             wav_data = response.read()
@@ -836,7 +844,10 @@ def send_voice(from_wx_id: str, audio_data: bytes, audio_format: str) -> None:
         req = urllib.request.Request(
             send_url,
             data=body,
-            headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
+            headers={
+                "Content-Type": f"multipart/form-data; boundary={boundary}",
+                "X-Private-Token": _client_private_token(),
+            },
             method="POST",
         )
         try:

skills/web-page/scripts/web_page.ts

@@ -1703,6 +1703,9 @@ function postJson(
   timeoutMs: number,
 ): Promise<string> {
   return new Promise<string>((resolve, reject) => {
+    const privateToken = (
+      process.env.ROBOT_CLIENT_PRIVATE_TOKEN ?? ""
+    ).trim();
     const parsed = new URL(url);
     const payload = Buffer.from(JSON.stringify(body), "utf8");
     const request = http.request(
@@ -1714,6 +1717,7 @@ function postJson(
         headers: {
           "Content-Type": "application/json",
           "Content-Length": payload.length,
+          "X-Private-Token": privateToken,
         },
         timeout: timeoutMs,
       },

tests/test_private_token_headers.py

@@ -0,0 +1,27 @@
+import unittest
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+
+
+class PrivateTokenHeaderTests(unittest.TestCase):
+    def test_every_local_client_api_script_supports_optional_private_token(self) -> None:
+        client_api_scripts: list[Path] = []
+        for pattern in ("*.py", "*.ts"):
+            for script in (REPO_ROOT / "skills").glob(f"*/scripts/{pattern}"):
+                content = script.read_text(encoding="utf-8")
+                if "/api/v1/robot" not in content:
+                    continue
+                client_api_scripts.append(script)
+                self.assertIn("ROBOT_CLIENT_PRIVATE_TOKEN", content, script)
+                self.assertIn("X-Private-Token", content, script)
+                self.assertNotIn(
+                    "环境变量 ROBOT_CLIENT_PRIVATE_TOKEN 未配置", content, script
+                )
+
+        self.assertTrue(client_api_scripts, "no local client API scripts were found")
+
+
+if __name__ == "__main__":
+    unittest.main()