diff --git a/agent_reach/channels/__init__.py b/agent_reach/channels/__init__.py index 3e79dbd..627cb66 100644 --- a/agent_reach/channels/__init__.py +++ b/agent_reach/channels/__init__.py @@ -20,6 +20,7 @@ from .douyin import DouyinChannel from .linkedin import LinkedInChannel from .bosszhipin import BossZhipinChannel from .wechat import WeChatChannel +from .weibo import WeiboChannel # Channel registry @@ -34,6 +35,7 @@ ALL_CHANNELS: List[Channel] = [ LinkedInChannel(), BossZhipinChannel(), WeChatChannel(), + WeiboChannel(), RSSChannel(), ExaSearchChannel(), WebChannel(), diff --git a/agent_reach/channels/weibo.py b/agent_reach/channels/weibo.py new file mode 100644 index 0000000..198f89b --- /dev/null +++ b/agent_reach/channels/weibo.py @@ -0,0 +1,54 @@ +# -*- coding: utf-8 -*- +"""Weibo (微博) — check if mcporter + mcp-server-weibo is available.""" + +import shutil +import subprocess +from .base import Channel + + +class WeiboChannel(Channel): + name = "weibo" + description = "微博动态与热搜" + backends = ["mcp-server-weibo"] + tier = 1 + + def can_handle(self, url: str) -> bool: + from urllib.parse import urlparse + d = urlparse(url).netloc.lower() + return "weibo.com" in d or "weibo.cn" in d + + def check(self, config=None): + mcporter = shutil.which("mcporter") + if not mcporter: + return "off", ( + "需要 mcporter + mcp-server-weibo。安装步骤:\n" + " 1. npm install -g mcporter\n" + " 2. uvx mcp-server-weibo(或 pip install mcp-server-weibo)\n" + " 3. mcporter config add weibo --command 'uvx mcp-server-weibo'\n" + " 详见 https://github.com/qinyuanpei/mcp-server-weibo" + ) + try: + r = subprocess.run( + [mcporter, "config", "list"], capture_output=True, + encoding="utf-8", errors="replace", timeout=5 + ) + if "weibo" not in r.stdout: + return "off", ( + "mcporter 已装但微博 MCP 未配置。运行:\n" + " uvx mcp-server-weibo # 或 pip install mcp-server-weibo\n" + " mcporter config add weibo --command 'uvx mcp-server-weibo'\n" + " # 海外服务器需配置 WEIBO_COOKIE 环境变量,\n" + " # 详见 SKILL.md 微博章节" + ) + except Exception: + return "off", "mcporter 连接异常" + try: + r = subprocess.run( + [mcporter, "list", "weibo"], capture_output=True, + encoding="utf-8", errors="replace", timeout=15 + ) + if r.returncode == 0 and "search_users" in r.stdout: + return "ok", "完整可用(热搜、搜索、用户动态、评论)" + return "warn", "MCP 已配置但工具加载失败,检查 mcp-server-weibo 版本" + except Exception: + return "warn", "MCP 连接异常,检查 mcp-server-weibo 是否可用" diff --git a/tests/test_channel_contracts.py b/tests/test_channel_contracts.py index 5cfb092..87feda3 100644 --- a/tests/test_channel_contracts.py +++ b/tests/test_channel_contracts.py @@ -116,6 +116,7 @@ def test_channel_can_handle_contract(): "douyin": "https://www.douyin.com/video/123", "linkedin": "https://www.linkedin.com/in/test", "bosszhipin": "https://www.zhipin.com/web/geek/job?query=python", + "weibo": "https://weibo.com/u/1749127163", "rss": "https://example.com/feed.xml", "exa_search": "https://example.com", "web": "https://example.com",