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

fix(xiaohongshu): switch from Docker MCP to xhs-cli (pipx install) (#236)

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
Pnant
2026-03-31 17:17:19 +08:00
committed by GitHub
parent 15f161e5b5
commit bc2d554e13
4 changed files with 88 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent_reach/channels/xiaohongshu.py
+29 -86
View File
@@ -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 已安装但连接失败"
agent_reach/cli.py
+14 -20
View File
@@ -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():
agent_reach/skill/references/social.md
+21 -9
View File
@@ -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
tests/test_channels.py
+24 -11
View File
@@ -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