Merge pull request #23 from Panniantong/refactor/simplify-to-installer

refactor: 回归安装器+文档定位，砍掉 read/search 包装层

README.md

@@ -11,7 +11,7 @@
 </p>
 
 <p align="center">
-  <a href="#30-秒上手">快速开始</a> · <a href="docs/README_en.md">English</a> · <a href="#支持的平台">支持平台</a> · <a href="#设计理念">设计理念</a>
+  <a href="#快速上手">快速开始</a> · <a href="docs/README_en.md">English</a> · <a href="#支持的平台">支持平台</a> · <a href="#设计理念">设计理念</a>
 </p>
 
 ---
@@ -104,7 +104,7 @@ AI Agent 已经能帮你写代码、改文档、管项目——但你让它去
 2. **安装系统依赖** — 自动检测并安装 Node.js、gh CLI、mcporter、bird 等
 3. **配置搜索引擎** — 通过 MCP 接入 Exa（免费，无需 API Key）
 4. **检测环境** — 判断是本地电脑还是服务器，给出对应的配置建议
-5. **注册 Skill** — 在 Agent 的 skills 目录安装 SKILL.md，以后 Agent 遇到"搜推特"、"看视频"这类需求，会自动知道调用 Agent Reach
+5. **注册 SKILL.md** — 在 Agent 的 skills 目录安装使用指南，以后 Agent 遇到"搜推特"、"看视频"这类需求，会自动知道该调哪个上游工具
 
 安装完之后，`agent-reach doctor` 一条命令告诉你每个渠道的状态。
 </details>
@@ -115,14 +115,14 @@ AI Agent 已经能帮你写代码、改文档、管项目——但你让它去
 
 不需要任何配置，告诉 Agent 就行：
 
-- "帮我看看这个链接" → 任意网页
+- "帮我看看这个链接" → `curl https://r.jina.ai/URL` 读任意网页
-- "这个 GitHub 仓库是做什么的" → GitHub 仓库、Issue、代码
+- "这个 GitHub 仓库是做什么的" → `gh repo view owner/repo`
-- "这个视频讲了什么" → YouTube / B站字幕提取
+- "这个视频讲了什么" → `yt-dlp --dump-json URL` 提取字幕
-- "帮我看看这条推文" → Twitter 推文
+- "帮我看看这条推文" → `bird read URL --json`
-- "订阅这个 RSS" → RSS / Atom 源
+- "订阅这个 RSS" → `feedparser` 解析
-- "搜一下 GitHub 上有什么 LLM 框架" → GitHub 搜索
+- "搜一下 GitHub 上有什么 LLM 框架" → `gh search repos "LLM framework"`
 
-**不需要记命令。** Agent 自己知道该调什么。
+**不需要记命令。** Agent 读了 SKILL.md 之后自己知道该调什么。
 
 ---
 
@@ -134,26 +134,30 @@ AI Agent 已经能帮你写代码、改文档、管项目——但你让它去
 
 Agent Reach 做的事情很简单：**帮你把这些选型和配置的活儿做完了。**
 
+安装完成后，Agent 直接调用上游工具（bird CLI、yt-dlp、mcporter、gh CLI 等），不需要经过 Agent Reach 的包装层。
+
 ### 🔌 每个渠道都是可插拔的
 
-每个平台对应一个独立的 Python 文件，实现统一接口。**后端工具随时可以换**——哪天出了更好的工具，改一个文件就行，其他不用动。
+每个平台背后是一个独立的上游工具。**不满意？换掉就行。**
 
 ```
 channels/
 ├── web.py          → Jina Reader     ← 可以换成 Firecrawl、Crawl4AI……
-├── twitter.py      → bird           ← 可以换成 Nitter、官方 API……
+├── twitter.py      → bird            ← 可以换成 Nitter、官方 API……
-├── youtube.py      → yt-dlp           ← 可以换成 YouTube API、Whisper……
+├── youtube.py      → yt-dlp          ← 可以换成 YouTube API、Whisper……
 ├── github.py       → gh CLI          ← 可以换成 REST API、PyGithub……
-├── bilibili.py     → yt-dlp           ← 可以换成 bilibili-api……
+├── bilibili.py     → yt-dlp          ← 可以换成 bilibili-api……
 ├── reddit.py       → JSON API + Exa  ← 可以换成 PRAW、Pushshift……
 ├── xiaohongshu.py  → mcporter MCP    ← 可以换成其他 XHS 工具……
 ├── linkedin.py     → linkedin-mcp    ← 可以换成 LinkedIn API……
 ├── bosszhipin.py   → mcp-bosszp      ← 可以换成其他招聘工具……
-├── rss.py          → feedparser       ← 可以换成 atoma……
+├── rss.py          → feedparser      ← 可以换成 atoma……
 ├── exa_search.py   → mcporter MCP    ← 可以换成 Tavily、SerpAPI……
-└── __init__.py     → 渠道注册
+└── __init__.py     → 渠道注册（doctor 检测用）
 ```
 
+每个渠道文件只负责检测对应上游工具是否可用（`check()` 方法），给 `agent-reach doctor` 提供状态信息。实际的读取和搜索由 Agent 直接调用上游工具完成。
+
 ### 当前选型
 
 | 场景 | 选型 | 为什么选它 |
@@ -229,13 +233,13 @@ Star 一下，下次需要的时候能找到。⭐
 <details>
 <summary><strong>AI Agent 怎么搜索 Twitter / X？不想付 API 费用</strong></summary>
 
-Agent Reach 使用 [bird CLI](https://www.npmjs.com/package/@steipete/bird) 通过 Cookie 认证访问 Twitter，完全免费。安装 Agent Reach 后，用 Cookie-Editor 导出你的 Twitter Cookie，运行 `agent-reach configure twitter-cookies "your_cookies"` 即可。之后 Agent 就可以用 `agent-reach search-twitter "关键词"` 搜索推文了。
+Agent Reach 使用 [bird CLI](https://www.npmjs.com/package/@steipete/bird) 通过 Cookie 认证访问 Twitter，完全免费。安装 Agent Reach 后，用 Cookie-Editor 导出你的 Twitter Cookie，运行 `agent-reach configure twitter-cookies "your_cookies"` 即可。之后 Agent 就可以用 `bird search "关键词" --json` 搜索推文了。
 </details>
 
 <details>
 <summary><strong>How to search Twitter/X with AI agent for free (no API)?</strong></summary>
 
-Agent Reach uses the bird CLI with cookie auth — zero API fees. After installing, export your Twitter cookies with the Cookie-Editor extension, run `agent-reach configure twitter-cookies "your_cookies"`, then your agent can search with `agent-reach search-twitter "query"`.
+Agent Reach uses the bird CLI with cookie auth — zero API fees. After installing, export your Twitter cookies with the Cookie-Editor extension, run `agent-reach configure twitter-cookies "your_cookies"`, then your agent can search with `bird search "query" --json`.
 </details>
 
 <details>
@@ -247,19 +251,19 @@ Reddit 封锁数据中心 IP。配置一个住宅代理即可解决：`agent-rea
 <details>
 <summary><strong>How to get YouTube video transcripts for AI?</strong></summary>
 
-`agent-reach read https://youtube.com/watch?v=xxx` automatically extracts the transcript. Uses yt-dlp under the hood, supports multiple languages. No API key needed.
+`yt-dlp --dump-json "https://youtube.com/watch?v=xxx"` extracts video metadata; `yt-dlp --write-sub --skip-download "URL"` extracts subtitles. Uses yt-dlp under the hood, supports multiple languages. No API key needed.
 </details>
 
 <details>
 <summary><strong>怎么让 AI Agent 读小红书？</strong></summary>
 
-小红书需要通过 Docker 运行一个 MCP 服务。安装 Docker 后，运行 `agent-reach install` 会自动配置。之后 Agent 就能用 `agent-reach read <小红书链接>` 或 `agent-reach search-xhs "关键词"` 了。
+小红书需要通过 Docker 运行一个 MCP 服务。安装 Docker 后，运行 `agent-reach install` 会自动配置。之后 Agent 就能用 `mcporter call 'xiaohongshu.get_feed_detail(...)'` 读取笔记或 `mcporter call 'xiaohongshu.search_feeds(keyword: "关键词")'` 搜索了。
 </details>
 
 <details>
 <summary><strong>Compatible with Claude Code / Cursor / OpenClaw / Windsurf?</strong></summary>
 
-Yes! Agent Reach is a standard CLI tool — any AI coding agent that can run shell commands can use it. Works with Claude Code, Cursor, OpenClaw, Windsurf, Codex, and more. Just `pip install agent-reach` and the agent can start using it immediately.
+Yes! Agent Reach is an installer + configuration tool — any AI coding agent that can run shell commands can use it. Works with Claude Code, Cursor, OpenClaw, Windsurf, Codex, and more. Just `pip install agent-reach`, run `agent-reach install`, and the agent can start using the upstream tools immediately.
 </details>
 
 <details>

agent_reach/channels/__init__.py

@@ -1,14 +1,10 @@
 # -*- coding: utf-8 -*-
 """
-Channel registry — routes URLs to the right channel.
+Channel registry — lists all supported platforms for doctor checks.
-
-This is the core of Agent Reach' pluggable architecture.
-Add a new channel: just create a file and register it here.
-Swap a backend: just change the implementation inside the channel file.
 """
 
-from typing import Dict, List, Optional
+from typing import List, Optional
-from .base import Channel, ReadResult, SearchResult
+from .base import Channel
 
 # Import all channels
 from .web import WebChannel
@@ -24,7 +20,7 @@ from .linkedin import LinkedInChannel
 from .bosszhipin import BossZhipinChannel
 
 
-# Channel registry — order matters (first match wins, web is last as fallback)
+# Channel registry
 ALL_CHANNELS: List[Channel] = [
     GitHubChannel(),
     TwitterChannel(),
@@ -36,22 +32,9 @@ ALL_CHANNELS: List[Channel] = [
     BossZhipinChannel(),
     RSSChannel(),
     ExaSearchChannel(),
-    WebChannel(),        # Fallback — handles any URL
+    WebChannel(),
 ]
 
-# Search-capable channels
-SEARCH_CHANNELS: Dict[str, Channel] = {
-    ch.name: ch for ch in ALL_CHANNELS if ch.can_search()
-}
-
-
-def get_channel_for_url(url: str) -> Channel:
-    """Find the right channel for a URL."""
-    for channel in ALL_CHANNELS:
-        if channel.can_handle(url):
-            return channel
-    return WebChannel()  # Should never reach here, but just in case
-
 
 def get_channel(name: str) -> Optional[Channel]:
     """Get a channel by name."""
@@ -67,7 +50,7 @@ def get_all_channels() -> List[Channel]:
 
 
 __all__ = [
-    "Channel", "ReadResult", "SearchResult",
+    "Channel",
-    "ALL_CHANNELS", "SEARCH_CHANNELS",
+    "ALL_CHANNELS",
-    "get_channel_for_url", "get_channel", "get_all_channels",
+    "get_channel", "get_all_channels",
 ]

agent_reach/channels/base.py

@@ -1,110 +1,28 @@
 # -*- coding: utf-8 -*-
 """
-Channel base class — the universal interface for all platforms.
+Channel base class — platform availability checking.
 
-Every channel (YouTube, Twitter, GitHub, etc.) implements this interface.
+Each channel represents a platform (YouTube, Twitter, GitHub, etc.)
-The backend tool can be swapped anytime without changing anything else.
+and provides:
+  - can_handle(url) → does this URL belong to this platform?
+  - check(config) → is the upstream tool installed and configured?
 
-Example:
+After installation, agents call upstream tools directly.
-    class YouTubeChannel(Channel):
-        name = "youtube"
-        backends = ["yt-dlp"]  # current backend, can be swapped
-        
-        async def read(self, url, config):
-            # Just call yt-dlp, return standardized dict
-            ...
 """
 
 import shutil
 from abc import ABC, abstractmethod
-from dataclasses import dataclass
+from typing import List, Tuple
-from typing import Any, Dict, List, Optional, Tuple
-
-
-@dataclass
-class ReadResult:
-    """Standardized read result. Every channel returns this."""
-    title: str
-    content: str
-    url: str
-    author: str = ""
-    date: str = ""
-    platform: str = ""
-    extra: dict = None
-
-    def __post_init__(self):
-        self.extra = self.extra or {}
-
-    def to_dict(self) -> dict:
-        d = {
-            "title": self.title,
-            "content": self.content,
-            "url": self.url,
-            "platform": self.platform,
-        }
-        if self.author:
-            d["author"] = self.author
-        if self.date:
-            d["date"] = self.date
-        if self.extra:
-            d["extra"] = self.extra
-        return d
-
-
-@dataclass
-class SearchResult:
-    """Standardized search result."""
-    title: str
-    url: str
-    snippet: str = ""
-    author: str = ""
-    date: str = ""
-    score: float = 0
-    extra: dict = None
-
-    def __post_init__(self):
-        self.extra = self.extra or {}
-
-    def to_dict(self) -> dict:
-        d = {
-            "title": self.title,
-            "url": self.url,
-            "snippet": self.snippet,
-        }
-        if self.author:
-            d["author"] = self.author
-        if self.date:
-            d["date"] = self.date
-        if self.extra:
-            d["extra"] = self.extra
-        return d
 
 
 class Channel(ABC):
-    """
+    """Base class for all channels."""
-    Base class for all channels.
-    
-    Subclasses just need to implement:
-    - read(url, config) → ReadResult
-    - can_handle(url) → bool
-    - check(config) → (status, message)
-    
-    Optionally:
-    - search(query, config, **kwargs) → list[SearchResult]
-    """
 
     name: str = ""                    # e.g. "youtube"
-    description: str = ""             # e.g. "YouTube video transcripts"
+    description: str = ""             # e.g. "YouTube 视频和字幕"
-    backends: List[str] = []          # e.g. ["yt-dlp"] — what external tool is used
+    backends: List[str] = []          # e.g. ["yt-dlp"] — what upstream tool is used
-    requires_config: List[str] = []   # e.g. ["reddit_proxy"]
-    requires_tools: List[str] = []    # e.g. ["yt-dlp"]
     tier: int = 0                     # 0=zero-config, 1=needs free key, 2=needs setup
 
-    @abstractmethod
-    async def read(self, url: str, config=None) -> ReadResult:
-        """Read content from a URL. Must return ReadResult."""
-        ...
-
     @abstractmethod
     def can_handle(self, url: str) -> bool:
         """Check if this channel can handle this URL."""
@@ -112,29 +30,7 @@ class Channel(ABC):
 
     def check(self, config=None) -> Tuple[str, str]:
         """
-        Check if this channel is available.
+        Check if this channel's upstream tool is available.
         Returns (status, message) where status is 'ok'/'warn'/'off'/'error'.
         """
-        # Check required tools
-        for tool in self.requires_tools:
-            if not shutil.which(tool):
-                return "off", f"需要安装：pip install {tool}"
-
-        # Check required config
-        for key in self.requires_config:
-            if config and not config.get(key):
-                return "off", f"需要配置 {key}，运行 agent-reach setup"
-
         return "ok", f"{'、'.join(self.backends) if self.backends else '内置'}"
-
-    async def search(self, query: str, config=None, **kwargs) -> List[SearchResult]:
-        """Search this platform. Override if supported."""
-        raise NotImplementedError(f"{self.name} does not support search")
-
-    def can_search(self) -> bool:
-        """Whether this channel supports search."""
-        try:
-            # Check if search is overridden
-            return type(self).search is not Channel.search
-        except:
-            return False

agent_reach/channels/bilibili.py

@@ -1,207 +1,26 @@
 # -*- coding: utf-8 -*-
-"""Bilibili — via yt-dlp (same backend as YouTube).
+"""Bilibili — check if yt-dlp is available."""
 
-Backend: yt-dlp (https://github.com/yt-dlp/yt-dlp)
+import os
-yt-dlp natively supports Bilibili — video info, subtitles, and search.
-"""
-
-import json
 import shutil
-import subprocess
+from .base import Channel
-from urllib.parse import urlparse
-from .base import Channel, ReadResult, SearchResult
-from typing import List
 
 
 class BilibiliChannel(Channel):
     name = "bilibili"
-    description = "B站视频信息和字幕"
+    description = "B站视频和字幕"
     backends = ["yt-dlp"]
-    requires_tools = ["yt-dlp"]
+    tier = 1
-    tier = 0
 
     def can_handle(self, url: str) -> bool:
+        from urllib.parse import urlparse
         d = urlparse(url).netloc.lower()
         return "bilibili.com" in d or "b23.tv" in d
 
     def check(self, config=None):
         if not shutil.which("yt-dlp"):
             return "off", "yt-dlp 未安装。安装：pip install yt-dlp"
-        proxy = config.get("bilibili_proxy") if config else None
+        proxy = (config.get("bilibili_proxy") if config else None) or os.environ.get("BILIBILI_PROXY")
         if proxy:
-            return "ok", "已配置代理，完整可用"
+            return "ok", "可提取视频信息和字幕（代理已配置）"
-        import os
+        return "ok", "可提取视频信息和字幕（本地环境）。服务器可能需要代理"
-        is_server = bool(os.environ.get("SSH_CONNECTION") or os.path.exists("/etc/cloud"))
-        if is_server:
-            return "warn", "服务器 IP 可能被封，配置代理即可解决：agent-reach configure proxy URL"
-        return "ok", "本地直连可用"
-
-    async def read(self, url: str, config=None) -> ReadResult:
-        if not shutil.which("yt-dlp"):
-            raise RuntimeError("yt-dlp not installed. Install: pip install yt-dlp")
-
-        proxy = config.get("bilibili_proxy") if config else None
-
-        # Get video info via yt-dlp
-        info = self._get_info(url, proxy)
-        if not info:
-            return ReadResult(
-                title="Bilibili",
-                content=f"⚠️ 无法获取视频信息: {url}\n服务器 IP 可能被封，配个代理：agent-reach configure proxy URL",
-                url=url, platform="bilibili",
-            )
-
-        title = info.get("title", url)
-        author = info.get("uploader", "")
-        desc = info.get("description", "")
-
-        # Try subtitles
-        subtitle = self._get_subtitles(url, proxy)
-        content = desc
-        if subtitle:
-            content += f"\n\n## 字幕\n{subtitle}"
-
-        return ReadResult(
-            title=title, content=content, url=url,
-            author=author, platform="bilibili",
-            extra={
-                "view_count": info.get("view_count"),
-                "like_count": info.get("like_count"),
-                "duration": info.get("duration_string"),
-            },
-        )
-
-    async def search(self, query: str, config=None, **kwargs) -> List[SearchResult]:
-        """Search Bilibili.
-
-        Strategy:
-        1. Try yt-dlp bilisearch (works on local machines)
-        2. Fallback to Exa site:bilibili.com (works on servers)
-        """
-        if not shutil.which("yt-dlp"):
-            raise RuntimeError("yt-dlp not installed. Install: pip install yt-dlp")
-
-        limit = kwargs.get("limit", 5)
-        proxy = config.get("bilibili_proxy") if config else None
-
-        # Strategy 1: yt-dlp bilisearch
-        results = self._search_ytdlp(query, limit, proxy)
-        if results:
-            return results
-
-        # Strategy 2: Exa fallback (server-friendly)
-        results = self._search_exa(query, limit)
-        if results:
-            return results
-
-        return []
-
-    def _search_ytdlp(self, query: str, limit: int, proxy: str = None) -> List[SearchResult]:
-        """Search via yt-dlp bilisearch (needs local/Chinese IP)."""
-        cmd = [
-            "yt-dlp", "--dump-json", "--no-download",
-            f"bilisearch{limit}:{query}",
-        ]
-        if proxy:
-            cmd += ["--proxy", proxy]
-
-        try:
-            r = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
-            if r.returncode != 0:
-                return []
-            results = []
-            for line in r.stdout.strip().split("\n"):
-                if not line.strip():
-                    continue
-                try:
-                    d = json.loads(line)
-                    vid = d.get("id", "")
-                    url = d.get("webpage_url", f"https://www.bilibili.com/video/av{vid}")
-                    results.append(SearchResult(
-                        title=d.get("title", f"av{vid}"),
-                        url=url,
-                        snippet=f"👤 {d.get('uploader', '?')} · 👁 {d.get('view_count', '?')}",
-                        extra={
-                            "view_count": d.get("view_count"),
-                            "uploader": d.get("uploader"),
-                            "duration": d.get("duration_string"),
-                        },
-                    ))
-                except json.JSONDecodeError:
-                    continue
-            return results
-        except subprocess.TimeoutExpired:
-            return []
-
-    def _search_exa(self, query: str, limit: int) -> List[SearchResult]:
-        """Fallback: search via Exa (site:bilibili.com). Works on any IP."""
-        try:
-            r = subprocess.run(
-                ["mcporter", "call",
-                 f'exa.web_search_exa(query: "site:bilibili.com {query}", numResults: {limit})'],
-                capture_output=True, text=True, timeout=30,
-            )
-            if r.returncode != 0:
-                return []
-
-            results = []
-            # Parse mcporter output: Title: / Author: / URL: / Text: blocks
-            title, author, url = "", "", ""
-            for line in r.stdout.split("\n"):
-                if line.startswith("Title: "):
-                    title = line[7:].strip()
-                elif line.startswith("Author: "):
-                    author = line[8:].strip()
-                elif line.startswith("URL: "):
-                    url = line[5:].strip()
-                    if url and "bilibili.com" in url:
-                        results.append(SearchResult(
-                            title=title or url,
-                            url=url,
-                            snippet=f"👤 {author}" if author else "(via Exa search)",
-                        ))
-                    title, author, url = "", "", ""
-            return results
-        except Exception:
-            return []
-
-    def _get_info(self, url: str, proxy: str = None) -> dict:
-        cmd = ["yt-dlp", "--dump-json", "--no-download", url]
-        if proxy:
-            cmd += ["--proxy", proxy]
-        try:
-            r = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
-            if r.returncode == 0:
-                return json.loads(r.stdout)
-        except (subprocess.TimeoutExpired, json.JSONDecodeError):
-            pass
-        return {}
-
-    def _get_subtitles(self, url: str, proxy: str = None) -> str:
-        import tempfile
-        from pathlib import Path
-
-        with tempfile.TemporaryDirectory() as tmpdir:
-            cmd = [
-                "yt-dlp", "--write-sub", "--write-auto-sub",
-                "--sub-lang", "zh-Hans,zh,en",
-                "--skip-download", "--sub-format", "vtt",
-                "-o", f"{tmpdir}/%(id)s.%(ext)s", url,
-            ]
-            if proxy:
-                cmd += ["--proxy", proxy]
-            try:
-                subprocess.run(cmd, capture_output=True, text=True, timeout=30)
-                for f in Path(tmpdir).glob("*.vtt"):
-                    text = f.read_text(errors="replace")
-                    lines = []
-                    for line in text.split("\n"):
-                        line = line.strip()
-                        if not line or line.startswith("WEBVTT") or "-->" in line or line.isdigit():
-                            continue
-                        if line not in lines[-1:]:
-                            lines.append(line)
-                    return "\n".join(lines)
-            except subprocess.TimeoutExpired:
-                pass
-        return ""

agent_reach/channels/bosszhipin.py

@@ -1,62 +1,9 @@
 # -*- coding: utf-8 -*-
-"""Boss直聘 (BOSS Zhipin) — via mcp-bosszp (MCP) or Jina Reader fallback.
+"""Boss直聘 — check if mcp-bosszp is available."""
 
-Backend: mcp-bosszp (161 stars, FastMCP + Playwright)
-Swap to: any Boss直聘 access tool
-"""
-
-import json
 import shutil
 import subprocess
-from urllib.parse import urlparse
+from .base import Channel
-from .base import Channel, ReadResult, SearchResult
-from typing import List
-import requests
-
-
-def _mcporter_has_bosszhipin() -> bool:
-    """Check if mcporter has Boss直聘 MCP configured."""
-    if not shutil.which("mcporter"):
-        return False
-    try:
-        r = subprocess.run(
-            ["mcporter", "list"], capture_output=True, text=True, timeout=10
-        )
-        # Check for various possible config names
-        out = r.stdout.lower()
-        return "boss" in out or "zhipin" in out or "bosszhipin" in out
-    except Exception:
-        return False
-
-
-def _mcporter_call(expr: str, timeout: int = 30) -> str:
-    """Call a Boss直聘 MCP tool via mcporter."""
-    r = subprocess.run(
-        ["mcporter", "call", expr],
-        capture_output=True, text=True, timeout=timeout,
-    )
-    if r.returncode != 0:
-        raise RuntimeError(r.stderr or r.stdout)
-    return r.stdout
-
-
-def _get_mcp_name() -> str:
-    """Get the actual MCP server name configured in mcporter."""
-    try:
-        r = subprocess.run(
-            ["mcporter", "list"], capture_output=True, text=True, timeout=10
-        )
-        for line in r.stdout.split("\n"):
-            line_lower = line.strip().lower()
-            for name in ["bosszhipin", "boss-zp", "bosszp", "boss"]:
-                if name in line_lower:
-                    # Extract the actual server name
-                    parts = line.strip().split()
-                    if parts:
-                        return parts[0]
-        return "bosszhipin"
-    except Exception:
-        return "bosszhipin"
 
 
 class BossZhipinChannel(Channel):
@@ -66,118 +13,29 @@ class BossZhipinChannel(Channel):
     tier = 2
 
     def can_handle(self, url: str) -> bool:
+        from urllib.parse import urlparse
         domain = urlparse(url).netloc.lower()
         return "zhipin.com" in domain or "boss.com" in domain
 
     def check(self, config=None):
-        if _mcporter_has_bosszhipin():
+        if not shutil.which("mcporter"):
-            return "ok", "可搜索职位、向 HR 打招呼"
+            return "off", (
-
+                "可通过 Jina Reader 读取职位页面。完整功能需要：\n"
+                "  1. git clone https://github.com/mucsbr/mcp-bosszp.git\n"
+                "  2. cd mcp-bosszp && pip install -r requirements.txt && playwright install chromium\n"
+                "  3. python boss_zhipin_fastmcp_v2.py（启动后扫码登录）\n"
+                "  4. mcporter config add bosszhipin http://localhost:8000/mcp"
+            )
+        try:
+            r = subprocess.run(
+                ["mcporter", "list"], capture_output=True, text=True, timeout=10
+            )
+            out = r.stdout.lower()
+            if "boss" in out or "zhipin" in out:
+                return "ok", "可搜索职位、向 HR 打招呼"
+        except Exception:
+            pass
         return "off", (
-            "可通过 Jina Reader 读取职位页面。完整功能需要：\n"
+            "mcporter 已装但 Boss直聘 MCP 未配置。\n"
-            "  1. git clone https://github.com/mucsbr/mcp-bosszp.git\n"
-            "  2. cd mcp-bosszp && pip install -r requirements.txt && playwright install chromium\n"
-            "  3. python boss_zhipin_fastmcp_v2.py（启动后扫码登录）\n"
-            "  4. mcporter config add bosszhipin http://localhost:8000/mcp\n"
-            "  或用 Docker：docker-compose up -d\n"
             "  详见 https://github.com/mucsbr/mcp-bosszp"
         )
-
-    async def read(self, url: str, config=None) -> ReadResult:
-        # Boss直聘 pages mostly work with Jina Reader
-        return await self._read_jina(url)
-
-    async def _read_jina(self, url: str) -> ReadResult:
-        """Read Boss直聘 page via Jina Reader."""
-        try:
-            resp = requests.get(
-                f"https://r.jina.ai/{url}",
-                headers={"Accept": "text/markdown"},
-                timeout=15,
-            )
-            resp.raise_for_status()
-            text = resp.text
-
-            if len(text.strip()) < 50:
-                return ReadResult(
-                    title="Boss直聘",
-                    content=(
-                        f"⚠️ 无法读取此页面内容: {url}\n\n"
-                        "提示：\n"
-                        "- 安装 mcp-bosszp 可解锁职位搜索和自动打招呼\n"
-                        "- 详见 https://github.com/mucsbr/mcp-bosszp"
-                    ),
-                    url=url,
-                    platform="bosszhipin",
-                )
-
-            return ReadResult(
-                title=text[:100] if text else url,
-                content=text,
-                url=url,
-                platform="bosszhipin",
-            )
-        except Exception:
-            return ReadResult(
-                title="Boss直聘",
-                content=(
-                    f"⚠️ 无法读取此 Boss直聘页面: {url}\n\n"
-                    "提示：\n"
-                    "- Boss直聘部分页面需要登录\n"
-                    "- 安装 mcp-bosszp 可解锁完整功能\n"
-                    "- 详见 https://github.com/mucsbr/mcp-bosszp"
-                ),
-                url=url,
-                platform="bosszhipin",
-            )
-
-    async def search(self, query: str, config=None, **kwargs) -> List[SearchResult]:
-        limit = kwargs.get("limit", 10)
-
-        # Try MCP search first
-        if _mcporter_has_bosszhipin():
-            try:
-                return await self._search_mcp(query, limit, config)
-            except Exception:
-                pass
-
-        # Fallback to Exa
-        from agent_reach.channels.exa_search import ExaSearchChannel
-        exa = ExaSearchChannel()
-        return await exa.search(f"site:zhipin.com {query}", config=config, limit=limit)
-
-    async def _search_mcp(self, query: str, limit: int, config=None) -> List[SearchResult]:
-        """Search Boss直聘 via MCP."""
-        server = _get_mcp_name()
-        try:
-            out = _mcporter_call(
-                f'{server}.get_recommend_jobs_tool(page: 1)',
-                timeout=30,
-            )
-            return self._parse_jobs(out, limit)
-        except Exception:
-            return []
-
-    def _parse_jobs(self, text: str, limit: int) -> List[SearchResult]:
-        """Parse MCP job search output into SearchResults."""
-        results = []
-        try:
-            data = json.loads(text)
-            jobs = data if isinstance(data, list) else data.get("jobs", data.get("results", []))
-            for job in jobs[:limit]:
-                if isinstance(job, dict):
-                    title = job.get("title") or job.get("jobName", "")
-                    company = job.get("company") or job.get("brandName", "")
-                    salary = job.get("salary") or job.get("salaryDesc", "")
-                    url = job.get("url", "")
-                    snippet = f"🏢 {company}" if company else ""
-                    if salary:
-                        snippet += f" · 💰 {salary}"
-                    results.append(SearchResult(
-                        title=title,
-                        url=url,
-                        snippet=snippet,
-                    ))
-        except (json.JSONDecodeError, KeyError):
-            pass
-        return results

agent_reach/channels/exa_search.py

@@ -1,110 +1,36 @@
 # -*- coding: utf-8 -*-
-"""Exa semantic search — via mcporter + Exa MCP server.
+"""Exa Search — check if mcporter + Exa MCP is available."""
 
-Backend: Exa MCP at mcp.exa.ai (OAuth, no API key needed)
-Requires: mcporter CLI
-"""
-
-import json
 import shutil
 import subprocess
-from .base import Channel, SearchResult
+from .base import Channel
-from typing import List
 
 
 class ExaSearchChannel(Channel):
     name = "exa_search"
-    description = "全网语义搜索（同时支持 Reddit/Twitter 搜索）"
+    description = "全网语义搜索"
-    backends = ["exa-mcp"]
+    backends = ["Exa via mcporter"]
-    tier = 1
+    tier = 0
-
-    def _mcporter_ok(self) -> bool:
-        if not shutil.which("mcporter"):
-            return False
-        try:
-            r = subprocess.run(
-                ["mcporter", "list"], capture_output=True, text=True, timeout=10
-            )
-            return "exa" in r.stdout
-        except Exception:
-            return False
-
-    def _call(self, expr: str, timeout: int = 30) -> str:
-        r = subprocess.run(
-            ["mcporter", "call", expr],
-            capture_output=True, text=True, timeout=timeout,
-        )
-        if r.returncode != 0:
-            raise RuntimeError(r.stderr or r.stdout)
-        return r.stdout
-
-    # ── Channel interface ──
 
     def can_handle(self, url: str) -> bool:
-        return False  # search-only
+        return False  # Search-only channel
-
-    async def read(self, url: str, config=None):
-        raise NotImplementedError("Exa is a search engine, not a reader")
 
     def check(self, config=None):
         if not shutil.which("mcporter"):
             return "off", (
-                "需要 mcporter。安装：npm install -g mcporter && "
+                "需要 mcporter + Exa MCP。安装：\n"
-                "mcporter config add exa https://mcp.exa.ai/mcp"
-            )
-        if not self._mcporter_ok():
-            return "off", "mcporter 已装但 Exa 未配置。运行：mcporter config add exa https://mcp.exa.ai/mcp"
-        return "ok", "MCP 已连接，免 Key 直接可用（全网搜索 + Reddit + Twitter）"
-
-    async def search(self, query: str, config=None, **kwargs) -> List[SearchResult]:
-        if not self._mcporter_ok():
-            raise ValueError(
-                "Exa 搜索需要 mcporter。安装：\n"
                 "  npm install -g mcporter\n"
                 "  mcporter config add exa https://mcp.exa.ai/mcp"
             )
-
+        try:
-        limit = kwargs.get("limit", 5)
+            r = subprocess.run(
-        safe_q = query.replace('"', '\\"')
+                ["mcporter", "list"], capture_output=True, text=True, timeout=10
-        out = self._call(
+            )
-            f'exa.web_search_exa(query: "{safe_q}", numResults: {min(limit, 10)})',
+            if "exa" in r.stdout.lower():
-            timeout=30,
+                return "ok", "全网语义搜索可用（免费，无需 API Key）"
-        )
+            return "off", (
-        return self._parse_output(out, limit)
+                "mcporter 已装但 Exa 未配置。运行：\n"
-
+                "  mcporter config add exa https://mcp.exa.ai/mcp"
-    # ── Parse mcporter text output ──
+            )
-
+        except Exception:
-    def _parse_output(self, text: str, limit: int) -> List[SearchResult]:
+            return "off", "mcporter 连接异常"
-        """Parse mcporter's Title/URL/Text block format."""
-        results = []
-        cur = {}
-
-        for line in text.split("\n"):
-            line = line.strip()
-            if line.startswith("Title: "):
-                if cur.get("title"):
-                    results.append(self._make_result(cur))
-                cur = {"title": line[7:]}
-            elif line.startswith("URL: "):
-                cur["url"] = line[5:]
-            elif line.startswith("Published Date: "):
-                cur["date"] = line[16:]
-            elif line.startswith("Text: "):
-                cur["text"] = line[6:]
-            elif "text" in cur and line:
-                cur["text"] += " " + line
-
-        if cur.get("title"):
-            results.append(self._make_result(cur))
-
-        return results[:limit]
-
-    @staticmethod
-    def _make_result(d: dict) -> SearchResult:
-        return SearchResult(
-            title=d.get("title", ""),
-            url=d.get("url", ""),
-            snippet=d.get("text", "")[:500],
-            date=d.get("date", ""),
-            score=0,
-        )

agent_reach/channels/github.py

@@ -1,16 +1,9 @@
 # -*- coding: utf-8 -*-
-"""GitHub — via gh CLI.
+"""GitHub — check if gh CLI is available."""
 
-Backend: gh CLI (https://cli.github.com)
-Swap to: GitHub REST API
-"""
-
-import json
 import shutil
 import subprocess
-from urllib.parse import urlparse
+from .base import Channel
-from .base import Channel, ReadResult, SearchResult
-from typing import List
 
 
 class GitHubChannel(Channel):
@@ -19,121 +12,18 @@ class GitHubChannel(Channel):
     backends = ["gh CLI"]
     tier = 0
 
-    def _gh(self, args: list, timeout: int = 15) -> str:
-        r = subprocess.run(
-            ["gh"] + args,
-            capture_output=True, text=True, timeout=timeout,
-        )
-        if r.returncode != 0:
-            raise RuntimeError(r.stderr or r.stdout)
-        return r.stdout
-
-    def _gh_json(self, args: list, timeout: int = 15) -> dict:
-        return json.loads(self._gh(args + ["--json"], timeout))
-
     def can_handle(self, url: str) -> bool:
+        from urllib.parse import urlparse
         return "github.com" in urlparse(url).netloc.lower()
 
     def check(self, config=None):
         if not shutil.which("gh"):
-            return "warn", "gh CLI 未安装。安装：https://cli.github.com 。公开仓库仍可通过 Jina Reader 读取"
+            return "warn", "gh CLI 未安装。安装：https://cli.github.com"
         try:
-            self._gh(["auth", "status"], timeout=5)
+            subprocess.run(
+                ["gh", "auth", "status"],
+                capture_output=True, text=True, timeout=5
+            )
             return "ok", "完整可用（读取、搜索、Fork、Issue、PR 等）"
         except Exception:
             return "ok", "gh CLI 已装但未认证。运行 gh auth login 可解锁完整功能"
-
-    async def read(self, url: str, config=None) -> ReadResult:
-        if not shutil.which("gh"):
-            # Fallback to Jina Reader for public repos
-            from agent_reach.channels.web import WebChannel
-            return await WebChannel().read(url, config)
-
-        path = urlparse(url).path.strip("/").split("/")
-        if len(path) < 2:
-            from agent_reach.channels.web import WebChannel
-            return await WebChannel().read(url, config)
-
-        owner, repo = path[0], path[1]
-
-        # Issues / PRs
-        if len(path) >= 4 and path[2] in ("issues", "pull"):
-            return await self._read_issue(owner, repo, path[3], url)
-
-        # Repo
-        return await self._read_repo(owner, repo, url)
-
-    async def _read_repo(self, owner: str, repo: str, url: str) -> ReadResult:
-        slug = f"{owner}/{repo}"
-        try:
-            # Get repo info
-            info = self._gh(["repo", "view", slug])
-            # Get README
-            try:
-                readme = self._gh(
-                    ["api", f"repos/{slug}/readme", "--jq", ".content"],
-                    timeout=10,
-                )
-                import base64
-                readme_text = base64.b64decode(readme).decode("utf-8", errors="replace")
-            except Exception:
-                readme_text = ""
-
-            content = readme_text or info
-            return ReadResult(
-                title=slug, content=content, url=url,
-                author=owner, platform="github",
-            )
-        except Exception:
-            from agent_reach.channels.web import WebChannel
-            return await WebChannel().read(url)
-
-    async def _read_issue(self, owner: str, repo: str, num: str, url: str) -> ReadResult:
-        slug = f"{owner}/{repo}"
-        try:
-            out = self._gh(["issue", "view", num, "-R", slug])
-            return ReadResult(
-                title=f"{slug}#{num}", content=out, url=url,
-                platform="github",
-            )
-        except Exception:
-            # Might be a PR
-            try:
-                out = self._gh(["pr", "view", num, "-R", slug])
-                return ReadResult(
-                    title=f"{slug}#{num}", content=out, url=url,
-                    platform="github",
-                )
-            except Exception:
-                from agent_reach.channels.web import WebChannel
-                return await WebChannel().read(url)
-
-    async def search(self, query: str, config=None, **kwargs) -> List[SearchResult]:
-        if not shutil.which("gh"):
-            raise ValueError("GitHub search requires gh CLI. Install: https://cli.github.com")
-
-        language = kwargs.get("language")
-        limit = kwargs.get("limit", 5)
-
-        args = ["search", "repos", query, "--sort", "stars", f"--limit={limit}"]
-        if language:
-            args += [f"--language={language}"]
-
-        out = self._gh(args, timeout=15)
-        results = []
-        for line in out.strip().split("\n"):
-            if not line.strip():
-                continue
-            parts = line.split("\t")
-            if len(parts) >= 1:
-                slug = parts[0].strip()
-                desc = parts[1].strip() if len(parts) > 1 else ""
-                stars = parts[3].strip() if len(parts) > 3 else ""
-                lang = parts[5].strip() if len(parts) > 5 else ""
-                results.append(SearchResult(
-                    title=slug,
-                    url=f"https://github.com/{slug}",
-                    snippet=desc,
-                    extra={"stars": stars, "language": lang},
-                ))
-        return results

agent_reach/channels/linkedin.py

@@ -1,268 +1,39 @@
 # -*- coding: utf-8 -*-
-"""LinkedIn — via linkedin-scraper-mcp (MCP) or Jina Reader fallback.
+"""LinkedIn — check if linkedin-scraper-mcp is available."""
-
-Backend: linkedin-scraper-mcp (916 stars, Patchright browser automation)
-Swap to: any LinkedIn access tool
-"""
 
 import shutil
 import subprocess
-from urllib.parse import urlparse
+from .base import Channel
-from .base import Channel, ReadResult, SearchResult
-from typing import List
-import requests
-
-
-def _mcporter_has_linkedin() -> bool:
-    """Check if mcporter has linkedin MCP configured."""
-    if not shutil.which("mcporter"):
-        return False
-    try:
-        r = subprocess.run(
-            ["mcporter", "list"], capture_output=True, text=True, timeout=10
-        )
-        return "linkedin" in r.stdout.lower()
-    except Exception:
-        return False
-
-
-def _mcporter_call(expr: str, timeout: int = 30) -> str:
-    """Call a LinkedIn MCP tool via mcporter."""
-    r = subprocess.run(
-        ["mcporter", "call", expr],
-        capture_output=True, text=True, timeout=timeout,
-    )
-    if r.returncode != 0:
-        raise RuntimeError(r.stderr or r.stdout)
-    return r.stdout
 
 
 class LinkedInChannel(Channel):
     name = "linkedin"
-    description = "LinkedIn 个人/公司 Profile 和职位"
+    description = "LinkedIn 职业社交"
     backends = ["linkedin-scraper-mcp", "Jina Reader"]
     tier = 2
 
     def can_handle(self, url: str) -> bool:
-        domain = urlparse(url).netloc.lower()
+        from urllib.parse import urlparse
-        return "linkedin.com" in domain
+        return "linkedin.com" in urlparse(url).netloc.lower()
 
     def check(self, config=None):
-        if _mcporter_has_linkedin():
+        if not shutil.which("mcporter"):
-            return "ok", "完整可用（Profile、公司、职位搜索）"
+            return "off", (
-
+                "基本内容可通过 Jina Reader 读取。完整功能需要：\n"
-        # Check if linkedin-scraper-mcp is installed as CLI
+                "  pip install linkedin-scraper-mcp\n"
-        if shutil.which("linkedin-scraper-mcp"):
+                "  mcporter config add linkedin http://localhost:3000/mcp\n"
-            return "warn", (
+                "  详见 https://github.com/stickerdaniel/linkedin-mcp-server"
-                "linkedin-scraper-mcp 已安装但未接入 mcporter。运行：\n"
-                "  1. linkedin-scraper-mcp --login（在有浏览器的机器上登录）\n"
-                "  2. linkedin-scraper-mcp --transport streamable-http --port 8001\n"
-                "  3. mcporter config add linkedin http://localhost:8001/mcp"
             )
-
+        try:
+            r = subprocess.run(
+                ["mcporter", "list"], capture_output=True, text=True, timeout=10
+            )
+            if "linkedin" in r.stdout.lower():
+                return "ok", "完整可用（Profile、公司、职位搜索）"
+        except Exception:
+            pass
         return "off", (
-            "可通过 Jina Reader 读取部分内容。完整功能需要：\n"
+            "mcporter 已装但 LinkedIn MCP 未配置。运行：\n"
-            "  1. pip install linkedin-scraper-mcp\n"
+            "  pip install linkedin-scraper-mcp\n"
-            "  2. linkedin-scraper-mcp --login（在有浏览器的机器上登录）\n"
+            "  mcporter config add linkedin http://localhost:3000/mcp"
-            "  3. linkedin-scraper-mcp --transport streamable-http --port 8001\n"
-            "  4. mcporter config add linkedin http://localhost:8001/mcp\n"
-            "  详见 https://github.com/stickerdaniel/linkedin-mcp-server"
         )
-
-    async def read(self, url: str, config=None) -> ReadResult:
-        path = urlparse(url).path.strip("/")
-
-        # Try MCP first
-        if _mcporter_has_linkedin():
-            try:
-                if "/in/" in url:
-                    return await self._read_profile_mcp(url)
-                elif "/company/" in url:
-                    return await self._read_company_mcp(url)
-                elif "/jobs/view/" in url:
-                    return await self._read_job_mcp(url)
-            except Exception:
-                pass  # Fall through to Jina
-
-        # Fallback: Jina Reader
-        return await self._read_jina(url)
-
-    async def _read_profile_mcp(self, url: str) -> ReadResult:
-        """Read a LinkedIn profile via MCP."""
-        import re
-        # Extract username from URL: /in/username/
-        match = re.search(r"/in/([^/]+)", url)
-        if not match:
-            return await self._read_jina(url)
-        username = match.group(1)
-        safe_username = username.replace('"', '\\"')
-        out = _mcporter_call(
-            f'linkedin.get_person_profile(linkedin_username: "{safe_username}")',
-            timeout=60,
-        )
-        return ReadResult(
-            title=self._extract_title(out) or f"LinkedIn Profile - {username}",
-            content=out.strip(),
-            url=url,
-            platform="linkedin",
-        )
-
-    async def _read_company_mcp(self, url: str) -> ReadResult:
-        """Read a LinkedIn company page via MCP."""
-        import re
-        # Extract company name from URL: /company/name/
-        match = re.search(r"/company/([^/]+)", url)
-        if not match:
-            return await self._read_jina(url)
-        company = match.group(1)
-        safe_company = company.replace('"', '\\"')
-        out = _mcporter_call(
-            f'linkedin.get_company_profile(company_name: "{safe_company}")',
-            timeout=60,
-        )
-        return ReadResult(
-            title=self._extract_title(out) or "LinkedIn Company",
-            content=out.strip(),
-            url=url,
-            platform="linkedin",
-        )
-
-    async def _read_job_mcp(self, url: str) -> ReadResult:
-        """Read a LinkedIn job posting via MCP."""
-        import re
-        match = re.search(r"/jobs/view/(\d+)", url)
-        if not match:
-            return await self._read_jina(url)
-
-        job_id = match.group(1)
-        out = _mcporter_call(
-            f'linkedin.get_job_details(job_id: "{job_id}")',
-            timeout=30,
-        )
-        return ReadResult(
-            title=self._extract_title(out) or f"LinkedIn Job {job_id}",
-            content=out.strip(),
-            url=url,
-            platform="linkedin",
-        )
-
-    async def _read_jina(self, url: str) -> ReadResult:
-        """Fallback: use Jina Reader."""
-        try:
-            resp = requests.get(
-                f"https://r.jina.ai/{url}",
-                headers={"Accept": "text/markdown"},
-                timeout=15,
-            )
-            resp.raise_for_status()
-            text = resp.text
-
-            # Check if content is usable
-            if len(text.strip()) < 100 or "Sign in" in text[:200]:
-                return ReadResult(
-                    title="LinkedIn",
-                    content=(
-                        f"⚠️ LinkedIn 页面需要登录才能完整查看。\n\n"
-                        f"URL: {url}\n\n"
-                        "完整功能需安装 linkedin-scraper-mcp：\n"
-                        "  pip install linkedin-scraper-mcp\n"
-                        "  uvx linkedin-scraper-mcp --login\n"
-                        "  详见 https://github.com/stickerdaniel/linkedin-mcp-server"
-                    ),
-                    url=url,
-                    platform="linkedin",
-                )
-
-            return ReadResult(
-                title=text[:100] if text else url,
-                content=text,
-                url=url,
-                platform="linkedin",
-            )
-        except Exception:
-            return ReadResult(
-                title="LinkedIn",
-                content=(
-                    f"⚠️ 无法读取此 LinkedIn 页面: {url}\n\n"
-                    "提示：\n"
-                    "- LinkedIn 需要登录才能查看大部分内容\n"
-                    "- 安装 linkedin-scraper-mcp 解锁完整功能\n"
-                    "- 详见 https://github.com/stickerdaniel/linkedin-mcp-server"
-                ),
-                url=url,
-                platform="linkedin",
-            )
-
-    async def search(self, query: str, config=None, **kwargs) -> List[SearchResult]:
-        limit = kwargs.get("limit", 10)
-
-        # Try MCP search first
-        if _mcporter_has_linkedin():
-            try:
-                return await self._search_mcp(query, limit)
-            except Exception:
-                pass
-
-        # Fallback to Exa
-        from agent_reach.channels.exa_search import ExaSearchChannel
-        exa = ExaSearchChannel()
-        return await exa.search(f"site:linkedin.com {query}", config=config, limit=limit)
-
-    async def _search_mcp(self, query: str, limit: int) -> List[SearchResult]:
-        """Search LinkedIn via MCP."""
-        safe_q = query.replace('"', '\\"')
-        # Try job search first (most common use case)
-        try:
-            out = _mcporter_call(
-                f'linkedin.search_jobs(keywords: "{safe_q}")',
-                timeout=60,
-            )
-            results = self._parse_search_results(out, "job")
-            if results:
-                return results[:limit]
-        except Exception:
-            pass
-
-        # Try people search
-        try:
-            out = _mcporter_call(
-                f'linkedin.search_people(keywords: "{safe_q}")',
-                timeout=60,
-            )
-            results = self._parse_search_results(out, "people")
-            if results:
-                return results
-        except Exception:
-            pass
-
-        return []
-
-    def _parse_search_results(self, text: str, result_type: str) -> List[SearchResult]:
-        """Parse MCP search output into SearchResults."""
-        import json
-        results = []
-        try:
-            data = json.loads(text)
-            items = data if isinstance(data, list) else data.get("results", data.get("jobs", []))
-            for item in items:
-                if isinstance(item, dict):
-                    title = item.get("title") or item.get("name") or item.get("headline", "")
-                    url = item.get("url") or item.get("link", "")
-                    snippet = item.get("description") or item.get("company", "")
-                    results.append(SearchResult(
-                        title=title,
-                        url=url,
-                        snippet=snippet[:200] if snippet else "",
-                    ))
-        except (json.JSONDecodeError, KeyError):
-            # Try line-by-line parsing
-            pass
-        return results
-
-    def _extract_title(self, text: str) -> str:
-        """Extract a title from MCP output."""
-        for line in text.split("\n"):
-            line = line.strip()
-            if line and not line.startswith(("{", "[", "#", "http")):
-                return line[:80]
-        return ""

agent_reach/channels/reddit.py

@@ -1,178 +1,26 @@
 # -*- coding: utf-8 -*-
-"""Reddit — via Reddit JSON API + optional proxy.
+"""Reddit — check if proxy and credentials are configured."""
-
-Backend: Reddit public JSON API (append .json to any URL)
-Swap to: any Reddit access method
-"""
 
 import os
-import requests
+from .base import Channel
-from urllib.parse import urlparse
-from .base import Channel, ReadResult
 
 
 class RedditChannel(Channel):
     name = "reddit"
     description = "Reddit 帖子和评论"
-    backends = ["Reddit JSON API"]
+    backends = ["JSON API", "Exa"]
-    tier = 2
+    tier = 1
-
-    USER_AGENT = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
 
     def can_handle(self, url: str) -> bool:
-        domain = urlparse(url).netloc.lower()
+        from urllib.parse import urlparse
-        return "reddit.com" in domain or "redd.it" in domain
+        d = urlparse(url).netloc.lower()
+        return "reddit.com" in d or "redd.it" in d
 
     def check(self, config=None):
-        proxy = config.get("reddit_proxy") if config else None
+        proxy = (config.get("reddit_proxy") if config else None) or os.environ.get("REDDIT_PROXY")
-        has_bot = bool(os.environ.get("REDDIT_CLIENT_ID"))
+        if proxy:
-        if proxy and has_bot:
+            return "ok", "代理已配置，可读取帖子。搜索走 Exa"
-            return "ok", "完整可用（代理 + OAuth Bot）"
+        return "warn", (
-        elif proxy:
+            "无代理。服务器 IP 可能被 Reddit 封锁。配置代理：\n"
-            return "ok", "代理已配置，可读取帖子。配置 REDDIT_CLIENT_ID/SECRET 可解锁高级搜索和发帖"
+            "  agent-reach configure proxy http://user:pass@ip:port"
-        elif has_bot:
-            return "warn", "OAuth Bot 已配置，但服务器直连可能被封。配个代理更稳定：agent-reach configure proxy URL"
-        else:
-            return "off", "搜索用 Exa 免费可用。读帖子需配个代理：agent-reach configure proxy URL"
-
-    async def read(self, url: str, config=None) -> ReadResult:
-        proxy = config.get("reddit_proxy") if config else None
-        proxies = {"http": proxy, "https": proxy} if proxy else None
-
-        # Clean URL: remove query params, trailing slash, then add .json
-        parsed = urlparse(url)
-        clean_path = parsed.path.rstrip("/")
-        # Remove trailing .json if already present (avoid double .json)
-        if clean_path.endswith(".json"):
-            clean_path = clean_path[:-5]
-        json_url = f"https://www.reddit.com{clean_path}.json"
-
-        try:
-            resp = requests.get(
-                json_url,
-                headers={"User-Agent": self.USER_AGENT},
-                proxies=proxies,
-                params={"limit": 50},
-                timeout=15,
-            )
-            resp.raise_for_status()
-        except requests.exceptions.HTTPError as e:
-            status = e.response.status_code if e.response is not None else 0
-            if status in (403, 429):
-                return ReadResult(
-                    title="Reddit",
-                    content="⚠️ Reddit blocked this request (403 Forbidden). "
-                            "Reddit blocks most server IPs.\n"
-                            "Fix: agent-reach configure proxy http://user:pass@ip:port\n"
-                            "Cheap option: https://www.webshare.io ($1/month)\n\n"
-                            "Alternatively, search Reddit via Exa (free, no proxy needed): "
-                            "agent-reach search-reddit \"your query\"",
-                    url=url,
-                    platform="reddit",
-                )
-            raise
-
-        data = resp.json()
-
-        # Subreddit listing page: /r/sub/, /r/sub/hot, /r/sub/new, /r/sub/top
-        if isinstance(data, dict) and data.get("kind") == "Listing":
-            return self._parse_listing(data, url)
-
-        if isinstance(data, list) and len(data) >= 1:
-            # Post page: [post_listing, comments_listing]
-            post = data[0]["data"]["children"][0]["data"]
-            title = post.get("title", "")
-            author = post.get("author", "")
-            selftext = post.get("selftext", "")
-            score = post.get("score", 0)
-            subreddit = post.get("subreddit", "")
-
-            # Extract comments
-            comments_text = ""
-            if len(data) >= 2:
-                comments_text = self._extract_comments(data[1])
-
-            content = selftext
-            if comments_text:
-                content += f"\n\n---\n## Comments\n{comments_text}"
-
-            return ReadResult(
-                title=title,
-                content=content,
-                url=url,
-                author=f"u/{author}",
-                platform="reddit",
-                extra={"subreddit": subreddit, "score": score},
-            )
-
-        raise ValueError(f"Could not parse Reddit response for: {url}")
-
-    def _parse_listing(self, data: dict, url: str) -> ReadResult:
-        """Parse a subreddit listing (hot/new/top/rising)."""
-        children = data.get("data", {}).get("children", [])
-
-        # Extract subreddit name and sort from URL
-        parsed = urlparse(url)
-        path_parts = [p for p in parsed.path.strip("/").split("/") if p]
-        subreddit = path_parts[1] if len(path_parts) >= 2 else "reddit"
-        sort_type = path_parts[2] if len(path_parts) >= 3 else "hot"
-
-        lines = []
-        for i, child in enumerate(children, 1):
-            if child.get("kind") != "t3":
-                continue
-            post = child.get("data", {})
-            title = post.get("title", "")
-            author = post.get("author", "")
-            score = post.get("score", 0)
-            num_comments = post.get("num_comments", 0)
-            permalink = post.get("permalink", "")
-            post_url = post.get("url", "")
-            is_self = post.get("is_self", False)
-
-            lines.append(f"### {i}. {title}")
-            lines.append(f"👤 u/{author} · ⬆ {score} · 💬 {num_comments}")
-            if not is_self and post_url:
-                lines.append(f"🔗 {post_url}")
-            lines.append(f"📎 https://www.reddit.com{permalink}")
-            # Add selftext preview (first 200 chars)
-            selftext = post.get("selftext", "")
-            if selftext:
-                preview = selftext[:200].replace("\n", " ")
-                if len(selftext) > 200:
-                    preview += "..."
-                lines.append(f"> {preview}")
-            lines.append("")
-
-        content = "\n".join(lines) if lines else "No posts found."
-        return ReadResult(
-            title=f"r/{subreddit} — {sort_type}",
-            content=content,
-            url=url,
-            platform="reddit",
-            extra={"subreddit": subreddit, "sort": sort_type, "count": len(children)},
         )
-
-    def _extract_comments(self, comments_data: dict, depth: int = 0, max_depth: int = 3) -> str:
-        """Recursively extract comments."""
-        lines = []
-        children = comments_data.get("data", {}).get("children", [])
-
-        for child in children:
-            if child.get("kind") != "t1":
-                continue
-            data = child.get("data", {})
-            author = data.get("author", "[deleted]")
-            body = data.get("body", "")
-            score = data.get("score", 0)
-            indent = "  " * depth
-
-            lines.append(f"{indent}**u/{author}** ({score} points):")
-            lines.append(f"{indent}{body}")
-            lines.append("")
-
-            # Recurse into replies
-            if depth < max_depth and data.get("replies") and isinstance(data["replies"], dict):
-                lines.append(self._extract_comments(data["replies"], depth + 1, max_depth))
-
-        return "\n".join(lines)

agent_reach/channels/rss.py

@@ -1,13 +1,7 @@
 # -*- coding: utf-8 -*-
-"""RSS feeds — via feedparser (free, pip dependency).
+"""RSS — check if feedparser is available."""
 
-Backend: feedparser (https://github.com/kurtmckee/feedparser)
+from .base import Channel
-Swap to: any RSS parser
-"""
-
-import feedparser
-from urllib.parse import urlparse
-from .base import Channel, ReadResult
 
 
 class RSSChannel(Channel):
@@ -17,41 +11,11 @@ class RSSChannel(Channel):
     tier = 0
 
     def can_handle(self, url: str) -> bool:
-        lower = url.lower()
+        return any(x in url.lower() for x in ["/feed", "/rss", ".xml", "atom"])
-        domain = urlparse(url).netloc.lower()
-        return (lower.endswith(".xml") or "/rss" in lower or "/feed" in lower
-                or "/atom" in lower or "rss" in domain)
 
-    async def read(self, url: str, config=None) -> ReadResult:
+    def check(self, config=None):
-        feed = feedparser.parse(url)
+        try:
-
+            import feedparser
-        if feed.bozo and not feed.entries:
+            return "ok", "可读取 RSS/Atom 源"
-            raise ValueError(f"Failed to parse RSS feed: {url}")
+        except ImportError:
-
+            return "off", "feedparser 未安装。安装：pip install feedparser"
-        if not feed.entries:
-            raise ValueError(f"No entries in RSS feed: {url}")
-
-        # Return latest entry
-        entry = feed.entries[0]
-        content = entry.get("summary", "") or entry.get("description", "")
-
-        # If multiple entries, summarize all
-        if len(feed.entries) > 1:
-            lines = [f"# {feed.feed.get('title', 'RSS Feed')}\n"]
-            for i, e in enumerate(feed.entries[:20], 1):
-                title = e.get("title", "Untitled")
-                link = e.get("link", "")
-                summary = e.get("summary", "")[:200]
-                lines.append(f"## {i}. {title}")
-                lines.append(f"🔗 {link}")
-                if summary:
-                    lines.append(summary)
-                lines.append("")
-            content = "\n".join(lines)
-
-        return ReadResult(
-            title=feed.feed.get("title", entry.get("title", url)),
-            content=content,
-            url=url,
-            platform="rss",
-        )

agent_reach/channels/twitter.py

@@ -1,286 +1,38 @@
 # -*- coding: utf-8 -*-
-"""Twitter/X — via bird CLI (free) or Jina Reader fallback.
+"""Twitter/X — check if bird CLI is available."""
-
-Backend: bird (@steipete/bird npm package) for search/timeline
-         Jina Reader for single tweets
-Swap to: any Twitter access tool
-"""
 
 import shutil
 import subprocess
-from urllib.parse import urlparse
+from .base import Channel
-from .base import Channel, ReadResult, SearchResult
-from typing import List
-import requests
-
-
-def _bird_cmd():
-    """Find bird CLI binary."""
-    return shutil.which("bird") or shutil.which("birdx")
-
-
-def _bird_env(config=None):
-    """Build env dict with Twitter cookies and proxy support for bird CLI.
-
-    Node.js native fetch() doesn't respect HTTP_PROXY/HTTPS_PROXY.
-    We inject undici's EnvHttpProxyAgent via NODE_OPTIONS so bird
-    automatically routes through the user's proxy.
-    """
-    import os
-    import tempfile
-    env = os.environ.copy()
-    if config:
-        auth_token = config.get("twitter_auth_token")
-        ct0 = config.get("twitter_ct0")
-        if auth_token:
-            env["AUTH_TOKEN"] = auth_token
-        if ct0:
-            env["CT0"] = ct0
-
-    # Auto-inject undici proxy support if HTTP_PROXY/HTTPS_PROXY is set
-    has_proxy = env.get("HTTPS_PROXY") or env.get("HTTP_PROXY") or env.get("https_proxy") or env.get("http_proxy")
-    if has_proxy:
-        bootstrap = _get_proxy_bootstrap_path()
-        if bootstrap:
-            npm_root = subprocess.run(
-                ["npm", "root", "-g"],
-                capture_output=True, text=True, timeout=5,
-            ).stdout.strip()
-            existing_opts = env.get("NODE_OPTIONS", "")
-            env["NODE_OPTIONS"] = f"--require {bootstrap} {existing_opts}".strip()
-            env["NODE_PATH"] = npm_root
-
-    return env
-
-
-def _get_proxy_bootstrap_path():
-    """Create/return a bootstrap JS file that sets up undici proxy for fetch."""
-    import os
-    import tempfile
-    bootstrap_path = os.path.join(tempfile.gettempdir(), "agent-reach-undici-proxy.js")
-    if not os.path.exists(bootstrap_path):
-        # Check if undici is available
-        npm_root = subprocess.run(
-            ["npm", "root", "-g"],
-            capture_output=True, text=True, timeout=5,
-        ).stdout.strip()
-        undici_path = os.path.join(npm_root, "undici", "index.js")
-        if not os.path.exists(undici_path):
-            return None
-        with open(bootstrap_path, "w") as f:
-            f.write(
-                "try {\n"
-                "  const { EnvHttpProxyAgent, setGlobalDispatcher } = require('undici');\n"
-                "  if (process.env.HTTPS_PROXY || process.env.HTTP_PROXY) {\n"
-                "    setGlobalDispatcher(new EnvHttpProxyAgent());\n"
-                "  }\n"
-                "} catch(e) {}\n"
-            )
-    return bootstrap_path
 
 
 class TwitterChannel(Channel):
     name = "twitter"
     description = "Twitter/X 推文"
-    backends = ["bird", "Jina Reader"]
+    backends = ["bird CLI"]
-    tier = 0  # Single tweet reading is zero-config
+    tier = 1
 
     def can_handle(self, url: str) -> bool:
-        domain = urlparse(url).netloc.lower()
+        from urllib.parse import urlparse
-        return "x.com" in domain or "twitter.com" in domain
+        d = urlparse(url).netloc.lower()
+        return "x.com" in d or "twitter.com" in d
 
     def check(self, config=None):
-        # Basic reading always works (Jina fallback)
+        bird = shutil.which("bird") or shutil.which("birdx")
-        bird = _bird_cmd()
+        if not bird:
-        if bird:
+            return "warn", (
-            # Actually test bird connectivity
+                "bird CLI 未安装。搜索可通过 Exa 替代。安装：\n"
-            try:
+                "  npm install -g @steipete/bird"
-                result = subprocess.run(
-                    [bird, "whoami"],
-                    capture_output=True, timeout=15,
-                    encoding='utf-8', errors='replace',
-                    env=_bird_env(config),
-                )
-                if result.returncode == 0 and "fetch failed" not in result.stdout.lower() and "fetch failed" not in result.stderr.lower():
-                    return "ok", "搜索、时间线、发推全部可用"
-                else:
-                    error_hint = (result.stderr or result.stdout).strip()[:100]
-                    if "fetch failed" in (error_hint + result.stdout).lower():
-                        return "warn", (
-                            f"bird 已安装但连接失败（fetch failed）。可能原因：\n"
-                            "  1. Cookie 无效或过期 → 重新导出 Cookie\n"
-                            "  2. 需要代理但 Node.js fetch 不走系统代理 → 使用全局/透明代理（如 Clash TUN 模式、Proxifier）\n"
-                            "  3. 网络无法直连 x.com\n"
-                            "  搜索功能暂不可用，将使用 Exa 搜索作为替代"
-                        )
-                    return "warn", f"bird 连接异常：{error_hint}。搜索将使用 Exa 替代"
-            except (subprocess.TimeoutExpired, FileNotFoundError):
-                return "warn", "bird 已安装但连接超时。搜索将使用 Exa 替代"
-        return "ok", "可读取推文。安装 bird + 配置 Cookie 可解锁搜索和发推"
-
-    async def read(self, url: str, config=None) -> ReadResult:
-        # Try bird first
-        bird = _bird_cmd()
-        if bird:
-            return await self._read_bird(url, bird, config)
-        # Fallback: Jina Reader
-        return await self._read_jina(url)
-
-    async def _read_bird(self, url: str, bird: str, config=None) -> ReadResult:
-        result = subprocess.run(
-            [bird, "read", url],
-            capture_output=True, timeout=30,
-            encoding='utf-8', errors='replace',
-            env=_bird_env(config),
-        )
-        if result.returncode != 0:
-            return await self._read_jina(url)
-
-        text = result.stdout.strip()
-        # Extract author from first line
-        author = ""
-        lines = text.split("\n")
-        if lines and lines[0].startswith("@"):
-            author = lines[0].split()[0]
-
-        return ReadResult(
-            title=text[:100],
-            content=text,
-            url=url,
-            author=author,
-            platform="twitter",
-        )
-
-    async def _read_jina(self, url: str) -> ReadResult:
-        try:
-            resp = requests.get(
-                f"https://r.jina.ai/{url}",
-                headers={"Accept": "text/markdown"},
-                timeout=15,
             )
-            resp.raise_for_status()
+        try:
-            text = resp.text
+            r = subprocess.run(
-
+                [bird, "whoami"], capture_output=True, text=True, timeout=10
-            # Detect unusable Jina responses for X/Twitter (JS-required pages)
+            )
-            unusable_indicators = [
+            if r.returncode == 0:
-                "page doesn",  # "this page doesn't exist" (handles both ' and ')
+                return "ok", "完整可用（读取、搜索推文）"
-                "miss what",   # "Don't miss what's happening"
+            return "warn", (
-                "Something went wrong. Try reloading",
+                "bird CLI 已安装但未配置 Cookie。运行：\n"
-                "Log in](",    # Markdown link: [Log in](...)
+                "  agent-reach configure twitter-cookies \"auth_token=xxx; ct0=yyy\""
-            ]
-            if any(indicator in text for indicator in unusable_indicators):
-                return ReadResult(
-                    title="Twitter/X",
-                    content="⚠️ Could not read this tweet.\n"
-                            "The tweet may have been deleted, or the account is private.\n\n"
-                            "Tips:\n"
-                            "- Make sure the URL is correct\n"
-                            "- Try: bird read <url> (if bird CLI is installed)\n"
-                            "- For protected tweets, configure Twitter cookies: "
-                            "agent-reach configure twitter-cookies AUTH_TOKEN CT0",
-                    url=url,
-                    platform="twitter",
-                )
-
-            title = text[:100] if text else url
-            return ReadResult(
-                title=title,
-                content=text,
-                url=url,
-                platform="twitter",
             )
         except Exception:
-            return ReadResult(
+            return "warn", "bird CLI 已安装但连接失败"
-                title="Twitter/X",
-                content="⚠️ Could not read this tweet.\n"
-                        "The tweet may have been deleted, or the account is private.\n\n"
-                        "Tips:\n"
-                        "- Make sure the URL is correct\n"
-                        "- Try: bird read <url> (if bird CLI is installed)\n"
-                        "- For protected tweets, configure Twitter cookies: "
-                        "agent-reach configure twitter-cookies AUTH_TOKEN CT0",
-                url=url,
-                platform="twitter",
-            )
-
-    async def search(self, query: str, config=None, **kwargs) -> List[SearchResult]:
-        limit = kwargs.get("limit", 10)
-
-        bird = _bird_cmd()
-        if bird:
-            return await self._search_bird(query, limit, bird, config)
-
-        # Fallback to Exa
-        return await self._search_exa(query, limit, config)
-
-    async def _search_bird(self, query: str, limit: int, bird: str, config=None) -> List[SearchResult]:
-        try:
-            result = subprocess.run(
-                [bird, "search", query, "-n", str(limit)],
-                capture_output=True, timeout=30,
-                encoding='utf-8', errors='replace',
-                env=_bird_env(config),
-            )
-            if result.returncode != 0:
-                stderr = (result.stderr or "").strip()
-                if "fetch failed" in stderr.lower() or "fetch failed" in (result.stdout or "").lower():
-                    # bird can't connect — fall back to Exa silently
-                    return await self._search_exa(query, limit, config)
-                return await self._search_exa(query, limit, config)
-
-            parsed = self._parse_bird_output(result.stdout)
-            if not parsed:
-                # bird returned nothing — try Exa
-                return await self._search_exa(query, limit, config)
-            return parsed
-        except (subprocess.TimeoutExpired, FileNotFoundError):
-            return await self._search_exa(query, limit, config)
-
-    def _parse_bird_output(self, text: str) -> List[SearchResult]:
-        """Parse bird text output into SearchResults."""
-        results = []
-        current = {}
-        text_lines = []
-
-        for line in text.strip().split("\n"):
-            line = line.strip()
-            if line.startswith("─"):
-                if current:
-                    current["text"] = "\n".join(text_lines).strip()
-                    results.append(SearchResult(
-                        title=current.get("text", "")[:80],
-                        url=current.get("url", ""),
-                        snippet=current.get("text", ""),
-                        author=current.get("author", ""),
-                        date=current.get("date", ""),
-                    ))
-                    current = {}
-                    text_lines = []
-                continue
-            if line.startswith("@") and line.endswith(":") and "(" in line:
-                current["author"] = line.split()[0]
-                continue
-            if line.startswith("date:"):
-                current["date"] = line[5:].strip()
-                continue
-            if line.startswith("url:"):
-                current["url"] = line[4:].strip()
-                continue
-            if current is not None:
-                text_lines.append(line)
-
-        if current and text_lines:
-            current["text"] = "\n".join(text_lines).strip()
-            results.append(SearchResult(
-                title=current.get("text", "")[:80],
-                url=current.get("url", ""),
-                snippet=current.get("text", ""),
-                author=current.get("author", ""),
-                date=current.get("date", ""),
-            ))
-        return results
-
-    async def _search_exa(self, query: str, limit: int, config=None) -> List[SearchResult]:
-        from agent_reach.channels.exa_search import ExaSearchChannel
-        exa = ExaSearchChannel()
-        return await exa.search(f"site:x.com {query}", config=config, limit=limit)

agent_reach/channels/web.py

@@ -1,49 +1,17 @@
 # -*- coding: utf-8 -*-
-"""Web pages — via Jina Reader API (free, no config needed).
+"""Web — any URL via Jina Reader. Always available."""
 
-Backend: Jina Reader (https://r.jina.ai)
+from .base import Channel
-Swap to: Firecrawl, Trafilatura, or any other reader API
-"""
-
-import requests
-from .base import Channel, ReadResult
 
 
 class WebChannel(Channel):
     name = "web"
-    description = "网页（任意 URL）"
+    description = "任意网页"
-    backends = ["Jina Reader API"]
+    backends = ["Jina Reader"]
     tier = 0
 
-    JINA_URL = "https://r.jina.ai/"
-
     def can_handle(self, url: str) -> bool:
-        # Fallback — handles any URL not matched by other channels
+        return True  # Fallback — handles any URL
-        return True
 
-    async def read(self, url: str, config=None) -> ReadResult:
+    def check(self, config=None):
-        resp = requests.get(
+        return "ok", "通过 Jina Reader 读取任意网页（curl https://r.jina.ai/URL）"
-            f"{self.JINA_URL}{url}",
-            headers={"Accept": "text/markdown"},
-            timeout=15,
-        )
-        resp.raise_for_status()
-        text = resp.text
-
-        # Extract title from first markdown heading
-        title = url
-        for line in text.split("\n"):
-            line = line.strip()
-            if line.startswith("# "):
-                title = line[2:].strip()
-                break
-            if line.startswith("Title:"):
-                title = line[6:].strip()
-                break
-
-        return ReadResult(
-            title=title,
-            content=text,
-            url=url,
-            platform="web",
-        )

agent_reach/channels/xiaohongshu.py

@@ -1,16 +1,9 @@
 # -*- coding: utf-8 -*-
-"""XiaoHongShu (小红书) — via mcporter + xiaohongshu MCP server.
+"""XiaoHongShu — check if mcporter + xiaohongshu MCP is available."""
 
-Backend: xiaohongshu-mcp server (internal API, reliable)
-Requires: mcporter CLI + xiaohongshu MCP server running
-"""
-
-import json
 import shutil
 import subprocess
-from urllib.parse import urlparse, parse_qs, urlencode
+from .base import Channel
-from .base import Channel, ReadResult, SearchResult
-from typing import List, Optional
 
 
 class XiaoHongShuChannel(Channel):
@@ -19,30 +12,8 @@ class XiaoHongShuChannel(Channel):
     backends = ["xiaohongshu-mcp"]
     tier = 2
 
-    def _mcporter_ok(self) -> bool:
-        """Check if mcporter + xiaohongshu MCP is available."""
-        if not shutil.which("mcporter"):
-            return False
-        try:
-            r = subprocess.run(
-                ["mcporter", "list"], capture_output=True, text=True, timeout=10
-            )
-            return "xiaohongshu" in r.stdout
-        except Exception:
-            return False
-
-    def _call(self, expr: str, timeout: int = 30) -> str:
-        r = subprocess.run(
-            ["mcporter", "call", expr],
-            capture_output=True, text=True, timeout=timeout,
-        )
-        if r.returncode != 0:
-            raise RuntimeError(r.stderr or r.stdout)
-        return r.stdout
-
-    # ── Channel interface ──
-
     def can_handle(self, url: str) -> bool:
+        from urllib.parse import urlparse
         d = urlparse(url).netloc.lower()
         return "xiaohongshu.com" in d or "xhslink.com" in d
 
@@ -55,190 +26,25 @@ class XiaoHongShuChannel(Channel):
                 "  3. mcporter config add xiaohongshu http://localhost:18060/mcp\n"
                 "  详见 https://github.com/xpzouying/xiaohongshu-mcp"
             )
-        if not self._mcporter_ok():
-            return "off", (
-                "mcporter 已装但小红书 MCP 未配置。运行：\n"
-                "  docker run -d --name xiaohongshu-mcp -p 18060:18060 xpzouying/xiaohongshu-mcp\n"
-                "  mcporter config add xiaohongshu http://localhost:18060/mcp"
-            )
         try:
-            out = self._call("xiaohongshu.check_login_status()", timeout=10)
+            r = subprocess.run(
-            if "已登录" in out or "logged" in out.lower():
+                ["mcporter", "list"], capture_output=True, text=True, timeout=10
+            )
+            if "xiaohongshu" not in r.stdout:
+                return "off", (
+                    "mcporter 已装但小红书 MCP 未配置。运行：\n"
+                    "  docker run -d --name xiaohongshu-mcp -p 18060:18060 xpzouying/xiaohongshu-mcp\n"
+                    "  mcporter config add xiaohongshu http://localhost:18060/mcp"
+                )
+        except Exception:
+            return "off", "mcporter 连接异常"
+        try:
+            r = subprocess.run(
+                ["mcporter", "call", "xiaohongshu.check_login_status()"],
+                capture_output=True, text=True, timeout=10
+            )
+            if "已登录" in r.stdout or "logged" in r.stdout.lower():
                 return "ok", "完整可用（阅读、搜索、发帖、评论、点赞）"
             return "warn", "MCP 已连接但未登录，需扫码登录"
         except Exception:
             return "warn", "MCP 连接异常，检查 xiaohongshu-mcp 服务是否在运行"
-
-    async def read(self, url: str, config=None) -> ReadResult:
-        if not self._mcporter_ok():
-            return ReadResult(
-                title="XiaoHongShu",
-                content=(
-                    "⚠️ 小红书需要 mcporter + xiaohongshu-mcp 才能使用。\n\n"
-                    "安装步骤：\n"
-                    "1. npm install -g mcporter\n"
-                    "2. docker run -d --name xiaohongshu-mcp -p 18060:18060 xpzouying/xiaohongshu-mcp\n"
-                    "3. mcporter config add xiaohongshu http://localhost:18060/mcp\n"
-                    "4. 运行 agent-reach doctor 检查状态\n\n"
-                    "详见 https://github.com/xpzouying/xiaohongshu-mcp"
-                ),
-                url=url, platform="xiaohongshu",
-            )
-
-        note_id = self._extract_note_id(url)
-        if not note_id:
-            return ReadResult(
-                title="XiaoHongShu",
-                content=f"⚠️ 无法从 URL 提取笔记 ID: {url}",
-                url=url, platform="xiaohongshu",
-            )
-
-        # Step 1: try xsec_token from URL query param (e.g. from search results)
-        xsec_token = self._extract_token_from_url(url)
-
-        # Step 2: try homepage feeds
-        if not xsec_token:
-            xsec_token = self._find_token_in_feeds(note_id)
-
-        # Step 3: search for the note to get a fresh token
-        if not xsec_token:
-            xsec_token = self._find_token_by_search(note_id)
-
-        # If no token found, fallback to Jina Reader
-        if not xsec_token:
-            return await self._read_jina(url)
-
-        # Get detail via MCP
-        out = self._call(
-            f'xiaohongshu.get_feed_detail(feed_id: "{note_id}", xsec_token: "{xsec_token}")',
-            timeout=15,
-        )
-
-        return ReadResult(
-            title=self._extract_title(out) or f"XHS {note_id}",
-            content=out.strip(),
-            url=url, platform="xiaohongshu",
-        )
-
-    async def search(self, query: str, config=None, **kwargs) -> List[SearchResult]:
-        if not self._mcporter_ok():
-            raise ValueError(
-                "小红书搜索需要 mcporter + xiaohongshu-mcp。\n"
-                "安装: npm install -g mcporter && mcporter config add xiaohongshu http://localhost:18060/mcp"
-            )
-        limit = kwargs.get("limit", 10)
-        safe_q = query.replace('"', '\\"')
-        out = self._call(f'xiaohongshu.search_feeds(keyword: "{safe_q}")', timeout=30)
-
-        results = []
-        try:
-            data = json.loads(out)
-            for item in data.get("feeds", [])[:limit]:
-                card = item.get("noteCard", {})
-                user = card.get("user", {})
-                interact = card.get("interactInfo", {})
-                note_id = item.get("id", "")
-                xsec_token = item.get("xsecToken", "")
-                note_url = f"https://www.xiaohongshu.com/explore/{note_id}"
-                if xsec_token:
-                    note_url += f"?xsec_token={xsec_token}"
-                results.append(SearchResult(
-                    title=card.get("displayTitle", ""),
-                    url=note_url,
-                    snippet=f"👤 {user.get('nickname', '')} · ❤ {interact.get('likedCount', '0')}",
-                    score=0,
-                ))
-        except (json.JSONDecodeError, KeyError):
-            pass
-        return results
-
-    # ── Helpers ──
-
-    def _extract_note_id(self, url: str) -> str:
-        """Extract note ID from URL path, ignoring query params."""
-        path = urlparse(url).path.strip("/").split("/")
-        return path[-1] if path else ""
-
-    def _extract_token_from_url(self, url: str) -> Optional[str]:
-        """Extract xsec_token from URL query parameter if present."""
-        qs = parse_qs(urlparse(url).query)
-        tokens = qs.get("xsec_token", [])
-        return tokens[0] if tokens else None
-
-    def _find_token_in_feeds(self, note_id: str) -> Optional[str]:
-        """Try to find xsec_token for a note from homepage feeds."""
-        try:
-            out = self._call("xiaohongshu.list_feeds()", timeout=15)
-            data = json.loads(out)
-            for feed in data.get("feeds", []):
-                if feed.get("id") == note_id:
-                    return feed.get("xsecToken") or None
-        except Exception:
-            pass
-        return None
-
-    def _find_token_by_search(self, note_id: str) -> Optional[str]:
-        """Search for the note ID to get a fresh xsec_token."""
-        try:
-            out = self._call(
-                f'xiaohongshu.search_feeds(keyword: "{note_id}")', timeout=20
-            )
-            data = json.loads(out)
-            for feed in data.get("feeds", []):
-                if feed.get("id") == note_id:
-                    return feed.get("xsecToken") or None
-            # If exact match not found but results exist, try the first one
-            # (search by note_id sometimes returns the note with a different key)
-        except Exception:
-            pass
-        return None
-
-    def _extract_title(self, text: str) -> str:
-        for line in text.split("\n"):
-            line = line.strip()
-            if line and not line.startswith(("{", "[", "#", "http")):
-                return line[:80]
-        return ""
-
-    async def _read_jina(self, url: str) -> ReadResult:
-        """Fallback: read XHS note via Jina Reader when xsec_token unavailable."""
-        import requests
-        try:
-            resp = requests.get(
-                f"https://r.jina.ai/{url}",
-                headers={"Accept": "text/markdown"},
-                timeout=15,
-            )
-            resp.raise_for_status()
-            text = resp.text
-            if len(text.strip()) < 50 or "登录" in text[:200]:
-                return ReadResult(
-                    title="XiaoHongShu",
-                    content=(
-                        f"⚠️ 无法获取笔记详情: {url}\n\n"
-                        "小红书需要 xsec_token 才能通过 MCP 读取笔记。\n"
-                        "请尝试先搜索相关关键词，再从结果中读取。"
-                    ),
-                    url=url, platform="xiaohongshu",
-                )
-            title = ""
-            for line in text.split("\n"):
-                line = line.strip()
-                if line and not line.startswith(("#", "http", "![", "[")):
-                    title = line[:80]
-                    break
-            return ReadResult(
-                title=title or "XiaoHongShu",
-                content=text.strip(),
-                url=url, platform="xiaohongshu",
-            )
-        except Exception:
-            return ReadResult(
-                title="XiaoHongShu",
-                content=(
-                    f"⚠️ 无法获取笔记详情: {url}\n\n"
-                    "小红书需要 xsec_token 才能通过 MCP 读取笔记。\n"
-                    "请尝试先搜索相关关键词，再从结果中读取。"
-                ),
-                url=url, platform="xiaohongshu",
-            )

agent_reach/channels/youtube.py

@@ -1,125 +1,22 @@
 # -*- coding: utf-8 -*-
-"""YouTube — via yt-dlp (video info, subtitles, and search).
+"""YouTube — check if yt-dlp is available."""
 
-Backend: yt-dlp (https://github.com/yt-dlp/yt-dlp)
-Supports: read (info + subtitles), search (ytsearch)
-"""
-
-import json
 import shutil
-import subprocess
+from .base import Channel
-import tempfile
-from pathlib import Path
-from urllib.parse import urlparse
-from .base import Channel, ReadResult, SearchResult
-from typing import List
 
 
 class YouTubeChannel(Channel):
     name = "youtube"
-    description = "YouTube 视频字幕"
+    description = "YouTube 视频和字幕"
     backends = ["yt-dlp"]
-    requires_tools = ["yt-dlp"]
     tier = 0
 
     def can_handle(self, url: str) -> bool:
+        from urllib.parse import urlparse
         d = urlparse(url).netloc.lower()
         return "youtube.com" in d or "youtu.be" in d
 
-    async def read(self, url: str, config=None) -> ReadResult:
+    def check(self, config=None):
-        if not shutil.which("yt-dlp"):
+        if shutil.which("yt-dlp"):
-            raise RuntimeError("yt-dlp not installed. Install: pip install yt-dlp")
+            return "ok", "可提取视频信息和字幕"
-
+        return "off", "yt-dlp 未安装。安装：pip install yt-dlp"
-        with tempfile.TemporaryDirectory() as tmpdir:
-            info = self._get_info(url)
-            title = info.get("title", url)
-            author = info.get("uploader", "")
-
-            transcript = self._get_subtitles(url, tmpdir)
-            if not transcript:
-                transcript = f"[Video: {title}]\n[No subtitles available.]"
-
-            return ReadResult(
-                title=title, content=transcript, url=url,
-                author=author, platform="youtube",
-                extra={
-                    "duration": info.get("duration_string"),
-                    "view_count": info.get("view_count"),
-                    "upload_date": info.get("upload_date"),
-                },
-            )
-
-    async def search(self, query: str, config=None, **kwargs) -> List[SearchResult]:
-        """Search YouTube via yt-dlp's ytsearch."""
-        if not shutil.which("yt-dlp"):
-            raise RuntimeError("yt-dlp not installed. Install: pip install yt-dlp")
-
-        limit = kwargs.get("limit", 10)
-
-        try:
-            r = subprocess.run(
-                ["yt-dlp", "--dump-json", "--flat-playlist",
-                 f"ytsearch{limit}:{query}"],
-                capture_output=True, text=True, timeout=30,
-            )
-            results = []
-            for line in r.stdout.strip().split("\n"):
-                if not line.strip():
-                    continue
-                try:
-                    d = json.loads(line)
-                    vid = d.get("id", "")
-                    results.append(SearchResult(
-                        title=d.get("title", ""),
-                        url=f"https://youtube.com/watch?v={vid}" if vid else "",
-                        snippet=(
-                            f"👤 {d.get('channel', '?')} · "
-                            f"⏱ {d.get('duration_string', '?')} · "
-                            f"👁 {d.get('view_count', '?')}"
-                        ),
-                        extra={
-                            "channel": d.get("channel"),
-                            "duration": d.get("duration_string"),
-                            "view_count": d.get("view_count"),
-                        },
-                    ))
-                except json.JSONDecodeError:
-                    continue
-            return results
-        except subprocess.TimeoutExpired:
-            return []
-
-    def _get_info(self, url: str) -> dict:
-        try:
-            r = subprocess.run(
-                ["yt-dlp", "--dump-json", "--no-download", url],
-                capture_output=True, text=True, timeout=30,
-            )
-            if r.returncode == 0:
-                return json.loads(r.stdout)
-        except (subprocess.TimeoutExpired, json.JSONDecodeError):
-            pass
-        return {}
-
-    def _get_subtitles(self, url: str, tmpdir: str) -> str:
-        try:
-            subprocess.run(
-                ["yt-dlp", "--write-auto-sub", "--write-sub",
-                 "--sub-lang", "en,zh-Hans,zh",
-                 "--skip-download", "--sub-format", "vtt",
-                 "-o", f"{tmpdir}/%(id)s.%(ext)s", url],
-                capture_output=True, text=True, timeout=30,
-            )
-            for f in Path(tmpdir).glob("*.vtt"):
-                text = f.read_text(errors="replace")
-                lines = []
-                for line in text.split("\n"):
-                    line = line.strip()
-                    if not line or line.startswith("WEBVTT") or "-->" in line or line.isdigit():
-                        continue
-                    if line not in lines[-1:]:
-                        lines.append(line)
-                return "\n".join(lines)
-        except subprocess.TimeoutExpired:
-            pass
-        return ""

agent_reach/cli.py

@@ -1,20 +1,15 @@
 # -*- coding: utf-8 -*-
 """
-Agent Reach CLI — command-line interface.
+Agent Reach CLI — installer, doctor, and configuration tool.
 
 Usage:
-    agent-reach read <url>
+    agent-reach install --env=auto
-    agent-reach search <query>
-    agent-reach search-reddit <query> [--sub <subreddit>]
-    agent-reach search-github <query> [--lang <language>]
-    agent-reach search-twitter <query>
-    agent-reach setup
     agent-reach doctor
-    agent-reach version
+    agent-reach configure twitter-cookies "auth_token=xxx; ct0=yyy"
+    agent-reach setup
 """
 
 import sys
-import asyncio
 import argparse
 import json
 import os
@@ -48,57 +43,6 @@ def main():
     sub = parser.add_subparsers(dest="command", help="Available commands")
 
     # ── read ──
-    p_read = sub.add_parser("read", help="Read content from a URL")
-    p_read.add_argument("url", help="URL to read")
-    p_read.add_argument("--json", dest="as_json", action="store_true", help="Output as JSON")
-
-    # ── search ──
-    p_search = sub.add_parser("search", help="Search the web (Exa)")
-    p_search.add_argument("query", nargs="+", help="Search query")
-    p_search.add_argument("-n", "--num", type=int, default=5, help="Number of results")
-
-    # ── search-reddit ──
-    p_sr = sub.add_parser("search-reddit", help="Search Reddit")
-    p_sr.add_argument("query", nargs="+", help="Search query")
-    p_sr.add_argument("--sub", help="Subreddit filter")
-    p_sr.add_argument("-n", "--num", type=int, default=10, help="Number of results")
-
-    # ── search-github ──
-    p_sg = sub.add_parser("search-github", help="Search GitHub")
-    p_sg.add_argument("query", nargs="+", help="Search query")
-    p_sg.add_argument("--lang", help="Language filter")
-    p_sg.add_argument("-n", "--num", type=int, default=5, help="Number of results")
-
-    # ── search-twitter ──
-    p_st = sub.add_parser("search-twitter", help="Search Twitter")
-    p_st.add_argument("query", nargs="+", help="Search query")
-    p_st.add_argument("-n", "--num", type=int, default=10, help="Number of results")
-
-    # ── search-youtube ──
-    p_sy = sub.add_parser("search-youtube", help="Search YouTube")
-    p_sy.add_argument("query", nargs="+", help="Search query")
-    p_sy.add_argument("-n", "--num", type=int, default=5, help="Number of results")
-
-    # ── search-bilibili ──
-    p_sb = sub.add_parser("search-bilibili", help="Search Bilibili")
-    p_sb.add_argument("query", nargs="+", help="Search query")
-    p_sb.add_argument("-n", "--num", type=int, default=5, help="Number of results")
-
-    # ── search-xhs ──
-    p_sx = sub.add_parser("search-xhs", help="Search XiaoHongShu")
-    p_sx.add_argument("query", nargs="+", help="Search query")
-    p_sx.add_argument("-n", "--num", type=int, default=10, help="Number of results")
-
-    # ── search-linkedin ──
-    p_sl = sub.add_parser("search-linkedin", help="Search LinkedIn")
-    p_sl.add_argument("query", nargs="+", help="Search query")
-    p_sl.add_argument("-n", "--num", type=int, default=10, help="Number of results")
-
-    # ── search-bosszhipin ──
-    p_sbz = sub.add_parser("search-bosszhipin", help="Search Boss直聘")
-    p_sbz.add_argument("query", nargs="+", help="Search query")
-    p_sbz.add_argument("-n", "--num", type=int, default=10, help="Number of results")
-
     # ── setup ──
     sub.add_parser("setup", help="Interactive configuration wizard")
 
@@ -161,10 +105,6 @@ def main():
         _cmd_install(args)
     elif args.command == "configure":
         _cmd_configure(args)
-    elif args.command == "read":
-        asyncio.run(_cmd_read(args))
-    elif args.command.startswith("search"):
-        asyncio.run(_cmd_search(args))
 
 
 # ── Command handlers ────────────────────────────────
@@ -849,98 +789,6 @@ def _cmd_setup():
     print()
 
 
-async def _cmd_read(args):
-    from agent_reach.core import AgentReach
-    eyes = AgentReach()
-    try:
-        result = await eyes.read(args.url)
-        if args.as_json:
-            print(json.dumps(result, ensure_ascii=False, indent=2))
-        else:
-            print(f"\n📖 {result.get('title', 'Untitled')}")
-            print(f"🔗 {result.get('url', '')}")
-            if result.get("author"):
-                print(f"👤 {result['author']}")
-            print(f"\n{result.get('content', '')}")
-    except Exception as e:
-        error_str = str(e)
-        if "400" in error_str and "Bad Request" in error_str:
-            print(f"❌ Invalid URL: {args.url}", file=sys.stderr)
-            print("   Please provide a valid URL (e.g., https://example.com)", file=sys.stderr)
-        elif "ConnectionError" in type(e).__name__ or "Timeout" in type(e).__name__:
-            print(f"❌ Could not connect to: {args.url}", file=sys.stderr)
-            print("   Check your internet connection or the URL.", file=sys.stderr)
-        else:
-            print(f"❌ Error: {e}", file=sys.stderr)
-        sys.exit(1)
-
-
-async def _cmd_search(args):
-    from agent_reach.core import AgentReach
-    eyes = AgentReach()
-    query = " ".join(args.query).strip()
-    num = args.num
-
-    if not query:
-        print("Please provide a search query.", file=sys.stderr)
-        sys.exit(1)
-
-    try:
-        if args.command == "search":
-            results = await eyes.search(query, num_results=num)
-        elif args.command == "search-reddit":
-            results = await eyes.search_reddit(query, subreddit=getattr(args, "sub", None), limit=num)
-        elif args.command == "search-github":
-            results = await eyes.search_github(query, language=getattr(args, "lang", None), limit=num)
-        elif args.command == "search-twitter":
-            results = await eyes.search_twitter(query, limit=num)
-        elif args.command == "search-youtube":
-            results = await eyes.search_youtube(query, limit=num)
-        elif args.command == "search-bilibili":
-            results = await eyes.search_bilibili(query, limit=num)
-        elif args.command == "search-xhs":
-            results = await eyes.search_xhs(query, limit=num)
-        elif args.command == "search-linkedin":
-            results = await eyes.search_linkedin(query, limit=num)
-        elif args.command == "search-bosszhipin":
-            results = await eyes.search_bosszhipin(query, limit=num)
-        else:
-            print(f"Unknown command: {args.command}", file=sys.stderr)
-            sys.exit(1)
-    except Exception as e:
-        error_str = str(e)
-        if "401" in error_str or "Unauthorized" in error_str:
-            print("⚠️  Exa API key not configured or invalid.")
-            print("Get a free key at https://exa.ai (1000 searches/month free)")
-            print("Then run: agent-reach configure exa-key YOUR_KEY")
-            sys.exit(1)
-        elif "exa" in error_str.lower() or "api_key" in error_str.lower():
-            print("⚠️  Exa API key not configured.")
-            print("Get a free key at https://exa.ai")
-            print("Then run: agent-reach configure exa-key YOUR_KEY")
-            sys.exit(1)
-        else:
-            print(f"❌ Error: {e}", file=sys.stderr)
-            sys.exit(1)
-
-    if not results:
-        print("No results found.")
-        return
-
-    for i, r in enumerate(results, 1):
-        title = r.get("title") or r.get("name") or r.get("text", "")[:60]
-        url = r.get("url", "")
-        snippet = r.get("snippet") or r.get("description") or r.get("text", "")
-        print(f"\n{i}. {title}")
-        print(f"   🔗 {url}")
-        if snippet:
-            print(f"   {snippet[:200]}")
-        # Extra info for GitHub
-        extra = r.get("extra", {})
-        if extra.get("stars"):
-            print(f"   ⭐ {extra['stars']}  🍴 {extra.get('forks', 0)}  📝 {extra.get('language', '')}")
-
-
 def _cmd_check_update():
     """Check for newer versions on GitHub."""
     import requests

agent_reach/core.py

@@ -1,120 +1,36 @@
 # -*- coding: utf-8 -*-
 """
-AgentReach — the unified entry point.
+AgentReach — installer, doctor, and configuration tool.
 
-Pure glue: routes URLs to the right channel, routes searches to the right engine.
+Agent Reach helps AI agents install and configure upstream platform tools
-Every channel is a thin wrapper around an external tool. Swap any backend anytime.
+(bird CLI, yt-dlp, mcporter, gh CLI, etc.). After installation, agents
+call the upstream tools directly — no wrapper layer needed.
 
 Usage:
-    from agent_reach import AgentReach
+    from agent_reach.doctor import check_all, format_report
+    from agent_reach.config import Config
 
-    eyes = AgentReach()
+    config = Config()
-    content = await eyes.read("https://github.com/openai/gpt-4")
+    results = check_all(config)
-    results = await eyes.search("AI agent framework")
+    print(format_report(results))
 """
 
-import asyncio
+from typing import Dict, Optional
-from typing import Any, Dict, List, Optional
 
 from agent_reach.config import Config
-from agent_reach.channels import get_channel_for_url, get_channel, get_all_channels
 
 
 class AgentReach:
-    """Give your AI Agent eyes to see the entire internet."""
+    """Give your AI Agent eyes to see the entire internet.
+
+    This class provides health-check functionality.
+    For reading/searching, use the upstream tools directly
+    (see SKILL.md for commands).
+    """
 
     def __init__(self, config: Optional[Config] = None):
         self.config = config or Config()
 
-    # ── Reading ─────────────────────────────────────────
-
-    async def read(self, url: str) -> Dict[str, Any]:
-        """
-        Read content from any URL. Auto-detects platform.
-
-        Supported: Web, GitHub, Reddit, Twitter, YouTube,
-        Bilibili, RSS, and more.
-
-        Returns:
-            Dict with title, content, url, author, platform, etc.
-        """
-        if not url.startswith(("http://", "https://")):
-            url = f"https://{url}"
-
-        channel = get_channel_for_url(url)
-        result = await channel.read(url, config=self.config)
-        return result.to_dict()
-
-    async def read_batch(self, urls: List[str]) -> List[Dict[str, Any]]:
-        """Read multiple URLs concurrently."""
-        tasks = [self.read(url) for url in urls]
-        results = await asyncio.gather(*tasks, return_exceptions=True)
-        return [r for r in results if not isinstance(r, Exception)]
-
-    def detect_platform(self, url: str) -> str:
-        """Detect what platform a URL belongs to."""
-        channel = get_channel_for_url(url)
-        return channel.name
-
-    # ── Searching ───────────────────────────────────────
-
-    async def search(self, query: str, num_results: int = 5) -> List[Dict[str, Any]]:
-        """Semantic web search via Exa."""
-        ch = get_channel("exa_search")
-        results = await ch.search(query, config=self.config, limit=num_results)
-        return [r.to_dict() for r in results]
-
-    async def search_reddit(self, query: str, subreddit: Optional[str] = None, limit: int = 10) -> List[Dict[str, Any]]:
-        """Search Reddit via Exa (bypasses IP blocks)."""
-        ch = get_channel("exa_search")
-        q = f"site:reddit.com/r/{subreddit} {query}" if subreddit else f"site:reddit.com {query}"
-        results = await ch.search(q, config=self.config, limit=limit)
-        return [r.to_dict() for r in results]
-
-    async def search_github(self, query: str, language: Optional[str] = None, limit: int = 5) -> List[Dict[str, Any]]:
-        """Search GitHub repositories."""
-        ch = get_channel("github")
-        results = await ch.search(query, config=self.config, language=language, limit=limit)
-        return [r.to_dict() for r in results]
-
-    async def search_twitter(self, query: str, limit: int = 10) -> List[Dict[str, Any]]:
-        """Search Twitter. Uses bird CLI if available, else Exa."""
-        ch = get_channel("twitter")
-        results = await ch.search(query, config=self.config, limit=limit)
-        return [r.to_dict() for r in results]
-
-    async def search_youtube(self, query: str, limit: int = 5) -> List[Dict[str, Any]]:
-        """Search YouTube via yt-dlp."""
-        ch = get_channel("youtube")
-        results = await ch.search(query, config=self.config, limit=limit)
-        return [r.to_dict() for r in results]
-
-    async def search_bilibili(self, query: str, limit: int = 5) -> List[Dict[str, Any]]:
-        """Search Bilibili. Tries yt-dlp first, falls back to Exa."""
-        ch = get_channel("bilibili")
-        results = await ch.search(query, config=self.config, limit=limit)
-        return [r.to_dict() for r in results]
-
-    async def search_xhs(self, query: str, limit: int = 10) -> List[Dict[str, Any]]:
-        """Search XiaoHongShu via mcporter."""
-        ch = get_channel("xiaohongshu")
-        results = await ch.search(query, config=self.config, limit=limit)
-        return [r.to_dict() for r in results]
-
-    async def search_linkedin(self, query: str, limit: int = 10) -> List[Dict[str, Any]]:
-        """Search LinkedIn via MCP or Exa."""
-        ch = get_channel("linkedin")
-        results = await ch.search(query, config=self.config, limit=limit)
-        return [r.to_dict() for r in results]
-
-    async def search_bosszhipin(self, query: str, limit: int = 10) -> List[Dict[str, Any]]:
-        """Search Boss直聘 via MCP or Exa."""
-        ch = get_channel("bosszhipin")
-        results = await ch.search(query, config=self.config, limit=limit)
-        return [r.to_dict() for r in results]
-
-    # ── Health ──────────────────────────────────────────
-
     def doctor(self) -> Dict[str, dict]:
         """Check all channel availability."""
         from agent_reach.doctor import check_all
@@ -124,13 +40,3 @@ class AgentReach:
         """Get formatted health report."""
         from agent_reach.doctor import check_all, format_report
         return format_report(check_all(self.config))
-
-    # ── Sync wrappers ───────────────────────────────────
-
-    def read_sync(self, url: str) -> Dict[str, Any]:
-        """Synchronous version of read()."""
-        return asyncio.run(self.read(url))
-
-    def search_sync(self, query: str, num_results: int = 5) -> List[Dict[str, Any]]:
-        """Synchronous version of search()."""
-        return asyncio.run(self.search(query, num_results))

agent_reach/guides/setup-exa.md

@@ -22,8 +22,8 @@ mcporter config add exa https://mcp.exa.ai/mcp
 
 ### 3. 验证
 ```bash
-agent-reach doctor | grep "Web Search"
+agent-reach doctor | grep "Search"
-agent-reach search "test" -n 1
+mcporter call 'exa.web_search_exa(query: "test", numResults: 1)'
 ```
 
 ## 需要用户手动做的步骤

agent_reach/guides/setup-wechat.md

@@ -18,7 +18,7 @@ playwright install chromium
 
 3. 安装完成后测试：
 ```bash
-agent-reach read "https://mp.weixin.qq.com/s/一个测试链接"
+curl -s "https://r.jina.ai/https://mp.weixin.qq.com/s/一个测试链接" -H "Accept: text/markdown"
 ```
 
 ## 需要用户手动做的步骤

agent_reach/integrations/mcp_server.py

@@ -1,10 +1,11 @@
 # -*- coding: utf-8 -*-
 """
-Agent Reach MCP Server — expose all capabilities as MCP tools.
+Agent Reach MCP Server — expose doctor/status as MCP tool.
 
 Run: python -m agent_reach.integrations.mcp_server
 
-8 tools for any MCP-compatible AI Agent.
+Agent Reach is an installer + doctor tool. For actual reading/searching,
+agents should call upstream tools directly (bird, yt-dlp, mcporter, etc.).
 """
 
 import asyncio
@@ -35,50 +36,15 @@ def create_server():
     @server.list_tools()
     async def list_tools():
         return [
-            Tool(name="read_url",
-                 description="Read content from any URL. Supports: web, GitHub, Reddit, Twitter, YouTube, Bilibili, RSS.",
-                 inputSchema={"type": "object", "properties": {"url": {"type": "string"}}, "required": ["url"]}),
-            Tool(name="read_batch",
-                 description="Read multiple URLs concurrently.",
-                 inputSchema={"type": "object", "properties": {"urls": {"type": "array", "items": {"type": "string"}}}, "required": ["urls"]}),
-            Tool(name="detect_platform",
-                 description="Detect what platform a URL belongs to.",
-                 inputSchema={"type": "object", "properties": {"url": {"type": "string"}}, "required": ["url"]}),
-            Tool(name="search",
-                 description="Semantic web search via Exa.",
-                 inputSchema={"type": "object", "properties": {"query": {"type": "string"}, "num_results": {"type": "integer", "default": 5}}, "required": ["query"]}),
-            Tool(name="search_reddit",
-                 description="Search Reddit posts.",
-                 inputSchema={"type": "object", "properties": {"query": {"type": "string"}, "subreddit": {"type": "string"}, "limit": {"type": "integer", "default": 10}}, "required": ["query"]}),
-            Tool(name="search_github",
-                 description="Search GitHub repositories.",
-                 inputSchema={"type": "object", "properties": {"query": {"type": "string"}, "language": {"type": "string"}, "limit": {"type": "integer", "default": 5}}, "required": ["query"]}),
-            Tool(name="search_twitter",
-                 description="Search Twitter/X posts.",
-                 inputSchema={"type": "object", "properties": {"query": {"type": "string"}, "limit": {"type": "integer", "default": 10}}, "required": ["query"]}),
             Tool(name="get_status",
-                 description="Get Agent Reach status: which channels are active.",
+                 description="Get Agent Reach status: which channels are installed and active.",
                  inputSchema={"type": "object", "properties": {}}),
         ]
 
     @server.call_tool()
     async def call_tool(name: str, arguments: dict):
         try:
-            if name == "read_url":
+            if name == "get_status":
-                result = await eyes.read(arguments["url"])
-            elif name == "read_batch":
-                result = await eyes.read_batch(arguments["urls"])
-            elif name == "detect_platform":
-                result = eyes.detect_platform(arguments["url"])
-            elif name == "search":
-                result = await eyes.search(arguments["query"], arguments.get("num_results", 5))
-            elif name == "search_reddit":
-                result = await eyes.search_reddit(arguments["query"], arguments.get("subreddit"), arguments.get("limit", 10))
-            elif name == "search_github":
-                result = await eyes.search_github(arguments["query"], arguments.get("language"), arguments.get("limit", 5))
-            elif name == "search_twitter":
-                result = await eyes.search_twitter(arguments["query"], arguments.get("limit", 10))
-            elif name == "get_status":
                 result = eyes.doctor_report()
             else:
                 result = f"Unknown tool: {name}"

agent_reach/skill/SKILL.md

@@ -1,72 +1,39 @@
 ---
 name: agent-reach
 description: >
-  Give your AI agent eyes to see the entire internet. Read and search across
+  Give your AI agent eyes to see the entire internet. Install and configure
-  Twitter/X, Reddit, YouTube, GitHub, Bilibili, XiaoHongShu, LinkedIn,
+  upstream tools for Twitter/X, Reddit, YouTube, GitHub, Bilibili, XiaoHongShu,
-  Boss直聘, RSS, and any web page — all from a single CLI.
+  LinkedIn, Boss直聘, RSS, and any web page — then call them directly.
-  Use when: (1) reading content from URLs (tweets, Reddit posts, articles, videos),
+  Use when: (1) setting up platform access tools for the first time,
-  (2) searching across platforms (web, Twitter, Reddit, GitHub, YouTube, Bilibili,
+  (2) checking which platforms are available,
-  XiaoHongShu, LinkedIn, Boss直聘),
+  (3) user asks to configure/enable a platform channel.
-  (3) user asks to configure/enable a platform channel,
+  Triggers: "帮我配", "帮我添加", "帮我安装", "agent reach", "install channels",
-  (4) checking channel health or updating Agent Reach.
+  "configure twitter", "enable reddit".
-  Triggers: "search Twitter/Reddit/YouTube", "read this URL", "find posts about",
-  "搜索", "读取", "查一下", "看看这个链接",
-  "帮我配", "帮我添加", "帮我安装".
 ---
 
 # Agent Reach
 
-Read and search the internet across 12+ platforms via unified CLI.
+Install and configure upstream tools for 12+ platforms. After setup, call them directly — no wrapper layer.
 
 ## Setup
 
-First check if agent-reach is installed:
-```bash
-agent-reach doctor
-```
-
-If command not found, install it:
 ```bash
 pip install https://github.com/Panniantong/agent-reach/archive/main.zip
 agent-reach install --env=auto
+agent-reach doctor
 ```
 
-`install` auto-detects your environment and installs core dependencies (Node.js, mcporter, bird CLI, gh CLI). Read the output and run `agent-reach doctor` to see what's active.
+`install` auto-detects your environment and installs core dependencies (Node.js, mcporter, bird CLI, gh CLI, yt-dlp, feedparser). Run `doctor` to see what's active.
 
-## Commands
+## Management
-
-### Read any URL
-```bash
-agent-reach read <url>
-agent-reach read <url> --json    # structured output
-```
-Handles: tweets, Reddit posts, articles, YouTube/Bilibili (transcripts), GitHub repos, LinkedIn profiles, Boss直聘 jobs, XiaoHongShu notes, RSS feeds, and any web page.
-
-### Search
-
-```bash
-agent-reach search "query"             # web search (Exa)
-agent-reach search-twitter "query"     # Twitter/X
-agent-reach search-reddit "query"      # Reddit (--sub <subreddit>)
-agent-reach search-github "query"      # GitHub (--lang <language>)
-agent-reach search-youtube "query"     # YouTube
-agent-reach search-bilibili "query"    # Bilibili (B站)
-agent-reach search-xhs "query"        # XiaoHongShu (小红书)
-agent-reach search-linkedin "query"    # LinkedIn
-agent-reach search-bosszhipin "query"  # Boss直聘
-```
-
-All search commands support `-n <count>` for number of results.
-
-### Management
 
 ```bash
 agent-reach doctor        # channel status overview
-agent-reach watch         # quick health + update check (for scheduled tasks)
+agent-reach watch         # quick health + update check
 agent-reach check-update  # check for new versions
 ```
 
-### Configure channels
+## Configure channels
 
 ```bash
 agent-reach configure twitter-cookies "auth_token=xxx; ct0=yyy"
@@ -76,45 +43,188 @@ agent-reach configure --from-browser chrome    # auto-extract cookies from local
 
 ## Configuring a channel ("帮我配 XXX")
 
-All channels follow the same flow. When a user asks to configure/enable any channel:
+When a user asks to configure/enable any channel:
 
 1. Run `agent-reach doctor`
-2. Find the channel in the output — it shows the current status (✅/⚠️/⬜) and **what to do next**
+2. Find the channel — it shows status (✅/⚠️/⬜) and **what to do next**
-3. Execute whatever you can automatically (install packages, start services, register MCP)
+3. Execute what you can automatically (install packages, start services)
-4. For steps that require human action (scan QR code, paste cookies, login), tell the user exactly what to do
+4. For human-required steps (scan QR, paste cookies), tell the user what to do
-5. After the user completes their part, run `agent-reach doctor` again to verify
+5. Run `agent-reach doctor` again to verify
 
-**Do NOT memorize per-channel steps.** Always rely on `doctor` output — it stays up-to-date even when backends change.
+**Do NOT memorize per-channel steps.** Always rely on `doctor` output.
 
 ### Common human actions
 
-These are things only the user can do. When `doctor` indicates one is needed, explain it clearly:
+- **Paste cookies:** User installs [Cookie-Editor](https://chromewebstore.google.com/detail/cookie-editor/hlkenndednhfkekhgcdicdfddnkalmdm) → goes to the website → exports Header String → sends it to you → you run `agent-reach configure <platform>-cookies "..."`
+- **Scan QR code:** User opens the URL on their phone/browser and scans with the platform's app
+- **Proxy:** Reddit/Bilibili/XiaoHongShu may block server IPs — suggest a residential proxy if on a server
 
-- **Paste cookies:** User installs [Cookie-Editor](https://chromewebstore.google.com/detail/cookie-editor/hlkenndednhfkekhgcdicdfddnkalmdm) Chrome extension → goes to the website → exports Header String → sends it to you → you run `agent-reach configure <platform>-cookies "..."`
+---
-- **Scan QR code:** User opens the URL shown in `doctor` output on their phone/browser and scans with the platform's app
-- **Browser login:** Some MCP services need a one-time browser login; on servers without a display, user may need VNC
-- **Proxy:** Reddit/Bilibili/XiaoHongShu block server IPs — suggest a residential proxy (~$1/month) if on a server
 
-## Tips
+## Using Upstream Tools Directly
 
-- Always try `agent-reach read <url>` first for any URL — it auto-detects the platform
+After `agent-reach install`, call the upstream tools directly. No need for `agent-reach read` or `agent-reach search`.
-- If a channel is ⬜ but the user hasn't asked for it, don't push — let them opt in
+
-- If a channel breaks, run `agent-reach doctor` to diagnose
+### Twitter/X (bird CLI)
-- LinkedIn and Boss直聘 have Jina Reader fallback even without full setup
+
-- Twitter search 在 bird 失败时会自动 fallback 到 Exa 搜索
+```bash
+# Search tweets
+bird search "query" --json -n 10
+
+# Read a specific tweet
+bird read https://x.com/user/status/123 --json
+
+# Read a user's timeline
+bird timeline @username --json -n 20
+```
+
+### YouTube (yt-dlp)
+
+```bash
+# Get video metadata
+yt-dlp --dump-json "https://www.youtube.com/watch?v=xxx"
+
+# Download subtitles only
+yt-dlp --write-sub --write-auto-sub --sub-lang "zh-Hans,zh,en" --skip-download -o "/tmp/%(id)s" "URL"
+# Then read the .vtt file
+
+# Search (yt-dlp ytsearch)
+yt-dlp --dump-json "ytsearch5:query"
+```
+
+### Bilibili (yt-dlp)
+
+```bash
+# Get video metadata
+yt-dlp --dump-json "https://www.bilibili.com/video/BVxxx"
+
+# Download subtitles
+yt-dlp --write-sub --write-auto-sub --sub-lang "zh-Hans,zh,en" --convert-subs vtt --skip-download -o "/tmp/%(id)s" "URL"
+```
+
+### Reddit (JSON API)
+
+```bash
+# Read a subreddit
+curl -s "https://www.reddit.com/r/python/hot.json?limit=10" -H "User-Agent: agent-reach/1.0"
+
+# Read a post with comments
+curl -s "https://www.reddit.com/r/python/comments/POST_ID.json" -H "User-Agent: agent-reach/1.0"
+
+# Search
+curl -s "https://www.reddit.com/search.json?q=query&limit=10" -H "User-Agent: agent-reach/1.0"
+```
+
+Note: On servers, Reddit may block your IP. Use proxy or search via Exa instead.
+
+### 小红书 / XiaoHongShu (mcporter + xiaohongshu-mcp)
+
+```bash
+# Search notes
+mcporter call 'xiaohongshu.search_feeds(keyword: "query")'
+
+# Read a note
+mcporter call 'xiaohongshu.get_feed_detail(feed_id: "xxx", xsec_token: "yyy")'
+
+# Get comments
+mcporter call 'xiaohongshu.get_feed_comments(feed_id: "xxx", xsec_token: "yyy")'
+
+# Post a note
+mcporter call 'xiaohongshu.create_image_feed(title: "标题", desc: "内容", image_paths: ["/path/to/img.jpg"])'
+```
+
+### GitHub (gh CLI)
+
+```bash
+# Search repos
+gh search repos "query" --sort stars --limit 10
+
+# View a repo
+gh repo view owner/repo
+
+# Search code
+gh search code "query" --language python
+
+# List issues
+gh issue list -R owner/repo --state open
+
+# View a specific issue/PR
+gh issue view 123 -R owner/repo
+```
+
+### Web — Any URL (Jina Reader)
+
+```bash
+# Read any webpage as markdown
+curl -s "https://r.jina.ai/URL" -H "Accept: text/markdown"
+
+# Search the web
+curl -s "https://s.jina.ai/query" -H "Accept: text/markdown"
+```
+
+### Exa Search (mcporter + exa MCP)
+
+```bash
+# Web search
+mcporter call 'exa.web_search_exa(query: "query", numResults: 5)'
+
+# Code search (GitHub, StackOverflow, docs)
+mcporter call 'exa.get_code_context_exa(query: "how to parse JSON in Python", tokensNum: 3000)'
+
+# Company research
+mcporter call 'exa.company_research_exa(companyName: "OpenAI")'
+```
+
+### LinkedIn (mcporter + linkedin-scraper-mcp)
+
+```bash
+# View a profile
+mcporter call 'linkedin.get_person_profile(linkedin_url: "https://linkedin.com/in/username")'
+
+# Search people
+mcporter call 'linkedin.search_people(keyword: "AI engineer", limit: 10)'
+
+# View company
+mcporter call 'linkedin.get_company_profile(linkedin_url: "https://linkedin.com/company/xxx")'
+```
+
+Fallback: `curl -s "https://r.jina.ai/https://linkedin.com/in/username"`
+
+### Boss直聘 (mcporter + mcp-bosszp)
+
+```bash
+# Browse recommended jobs
+mcporter call 'bosszhipin.get_recommend_jobs_tool(page: 1)'
+
+# Search jobs
+mcporter call 'bosszhipin.search_jobs_tool(keyword: "Python", city: "北京", page: 1)'
+
+# View job details
+mcporter call 'bosszhipin.get_job_detail_tool(job_url: "https://www.zhipin.com/job_detail/xxx")'
+```
+
+Fallback: `curl -s "https://r.jina.ai/https://www.zhipin.com/job_detail/xxx"`
+
+### RSS (feedparser)
+
+```python
+python3 -c "
+import feedparser
+d = feedparser.parse('https://example.com/feed')
+for e in d.entries[:5]:
+    print(f'{e.title} — {e.link}')
+"
+```
 
 ## Troubleshooting
 
 ### Twitter "fetch failed"
 
-bird CLI 用 Node.js 原生 `fetch()`，不走 `HTTP_PROXY`/`HTTPS_PROXY` 环境变量。
+bird CLI uses Node.js native `fetch()`, which doesn't respect `HTTP_PROXY`. Solutions:
-Agent Reach 已内置解决方案：通过 `undici` 的 `EnvHttpProxyAgent` 自动注入代理。
+1. Ensure `undici` is installed: `npm install -g undici`
+2. Configure proxy: `agent-reach configure proxy http://user:pass@ip:port`
+3. If still failing, use transparent proxy (Clash TUN, Proxifier)
 
-**你（Agent）需要确认：**
+### Channel broken?
-1. `undici` 已全局安装：`npm list -g undici`，没有就装：`npm install -g undici`
-2. 用户已配置代理：`agent-reach configure proxy http://user:pass@ip:port`
-3. Agent Reach 会自动处理注入，不需要用户额外操作
 
-如果以上都没问题仍然 "fetch failed"，建议用户使用透明代理（如 Clash TUN 模式、Proxifier）。
+Run `agent-reach doctor` — it shows what's wrong and how to fix it.
-
-详见 https://github.com/Panniantong/Agent-Reach/blob/main/docs/troubleshooting.md

docs/README_en.md

@@ -108,14 +108,14 @@ After the Skill is installed, the Agent will auto-detect whether `agent-reach` C
 
 No configuration needed — just tell your Agent:
 
-- "Read this link" → any web page
+- "Read this link" → `curl https://r.jina.ai/URL` for any web page
-- "What's this GitHub repo about?" → repos, issues, code
+- "What's this GitHub repo about?" → `gh repo view owner/repo`
-- "What does this video cover?" → YouTube / Bilibili subtitles
+- "What does this video cover?" → `yt-dlp --dump-json URL` for subtitles
-- "Read this tweet" → Twitter posts
+- "Read this tweet" → `bird read URL --json`
-- "Subscribe to this RSS" → RSS / Atom feeds
+- "Subscribe to this RSS" → `feedparser` to parse feeds
-- "Search GitHub for LLM frameworks" → GitHub search
+- "Search GitHub for LLM frameworks" → `gh search repos "LLM framework"`
 
-**No commands to remember.** The Agent knows what to call.
+**No commands to remember.** The Agent reads SKILL.md and knows what to call.
 
 ---
 
@@ -171,26 +171,30 @@ Every time you spin up a new Agent, you spend time finding tools, installing dep
 
 Agent Reach does one simple thing: **it makes those tool selection and configuration decisions for you.**
 
+After installation, your Agent calls the upstream tools directly (bird CLI, yt-dlp, mcporter, gh CLI, etc.) — no wrapper layer in between.
+
 ### 🔌 Every Channel is Pluggable
 
-Each platform is a single Python file implementing a unified interface. **Backends can be swapped anytime** — when a better tool comes along, change one file and nothing else breaks.
+Each platform maps to an upstream tool. **Don't like one? Swap it out.**
 
 ```
 channels/
 ├── web.py          → Jina Reader     ← swap to Firecrawl, Crawl4AI…
-├── twitter.py      → bird           ← swap to Nitter, official API…
+├── twitter.py      → bird            ← swap to Nitter, official API…
-├── youtube.py      → yt-dlp           ← swap to YouTube API, Whisper…
+├── youtube.py      → yt-dlp          ← swap to YouTube API, Whisper…
 ├── github.py       → gh CLI          ← swap to REST API, PyGithub…
-├── bilibili.py     → yt-dlp           ← swap to bilibili-api…
+├── bilibili.py     → yt-dlp          ← swap to bilibili-api…
 ├── reddit.py       → JSON API + Exa  ← swap to PRAW, Pushshift…
 ├── xiaohongshu.py  → mcporter MCP    ← swap to other XHS tools…
 ├── linkedin.py     → linkedin-mcp    ← swap to LinkedIn API…
 ├── bosszhipin.py   → mcp-bosszp      ← swap to other job tools…
-├── rss.py          → feedparser       ← swap to atoma…
+├── rss.py          → feedparser      ← swap to atoma…
 ├── exa_search.py   → mcporter MCP    ← swap to Tavily, SerpAPI…
-└── __init__.py     → Channel registry
+└── __init__.py     → Channel registry (for doctor checks)
 ```
 
+Each channel file only checks whether its upstream tool is installed and working (`check()` method for `agent-reach doctor`). The actual reading and searching is done by calling the upstream tools directly.
+
 ### Current Tool Choices
 
 | Scenario | Tool | Why |
@@ -226,13 +230,13 @@ This project was entirely vibe-coded 🎸 There might be rough edges here and th
 <details>
 <summary><strong>How to search Twitter/X with AI agent without paying for API?</strong></summary>
 
-Agent Reach uses the [bird CLI](https://www.npmjs.com/package/@steipete/bird) with cookie-based authentication — completely free, no Twitter API subscription needed. After installing Agent Reach, export your Twitter cookies using the Cookie-Editor Chrome extension, run `agent-reach configure twitter-cookies "your_cookies"`, and your agent can search with `agent-reach search-twitter "query"`.
+Agent Reach uses the [bird CLI](https://www.npmjs.com/package/@steipete/bird) with cookie-based authentication — completely free, no Twitter API subscription needed. After installing Agent Reach, export your Twitter cookies using the Cookie-Editor Chrome extension, run `agent-reach configure twitter-cookies "your_cookies"`, and your agent can search with `bird search "query" --json`.
 </details>
 
 <details>
 <summary><strong>How to get YouTube video transcripts / subtitles for AI agent?</strong></summary>
 
-Simply run `agent-reach read https://youtube.com/watch?v=xxx`. It automatically extracts transcripts using yt-dlp. Supports multiple languages, no API key required.
+`yt-dlp --dump-json "https://youtube.com/watch?v=xxx"` extracts video metadata; `yt-dlp --write-sub --skip-download "URL"` extracts subtitles. Supports multiple languages, no API key required.
 </details>
 
 <details>
@@ -244,7 +248,7 @@ Reddit blocks datacenter IPs. Configure a residential proxy: `agent-reach config
 <details>
 <summary><strong>Does Agent Reach work with Claude Code / Cursor / Windsurf / OpenClaw?</strong></summary>
 
-Yes! Agent Reach is a standard CLI tool. Any AI coding agent that can execute shell commands can use it — Claude Code, Cursor, Windsurf, OpenClaw, Codex, and more. Just `pip install agent-reach` and start using it.
+Yes! Agent Reach is an installer + configuration tool. Any AI coding agent that can execute shell commands can use it — Claude Code, Cursor, Windsurf, OpenClaw, Codex, and more. Just `pip install agent-reach`, run `agent-reach install`, and the agent can start using the upstream tools immediately.
 </details>
 
 <details>
@@ -262,7 +266,7 @@ Agent Reach uses bird CLI which accesses Twitter via cookie auth — same as you
 <details>
 <summary><strong>How to read XiaoHongShu / 小红书 content programmatically?</strong></summary>
 
-Agent Reach integrates with xiaohongshu-mcp (runs in Docker). After setup, use `agent-reach read <xiaohongshu_url>` or `agent-reach search-xhs "query"` to search and read XiaoHongShu notes.
+Agent Reach integrates with xiaohongshu-mcp (runs in Docker). After setup, use `mcporter call 'xiaohongshu.get_feed_detail(...)'` to read notes or `mcporter call 'xiaohongshu.search_feeds(keyword: "query")'` to search.
 </details>
 
 ---

docs/install.md

@@ -20,7 +20,7 @@ Copy this to your AI Agent:
 
 ### Goal
 
-Install Agent Reach and make sure **all channels are working**. Your user expects you to have full internet access capabilities when this is done — reading tweets, searching Reddit, watching YouTube, browsing GitHub, and more.
+Install Agent Reach and all upstream tools so your user's agent has full internet access. After installation, you'll use the upstream tools directly (bird CLI, yt-dlp, mcporter, gh CLI, etc.) — Agent Reach is just the installer and health checker.
 
 ### ⚠️ Boundaries
 
@@ -228,13 +228,19 @@ If the user wants a different agent to handle it, let them choose.
 | `agent-reach check-update` | Check for new versions |
 | `agent-reach configure twitter-cookies "..."` | Unlock Twitter search + posting |
 | `agent-reach configure proxy URL` | Unlock Reddit + Bilibili on servers |
-| `agent-reach read <url>` | Read any URL |
+
-| `agent-reach search "query"` | Search the web |
+After installation, use upstream tools directly. See SKILL.md for the full command reference:
-| `agent-reach search-reddit "query"` | Search Reddit |
+
-| `agent-reach search-twitter "query"` | Search Twitter |
+| Platform | Upstream Tool | Example |
-| `agent-reach search-github "query"` | Search GitHub |
+|----------|--------------|---------|
-| `agent-reach search-youtube "query"` | Search YouTube |
+| Twitter/X | `bird` | `bird search "query" --json` |
-| `agent-reach search-bilibili "query"` | Search Bilibili |
+| YouTube | `yt-dlp` | `yt-dlp --dump-json URL` |
-| `agent-reach search-xhs "query"` | Search XiaoHongShu |
+| Bilibili | `yt-dlp` | `yt-dlp --dump-json URL` |
-| `agent-reach search-linkedin "query"` | Search LinkedIn |
+| Reddit | `curl` | `curl -s "https://reddit.com/r/xxx.json"` |
-| `agent-reach search-bosszhipin "query"` | Search Boss直聘 |
+| GitHub | `gh` | `gh search repos "query"` |
+| Web | `curl` + Jina | `curl -s "https://r.jina.ai/URL"` |
+| Exa Search | `mcporter` | `mcporter call 'exa.web_search_exa(...)'` |
+| 小红书 | `mcporter` | `mcporter call 'xiaohongshu.search_feeds(...)'` |
+| LinkedIn | `mcporter` | `mcporter call 'linkedin.get_person_profile(...)'` |
+| Boss直聘 | `mcporter` | `mcporter call 'bosszhipin.search_jobs_tool(...)'` |
+| RSS | `feedparser` | `python3 -c "import feedparser; ..."` |

docs/troubleshooting.md

@@ -26,11 +26,10 @@
 
 ### 方案 3：不用 bird，用 Exa 搜索替代
 
-Agent Reach 在 bird 失败时会自动 fallback 到 Exa 搜索。Exa 支持搜索 x.com 上的内容，虽然不如 bird 实时，但不受代理限制：
+bird 不可用时，可以直接用 Exa 搜索 Twitter 内容：
 
 ```bash
-agent-reach search-twitter "query"  # bird 失败时自动用 Exa
+mcporter call 'exa.web_search_exa(query: "site:x.com query", numResults: 10)'
-agent-reach search "site:x.com query"  # 直接用 Exa 搜索
 ```
 
 ### 方案 4：配置 Node.js 全局代理（高级）