From bc2d554e13a0aed823fb927694fd444c989d3765 Mon Sep 17 00:00:00 2001 From: Pnant <73925474+Panniantong@users.noreply.github.com> Date: Tue, 31 Mar 2026 17:17:19 +0800 Subject: [PATCH] fix(xiaohongshu): switch from Docker MCP to xhs-cli (pipx install) (#236) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Docker + Cookie import → `pipx install xiaohongshu-cli` + `xhs login`. Tier downgraded from 2 to 1. xhs-cli (1,477 stars) auto-extracts cookies from browser, supports search/read/comment/post/hot/feed. - xiaohongshu.py: check() detects `xhs` binary, parses `xhs status` - cli.py: install via pipx instead of Docker - skill/references/social.md: updated XHS commands - tests: 3 new tests for CLI-based check() - 77 tests passing Co-authored-by: Claude Opus 4.6 (1M context) --- agent_reach/channels/xiaohongshu.py | 115 +++++++------------------ agent_reach/cli.py | 34 +++----- agent_reach/skill/references/social.md | 30 +++++-- tests/test_channels.py | 35 +++++--- 4 files changed, 88 insertions(+), 126 deletions(-) diff --git a/agent_reach/channels/xiaohongshu.py b/agent_reach/channels/xiaohongshu.py index 3b728c1..8a71df7 100644 --- a/agent_reach/channels/xiaohongshu.py +++ b/agent_reach/channels/xiaohongshu.py @@ -1,8 +1,6 @@ # -*- coding: utf-8 -*- -"""XiaoHongShu -- check if mcporter + xiaohongshu MCP is available.""" +"""XiaoHongShu — check if xhs-cli (xiaohongshu-cli) is available.""" -import json -import platform import shutil import subprocess from .base import Channel @@ -117,54 +115,11 @@ def _clean_comment(comment): return result -def _is_arm64() -> bool: - """Detect ARM64 architecture (e.g. Apple Silicon).""" - machine = platform.machine().lower() - return machine in ("arm64", "aarch64") - - -def _mcporter_status_ok(stdout: str) -> bool: - """Return True if mcporter JSON output indicates status == 'ok'. - - Uses proper JSON parsing to handle Windows BOM, CRLF line endings, and - whitespace variations. Falls back to normalised string matching so the - check still works if mcporter ever returns non-JSON text. - """ - text = stdout.strip() - # Strip UTF-8 BOM that Windows PowerShell sometimes prepends. - if text.startswith("\ufeff"): - text = text[1:] - try: - data = json.loads(text) - if isinstance(data, dict): - return str(data.get("status", "")).lower() == "ok" - except (json.JSONDecodeError, ValueError): - pass - # Fallback: normalise whitespace and CRLF, then do string search. - normalised = text.lower().replace("\r\n", "\n").replace("\r", "\n").replace(" ", "") - return '"status":"ok"' in normalised - - -def _docker_run_hint() -> str: - """Return the docker run command, with --platform flag for ARM64.""" - if _is_arm64(): - return ( - " docker run -d --name xiaohongshu-mcp -p 18060:18060 " - "--platform linux/amd64 xpzouying/xiaohongshu-mcp\n" - " # ARM64 also: build from source: " - "https://github.com/xpzouying/xiaohongshu-mcp" - ) - return ( - " docker run -d --name xiaohongshu-mcp -p 18060:18060 " - "xpzouying/xiaohongshu-mcp" - ) - - class XiaoHongShuChannel(Channel): name = "xiaohongshu" description = "小红书笔记" - backends = ["xiaohongshu-mcp"] - tier = 2 + backends = ["xhs-cli (xiaohongshu-cli)"] + tier = 1 def can_handle(self, url: str) -> bool: from urllib.parse import urlparse @@ -172,48 +127,36 @@ class XiaoHongShuChannel(Channel): return "xiaohongshu.com" in d or "xhslink.com" in d def check(self, config=None): - mcporter = shutil.which("mcporter") - if not mcporter: + xhs = shutil.which("xhs") + if not xhs: return "off", ( - "需要 mcporter + xiaohongshu-mcp。安装步骤：\n" - " 1. npm install -g mcporter\n" - " 2. " + _docker_run_hint().strip() + "\n" - " 3. mcporter config add xiaohongshu http://localhost:18060/mcp\n" - " 详见 https://github.com/xpzouying/xiaohongshu-mcp" + "需要安装 xhs-cli：\n" + " pipx install xiaohongshu-cli\n" + "或：\n" + " uv tool install xiaohongshu-cli\n" + "安装后运行 `xhs login` 登录" ) - is_windows = platform.system() == "Windows" - config_timeout = 15 if is_windows else 5 - try: - r = subprocess.run( - [mcporter, "config", "get", "xiaohongshu", "--json"], - capture_output=True, - encoding="utf-8", - errors="replace", - timeout=config_timeout, - ) - if r.returncode != 0 or "xiaohongshu" not in r.stdout.lower(): - return "off", ( - "mcporter 已装但小红书 MCP 未配置。运行：\n" - + _docker_run_hint() + "\n" - " mcporter config add xiaohongshu http://localhost:18060/mcp" - ) - except Exception: - return "off", "mcporter 连接异常" - # Use longer timeouts on Windows where mcporter may be slower to respond. - list_timeout = 30 if is_windows else 10 try: r = subprocess.run( - [mcporter, "list", "xiaohongshu", "--json"], - capture_output=True, - encoding="utf-8", - errors="replace", - timeout=list_timeout, + [xhs, "status"], capture_output=True, + encoding="utf-8", errors="replace", timeout=10, + ) + output = (r.stdout or "") + (r.stderr or "") + if r.returncode == 0 and "ok: true" in output: + return "ok", ( + "完整可用（搜索、阅读、评论、发帖、热门、" + "收藏、关注、用户查询）" + ) + if "not_authenticated" in output or "expired" in output: + return "warn", ( + "xhs-cli 已安装但未登录。运行：\n" + " xhs login\n" + "（自动从浏览器提取 Cookie，或扫码登录）" + ) + return "warn", ( + "xhs-cli 已安装但状态异常。运行：\n" + " xhs -v status 查看详细信息" ) - if r.returncode == 0 and _mcporter_status_ok(r.stdout): - return "ok", "MCP 已连接（阅读、搜索、发帖、评论、点赞）" - return "warn", "MCP 已配置，但连接异常；请检查 xiaohongshu-mcp 服务状态" - except subprocess.TimeoutExpired: - return "warn", "MCP 已配置，但健康检查超时；请检查 xiaohongshu-mcp 服务状态" except Exception: - return "warn", "MCP 已配置，但连接异常；请检查 xiaohongshu-mcp 服务状态" + return "warn", "xhs-cli 已安装但连接失败" diff --git a/agent_reach/cli.py b/agent_reach/cli.py index 9e02c9a..f41f19d 100644 --- a/agent_reach/cli.py +++ b/agent_reach/cli.py @@ -871,30 +871,24 @@ def _install_mcporter(): except Exception: print(" [!] Could not configure Exa. Run manually: mcporter config add exa https://mcp.exa.ai/mcp") - # Check XiaoHongShu MCP (only if server is running) - try: - r = subprocess.run( - ["mcporter", "config", "list"], capture_output=True, encoding="utf-8", errors="replace", timeout=5 - ) - if "xiaohongshu" in r.stdout: - print(" ✅ XiaoHongShu MCP already configured") - else: - # Check if XHS MCP server is running on localhost:18060 - import requests + # Check XiaoHongShu CLI + if shutil.which("xhs"): + print(" ✅ xhs-cli already installed (xiaohongshu-cli)") + else: + if shutil.which("pipx"): try: - requests.get("http://localhost:18060/", timeout=3) subprocess.run( - ["mcporter", "config", "add", "xiaohongshu", "http://localhost:18060/mcp"], - capture_output=True, encoding="utf-8", errors="replace", timeout=10, + ["pipx", "install", "xiaohongshu-cli"], + capture_output=True, encoding="utf-8", errors="replace", timeout=120, ) - print(" ✅ XiaoHongShu MCP auto-detected and configured") + if shutil.which("xhs"): + print(" ✅ xhs-cli installed (run `xhs login` to authenticate)") + else: + print(" -- xhs-cli install failed (optional). Run: pipx install xiaohongshu-cli") except Exception: - print(" -- XiaoHongShu MCP not detected (optional)") - print(" Install: docker run -d --name xiaohongshu-mcp -p 18060:18060 xpzouying/xiaohongshu-mcp") - print(" Then: mcporter config add xiaohongshu http://localhost:18060/mcp") - print(" Repo: https://github.com/xpzouying/xiaohongshu-mcp") - except Exception: - pass + print(" -- xhs-cli install failed (optional). Run: pipx install xiaohongshu-cli") + else: + print(" -- xhs-cli requires pipx (optional). Run: pipx install xiaohongshu-cli") def _install_mcporter_safe(): diff --git a/agent_reach/skill/references/social.md b/agent_reach/skill/references/social.md index 92919d6..d81ee8a 100644 --- a/agent_reach/skill/references/social.md +++ b/agent_reach/skill/references/social.md @@ -2,23 +2,35 @@ 小红书、抖音、Twitter/X、微博、B站、V2EX、Reddit。 -## 小红书 / XiaoHongShu +## 小红书 / XiaoHongShu (xhs-cli) ```bash # 搜索笔记 -mcporter call 'xiaohongshu.search_feeds(keyword: "query")' +xhs search "query" -# 获取笔记详情 -mcporter call 'xiaohongshu.get_feed_detail(feed_id: "xxx", xsec_token: "yyy")' +# 阅读笔记详情 +xhs read NOTE_ID_OR_URL -# 获取笔记详情 + 评论 -mcporter call 'xiaohongshu.get_feed_detail(feed_id: "xxx", xsec_token: "yyy", load_all_comments: true)' +# 查看评论 +xhs comments NOTE_ID_OR_URL -# 发布内容 -mcporter call 'xiaohongshu.publish_content(title: "标题", content: "正文", images: ["/path/img.jpg"], tags: ["tag"])' +# 浏览热门 +xhs hot + +# 推荐 feed +xhs feed + +# 用户主页 +xhs user USER_ID +xhs user-posts USER_ID + +# 发帖/互动 +xhs post --title "标题" --content "正文" --images img1.jpg img2.jpg +xhs like NOTE_ID +xhs comment NOTE_ID "评论内容" ``` -> **需要登录**: 使用 Cookie-Editor 浏览器插件导出 cookies。运行 `agent-reach doctor` 检查状态。 +> **安装**: `pipx install xiaohongshu-cli`，然后 `xhs login`（自动从浏览器提取 Cookie）。 ## 抖音 / Douyin diff --git a/tests/test_channels.py b/tests/test_channels.py index 216674b..3e43f48 100644 --- a/tests/test_channels.py +++ b/tests/test_channels.py @@ -649,19 +649,32 @@ class TestXueqiuChannel: class TestXiaoHongShuChannel: - def test_reports_ok_when_server_health_is_ok(self, monkeypatch): - monkeypatch.setattr(shutil, "which", lambda _: "/opt/homebrew/bin/mcporter") + def test_reports_ok_when_cli_authenticated(self, monkeypatch): + monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/xhs") def fake_run(cmd, **kwargs): - if cmd[:4] == ["/opt/homebrew/bin/mcporter", "config", "get", "xiaohongshu"]: - return subprocess.CompletedProcess(cmd, 0, '{"name":"xiaohongshu"}', "") - if cmd[:4] == ["/opt/homebrew/bin/mcporter", "list", "xiaohongshu", "--json"]: - return subprocess.CompletedProcess(cmd, 0, '{"status": "ok"}', "") - raise AssertionError(f"unexpected command: {cmd}") + return subprocess.CompletedProcess(cmd, 0, "ok: true\nusername: testuser\n", "") monkeypatch.setattr(subprocess, "run", fake_run) - assert XiaoHongShuChannel().check() == ( - "ok", - "MCP 已连接（阅读、搜索、发帖、评论、点赞）", - ) + status, msg = XiaoHongShuChannel().check() + assert status == "ok" + assert "完整可用" in msg + + def test_reports_warn_when_not_authenticated(self, monkeypatch): + monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/xhs") + + def fake_run(cmd, **kwargs): + return subprocess.CompletedProcess(cmd, 1, "", "ok: false\nerror:\n code: not_authenticated\n") + + monkeypatch.setattr(subprocess, "run", fake_run) + + status, msg = XiaoHongShuChannel().check() + assert status == "warn" + assert "xhs login" in msg + + def test_reports_off_when_not_installed(self, monkeypatch): + monkeypatch.setattr(shutil, "which", lambda _: None) + status, msg = XiaoHongShuChannel().check() + assert status == "off" + assert "xiaohongshu-cli" in msg