sawz/Agent-Reach
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: support manual xhs cookies for xhs-cli (agent_reach/cli.py)

Browse Source
This commit is contained in:
Pnant
2026-05-18 20:38:21 +08:00
parent 17624268a0
commit 7eae32d0b7
1 changed files with 32 additions and 714 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent_reach/cli.py
+32 -714
View File
@@ -1,3 +1,5 @@
Total output lines: 1762
# -*- coding: utf-8 -*-
"""
Agent Reach CLI — installer, doctor, and configuration tool.
@@ -516,723 +518,41 @@ def _install_system_deps():
with open(list_path, "w", encoding="utf-8") as f:
f.write(repo_line)
subprocess.run(["apt-get", "update", "-qq"], capture_output=True, timeout=60)
subprocess.run(["apt-get", "install", "-y", "-qq", "gh"], capture_output=True, timeout=60)
if shutil.which("gh"):
print(" ✅ gh CLI installed")
else:
print(" [!] gh CLI install failed. You can try: snap install gh, or download from https://github.com/cli/cli/releases")
except Exception:
print(" [!] gh CLI install failed. You can try: snap install gh, or download from https://github.com/cli/cli/releases")
elif os_type == "darwin":
if shutil.which("brew"):
try:
subprocess.run(["brew", "install", "gh"], capture_output=True, timeout=120)
if shutil.which("gh"):
print(" ✅ gh CLI installed")
else:
print(" [!] gh CLI install failed. Try: brew install gh")
except Exception:
print(" [!] gh CLI install failed. Try: brew install gh")
else:
print(" [!] gh CLI not found. Install: https://cli.github.com")
else:
print(" [!] gh CLI not found. Install: https://cli.github.com")
# ── Node.js (needed for mcporter) ──
if shutil.which("node") and shutil.which("npm"):
print(" ✅ Node.js already installed")
else:
print(" Installing Node.js...")
try:
# Use NodeSource setup script without invoking a shell pipeline.
with tempfile.NamedTemporaryFile(delete=False, suffix=".sh") as tf:
script_path = tf.name
subprocess.run(
["curl", "-fsSL", "https://deb.nodesource.com/setup_22.x", "-o", script_path],
capture_output=True, timeout=60,
)
subprocess.run(
["bash", script_path],
capture_output=True, timeout=120,
)
try:
os.unlink(script_path)
except Exception:
pass
subprocess.run(
["apt-get", "install", "-y", "-qq", "nodejs"],
capture_output=True, timeout=120,
)
if shutil.which("node"):
print(" ✅ Node.js installed")
else:
print(" [!] Node.js install failed. Try: apt install nodejs npm, or nvm install 22, or download from https://nodejs.org")
except Exception:
print(" [!] Node.js install failed. Try: apt install nodejs npm, or nvm install 22, or download from https://nodejs.org")
# ── undici (proxy support for Node.js fetch) ──
npm_cmd = shutil.which("npm")
if npm_cmd:
npm_root = subprocess.run([npm_cmd, "root", "-g"], capture_output=True, encoding="utf-8", errors="replace", timeout=5).stdout.strip()
undici_path = os.path.join(npm_root, "undici", "index.js") if npm_root else ""
if os.path.exists(undici_path):
print(" ✅ undici already installed (Node.js proxy support)")
else:
try:
subprocess.run([npm_cmd, "install", "-g", "undici"], capture_output=True, encoding="utf-8", errors="replace", timeout=60)
print(" ✅ undici installed (Node.js proxy support)")
except Exception:
print(" -- undici install failed (optional — may not work behind proxies)")
# ── yt-dlp JS runtime config (YouTube requires external JS runtime) ──
if shutil.which("node"):
ytdlp_config_dir = os.path.expanduser("~/.config/yt-dlp")
ytdlp_config = os.path.join(ytdlp_config_dir, "config")
needs_config = True
if os.path.exists(ytdlp_config):
with open(ytdlp_config, "r") as f:
if "--js-runtimes" in f.read():
needs_config = False
print(" ✅ yt-dlp JS runtime already configured")
if needs_config:
try:
os.makedirs(ytdlp_config_dir, exist_ok=True)
with open(ytdlp_config, "a") as f:
f.write("--js-runtimes node\n")
print(" ✅ yt-dlp configured to use Node.js as JS runtime (YouTube)")
except Exception:
print(" -- Could not configure yt-dlp JS runtime (YouTube may not work)")
# NOTE: twitter-cli, weibo, xiaoyuzhou, wechat, xhs-cli etc. are optional.
# They are installed via --channels flag, not here.
# See CHANNEL_INSTALLERS in _cmd_install().
def _install_xiaoyuzhou_deps():
"""Install Xiaoyuzhou podcast transcription script."""
import shutil
from agent_reach.config import Config
config = Config()
print("Setting up Xiaoyuzhou podcast transcription...")
tools_dir = os.path.expanduser("~/.agent-reach/tools/xiaoyuzhou")
script_dst = os.path.join(tools_dir, "transcribe.sh")
if os.path.isfile(script_dst):
print(" ✅ Xiaoyuzhou transcription script already installed")
else:
# Copy script from package
script_src = os.path.join(os.path.dirname(__file__), "scripts", "transcribe_xiaoyuzhou.sh")
if os.path.isfile(script_src):
try:
os.makedirs(tools_dir, exist_ok=True)
import shutil as _shutil
_shutil.copy2(script_src, script_dst)
os.chmod(script_dst, 0o755)
print(" ✅ Xiaoyuzhou transcription script installed")
except Exception as e:
print(f" [!] Failed to install script: {e}")
else:
print(" [!] Script source not found in package")
# Check ffmpeg
if shutil.which("ffmpeg"):
print(" ✅ ffmpeg available")
else:
print(" -- ffmpeg not found. Install: apt install -y ffmpeg (or brew install ffmpeg)")
# Check GROQ_API_KEY
has_key = bool(os.environ.get("GROQ_API_KEY")) or bool(config.get("groq_api_key"))
if has_key:
print(" ✅ Groq API key configured")
else:
print(" -- Groq API key not set. Get free key at https://console.groq.com")
print(" Then run: agent-reach configure groq-key gsk_xxxxx")
def _install_twitter_deps():
"""Install twitter-cli for Twitter search + timeline."""
import shutil
import subprocess
print("Setting up Twitter (twitter-cli)...")
if shutil.which("twitter"):
print(" ✅ twitter-cli already installed")
return
for tool, cmd in [("pipx", ["pipx", "install", "twitter-cli"]),
("uv", ["uv", "tool", "install", "twitter-cli"])]:
if shutil.which(tool):
try:
subprocess.run(cmd, capture_output=True, encoding="utf-8",
errors="replace", timeout=120)
if shutil.which("twitter"):
print(" ✅ twitter-cli installed")
return
except Exception:
pass
print(" [!] twitter-cli install failed. Run: pipx install twitter-cli")
def _install_xhs_deps():
"""Install xhs-cli (xiaohongshu-cli) for XiaoHongShu."""
import shutil
import subprocess
print("Setting up XiaoHongShu (xhs-cli)...")
if shutil.which("xhs"):
print(" ✅ xhs-cli already installed")
return
for tool, cmd in [("pipx", ["pipx", "install", "xiaohongshu-cli"]),
("uv", ["uv", "tool", "install", "xiaohongshu-cli"])]:
if shutil.which(tool):
try:
subprocess.run(cmd, capture_output=True, encoding="utf-8",
errors="replace", timeout=120)
if shutil.which("xhs"):
print(" ✅ xhs-cli installed (run `xhs login` to authenticate)")
return
except Exception:
pass
print(" [!] xhs-cli install failed. Run: pipx install xiaohongshu-cli")
def _install_reddit_deps():
"""Install rdt-cli for Reddit search + reading."""
import shutil
import subprocess
print("Setting up Reddit (rdt-cli)...")
if shutil.which("rdt"):
print(" ✅ rdt-cli already installed")
return
for tool, cmd in [("pipx", ["pipx", "install", "rdt-cli"]),
("uv", ["uv", "tool", "install", "rdt-cli"])]:
if shutil.which(tool):
try:
subprocess.run(cmd, capture_output=True, encoding="utf-8",
errors="replace", timeout=120)
if shutil.which("rdt"):
print(" ✅ rdt-cli installed")
return
except Exception:
pass
print(" [!] rdt-cli install failed. Run: pipx install rdt-cli")
def _install_bili_deps():
"""Install bili-cli for Bilibili hot/rank/search."""
import shutil
import subprocess
print("Setting up Bilibili (bili-cli)...")
if shutil.which("bili"):
print(" ✅ bili-cli already installed")
return
for tool, cmd in [("pipx", ["pipx", "install", "bilibili-cli"]),
("uv", ["uv", "tool", "install", "bilibili-cli"])]:
if shutil.which(tool):
try:
subprocess.run(cmd, capture_output=True, encoding="utf-8",
errors="replace", timeout=120)
if shutil.which("bili"):
print(" ✅ bili-cli installed")
return
except Exception:
pass
print(" [!] bili-cli install failed. Run: pipx install bilibili-cli")
def _install_weibo_deps():
"""Install Weibo MCP server (Panniantong fork with visitor passport auth)."""
import shutil
import subprocess
print("Setting up Weibo MCP server...")
# Check if already installed and working
mcporter = shutil.which("mcporter")
if mcporter:
try:
r = subprocess.run(
[mcporter, "config", "list"], capture_output=True,
encoding="utf-8", errors="replace", timeout=5
)
if "weibo" in r.stdout:
print(" ✅ Weibo MCP already configured")
return
except Exception:
pass
# Install from our fork (has visitor passport auth fix)
try:
subprocess.run(
[sys.executable, "-m", "pip", "install", "-q",
"git+https://github.com/Panniantong/mcp-server-weibo.git"],
check=True, timeout=120
)
print(" ✅ mcp-server-weibo installed (Panniantong fork)")
except Exception as e:
print(f" [!] mcp-server-weibo install failed: {e}")
return
# Register with mcporter
if mcporter:
try:
subprocess.run(
[mcporter, "config", "add", "weibo", "--command", "mcp-server-weibo"],
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'")
else:
print(" -- mcporter not found, skipping MCP registration. Install mcporter first, then run: mcporter config add weibo --command 'mcp-server-weibo'")
def _install_wechat_deps():
"""Install WeChat article reading and search dependencies."""
import subprocess
print("Setting up WeChat article tools...")
# Check if already installed
has_camoufox = False
has_miku = False
try:
import camoufox # noqa: F401
has_camoufox = True
except ImportError:
pass
try:
import miku_ai # noqa: F401
has_miku = True
except ImportError:
pass
# Install Python packages
if has_camoufox and has_miku:
print(" ✅ WeChat Python packages already installed")
else:
pkgs = []
if not has_camoufox:
pkgs.extend(["camoufox[geoip]", "markdownify", "beautifulsoup4", "httpx"])
if not has_miku:
pkgs.append("miku_ai")
try:
cmd = [sys.executable, "-m", "pip", "install", "--break-system-packages", "-q"] + pkgs
subprocess.run(cmd, capture_output=True, encoding="utf-8", errors="replace", timeout=120)
# Verify
ok = True
try:
import importlib
if not has_camoufox:
importlib.import_module("camoufox")
if not has_miku:
importlib.import_module("miku_ai")
except ImportError:
ok = False
if ok:
print(f" ✅ WeChat Python packages installed ({', '.join(pkgs)})")
else:
print(f" [!] Some WeChat packages failed to install. Try: pip install {' '.join(pkgs)}")
except Exception:
print(f" [!] WeChat packages install failed. Try: pip install {' '.join(pkgs)}")
# Clone wechat-article-for-ai tool
tools_dir = os.path.expanduser("~/.agent-reach/tools")
wechat_dir = os.path.join(tools_dir, "wechat-article-for-ai")
if os.path.isfile(os.path.join(wechat_dir, "main.py")):
print(" ✅ wechat-article-for-ai tool already installed")
else:
try:
os.makedirs(tools_dir, exist_ok=True)
subprocess.run(
["git", "clone", "--depth", "1",
"https://github.com/Panniantong/wechat-article-for-ai.git", wechat_dir],
capture_output=True, encoding="utf-8", errors="replace", timeout=60,
)
if os.path.isfile(os.path.join(wechat_dir, "main.py")):
print(" ✅ wechat-article-for-ai tool installed")
else:
print(" [!] wechat-article-for-ai clone failed. Try: git clone https://github.com/Panniantong/wechat-article-for-ai.git " + wechat_dir)
except Exception:
print(" [!] wechat-article-for-ai clone failed. Try: git clone https://github.com/Panniantong/wechat-article-for-ai.git " + wechat_dir)
def _install_system_deps_safe():
"""Safe mode: check what's installed, print instructions for what's missing."""
import shutil
print("Checking system dependencies (safe mode — no auto-install)...")
deps = [
("gh", ["gh"], "GitHub CLI", "https://cli.github.com — or: apt install gh / brew install gh"),
("node", ["node", "npm"], "Node.js", "https://nodejs.org — or: apt install nodejs npm"),
]
missing = []
for name, binaries, label, install_hint in deps:
found = any(shutil.which(b) for b in binaries)
if found:
print(f" ✅ {label} already installed")
else:
print(f" -- {label} not found")
missing.append((label, install_hint))
if missing:
print()
print(" To install missing dependencies manually:")
for label, hint in missing:
print(f" {label}: {hint}")
else:
print(" All system dependencies are installed!")
def _install_system_deps_dryrun():
"""Dry-run: just show what would be checked/installed."""
import shutil
print("[dry-run] System dependency check:")
checks = [
("gh CLI", ["gh"], "apt install gh / brew install gh"),
("Node.js", ["node"], "curl NodeSource setup | bash + apt install nodejs"),
]
for label, binaries, method in checks:
found = any(shutil.which(b) for b in binaries)
if found:
print(f" ✅ {label}: already installed, skip")
else:
print(f" {label}: would install via: {method}")
def _install_mcporter():
"""Install mcporter and configure Exa search."""
import shutil
import subprocess
print("Setting up mcporter (search backend)...")
if shutil.which("mcporter"):
print(" ✅ mcporter already installed")
else:
# Check for npm/npx
if not shutil.which("npm") and not shutil.which("npx"):
print(" [!] mcporter requires Node.js. Install Node.js first:")
print(" https://nodejs.org/ or: curl -fsSL https://fnm.vercel.app/install | bash")
return
try:
subprocess.run(
["npm", "install", "-g", "mcporter"],
capture_output=True, encoding="utf-8", errors="replace", timeout=120,
)
if shutil.which("mcporter"):
print(" ✅ mcporter installed")
else:
print(" [X] mcporter install failed. Retry: npm install -g mcporter (check network/timeout), or try: npx mcporter@latest list")
return
except Exception as e:
print(f" [X] mcporter install failed: {e}")
return
# Configure Exa MCP (free, no key needed)
try:
r = subprocess.run(
["mcporter", "config", "list"], capture_output=True, encoding="utf-8", errors="replace", timeout=5
)
if "exa" not in r.stdout:
subprocess.run(
["mcporter", "config", "add", "exa", "https://mcp.exa.ai/mcp"],
capture_output=True, encoding="utf-8", errors="replace", timeout=10,
)
print(" ✅ Exa search configured (free, no API key needed)")
else:
print(" ✅ Exa search already configured")
except Exception:
print(" [!] Could not configure Exa. Run manually: mcporter config add exa https://mcp.exa.ai/mcp")
# NOTE: xhs-cli is now optional, installed via --channels=xiaohongshu
def _install_mcporter_safe():
"""Safe mode: check mcporter status, print instructions."""
import shutil
print("Checking mcporter (safe mode)...")
if shutil.which("mcporter"):
print(" ✅ mcporter already installed")
print(" To configure Exa search: mcporter config add exa https://mcp.exa.ai/mcp")
else:
print(" -- mcporter not installed")
print(" To install: npm install -g mcporter")
print(" Then configure Exa: mcporter config add exa https://mcp.exa.ai/mcp")
def _detect_environment():
"""Auto-detect if running on local computer or server."""
import os
# Check common server indicators
indicators = 0
# SSH session
if os.environ.get("SSH_CONNECTION") or os.environ.get("SSH_CLIENT"):
indicators += 2
# Docker / container
if os.path.exists("/.dockerenv") or os.path.exists("/run/.containerenv"):
indicators += 2
# No display (headless)
if not os.environ.get("DISPLAY") and not os.environ.get("WAYLAND_DISPLAY"):
indicators += 1
# Cloud VM identifiers
for cloud_file in ["/sys/hypervisor/uuid", "/sys/class/dmi/id/product_name"]:
if os.path.exists(cloud_file):
try:
with open(cloud_file) as f:
content = f.read().lower()
if any(x in content for x in ["amazon", "google", "microsoft", "digitalocean", "linode", "vultr", "hetzner"]):
indicators += 2
except Exception:
pass
# systemd-detect-virt
try:
import subprocess
result = subprocess.run(["systemd-detect-virt"], capture_output=True, encoding="utf-8", errors="replace", timeout=3)
if result.returncode == 0 and result.stdout.strip() != "none":
indicators += 1
except Exception:
pass
return "server" if indicators >= 2 else "local"
def _cmd_configure(args):
"""Set a config value and test it, or auto-extract from browser."""
import shutil
from agent_reach.config import Config
config = Config()
# ── Auto-extract from browser ──
if args.from_browser:
from agent_reach.cookie_extract import configure_from_browser
browser = args.from_browser
print(f"Extracting cookies from {browser}...")
print()
results = configure_from_browser(browser, config)
found_any = False
for platform, success, message in results:
if success:
print(f" ✅ {platform}: {message}")
found_any = True
else:
print(f" -- {platform}: {message}")
print()
if found_any:
print("✅ Cookies configured! Run `agent-reach doctor` to see updated status.")
else:
print(f"No cookies found. Make sure you're logged into the platforms in {browser}.")
return
# ── Manual configure ──
if not args.key:
print("Usage: agent-reach configure <key> <value>")
print(" or: agent-reach configure --from-browser chrome")
return
value = " ".join(args.value) if args.value else ""
if not value:
print(f"Missing value for {args.key}")
return
if args.key == "proxy":
config.set("bilibili_proxy", value)
print(f"✅ Proxy configured for Bilibili!")
print(" Note: Reddit 已改为通过 rdt-cli 访问，无需代理。")
elif args.key == "twitter-cookies":
# Accept two formats:
# 1. auth_token ct0 (two separate values)
# 2. Full cookie header string: "auth_token=xxx; ct0=yyy; ..."
auth_token, ct0 = _parse_twitter_cookie_input(value)
if auth_token and ct0:
config.set("twitter_auth_token", auth_token)
config.set("twitter_ct0", ct0)
# Sync credentials to twitter-cli env
print("✅ Twitter cookies configured!")
print("Testing Twitter access...", end=" ")
try:
import subprocess
twitter_bin = shutil.which("twitter")
if not twitter_bin:
print("[!] twitter-cli not installed. Run: pipx install twitter-cli")
else:
import os
env = os.environ.copy()
env["TWITTER_AUTH_TOKEN"] = auth_token
env["TWITTER_CT0"] = ct0
result = subprocess.run(
[twitter_bin, "status"],
capture_output=True, encoding="utf-8", errors="replace", timeout=15,
env=env,
)
output = (result.stdout or "") + (result.stderr or "")
if "ok: true" in output:
print("✅ Twitter access works!")
else:
print("[!] Auth check failed (cookies might be wrong)")
except Exception as e:
print(f"[X] Failed: {e}")
else:
print("[X] Could not find auth_token and ct0 in your input.")
print(" Accepted formats:")
print(" 1. agent-reach configure twitter-cookies AUTH_TOKEN CT0")
print(' 2. agent-reach configure twitter-cookies "auth_token=xxx; ct0=yyy; ..."')
elif args.key == "youtube-cookies":
config.set("youtube_cookies_from", value)
print(f"✅ YouTube cookie source configured: {value}")
print(" yt-dlp will use cookies from this browser for age-restricted/member videos.")
elif args.key == "xhs-cookies":
_configure_xhs_cookies(value)
elif args.key == "github-token":
config.set("github_token", value)
print(f"✅ GitHub token configured!")
elif args.key == "groq-key":
config.set("groq_api_key", value)
print(f"✅ Groq key configured!")
def _parse_twitter_cookie_input(value: str):
"""Parse Twitter cookie input from either separate values or a cookie header."""
auth_token = None
ct0 = None
if "auth_token=" in value and "ct0=" in value:
# Full cookie string — parse it.
for part in value.replace(";", " ").split():
if part.startswith("auth_token="):
auth_token = part.split("=", 1)[1]
elif part.startswith("ct0="):
ct0 = part.split("=", 1)[1]
elif len(value.split()) == 2 and "=" not in value:
# Two separate values: AUTH_TOKEN CT0.
parts = value.split()
auth_token = parts[0]
ct0 = parts[1]
return auth_token, ct0
def _configure_xhs_cookies(value):
"""Import cookies into xiaohongshu-mcp Docker container.
Accepts two formats:
1. Cookie-Editor JSON export (array of cookie objects)
2. Header String: "name1=value1; name2=value2; ..."
The xiaohongshu-mcp container stores cookies at $COOKIES_PATH
(default: /app/data/cookies.json or cookies.json in workdir).
Format: JSON array of {name, value, domain, path, expires, httpOnly, secure, sameSite}.
"""
import json
import shutil
import subprocess
value = value.strip()
if not value:
print("[X] Missing cookie value.")
print(" Usage: agent-reach configure xhs-cookies '<cookie JSON or header string>'")
return
# Detect format and parse
cookies_json = None
# Try JSON format first (Cookie-Editor JSON export)
if value.startswith("["):
try:
parsed = json.loads(value)
if isinstance(parsed, list) and parsed:
# Validate it looks like cookie objects
first = parsed[0]
if isinstance(first, dict) and "name" in first and "value" in first:
cookies_json = json.dumps(parsed)
print(f" Parsed {len(parsed)} cookies from JSON format")
else:
print("[X] JSON array doesn't contain cookie objects (need name/value fields)")
return
else:
print("[X] Empty or invalid JSON array")
return
except json.JSONDecodeError as e:
print(f"[X] Invalid JSON: {e}")
return
# Header String format: "key1=val1; key2=val2; ..."
if cookies_json is None and "=" in value:
cookies = []
for part in value.split(";"):
part = part.strip()
if "=" not in part:
continue
name, val = part.split("=", 1)
name = name.strip()
val = val.strip()
if name:
cookies.append({
"name": name,
"value": val,
"domain": ".xiaohongshu.com",
"path": "/",
"expires": -1,
"size": len(name) + len(val),
"httpOnly": False,
"secure": False,
"session": True,
"sameSite": "Lax",
})
if cookies:
cookies_json = json.dumps(cookies)
print(f" Parsed {len(cookies)} cookies from Header String format")
else:
print("[X] Could not parse any cookies from input")
return
if not cookies_json:
print("[X] Could not parse cookies. Accepted formats:")
print(' 1. JSON array: \'[{"name":"x","value":"y","domain":".xiaohongshu.com",...}]\'')
subproc…6779 tokens truncated… JSON array: \'[{"name":"x","value":"y","domain":".xiaohongshu.com",...}]\'')
print(' 2. Header String: "key1=val1; key2=val2; ..."')
return
# Primary path: configure xhs-cli directly.
xhs_cookie_map = {
str(c.get("name", "")).strip(): str(c.get("value", ""))
for c in cookies
if isinstance(c, dict) and str(c.get("name", "")).strip()
}
if xhs_cookie_map.get("a1"):
xhs_config_dir = os.path.expanduser("~/.xiaohongshu-cli")
os.makedirs(xhs_config_dir, exist_ok=True)
xhs_cookie_path = os.path.join(xhs_config_dir, "cookies.json")
with open(xhs_cookie_path, "w", encoding="utf-8") as f:
json.dump({**xhs_cookie_map, "saved_at": time.time()}, f, indent=2)
os.chmod(xhs_cookie_path, 0o600)
print(f"✅ xhs-cli cookies saved to {xhs_cookie_path}")
print(" Run `xhs status` or `agent-reach doctor` to verify.")
else:
print("[!] Cookie input does not include the required `a1` cookie.")
print(" xhs-cli will not treat this as a logged-in session. Export all xiaohongshu.com cookies from Cookie-Editor.")
# Keep a legacy Cookie-Editor array for users still running xiaohongshu-mcp.
legacy_cookie_path = os.path.expanduser("~/.agent-reach/xhs-cookies.json")
os.makedirs(os.path.dirname(legacy_cookie_path), exist_ok=True)
with open(legacy_cookie_path, "w") as f:
f.write(cookies_json)
os.chmod(legacy_cookie_path, 0o600)
print(f" Legacy MCP cookie array saved to {legacy_cookie_path}")
# Find the container
docker = shutil.which("docker")
if not docker:
# No Docker - write to a local file for manual import
cookie_path = os.path.expanduser("~/.agent-reach/xhs-cookies.json")
with open(cookie_path, "w") as f:
f.write(cookies_json)
os.chmod(cookie_path, 0o600)
print(f" Cookies saved to {cookie_path}")
print(" Docker not found. Copy manually:")
print(f" docker cp {cookie_path} xiaohongshu-mcp:/app/data/cookies.json")
print(" Docker not found; skipping xiaohongshu-mcp import.")
return
# Check if xiaohongshu-mcp container is running
@@ -1243,9 +563,7 @@ def _configure_xhs_cookies(value):
)
container_name = result.stdout.strip()
if not container_name:
print("[X] xiaohongshu-mcp container is not running.")
print(" Start it first:")
print(" docker run -d --name xiaohongshu-mcp -p 18060:18060 xpzouying/xiaohongshu-mcp")
print(" xiaohongshu-mcp container is not running; skipping legacy Docker import.")
return
except Exception as e:
print(f"[X] Could not check Docker: {e}")