fix(windows): force UTF-8 in channel subprocesses and MCP server env (#345)

Extracted from #318 (the UTF-8/doctor core, minus the env-wrapper feature):

- new agent_reach/utils/process.py: utf8_subprocess_env() +
  mcporter_utf8_env_args() — Windows GBK consoles otherwise corrupt
  Chinese output from mcporter/MCP child processes
- weibo/douyin/linkedin checks and weibo install/registration now pass
  the UTF-8 env (and register the MCP server with --env PYTHONUTF8=1)
- youtube: extract _has_js_runtime_config() with an OSError guard so an
  unreadable yt-dlp config can't crash doctor
- test_skill_command: open SKILL.md with explicit utf-8 (Windows GBK
  default broke these tests)

Co-authored-by: chidao <2980933590@qq.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

agent_reach/channels/douyin.py

@@ -3,6 +3,9 @@
 
 import shutil
 import subprocess
+
+from agent_reach.utils.process import utf8_subprocess_env
+
 from .base import Channel
 
 
@@ -31,7 +34,8 @@ class DouyinChannel(Channel):
         try:
             r = subprocess.run(
                 [mcporter, "config", "list"], capture_output=True,
-                encoding="utf-8", errors="replace", timeout=5
+                encoding="utf-8", errors="replace", timeout=5,
+                env=utf8_subprocess_env(),
             )
             if "douyin" not in r.stdout:
                 return "off", (
@@ -47,7 +51,8 @@ class DouyinChannel(Channel):
         try:
             r = subprocess.run(
                 [mcporter, "list", "douyin"],
-                capture_output=True, encoding="utf-8", errors="replace", timeout=15
+                capture_output=True, encoding="utf-8", errors="replace", timeout=15,
+                env=utf8_subprocess_env(),
             )
             if r.returncode == 0 and r.stdout.strip():
                 return "ok", "完整可用（视频解析、下载链接获取）"

agent_reach/channels/linkedin.py

@@ -3,6 +3,9 @@
 
 import shutil
 import subprocess
+
+from agent_reach.utils.process import utf8_subprocess_env
+
 from .base import Channel
 
 
@@ -28,7 +31,8 @@ class LinkedInChannel(Channel):
         try:
             r = subprocess.run(
                 [mcporter, "config", "list"], capture_output=True,
-                encoding="utf-8", errors="replace", timeout=5
+                encoding="utf-8", errors="replace", timeout=5,
+                env=utf8_subprocess_env(),
             )
             if "linkedin" in r.stdout.lower():
                 return "ok", "完整可用（Profile、公司、职位搜索）"

agent_reach/channels/weibo.py

@@ -3,6 +3,9 @@
 
 import shutil
 import subprocess
+
+from agent_reach.utils.process import utf8_subprocess_env
+
 from .base import Channel
 
 
@@ -24,26 +27,30 @@ class WeiboChannel(Channel):
                 "需要 mcporter + mcp-server-weibo。安装步骤：\n"
                 "  1. npm install -g mcporter\n"
                 "  2. pip install git+https://github.com/Panniantong/mcp-server-weibo.git\n"
-                "  3. mcporter config add weibo --command 'mcp-server-weibo'\n"
+                "  3. mcporter config add weibo --command 'mcp-server-weibo' "
+                "--env PYTHONUTF8=1 --env PYTHONIOENCODING=utf-8\n"
                 "  详见 https://github.com/Panniantong/mcp-server-weibo"
             )
         try:
             r = subprocess.run(
                 [mcporter, "config", "list"], capture_output=True,
-                encoding="utf-8", errors="replace", timeout=5
+                encoding="utf-8", errors="replace", timeout=5,
+                env=utf8_subprocess_env(),
             )
             if "weibo" not in r.stdout:
                 return "off", (
                     "mcporter 已装但微博 MCP 未配置。运行：\n"
                     "  pip install git+https://github.com/Panniantong/mcp-server-weibo.git\n"
-                    "  mcporter config add weibo --command 'mcp-server-weibo'"
+                    "  mcporter config add weibo --command 'mcp-server-weibo' "
+                    "--env PYTHONUTF8=1 --env PYTHONIOENCODING=utf-8"
                 )
         except Exception:
             return "off", "mcporter 连接异常"
         try:
             r = subprocess.run(
                 [mcporter, "list", "weibo"], capture_output=True,
-                encoding="utf-8", errors="replace", timeout=15
+                encoding="utf-8", errors="replace", timeout=15,
+                env=utf8_subprocess_env(),
             )
             if r.returncode == 0 and "search_users" in r.stdout:
                 return "ok", "完整可用（热搜、搜索、用户动态、评论）"

agent_reach/channels/youtube.py

@@ -9,6 +9,16 @@ from agent_reach.utils.text import read_utf8_text
 from .base import Channel
 
 
+def _has_js_runtime_config(config_path) -> bool:
+    """Return whether yt-dlp config explicitly enables a JS runtime."""
+    try:
+        if not config_path.exists():
+            return False
+        return "--js-runtimes" in read_utf8_text(config_path)
+    except OSError:
+        return False
+
+
 class YouTubeChannel(Channel):
     name = "youtube"
     description = "YouTube 视频和字幕"
@@ -36,10 +46,7 @@ class YouTubeChannel(Channel):
         has_deno = shutil.which("deno")
         if not has_deno:
             ytdlp_config = get_ytdlp_config_path()
-            has_js_config = False
+            if not _has_js_runtime_config(ytdlp_config):
-            if ytdlp_config.exists():
-                has_js_config = "--js-runtimes" in read_utf8_text(ytdlp_config)
-            if not has_js_config:
                 return "warn", (
                     f"yt-dlp 已安装但未配置 JS runtime。运行：\n  {render_ytdlp_fix_command()}"
                 )

agent_reach/cli.py

@@ -763,6 +763,8 @@ def _install_weibo_deps():
     import shutil
     import subprocess
 
+    from agent_reach.utils.process import mcporter_utf8_env_args, utf8_subprocess_env
+
     print("Setting up Weibo MCP server...")
 
     # Check if already installed and working
@@ -771,7 +773,8 @@ def _install_weibo_deps():
         try:
             r = subprocess.run(
                 [mcporter, "config", "list"], capture_output=True,
-                encoding="utf-8", errors="replace", timeout=5
+                encoding="utf-8", errors="replace", timeout=5,
+                env=utf8_subprocess_env(),
             )
             if "weibo" in r.stdout:
                 print("  ✅ Weibo MCP already configured")
@@ -784,25 +787,27 @@ def _install_weibo_deps():
         subprocess.run(
             [sys.executable, "-m", "pip", "install", "-q",
              "git+https://github.com/Panniantong/mcp-server-weibo.git"],
-            check=True, timeout=120
+            check=True, timeout=120, env=utf8_subprocess_env()
         )
         print("  ✅ mcp-server-weibo installed (Panniantong fork)")
     except Exception as e:
         print(f"  [!]  mcp-server-weibo install failed: {e}")
         return
 
-    # Register with mcporter
+    # Register with mcporter (force UTF-8 in the server's env — Windows GBK
+    # consoles otherwise corrupt the MCP server's Chinese output)
     if mcporter:
         try:
             subprocess.run(
-                [mcporter, "config", "add", "weibo", "--command", "mcp-server-weibo"],
+                [mcporter, "config", "add", "weibo", "--command", "mcp-server-weibo",
+                 *mcporter_utf8_env_args()],
                 check=True, capture_output=True, timeout=10
             )
             print("  ✅ Weibo MCP registered with mcporter")
         except Exception:
-            print("  [!]  mcporter config add failed. Run manually: mcporter config add weibo --command 'mcp-server-weibo'")
+            print("  [!]  mcporter config add failed. Run manually: mcporter config add weibo --command 'mcp-server-weibo' --env PYTHONUTF8=1 --env PYTHONIOENCODING=utf-8")
     else:
-        print("  -- mcporter not found, skipping MCP registration. Install mcporter first, then run: mcporter config add weibo --command 'mcp-server-weibo'")
+        print("  -- mcporter not found, skipping MCP registration. Install mcporter first, then run: mcporter config add weibo --command 'mcp-server-weibo' --env PYTHONUTF8=1 --env PYTHONIOENCODING=utf-8")
 
 
 def _install_wechat_deps():

agent_reach/utils/process.py

@@ -0,0 +1,26 @@
+"""Subprocess helpers for consistent cross-platform text handling."""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Mapping
+
+UTF8_ENV = {
+    "PYTHONUTF8": "1",
+    "PYTHONIOENCODING": "utf-8",
+}
+
+
+def utf8_subprocess_env(base: Mapping[str, str] | None = None) -> dict[str, str]:
+    """Return an environment that forces Python child processes into UTF-8 mode."""
+    env = dict(base or os.environ)
+    env.update(UTF8_ENV)
+    return env
+
+
+def mcporter_utf8_env_args() -> list[str]:
+    """Return mcporter --env arguments for UTF-8 Python stdio servers."""
+    args = []
+    for key, value in UTF8_ENV.items():
+        args.extend(["--env", f"{key}={value}"])
+    return args

tests/test_process.py

@@ -0,0 +1,18 @@
+from agent_reach.utils.process import mcporter_utf8_env_args, utf8_subprocess_env
+
+
+def test_utf8_subprocess_env_forces_python_utf8():
+    env = utf8_subprocess_env({"PYTHONUTF8": "0", "OTHER": "value"})
+
+    assert env["PYTHONUTF8"] == "1"
+    assert env["PYTHONIOENCODING"] == "utf-8"
+    assert env["OTHER"] == "value"
+
+
+def test_mcporter_utf8_env_args():
+    assert mcporter_utf8_env_args() == [
+        "--env",
+        "PYTHONUTF8=1",
+        "--env",
+        "PYTHONIOENCODING=utf-8",
+    ]

tests/test_skill_command.py

@@ -39,14 +39,11 @@ class TestSkillCommand(unittest.TestCase):
                 with patch.dict(os.environ, env, clear=True):
                     _install_skill()
 
-            target = os.path.join(skill_dir, "agent-reach", "SKILL.md")
             # Check at least one known skill dir pattern
-            found = False
             for dirpath, _, filenames in os.walk(tmpdir):
                 if "SKILL.md" in filenames:
-                    found = True
                     # Verify content is non-empty
-                    with open(os.path.join(dirpath, "SKILL.md")) as f:
+                    with open(os.path.join(dirpath, "SKILL.md"), encoding="utf-8") as f:
                         content = f.read()
                     self.assertIn("Agent Reach", content)
             # _install_skill may or may not find dirs depending on mock; just ensure no crash
@@ -58,7 +55,7 @@ class TestSkillCommand(unittest.TestCase):
             # Create a fake skill installation
             skill_path = os.path.join(tmpdir, ".openclaw", "skills", "agent-reach")
             os.makedirs(skill_path)
-            with open(os.path.join(skill_path, "SKILL.md"), "w") as f:
+            with open(os.path.join(skill_path, "SKILL.md"), "w", encoding="utf-8") as f:
                 f.write("test")
 
             self.assertTrue(os.path.exists(skill_path))
@@ -92,7 +89,7 @@ class TestSkillCommand(unittest.TestCase):
 
             target = os.path.join(skill_parent, "agent-reach", "SKILL.md")
             self.assertTrue(os.path.exists(target))
-            with open(target) as f:
+            with open(target, encoding="utf-8") as f:
                 content = f.read()
             self.assertIn("Agent Reach", content)
 
@@ -114,7 +111,7 @@ class TestSkillCommand(unittest.TestCase):
 
             target = os.path.join(skill_parent, "agent-reach", "SKILL.md")
             self.assertTrue(os.path.exists(target))
-            with open(target) as f:
+            with open(target, encoding="utf-8") as f:
                 content = f.read()
             self.assertTrue(content.strip())
             self.assertIn("WeChat Articles, Xiaoyuzhou Podcast", content)