代码验证 - 核心逻辑与集成示例
技术研究 代码示例 集成指南
展示 Agent-Reach 的核心代码逻辑、配置示例、Agent 集成方式。包括渠道检测实现、CLI 命令解析、MCP 服务配置等关键代码片段。
核心代码结构
项目文件组织
agent-reach/
├── agent_reach/
│ ├── __init__.py
│ ├── cli.py # CLI 入口
│ ├── config.py # 配置管理
│ ├── doctor.py # 诊断命令
│ ├── install.py # 安装命令
│ ├── uninstall.py # 卸载命令
│ ├── channels/ # 渠道检测模块
│ │ ├── __init__.py
│ │ ├── web.py
│ │ ├── twitter.py
│ │ ├── youtube.py
│ │ └── ...
│ ├── skill/ # Agent 技能文档
│ │ └── SKILL.md
│ └── guides/ # 配置指南
│ └── *.md
├── tests/
├── pyproject.toml
└── README.md核心实现解析
1. 渠道检测接口
抽象接口定义：
# agent_reach/channels/__init__.py
from dataclasses import dataclass
from enum import Enum
from typing import Protocol
class ChannelStatus(Enum):
READY = "ready" # 立即可用
MISSING = "missing" # 工具未安装
MISCONFIGURED = "misconfigured" # 需要配置
@dataclass
class CheckResult:
status: ChannelStatus
message: str
channel_name: str
upstream_tool: str
class Channel(Protocol):
"""渠道接口规范"""
def check(self) -> CheckResult:
"""检测渠道是否可用"""
...
def get_name(self) -> str:
"""获取渠道名称"""
...
def get_upstream_tool(self) -> str:
"""获取上游工具名称"""
...Twitter 渠道实现：
# agent_reach/channels/twitter.py
import subprocess
from . import Channel, CheckResult, ChannelStatus
class TwitterChannel:
"""Twitter/X 渠道 - 使用 xreach CLI"""
def get_name(self) -> str:
return "Twitter/X"
def get_upstream_tool(self) -> str:
return "xreach CLI"
def check(self) -> CheckResult:
"""检查 xreach CLI 是否可用"""
try:
# 检查 xreach 是否安装
result = subprocess.run(
["xreach", "--version"],
capture_output=True,
text=True,
timeout=5
)
if result.returncode == 0:
# xreach 已安装，检查是否需要配置 Cookie
if self._has_cookies():
return CheckResult(
status=ChannelStatus.READY,
message="xreach CLI 已安装，Cookie 已配置",
channel_name=self.get_name(),
upstream_tool=self.get_upstream_tool()
)
else:
return CheckResult(
status=ChannelStatus.MISCONFIGURED,
message="xreach CLI 已安装，但需要配置 Cookie",
channel_name=self.get_name(),
upstream_tool=self.get_upstream_tool()
)
else:
return CheckResult(
status=ChannelStatus.MISSING,
message="xreach CLI 未安装",
channel_name=self.get_name(),
upstream_tool=self.get_upstream_tool()
)
except subprocess.TimeoutExpired:
return CheckResult(
status=ChannelStatus.MISSING,
message="xreach CLI 检测超时",
channel_name=self.get_name(),
upstream_tool=self.get_upstream_tool()
)
except FileNotFoundError:
return CheckResult(
status=ChannelStatus.MISSING,
message="xreach CLI 未找到，请先安装",
channel_name=self.get_name(),
upstream_tool=self.get_upstream_tool()
)
def _has_cookies(self) -> bool:
"""检查是否配置了 Twitter Cookie"""
from agent_reach.config import load_config
config = load_config()
return bool(config.get("twitter", {}).get("cookies"))2. 诊断命令（doctor）
# agent_reach/doctor.py
from rich.console import Console
from rich.table import Table
from .channels import discover_all_channels
console = Console()
def run_doctor():
"""执行诊断检查"""
channels = discover_all_channels()
ready = []
missing = []
misconfigured = []
# 检测所有渠道
for channel in channels:
result = channel.check()
if result.status == ChannelStatus.READY:
ready.append(result)
elif result.status == ChannelStatus.MISSING:
missing.append(result)
else:
misconfigured.append(result)
# 生成报告
console.print("\n👁️ Agent Reach Status")
console.print("=" * 40)
if ready:
console.print("\n✅ Ready to use:")
for r in ready:
console.print(f" ✅ {r.channel_name} — {r.message}")
if misconfigured:
console.print("\n🔧 Configurable:")
for r in misconfigured:
console.print(f" ⚠️ {r.channel_name} — {r.message}")
if missing:
console.print("\n❌ Missing:")
for r in missing:
console.print(f" ❌ {r.channel_name} — {r.message}")
# 总结
total = len(channels)
available = len(ready)
console.print(f"\nStatus: {available}/{total} channels available")
return available == total3. 配置管理
# agent_reach/config.py
import os
import yaml
from pathlib import Path
from typing import Optional
CONFIG_DIR = Path.home() / ".agent-reach"
CONFIG_FILE = CONFIG_DIR / "config.yaml"
class ConfigManager:
"""配置管理器"""
def __init__(self):
self.config_path = CONFIG_FILE
self._config: Optional[dict] = None
def load(self) -> dict:
"""加载配置"""
if self._config is not None:
return self._config
if not self.config_path.exists():
self._config = {}
return self._config
with open(self.config_path, "r") as f:
self._config = yaml.safe_load(f) or {}
return self._config
def save(self, config: dict):
"""保存配置"""
# 确保目录存在
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
# 设置文件权限为 600（仅所有者可读写）
if not self.config_path.exists():
self.config_path.touch(mode=0o600)
with open(self.config_path, "w") as f:
yaml.safe_dump(config, f)
# 确保权限正确
os.chmod(self.config_path, 0o600)
self._config = config
def get(self, key: str, default=None):
"""获取配置项"""
config = self.load()
return config.get(key, default)
def set(self, key: str, value):
"""设置配置项"""
config = self.load()
config[key] = value
self.save(config)
def configure_twitter_cookies(self, cookies: str):
"""配置 Twitter Cookie"""
config = self.load()
if "twitter" not in config:
config["twitter"] = {}
config["twitter"]["cookies"] = cookies
self.save(config)
def configure_proxy(self, proxy_url: str):
"""配置代理"""
config = self.load()
if "proxy" not in config:
config["proxy"] = {}
config["proxy"]["http"] = proxy_url
config["proxy"]["https"] = proxy_url
self.save(config)
# 全局配置实例
_config_manager = ConfigManager()
def load_config() -> dict:
return _config_manager.load()
def save_config(config: dict):
_config_manager.save(config)4. 安装命令
# agent_reach/install.py
import subprocess
import sys
from pathlib import Path
from rich.console import Console
console = Console()
class Installer:
"""Agent-Reach 安装器"""
def __init__(self, env: str = "auto", safe: bool = False, dry_run: bool = False):
self.env = env
self.safe = safe
self.dry_run = dry_run
def install(self):
"""执行安装"""
console.print("👁️ Installing Agent Reach...\n")
# 1. 检测环境
detected_env = self.detect_environment()
console.print(f"🔍 Detected environment: {detected_env}")
# 2. 检测并安装依赖
self.check_dependencies()
# 3. 配置 MCP 服务
self.setup_mcp()
# 4. 注册 Agent 技能
self.register_skills()
# 5. 运行诊断
console.print("\n✅ Installation complete!")
console.print("Running doctor...\n")
from .doctor import run_doctor
run_doctor()
def detect_environment(self) -> str:
"""检测运行环境"""
if self.env != "auto":
return self.env
# 检测是否是服务器
if self._is_server():
return "server"
return "local"
def _is_server(self) -> bool:
"""检测是否是云服务器"""
# 检查云厂商元数据
cloud_indicators = [
"EC2", # AWS
"GCE", # GCP
"Azure", # Azure
"DigitalOcean",
"Linode"
]
try:
with open("/sys/class/dmi/product_name", "r") as f:
product_name = f.read().strip()
return any(indicator in product_name for indicator in cloud_indicators)
except FileNotFoundError:
return False
def check_dependencies(self):
"""检查并安装依赖"""
dependencies = [
{"name": "Node.js", "check": ["node", "--version"], "install": "See https://nodejs.org"},
{"name": "gh CLI", "check": ["gh", "--version"], "install": "See https://cli.github.com"},
{"name": "xreach CLI", "check": ["xreach", "--version"], "install": "npm install -g xreach-cli"},
{"name": "yt-dlp", "check": ["yt-dlp", "--version"], "install": "pip install yt-dlp"},
]
for dep in dependencies:
if self._check_command(dep["check"]):
console.print(f"✅ {dep['name']} already installed")
else:
if self.safe:
console.print(f"⚠️ {dep['name']} not found. Install manually: {dep['install']}")
elif self.dry_run:
console.print(f"🔍 Would install: {dep['name']}")
else:
console.print(f"📦 Installing {dep['name']}...")
self._install_dependency(dep)
def _check_command(self, cmd: list) -> bool:
"""检查命令是否可用"""
try:
result = subprocess.run(cmd, capture_output=True, timeout=5)
return result.returncode == 0
except (subprocess.TimeoutExpired, FileNotFoundError):
return False
def setup_mcp(self):
"""配置 MCP 服务"""
console.print("\n🔧 Setting up MCP services...")
# 检查 Docker 是否可用
if not self._check_command(["docker", "--version"]):
console.print("⚠️ Docker not found. MCP services will not be available.")
console.print(" Install Docker: https://docs.docker.com/get-docker/")
return
# 拉取并启动 MCP 服务容器
mcp_services = [
"xiaohongshu-mcp",
"douyin-mcp-server",
"mcporter"
]
for service in mcp_services:
if self.dry_run:
console.print(f"🔍 Would setup MCP service: {service}")
else:
console.print(f"📦 Setting up {service}...")
# 实际会执行 docker pull 和 docker run
self._setup_mcp_service(service)
def register_skills(self):
"""注册 Agent 技能"""
console.print("\n📚 Registering Agent skills...")
# 找到 Agent 的 skills 目录
agent_skills_dirs = self._find_agent_skills_dirs()
for skills_dir in agent_skills_dirs:
skill_file = skills_dir / "agent-reach.md"
if self.dry_run:
console.print(f"🔍 Would register skill at: {skill_file}")
else:
# 复制 SKILL.md 到 Agent 的 skills 目录
self._copy_skill_file(skills_dir)
console.print(f"✅ Registered skill at: {skill_file}")Agent 集成示例
Claude Code 集成
安装后自动注册：
# ~/.claude/skills/agent-reach.md
## Twitter/X 操作
### 搜索推文
```bash
xreach search "query" --json读取单条推文
xreach tweet "https://twitter.com/user/status/123" --json配置 Cookie
- 使用 Cookie-Editor 扩展导出 Twitter Cookie
- 运行：
agent-reach configure twitter-cookies "<cookies>"
YouTube 操作
提取字幕
yt-dlp --write-sub --skip-download "https://youtube.com/watch?v=xxx"获取视频元数据
yt-dlp --dump-json "https://youtube.com/watch?v=xxx"GitHub 操作
查看仓库信息
gh repo view owner/repo搜索仓库
gh search repos "LLM framework"认证（可选）
gh auth login
### Cursor 集成
Cursor 会自动读取项目中的 SKILL.md 文件：
```python
# .cursor/rules/agent-reach.mdc
---
description: Agent-Reach usage guidelines
globs: **
---
When user asks about social media or web content:
1. Check if agent-reach is installed
2. Use appropriate upstream tool:
- Twitter: xreach
- YouTube: yt-dlp
- GitHub: gh CLI
- Web: Jina Reader (curl https://r.jina.ai/URL)
3. If tool not available, suggest: agent-reach installWindsurf 集成
Windsurf 通过 Cascade 功能自动发现和使用 Agent-Reach：
# .windsurf/rules/agent-reach.yaml
tools:
- name: agent-reach
commands:
twitter_search: "xreach search {query} --json"
youtube_transcript: "yt-dlp --write-sub --skip-download {url}"
github_view: "gh repo view {repo}"
config:
check_command: "agent-reach doctor"
install_command: "pip install agent-reach"MCP 服务配置
小红书 MCP 配置
# mcporter 配置文件
mcpServers:
xiaohongshu:
command: docker
args:
- run
- --rm
- -p
- "3000:3000"
- xiaohongshu-mcp
env:
XHS_COOKIE: "${XHS_COOKIE}"调用示例
# 通过 mcporter 调用小红书 API
import subprocess
import json
def get_xiaohongshu_feed(note_id: str) -> dict:
"""获取小红书笔记详情"""
cmd = [
"mcporter", "call",
f"xiaohongshu.get_feed_detail(note_id: '{note_id}')"
]
result = subprocess.run(cmd, capture_output=True, text=True)
return json.loads(result.stdout)
def search_xiaohongshu(keyword: str) -> list:
"""搜索小红书笔记"""
cmd = [
"mcporter", "call",
f"xiaohongshu.search_feeds(keyword: '{keyword}')"
]
result = subprocess.run(cmd, capture_output=True, text=True)
return json.loads(result.stdout)配置命令示例
配置 Twitter Cookie
# 1. 使用 Cookie-Editor 导出 Cookie
# 2. 运行配置命令
agent-reach configure twitter-cookies "cookie_string_here"
# 3. 验证配置
agent-reach doctor
# 应显示：✅ Twitter/X — xreach CLI 已安装，Cookie 已配置配置代理
# 配置 HTTP/HTTPS 代理
agent-reach configure proxy "http://user:pass@ip:port"
# 验证
curl -x "http://user:pass@ip:port" https://www.reddit.com配置 GitHub Token（可选）
# 通过 gh CLI 认证
gh auth login
# 或手动配置 token
agent-reach configure github-token "ghp_xxx"诊断流程
doctor 命令完整输出
$ agent-reach doctor
👁️ Agent Reach Status
========================================
✅ Ready to use:
✅ Web pages — Jina Reader available
✅ YouTube — yt-dlp installed
✅ GitHub — gh CLI ready (public repos)
✅ Twitter/X — xreach CLI ready (read only)
✅ RSS — feedparser available
✅ Bilibili — yt-dlp available (local)
⚠️ Limited functionality:
⚠️ Twitter/X search — needs Cookie configuration
⚠️ Bilibili (server) — needs proxy for server IPs
❌ Not configured:
❌ Reddit posts — needs proxy
❌ XiaoHongShu — needs MCP service + Cookie
❌ LinkedIn — needs MCP service
🔧 Quick fixes:
- Twitter Cookie: agent-reach configure twitter-cookies "<cookies>"
- Proxy: agent-reach configure proxy "http://user:pass@ip:port"
- MCP services: Install Docker and run: agent-reach install
Status: 6/12 channels fully available安全模式示例
$ agent-reach install --safe
👁️ Agent Reach — Safe Mode
========================================
The following dependencies will be installed:
1. Node.js 18+
Current: Not installed
Install: See https://nodejs.org
2. gh CLI
Current: Not installed
Install: See https://cli.github.com
3. xreach CLI
Current: Not installed
Install: npm install -g xreach-cli
4. yt-dlp
Current: Not installed
Install: pip install yt-dlp
5. MCP Services (Docker required)
- xiaohongshu-mcp
- douyin-mcp-server
- mcporter
Remove --safe flag to auto-install, or install manually and run:
agent-reach install --skip-depsDry Run 示例
$ agent-reach install --dry-run
👁️ Agent Reach — Dry Run
========================================
The following actions will be performed:
[INSTALL]
- Install Node.js dependencies: xreach-cli
- Install Python packages: yt-dlp, feedparser
- Setup MCP services via Docker
[CONFIGURE]
- Configure Exa search (free, auto-configured)
- Register SKILL.md to Agent skills directory
- Create config file at ~/.agent-reach/config.yaml
[VERIFY]
- Run doctor to verify installation
No changes will be made. Remove --dry-run to proceed.结论
Agent-Reach 的核心代码设计遵循以下原则：
- 接口标准化：所有渠道实现统一接口
- 检测优先：安装前检测，配置前验证
- 安全默认：Cookie 本地存储，文件权限 600
- Agent 友好：自动生成技能文档
通过可插拔的渠道设计和上游工具集成，实现了零包装开销、易扩展、易维护的目标。