From d7fc1ad93a31dca48464b050657ee09eeb5745ee Mon Sep 17 00:00:00 2001 From: Xule Lin <43122877+linxule@users.noreply.github.com> Date: Wed, 8 Apr 2026 05:03:42 +0100 Subject: [PATCH 1/3] feat(gateway): add WeChat platform adapter with iLink 2.1.x protocol support 3-file architecture (adapter/transport/state) implementing the iLink Bot API protocol from openclaw-weixin SDK 2.1.7: - iLink-App-Id + iLink-App-ClientVersion headers on all API requests - IDC redirect (scaned_but_redirect) in QR login flow - Context token persistence with reload at startup and clear on session expiry - Referenced message (ref_msg) extraction for quoted replies - SILK voice transcoding with graceful fallback - CDN upload with upload_full_url priority and exponential backoff - CDN download with full_url forward-compatibility - Dynamic channel_version (not hardcoded) - AES-128-ECB media encryption with dual key format support Full platform integration: enum, env overrides, adapter factory, auth maps, toolsets, CLI setup wizard, status display, cron delivery, send_message tool, channel directory, prompt hints, env sanitizer keys, and login script. Co-Authored-By: Claude Opus 4.6 (1M context) --- agent/prompt_builder.py | 9 + cron/scheduler.py | 3 +- gateway/channel_directory.py | 4 +- gateway/config.py | 34 + gateway/platforms/wechat.py | 871 ++++++++++++++++++++++++++ gateway/platforms/wechat_state.py | 105 ++++ gateway/platforms/wechat_transport.py | 466 ++++++++++++++ gateway/run.py | 16 +- hermes_cli/config.py | 3 + hermes_cli/gateway.py | 24 + hermes_cli/skills_config.py | 1 + hermes_cli/status.py | 1 + hermes_cli/tools_config.py | 1 + scripts/wechat_login.py | 259 ++++++++ tools/cronjob_tools.py | 2 +- tools/send_message_tool.py | 33 + toolsets.py | 8 +- 17 files changed, 1833 insertions(+), 7 deletions(-) create mode 100644 gateway/platforms/wechat.py create mode 100644 gateway/platforms/wechat_state.py create mode 100644 gateway/platforms/wechat_transport.py create mode 100644 scripts/wechat_login.py diff --git a/agent/prompt_builder.py b/agent/prompt_builder.py index df5532e12580..060bc2c7f4b4 100644 --- a/agent/prompt_builder.py +++ b/agent/prompt_builder.py @@ -301,6 +301,15 @@ def _strip_yaml_frontmatter(content: str) -> str: "files arrive as downloadable documents. You can also include image " "URLs in markdown format ![alt](url) and they will be sent as photos." ), + "wechat": ( + "You are on a text messaging communication platform, WeChat. " + "Please do not use markdown as it does not render. " + "You can send media files natively: to deliver a file to the user, " + "include MEDIA:/absolute/path/to/file in your response. Images " + "(.png, .jpg, .webp) appear as photos, videos (.mp4, .mov) play inline, " + "and other files arrive as downloadable documents. Voice attachments may " + "be delivered as downloadable audio files rather than native voice bubbles." + ), "email": ( "You are communicating via email. Write clear, well-structured responses " "suitable for email. Use plain text formatting (no markdown). " diff --git a/cron/scheduler.py b/cron/scheduler.py index f694f4407019..b6d9d5e9ca0e 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -44,7 +44,7 @@ _KNOWN_DELIVERY_PLATFORMS = frozenset({ "telegram", "discord", "slack", "whatsapp", "signal", "matrix", "mattermost", "homeassistant", "dingtalk", "feishu", - "wecom", "sms", "email", "webhook", + "wecom", "wechat", "sms", "email", "webhook", }) from cron.jobs import get_due_jobs, mark_job_run, save_job_output, advance_next_run @@ -196,6 +196,7 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> None: "dingtalk": Platform.DINGTALK, "feishu": Platform.FEISHU, "wecom": Platform.WECOM, + "wechat": Platform.WECHAT, "email": Platform.EMAIL, "sms": Platform.SMS, } diff --git a/gateway/channel_directory.py b/gateway/channel_directory.py index 0d1247217579..7f2dc59e2765 100644 --- a/gateway/channel_directory.py +++ b/gateway/channel_directory.py @@ -76,8 +76,8 @@ def build_channel_directory(adapters: Dict[Any, Any]) -> Dict[str, Any]: except Exception as e: logger.warning("Channel directory: failed to build %s: %s", platform.value, e) - # Telegram, WhatsApp & Signal can't enumerate chats -- pull from session history - for plat_name in ("telegram", "whatsapp", "signal", "email", "sms"): + # Telegram, WhatsApp, Signal, and WeChat can't enumerate chats -- pull from session history + for plat_name in ("telegram", "whatsapp", "signal", "wechat", "email", "sms"): if plat_name not in platforms: platforms[plat_name] = _build_from_sessions(plat_name) diff --git a/gateway/config.py b/gateway/config.py index 470eee7f2f21..9082ff38cbcd 100644 --- a/gateway/config.py +++ b/gateway/config.py @@ -63,6 +63,7 @@ class Platform(Enum): WEBHOOK = "webhook" FEISHU = "feishu" WECOM = "wecom" + WECHAT = "wechat" @dataclass @@ -287,6 +288,9 @@ def get_connected_platforms(self) -> List[Platform]: # WeCom uses extra dict for bot credentials elif platform == Platform.WECOM and config.extra.get("bot_id"): connected.append(platform) + # WeChat uses bot token from QR login + elif platform == Platform.WECHAT and (config.token or config.extra.get("bot_token")): + connected.append(platform) return connected def get_home_channel(self, platform: Platform) -> Optional[HomeChannel]: @@ -927,6 +931,36 @@ def _apply_env_overrides(config: GatewayConfig) -> None: name=os.getenv("WECOM_HOME_CHANNEL_NAME", "Home"), ) + # WeChat (Personal WeChat via iLink Bot API) + wechat_token = os.getenv("WECHAT_BOT_TOKEN") + if wechat_token: + if Platform.WECHAT not in config.platforms: + config.platforms[Platform.WECHAT] = PlatformConfig() + config.platforms[Platform.WECHAT].enabled = True + config.platforms[Platform.WECHAT].token = wechat_token + wechat_base_url = os.getenv("WECHAT_API_BASE_URL", "") + if wechat_base_url: + config.platforms[Platform.WECHAT].extra["base_url"] = wechat_base_url + wechat_cdn_url = os.getenv("WECHAT_CDN_BASE_URL", "") + if wechat_cdn_url: + config.platforms[Platform.WECHAT].extra["cdn_base_url"] = wechat_cdn_url + wechat_account = os.getenv("WECHAT_ACCOUNT_ID", "") + if wechat_account: + config.platforms[Platform.WECHAT].extra["account_id"] = wechat_account + wechat_app_id = os.getenv("WECHAT_ILINK_APP_ID", "") + if wechat_app_id: + config.platforms[Platform.WECHAT].extra["ilink_app_id"] = wechat_app_id + wechat_client_ver = os.getenv("WECHAT_ILINK_CLIENT_VERSION", "") + if wechat_client_ver: + config.platforms[Platform.WECHAT].extra["ilink_client_version"] = wechat_client_ver + wechat_home = os.getenv("WECHAT_HOME_CHANNEL") + if wechat_home: + config.platforms[Platform.WECHAT].home_channel = HomeChannel( + platform=Platform.WECHAT, + chat_id=wechat_home, + name=os.getenv("WECHAT_HOME_CHANNEL_NAME", "Home"), + ) + # Session settings idle_minutes = os.getenv("SESSION_IDLE_MINUTES") if idle_minutes: diff --git a/gateway/platforms/wechat.py b/gateway/platforms/wechat.py new file mode 100644 index 000000000000..55084e067ac3 --- /dev/null +++ b/gateway/platforms/wechat.py @@ -0,0 +1,871 @@ +""" +WeChat platform adapter using the iLink Bot API (openclaw-weixin 2.1.x protocol). + +Uses long-polling (getUpdates) for inbound messages and REST API for outbound. +Media is transmitted through the WeChat CDN with AES-128-ECB encryption. + +Protocol compliance (SDK 2.1.x): + - iLink-App-Id and iLink-App-ClientVersion headers on all API requests + - IDC redirect (scaned_but_redirect) in QR login flow + - Context token persistence with reload at startup + - Referenced message (ref_msg) extraction for quoted replies + - SILK voice transcoding with graceful fallback + - Streaming markdown filter for outbound text + - CDN upload with upload_full_url priority and exponential backoff + +Architecture: + - wechat.py (this file): Adapter lifecycle, message routing, platform API + - wechat_transport.py: HTTP layer, CDN, AES crypto, iLink headers + - wechat_state.py: Context token and sync buffer persistence + +Requires: + pip install httpx cryptography + WECHAT_BOT_TOKEN env var (from QR login) or config.yaml + +Optional: + pip install silk-python # For native SILK voice transcoding +""" + +import asyncio +import base64 +import logging +import os +import re +import time +import uuid +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +from gateway.config import Platform, PlatformConfig +from gateway.platforms.base import ( + BasePlatformAdapter, + MessageEvent, + MessageType, + SendResult, + cache_image_from_bytes, + cache_audio_from_bytes, + cache_document_from_bytes, + get_image_cache_dir, + get_document_cache_dir, +) +from gateway.platforms.wechat_state import ( + load_context_tokens, + save_context_tokens, + clear_context_tokens, + load_sync_buf, + save_sync_buf, +) +from gateway.platforms.wechat_transport import ( + WeChatTransport, + check_wechat_requirements, + parse_aes_key, + aes_ecb_decrypt, + mime_from_path, + UPLOAD_MEDIA_IMAGE, + UPLOAD_MEDIA_VIDEO, + UPLOAD_MEDIA_FILE, + DEFAULT_BASE_URL, + CDN_BASE_URL, +) + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# WeChat protocol constants +# --------------------------------------------------------------------------- + +WX_MSG_TYPE_USER = 1 +WX_MSG_TYPE_BOT = 2 +WX_MSG_STATE_FINISH = 2 +WX_ITEM_TEXT = 1 +WX_ITEM_IMAGE = 2 +WX_ITEM_VOICE = 3 +WX_ITEM_FILE = 4 +WX_ITEM_VIDEO = 5 + +# Polling +DEFAULT_LONG_POLL_TIMEOUT_S = 35 +MAX_CONSECUTIVE_FAILURES = 3 +BACKOFF_DELAY_S = 30 +RETRY_DELAY_S = 2 + +# Session expired +SESSION_EXPIRED_ERRCODE = -14 +SESSION_PAUSE_DURATION_S = 3600 # 1 hour + +# Message limits +MAX_MESSAGE_LENGTH = 4096 +DEDUP_WINDOW_S = 300 +DEDUP_MAX_SIZE = 1000 + + +# --------------------------------------------------------------------------- +# Markdown -> plain text (WeChat doesn't render markdown) +# --------------------------------------------------------------------------- + +def _markdown_to_plain(text: str) -> str: + """Strip markdown syntax for WeChat delivery.""" + result = text + result = re.sub(r"```[^\n]*\n?([\s\S]*?)```", lambda m: m.group(1).strip(), result) + result = re.sub(r"!\[[^\]]*\]\([^)]*\)", "", result) + result = re.sub(r"\[([^\]]+)\]\([^)]*\)", r"\1", result) + result = re.sub(r"^\|[\s:|\-]+\|$", "", result, flags=re.MULTILINE) + + def _table_row(m): + inner = m.group(1) + return " ".join(cell.strip() for cell in inner.split("|")) + result = re.sub(r"^\|(.+)\|$", _table_row, result, flags=re.MULTILINE) + + result = re.sub(r"^#{1,6}\s+", "", result, flags=re.MULTILINE) + result = re.sub(r"\*\*(.+?)\*\*", r"\1", result) + result = re.sub(r"\*(.+?)\*", r"\1", result) + result = re.sub(r"__(.+?)__", r"\1", result) + result = re.sub(r"_(.+?)_", r"\1", result) + result = re.sub(r"~~(.+?)~~", r"\1", result) + result = re.sub(r"`(.+?)`", r"\1", result) + return result + + +# --------------------------------------------------------------------------- +# SILK voice transcoding +# --------------------------------------------------------------------------- + +def _silk_to_wav(silk_buf: bytes) -> Optional[bytes]: + """Best-effort SILK -> WAV conversion. + + Tries: + 1. silk-decoder CLI (pip install silk-python) + 2. ffmpeg with SILK input format + + Returns WAV bytes on success, None on failure (caller falls back to + passing raw SILK or using voice-to-text from the WeChat API). + """ + import subprocess + import tempfile + + silk_path = None + wav_path = None + try: + with tempfile.NamedTemporaryFile(suffix=".silk", delete=False) as sf: + sf.write(silk_buf) + silk_path = sf.name + wav_path = silk_path.replace(".silk", ".wav") + + for cmd in [ + ["silk-decoder", silk_path, wav_path], + ["ffmpeg", "-y", "-i", silk_path, "-ar", "24000", "-ac", "1", wav_path], + ]: + try: + result = subprocess.run(cmd, capture_output=True, timeout=10) + if result.returncode == 0 and os.path.exists(wav_path): + wav_data = Path(wav_path).read_bytes() + if len(wav_data) > 44: + return wav_data + except FileNotFoundError: + continue + except subprocess.TimeoutExpired: + continue + except Exception: + pass + finally: + for p in (silk_path, wav_path): + if p: + try: + os.unlink(p) + except Exception: + pass + return None + + +# --------------------------------------------------------------------------- +# WeChatAdapter +# --------------------------------------------------------------------------- + +class WeChatAdapter(BasePlatformAdapter): + """WeChat chatbot adapter using the iLink Bot long-polling API. + + Message flow: + 1. connect() loads persisted state and starts the long-poll loop + 2. Inbound messages are parsed, media is downloaded/decrypted from CDN + 3. Referenced messages (ref_msg) are extracted for quote context + 4. MessageEvent is dispatched to self.handle_message() + 5. Outbound text has markdown stripped; media is encrypted + uploaded to CDN + """ + + MAX_MESSAGE_LENGTH = MAX_MESSAGE_LENGTH + + def __init__(self, config: PlatformConfig): + super().__init__(config, Platform.WECHAT) + + extra = config.extra or {} + token = config.token or os.getenv("WECHAT_BOT_TOKEN", "") + base_url = extra.get("base_url") or os.getenv("WECHAT_API_BASE_URL", DEFAULT_BASE_URL) + cdn_base_url = extra.get("cdn_base_url") or os.getenv("WECHAT_CDN_BASE_URL", CDN_BASE_URL) + ilink_app_id = extra.get("ilink_app_id") or os.getenv("WECHAT_ILINK_APP_ID", "bot") + ilink_version_raw = extra.get("ilink_client_version") + if ilink_version_raw is None or ilink_version_raw == "": + ilink_version_raw = os.getenv("WECHAT_ILINK_CLIENT_VERSION", "") + ilink_version = int(ilink_version_raw) if ilink_version_raw not in (None, "") else None + + self._account_id: str = extra.get("account_id") or os.getenv("WECHAT_ACCOUNT_ID", "") + self._transport = WeChatTransport( + token=token, + base_url=base_url, + cdn_base_url=cdn_base_url, + ilink_app_id=ilink_app_id, + ilink_client_version=ilink_version, + ) + + self._poll_task: Optional[asyncio.Task] = None + + # Context tokens: loaded from disk at startup, updated on each inbound message + self._context_tokens: Dict[str, str] = {} + # Typing tickets: user_id -> (ticket, fetched_at) + self._typing_tickets: Dict[str, Tuple[str, float]] = {} + self._typing_ticket_ttl_s: float = 12 * 3600 + + # Deduplication + self._seen_messages: Dict[str, float] = {} + + # Session pause (errcode -14) + self._paused_until: float = 0.0 + + # -- Connection lifecycle ----------------------------------------------- + + async def connect(self) -> bool: + """Start the long-poll loop for inbound messages.""" + if not check_wechat_requirements(): + logger.error("[WeChat] Missing dependencies (httpx, cryptography)") + return False + + if not self._transport._token: + logger.error("[WeChat] No token configured. Run scripts/wechat_login.py or set WECHAT_BOT_TOKEN") + self._set_fatal_error("no_token", "WeChat token not configured", retryable=False) + await self._notify_fatal_error() + return False + + try: + await self._transport.open() + + # Load persisted context tokens (survive gateway restarts) + self._context_tokens = load_context_tokens() + + self._poll_task = asyncio.create_task(self._poll_loop()) + self._mark_connected() + logger.info("[WeChat] Connected, starting poll loop (account=%s)", self._account_id or "default") + return True + except Exception as e: + logger.error("[WeChat] Failed to connect: %s", e) + return False + + async def disconnect(self) -> None: + """Stop polling and clean up.""" + self._running = False + self._mark_disconnected() + + # Cancel poll task first (may be blocked on a 40s HTTP timeout) + if self._poll_task: + self._poll_task.cancel() + try: + await self._poll_task + except asyncio.CancelledError: + pass + self._poll_task = None + + await self.cancel_background_tasks() + await self._transport.close() + + self._context_tokens.clear() + self._typing_tickets.clear() + self._seen_messages.clear() + logger.info("[WeChat] Disconnected") + + # -- Long-poll loop ----------------------------------------------------- + + async def _poll_loop(self) -> None: + """Long-poll getUpdates in a loop until disconnected.""" + get_updates_buf = load_sync_buf(self._account_id) + if get_updates_buf: + logger.info("[WeChat] Resuming from saved sync buf (%d bytes)", len(get_updates_buf)) + + consecutive_failures = 0 + poll_timeout_ms = DEFAULT_LONG_POLL_TIMEOUT_S * 1000 + + while self._running: + try: + # Session pause check + if self._paused_until > time.time(): + remaining = int(self._paused_until - time.time()) + logger.info("[WeChat] Session paused, %ds remaining", remaining) + await asyncio.sleep(min(remaining, 60)) + continue + + resp = await self._transport.get_updates( + get_updates_buf, + timeout_s=poll_timeout_ms / 1000 + 5, + ) + + suggested = resp.get("longpolling_timeout_ms") + if isinstance(suggested, (int, float)) and suggested > 0: + poll_timeout_ms = int(suggested) + + ret = resp.get("ret", 0) + errcode = resp.get("errcode", 0) + is_error = (ret != 0) or (errcode != 0) + + if is_error: + if errcode == SESSION_EXPIRED_ERRCODE or ret == SESSION_EXPIRED_ERRCODE: + self._paused_until = time.time() + SESSION_PAUSE_DURATION_S + # Clear context tokens on session expiry + self._context_tokens.clear() + clear_context_tokens() + logger.warning( + "[WeChat] Session expired (errcode %d), pausing for %d min, tokens cleared", + SESSION_EXPIRED_ERRCODE, SESSION_PAUSE_DURATION_S // 60, + ) + consecutive_failures = 0 + continue + + consecutive_failures += 1 + logger.warning( + "[WeChat] getUpdates error: ret=%s errcode=%s errmsg=%s (%d/%d)", + ret, errcode, resp.get("errmsg", ""), consecutive_failures, MAX_CONSECUTIVE_FAILURES, + ) + if consecutive_failures >= MAX_CONSECUTIVE_FAILURES: + consecutive_failures = 0 + await asyncio.sleep(BACKOFF_DELAY_S) + else: + await asyncio.sleep(RETRY_DELAY_S) + continue + + consecutive_failures = 0 + + new_buf = resp.get("get_updates_buf", "") + if new_buf: + get_updates_buf = new_buf + save_sync_buf(self._account_id, new_buf) + + msgs = resp.get("msgs") or [] + for msg in msgs: + try: + await self._on_message(msg) + except Exception as e: + logger.error("[WeChat] Error processing message: %s", e, exc_info=True) + + except asyncio.CancelledError: + return + except Exception as e: + if not self._running: + return + consecutive_failures += 1 + logger.warning("[WeChat] Poll loop error (%d/%d): %s", consecutive_failures, MAX_CONSECUTIVE_FAILURES, e) + if consecutive_failures >= MAX_CONSECUTIVE_FAILURES: + consecutive_failures = 0 + await asyncio.sleep(BACKOFF_DELAY_S) + else: + await asyncio.sleep(RETRY_DELAY_S) + + # -- Inbound message processing ----------------------------------------- + + async def _on_message(self, msg: Dict[str, Any]) -> None: + """Process a single inbound WeChat message.""" + from_user = msg.get("from_user_id", "") + if not from_user: + return + + msg_id = str(msg.get("message_id", "")) + seq = str(msg.get("seq", "")) + dedup_key = f"{from_user}:{msg_id}:{seq}" + if self._is_duplicate(dedup_key): + return + + if msg.get("message_type") != WX_MSG_TYPE_USER: + return + if self._account_id and from_user == self._account_id: + return + + # Cache context token + context_token = msg.get("context_token", "") + if context_token: + self._context_tokens[from_user] = context_token + save_context_tokens(self._context_tokens) + + items = msg.get("item_list") or [] + text = self._extract_text(items) + + hermes_msg_type = MessageType.TEXT + media_urls: List[str] = [] + media_types: List[str] = [] + + media_item = self._find_media_item(items) + if media_item: + try: + path, mime, mtype = await self._download_media(media_item) + if path: + media_urls.append(path) + media_types.append(mime) + hermes_msg_type = mtype + except Exception as e: + logger.error("[WeChat] Media download failed: %s", e) + + # Voice STT fallback + if not text and hermes_msg_type == MessageType.VOICE: + if media_item and media_item.get("type") == WX_ITEM_VOICE: + stt = (media_item.get("voice_item") or {}).get("text", "") + if stt: + text = stt + + if not text and not media_urls: + return + + # Fetch typing ticket (async, non-blocking) + task = asyncio.create_task(self._cache_typing_ticket(from_user, context_token)) + self._background_tasks.add(task) + task.add_done_callback(self._background_tasks.discard) + + source = self.build_source(chat_id=from_user, chat_type="dm", user_id=from_user) + + create_time_ms = msg.get("create_time_ms") + try: + timestamp = datetime.fromtimestamp( + create_time_ms / 1000, tz=timezone.utc + ) if create_time_ms else datetime.now(tz=timezone.utc) + except (ValueError, OSError, TypeError): + timestamp = datetime.now(tz=timezone.utc) + + event = MessageEvent( + text=text or "", + message_type=hermes_msg_type, + source=source, + message_id=msg_id or seq, + raw_message=msg, + media_urls=media_urls, + media_types=media_types, + timestamp=timestamp, + ) + + await self.handle_message(event) + + @staticmethod + def _extract_text(items: List[Dict[str, Any]]) -> str: + """Extract text from item_list, handling referenced messages (ref_msg).""" + for item in items: + if item.get("type") == WX_ITEM_TEXT: + text_item = item.get("text_item") or {} + text = text_item.get("text", "") + + ref = item.get("ref_msg") + if not ref: + return text + + # Quoted media: just return the text part + ref_item = ref.get("message_item") + if ref_item and ref_item.get("type") in (WX_ITEM_IMAGE, WX_ITEM_VIDEO, WX_ITEM_FILE, WX_ITEM_VOICE): + return text + + # Build quoted context from title and message content + parts = [] + title = ref.get("title", "") + if title: + parts.append(title) + if ref_item: + ref_text = WeChatAdapter._extract_text([ref_item]) + if ref_text: + parts.append(ref_text) + + if parts: + return f'[Quote: {" | ".join(parts)}]\n{text}' + return text + + # Voice with speech-to-text + if item.get("type") == WX_ITEM_VOICE: + voice = item.get("voice_item") or {} + if voice.get("text"): + return voice["text"] + + return "" + + @staticmethod + def _find_media_item(items: List[Dict[str, Any]]) -> Optional[Dict[str, Any]]: + """Find the first downloadable media item (priority: image > video > file > voice). + + Voice items with server-side STT (voice_item.text) are skipped — + the transcription is used as text instead. + """ + for item_type in (WX_ITEM_IMAGE, WX_ITEM_VIDEO, WX_ITEM_FILE, WX_ITEM_VOICE): + for item in items: + if item.get("type") == item_type: + if item_type == WX_ITEM_VOICE: + voice_data = item.get("voice_item") or {} + if voice_data.get("text"): + continue + type_key = { + WX_ITEM_IMAGE: "image_item", + WX_ITEM_VIDEO: "video_item", + WX_ITEM_FILE: "file_item", + WX_ITEM_VOICE: "voice_item", + }[item_type] + media_data = item.get(type_key) or {} + media_ref = media_data.get("media") or {} + if media_ref.get("encrypt_query_param") or media_ref.get("full_url"): + return item + + # Check quoted media in ref_msg + for item in items: + if item.get("type") == WX_ITEM_TEXT: + ref = item.get("ref_msg", {}) + ref_item = ref.get("message_item") + if ref_item and ref_item.get("type") in (WX_ITEM_IMAGE, WX_ITEM_VIDEO, WX_ITEM_FILE, WX_ITEM_VOICE): + type_key = { + WX_ITEM_IMAGE: "image_item", WX_ITEM_VIDEO: "video_item", + WX_ITEM_FILE: "file_item", WX_ITEM_VOICE: "voice_item", + }.get(ref_item["type"]) + if type_key: + media_ref = (ref_item.get(type_key) or {}).get("media") or {} + if media_ref.get("encrypt_query_param") or media_ref.get("full_url"): + return ref_item + + return None + + async def _download_media(self, item: Dict[str, Any]) -> Tuple[str, str, MessageType]: + """Download and decrypt media from CDN. Returns (local_path, mime, message_type).""" + item_type = item.get("type") + + if item_type == WX_ITEM_IMAGE: + img = item.get("image_item") or {} + media = img.get("media") or {} + eqp = media.get("encrypt_query_param", "") + fu = media.get("full_url", "") + if img.get("aeskey"): + aes_key_b64 = base64.b64encode(bytes.fromhex(img["aeskey"])).decode() + else: + aes_key_b64 = media.get("aes_key", "") + if not eqp and not fu: + return ("", "", MessageType.TEXT) + buf = await self._transport.cdn_download_decrypt(eqp, aes_key_b64, full_url=fu) if aes_key_b64 else await self._transport.cdn_download_plain(eqp, full_url=fu) + path = cache_image_from_bytes(buf, ".jpg") + return (path, "image/jpeg", MessageType.PHOTO) + + elif item_type == WX_ITEM_VOICE: + voice = item.get("voice_item") or {} + media = voice.get("media") or {} + eqp = media.get("encrypt_query_param", "") + fu = media.get("full_url", "") + aes_key_b64 = media.get("aes_key", "") + if (not eqp and not fu) or not aes_key_b64: + return ("", "", MessageType.TEXT) + silk_buf = await self._transport.cdn_download_decrypt(eqp, aes_key_b64, full_url=fu) + wav_buf = _silk_to_wav(silk_buf) + if wav_buf: + path = cache_audio_from_bytes(wav_buf, ".wav") + return (path, "audio/wav", MessageType.VOICE) + else: + path = cache_audio_from_bytes(silk_buf, ".silk") + return (path, "audio/silk", MessageType.VOICE) + + elif item_type == WX_ITEM_FILE: + file_item = item.get("file_item") or {} + media = file_item.get("media") or {} + eqp = media.get("encrypt_query_param", "") + fu = media.get("full_url", "") + aes_key_b64 = media.get("aes_key", "") + filename = file_item.get("file_name", "file.bin") + if (not eqp and not fu) or not aes_key_b64: + return ("", "", MessageType.TEXT) + buf = await self._transport.cdn_download_decrypt(eqp, aes_key_b64, full_url=fu) + path = cache_document_from_bytes(buf, filename) + return (path, mime_from_path(filename), MessageType.DOCUMENT) + + elif item_type == WX_ITEM_VIDEO: + video = item.get("video_item") or {} + media = video.get("media") or {} + eqp = media.get("encrypt_query_param", "") + fu = media.get("full_url", "") + aes_key_b64 = media.get("aes_key", "") + if (not eqp and not fu) or not aes_key_b64: + return ("", "", MessageType.TEXT) + buf = await self._transport.cdn_download_decrypt(eqp, aes_key_b64, full_url=fu) + path = cache_document_from_bytes(buf, f"video_{uuid.uuid4().hex[:8]}.mp4") + return (path, "video/mp4", MessageType.VIDEO) + + return ("", "", MessageType.TEXT) + + # -- Deduplication ------------------------------------------------------ + + def _is_duplicate(self, key: str) -> bool: + now = time.time() + if len(self._seen_messages) > DEDUP_MAX_SIZE: + cutoff = now - DEDUP_WINDOW_S + self._seen_messages = {k: v for k, v in self._seen_messages.items() if v > cutoff} + if key in self._seen_messages: + return True + self._seen_messages[key] = now + return False + + # -- Outbound: text ----------------------------------------------------- + + async def send( + self, + chat_id: str, + content: str, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + """Send a plain text message to a WeChat user.""" + context_token = self._context_tokens.get(chat_id) + if not context_token: + return SendResult(success=False, error="No context_token for this user (they haven't messaged yet)") + + plain = _markdown_to_plain(content).strip() + if not plain: + return SendResult(success=True, message_id="skipped-empty") + + chunks = self.truncate_message(plain, self.MAX_MESSAGE_LENGTH) + last_id = None + + try: + for chunk in chunks: + client_id = f"hermes-{uuid.uuid4().hex[:12]}" + body = { + "msg": { + "from_user_id": "", + "to_user_id": chat_id, + "client_id": client_id, + "message_type": WX_MSG_TYPE_BOT, + "message_state": WX_MSG_STATE_FINISH, + "item_list": [{"type": WX_ITEM_TEXT, "text_item": {"text": chunk}}], + "context_token": context_token, + }, + } + resp = await self._transport.send_message(body) + if resp.get("ret", 0) != 0: + raise RuntimeError(f"ret={resp.get('ret')} {resp.get('errmsg', '')}") + last_id = client_id + return SendResult(success=True, message_id=last_id) + except Exception as e: + logger.error("[WeChat] Send failed: %s", e) + return SendResult(success=False, error=str(e)) + + # -- Outbound: typing --------------------------------------------------- + + async def send_typing(self, chat_id: str, metadata=None) -> None: + """Send typing indicator via WeChat API.""" + entry = self._typing_tickets.get(chat_id) + if not entry: + return + ticket, _ = entry + await self._transport.send_typing(chat_id, ticket) + + async def _cache_typing_ticket(self, user_id: str, context_token: str) -> None: + """Fetch and cache the typing_ticket for a user (with TTL refresh).""" + entry = self._typing_tickets.get(user_id) + if entry: + _, fetched_at = entry + if time.time() - fetched_at < self._typing_ticket_ttl_s: + return + try: + resp = await self._transport.get_config(user_id, context_token) + ticket = resp.get("typing_ticket", "") + if ticket: + self._typing_tickets[user_id] = (ticket, time.time()) + except Exception as e: + logger.debug("[WeChat] Failed to get typing ticket: %s", e) + + # -- Outbound: media ---------------------------------------------------- + + async def send_image( + self, chat_id: str, image_url: str, + caption: Optional[str] = None, reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + try: + local_path, _ = await self._transport.download_remote_file(image_url, get_image_cache_dir()) + return await self._send_media_file(chat_id, local_path, caption, UPLOAD_MEDIA_IMAGE, WX_ITEM_IMAGE) + except Exception as e: + logger.error("[WeChat] send_image failed: %s", e) + text = f"{caption}\n{image_url}" if caption else image_url + return await self.send(chat_id, text) + + async def send_image_file( + self, chat_id: str, image_path: str, + caption: Optional[str] = None, reply_to: Optional[str] = None, **kwargs, + ) -> SendResult: + try: + return await self._send_media_file(chat_id, image_path, caption, UPLOAD_MEDIA_IMAGE, WX_ITEM_IMAGE) + except Exception as e: + logger.error("[WeChat] send_image_file failed: %s", e) + return SendResult(success=False, error=str(e)) + + async def send_video( + self, chat_id: str, video_path: str, + caption: Optional[str] = None, reply_to: Optional[str] = None, **kwargs, + ) -> SendResult: + try: + return await self._send_media_file(chat_id, video_path, caption, UPLOAD_MEDIA_VIDEO, WX_ITEM_VIDEO) + except Exception as e: + logger.error("[WeChat] send_video failed: %s", e) + return SendResult(success=False, error=str(e)) + + async def send_document( + self, chat_id: str, file_path: str, + caption: Optional[str] = None, file_name: Optional[str] = None, + reply_to: Optional[str] = None, **kwargs, + ) -> SendResult: + try: + return await self._send_media_file( + chat_id, file_path, caption, UPLOAD_MEDIA_FILE, WX_ITEM_FILE, + file_name=file_name or Path(file_path).name, + ) + except Exception as e: + logger.error("[WeChat] send_document failed: %s", e) + return SendResult(success=False, error=str(e)) + + async def send_voice( + self, chat_id: str, audio_path: str, + caption: Optional[str] = None, reply_to: Optional[str] = None, **kwargs, + ) -> SendResult: + """Send audio as a file attachment. + + Native voice bubbles are not supported by the iLink Bot API for + outbound — even Tencent's SDK sends audio as FILE attachments. + """ + try: + send_path = audio_path + compressed_path = None + if Path(audio_path).stat().st_size > 100_000: + compressed_path = self._compress_audio(audio_path) + if compressed_path: + send_path = compressed_path + try: + return await self._send_media_file( + chat_id, send_path, caption, + UPLOAD_MEDIA_FILE, WX_ITEM_FILE, + file_name=Path(audio_path).name, + ) + finally: + if compressed_path: + try: + os.unlink(compressed_path) + except OSError: + pass + except Exception as e: + logger.error("[WeChat] send_voice failed: %s", e) + return SendResult(success=False, error=str(e)) + + @staticmethod + def _compress_audio(audio_path: str) -> Optional[str]: + """Compress audio with ffmpeg for CDN upload. Returns temp path or None.""" + import subprocess + import tempfile + try: + fd, out_path = tempfile.mkstemp(suffix=Path(audio_path).suffix) + os.close(fd) + result = subprocess.run( + ["ffmpeg", "-y", "-i", audio_path, "-ar", "16000", "-ac", "1", "-b:a", "64k", out_path], + capture_output=True, timeout=15, + ) + if result.returncode == 0 and Path(out_path).stat().st_size > 0: + return out_path + os.unlink(out_path) + return None + except Exception: + return None + + async def _send_media_file( + self, + chat_id: str, + file_path: str, + caption: Optional[str], + upload_media_type: int, + wx_item_type: int, + file_name: Optional[str] = None, + ) -> SendResult: + """Upload a file to CDN and send it as a WeChat message.""" + context_token = self._context_tokens.get(chat_id) + if not context_token: + return SendResult(success=False, error="No context_token for this user") + + uploaded = await self._transport.cdn_upload(file_path, chat_id, upload_media_type) + + # Outbound aes_key: base64(hex string) for all media types + aes_key_b64 = base64.b64encode(uploaded["aeskey"].encode()).decode() + + if wx_item_type == WX_ITEM_IMAGE: + media_item = { + "type": WX_ITEM_IMAGE, + "image_item": { + "media": { + "encrypt_query_param": uploaded["download_param"], + "aes_key": aes_key_b64, + "encrypt_type": 1, + }, + "mid_size": uploaded["ciphertext_size"], + }, + } + elif wx_item_type == WX_ITEM_VIDEO: + media_item = { + "type": WX_ITEM_VIDEO, + "video_item": { + "media": { + "encrypt_query_param": uploaded["download_param"], + "aes_key": aes_key_b64, + "encrypt_type": 1, + }, + "video_size": uploaded["ciphertext_size"], + }, + } + elif wx_item_type == WX_ITEM_FILE: + media_item = { + "type": WX_ITEM_FILE, + "file_item": { + "media": { + "encrypt_query_param": uploaded["download_param"], + "aes_key": aes_key_b64, + "encrypt_type": 1, + }, + "file_name": file_name or Path(file_path).name, + "len": str(uploaded["plaintext_size"]), + }, + } + else: + return SendResult(success=False, error=f"Unsupported item type: {wx_item_type}") + + item_list = [] + if caption: + item_list.append({"type": WX_ITEM_TEXT, "text_item": {"text": _markdown_to_plain(caption)}}) + item_list.append(media_item) + + last_cid = None + for item in item_list: + cid = f"hermes-{uuid.uuid4().hex[:12]}" + body = { + "msg": { + "from_user_id": "", + "to_user_id": chat_id, + "client_id": cid, + "message_type": WX_MSG_TYPE_BOT, + "message_state": WX_MSG_STATE_FINISH, + "item_list": [item], + "context_token": context_token, + }, + } + resp = await self._transport.send_message(body) + if resp.get("ret", 0) != 0: + raise RuntimeError(f"sendmessage ret={resp.get('ret')} {resp.get('errmsg', '')}") + last_cid = cid + + return SendResult(success=True, message_id=last_cid) + + # -- Chat info ---------------------------------------------------------- + + async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: + return { + "name": chat_id[:12] + "..." if len(chat_id) > 12 else chat_id, + "type": "dm", + "chat_id": chat_id, + } + + def format_message(self, content: str) -> str: + """Strip markdown for WeChat (plain text only).""" + return _markdown_to_plain(content) diff --git a/gateway/platforms/wechat_state.py b/gateway/platforms/wechat_state.py new file mode 100644 index 000000000000..f6e0d4dcd1c5 --- /dev/null +++ b/gateway/platforms/wechat_state.py @@ -0,0 +1,105 @@ +""" +WeChat platform state management — context tokens and sync buffer persistence. + +Context tokens are issued per-message by the WeChat getUpdates API and must be +echoed verbatim in every outbound send. Tokens are cached in-memory and persisted +to disk so they survive gateway restarts. +""" + +import json +import logging +from pathlib import Path +from typing import Dict, List, Optional + +from hermes_constants import get_hermes_home + +logger = logging.getLogger(__name__) + + +def _wechat_state_dir() -> Path: + """Return (and create) the WeChat state directory under HERMES_HOME.""" + d = get_hermes_home() / "wechat" + d.mkdir(parents=True, exist_ok=True) + return d + + +# --------------------------------------------------------------------------- +# Sync buffer (poll cursor) persistence +# --------------------------------------------------------------------------- + +def _sync_buf_path(account_id: str) -> Path: + d = _wechat_state_dir() / "sync" + d.mkdir(parents=True, exist_ok=True) + return d / f"{account_id}.buf" + + +def load_sync_buf(account_id: str) -> str: + """Load the last getUpdates sync buffer for an account.""" + if not account_id: + return "" + p = _sync_buf_path(account_id) + try: + return p.read_text("utf-8").strip() if p.exists() else "" + except Exception: + return "" + + +def save_sync_buf(account_id: str, buf: str) -> None: + """Persist the getUpdates sync buffer for resume after restart.""" + if not account_id: + return + try: + _sync_buf_path(account_id).write_text(buf, "utf-8") + except Exception as e: + logger.warning("[WeChat] Failed to save sync buf: %s", e) + + +# --------------------------------------------------------------------------- +# Context token persistence +# --------------------------------------------------------------------------- + +def _context_tokens_path() -> Path: + return _wechat_state_dir() / "context_tokens.json" + + +def load_context_tokens() -> Dict[str, str]: + """Load persisted context tokens from disk. + + Called at adapter startup so tokens survive gateway restarts — + users don't need to re-message before the bot can reply. + """ + p = _context_tokens_path() + try: + if p.exists(): + data = json.loads(p.read_text("utf-8")) + if isinstance(data, dict): + count = len(data) + if count: + logger.info("[WeChat] Loaded %d context tokens from disk", count) + return {k: v for k, v in data.items() if isinstance(v, str)} + except Exception as e: + logger.warning("[WeChat] Failed to load context tokens: %s", e) + return {} + + +def save_context_tokens(tokens: Dict[str, str]) -> None: + """Persist all context tokens to disk with restricted permissions.""" + try: + p = _context_tokens_path() + p.write_text(json.dumps(tokens), "utf-8") + p.chmod(0o600) + except Exception as e: + logger.debug("[WeChat] Failed to persist context_tokens: %s", e) + + +def clear_context_tokens() -> None: + """Remove all persisted context tokens (called on session expiry).""" + try: + p = _context_tokens_path() + if p.exists(): + p.unlink() + logger.info("[WeChat] Cleared persisted context tokens") + except Exception as e: + logger.warning("[WeChat] Failed to clear context tokens: %s", e) + + diff --git a/gateway/platforms/wechat_transport.py b/gateway/platforms/wechat_transport.py new file mode 100644 index 000000000000..74c684b5cc88 --- /dev/null +++ b/gateway/platforms/wechat_transport.py @@ -0,0 +1,466 @@ +""" +WeChat platform transport layer — HTTP API, CDN upload/download, AES encryption. + +Handles all network communication with the WeChat iLink Bot API and CDN. +Implements protocol requirements from openclaw-weixin SDK 2.1.x: + - iLink-App-Id and iLink-App-ClientVersion headers on all requests + - X-WECHAT-UIN random header + - AES-128-ECB media encryption/decryption + - CDN upload with retry and exponential backoff + - upload_full_url priority over upload_param +""" + +import asyncio +import base64 +import hashlib +import logging +import os +import struct +import urllib.parse +import uuid +from pathlib import Path +from typing import Any, Dict, Optional, Tuple + +try: + import httpx + HTTPX_AVAILABLE = True +except ImportError: + HTTPX_AVAILABLE = False + httpx = None # type: ignore[assignment] + +try: + from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + from cryptography.hazmat.primitives import padding as crypto_padding + CRYPTO_AVAILABLE = True +except ImportError: + CRYPTO_AVAILABLE = False + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +DEFAULT_BASE_URL = "https://ilinkai.weixin.qq.com" +CDN_BASE_URL = "https://novac2c.cdn.weixin.qq.com/c2c" +ADAPTER_VERSION = "0.2.0" + +# Upload media type constants +UPLOAD_MEDIA_IMAGE = 1 +UPLOAD_MEDIA_VIDEO = 2 +UPLOAD_MEDIA_FILE = 3 +UPLOAD_MEDIA_VOICE = 4 + +# CDN limits +MEDIA_MAX_BYTES = 100 * 1024 * 1024 # 100 MB +CDN_UPLOAD_MAX_RETRIES = 3 + + +def check_wechat_requirements() -> bool: + """Check if WeChat adapter dependencies are available.""" + if not HTTPX_AVAILABLE: + logger.warning("WeChat: httpx not installed. Run: pip install httpx") + return False + if not CRYPTO_AVAILABLE: + logger.warning("WeChat: cryptography not installed. Run: pip install cryptography") + return False + return True + + +# --------------------------------------------------------------------------- +# iLink header computation (SDK 2.1.1+) +# --------------------------------------------------------------------------- + +def _build_client_version(version: str) -> int: + """Encode version as uint32: 0x00MMNNPP (major<<16 | minor<<8 | patch). + + e.g. "0.2.0" -> 0x00000200 = 512 + Matches the SDK's buildClientVersion() in api.ts. + """ + parts = version.split(".") + major = int(parts[0]) if len(parts) > 0 else 0 + minor = int(parts[1]) if len(parts) > 1 else 0 + patch = int(parts[2]) if len(parts) > 2 else 0 + return ((major & 0xFF) << 16) | ((minor & 0xFF) << 8) | (patch & 0xFF) + + +def _build_channel_version() -> str: + """Dynamic channel_version for base_info (not hardcoded).""" + return f"hermes-wechat/{ADAPTER_VERSION}" + + +# --------------------------------------------------------------------------- +# AES-128-ECB crypto (matches SDK aes-ecb.ts) +# --------------------------------------------------------------------------- + +def aes_ecb_encrypt(plaintext: bytes, key: bytes) -> bytes: + """Encrypt with AES-128-ECB + PKCS7 padding.""" + padder = crypto_padding.PKCS7(128).padder() + padded = padder.update(plaintext) + padder.finalize() + cipher = Cipher(algorithms.AES(key), modes.ECB()) + enc = cipher.encryptor() + return enc.update(padded) + enc.finalize() + + +def aes_ecb_decrypt(ciphertext: bytes, key: bytes) -> bytes: + """Decrypt AES-128-ECB with PKCS7 padding.""" + cipher = Cipher(algorithms.AES(key), modes.ECB()) + dec = cipher.decryptor() + padded = dec.update(ciphertext) + dec.finalize() + unpadder = crypto_padding.PKCS7(128).unpadder() + return unpadder.update(padded) + unpadder.finalize() + + +def aes_ecb_padded_size(plaintext_size: int) -> int: + """Compute AES-128-ECB ciphertext size (PKCS7 to 16-byte boundary).""" + return ((plaintext_size + 1 + 15) // 16) * 16 + + +def parse_aes_key(aes_key_b64: str) -> bytes: + """Parse CDNMedia.aes_key into a raw 16-byte key. + + Two encodings exist in the wild: + - base64(raw 16 bytes) -> images + - base64(hex string of 16 bytes) -> file / voice / video + """ + decoded = base64.b64decode(aes_key_b64) + if len(decoded) == 16: + return decoded + if len(decoded) == 32: + try: + hex_str = decoded.decode("ascii") + if all(c in "0123456789abcdefABCDEF" for c in hex_str): + return bytes.fromhex(hex_str) + except (UnicodeDecodeError, ValueError): + pass + raise ValueError( + f"aes_key must decode to 16 raw bytes or 32-char hex string, got {len(decoded)} bytes" + ) + + +# --------------------------------------------------------------------------- +# MIME helpers +# --------------------------------------------------------------------------- + +_MIME_MAP = { + ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png", + ".gif": "image/gif", ".webp": "image/webp", ".bmp": "image/bmp", + ".mp4": "video/mp4", ".mov": "video/quicktime", ".avi": "video/x-msvideo", + ".mkv": "video/x-matroska", ".webm": "video/webm", + ".pdf": "application/pdf", ".doc": "application/msword", + ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document", + ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + ".zip": "application/zip", ".txt": "text/plain", + ".wav": "audio/wav", ".mp3": "audio/mpeg", ".ogg": "audio/ogg", + ".opus": "audio/opus", ".m4a": "audio/mp4", +} + + +def mime_from_path(file_path: str) -> str: + ext = Path(file_path).suffix.lower() + return _MIME_MAP.get(ext, "application/octet-stream") + + +# --------------------------------------------------------------------------- +# WeChatTransport +# --------------------------------------------------------------------------- + +class WeChatTransport: + """Handles all HTTP communication with the WeChat iLink Bot API and CDN. + + Sends required iLink headers on all requests (SDK 2.1.1+): + - iLink-App-Id + - iLink-App-ClientVersion (uint32 encoded) + - X-WECHAT-UIN (random per-request) + - AuthorizationType: ilink_bot_token + """ + + def __init__( + self, + token: str, + base_url: str = DEFAULT_BASE_URL, + cdn_base_url: str = CDN_BASE_URL, + ilink_app_id: str = "bot", + ilink_client_version: Optional[int] = None, + ): + self._token = token + self._base_url = base_url.rstrip("/") + self._cdn_base_url = cdn_base_url.rstrip("/") + self._ilink_app_id = ilink_app_id + self._ilink_client_version = ( + ilink_client_version + if ilink_client_version is not None + else _build_client_version(ADAPTER_VERSION) + ) + self._http: Optional["httpx.AsyncClient"] = None + + async def open(self) -> None: + """Initialize the HTTP client.""" + self._http = httpx.AsyncClient( + timeout=httpx.Timeout(60.0, connect=10.0), + follow_redirects=True, + ) + + async def close(self) -> None: + """Close the HTTP client.""" + if self._http: + await self._http.aclose() + self._http = None + + @property + def is_open(self) -> bool: + return self._http is not None + + # -- Header construction ------------------------------------------------ + + def _build_common_headers(self) -> Dict[str, str]: + """Headers included on every API request (iLink 2.1.1+ protocol).""" + return { + "iLink-App-Id": self._ilink_app_id, + "iLink-App-ClientVersion": str(self._ilink_client_version), + } + + def _build_headers(self) -> Dict[str, str]: + """Full header set for POST API requests.""" + rand_uint32 = struct.unpack(">I", os.urandom(4))[0] + uin_b64 = base64.b64encode(str(rand_uint32).encode()).decode() + + headers = { + "Content-Type": "application/json", + "AuthorizationType": "ilink_bot_token", + "X-WECHAT-UIN": uin_b64, + **self._build_common_headers(), + } + if self._token: + headers["Authorization"] = f"Bearer {self._token}" + return headers + + # -- API methods -------------------------------------------------------- + + async def api_fetch(self, endpoint: str, body: dict, timeout_s: float = 15) -> dict: + """POST JSON to a WeChat API endpoint with all required headers.""" + if not self._http: + raise RuntimeError("HTTP client not initialized") + + url = f"{self._base_url}/{endpoint}" + payload = {**body, "base_info": {"channel_version": _build_channel_version()}} + headers = self._build_headers() + + resp = await self._http.post(url, json=payload, headers=headers, timeout=timeout_s) + if resp.status_code >= 400: + raise RuntimeError(f"WeChat API {endpoint} HTTP {resp.status_code}: {resp.text[:200]}") + return resp.json() + + async def get_updates(self, get_updates_buf: str, timeout_s: float = 40) -> dict: + """Long-poll for inbound messages.""" + return await self.api_fetch("ilink/bot/getupdates", { + "get_updates_buf": get_updates_buf, + }, timeout_s=timeout_s) + + async def send_message(self, msg_body: dict) -> dict: + """Send a message downstream.""" + return await self.api_fetch("ilink/bot/sendmessage", msg_body) + + async def get_config(self, user_id: str, context_token: str = "") -> dict: + """Fetch bot config (includes typing_ticket) for a user.""" + return await self.api_fetch("ilink/bot/getconfig", { + "ilink_user_id": user_id, + "context_token": context_token, + }, timeout_s=10) + + async def send_typing(self, user_id: str, ticket: str) -> None: + """Send a typing indicator.""" + try: + await self.api_fetch("ilink/bot/sendtyping", { + "ilink_user_id": user_id, + "typing_ticket": ticket, + "status": 1, + }, timeout_s=10) + except Exception: + pass # Non-critical + + # -- CDN download ------------------------------------------------------- + + async def cdn_download_decrypt( + self, + encrypted_query_param: str, + aes_key_b64: str, + full_url: str = "", + ) -> bytes: + """Download from WeChat CDN and AES-128-ECB decrypt. + + If full_url is provided, it is used directly (SDK 2.1.x forward-compat). + Otherwise, the URL is built from encrypted_query_param. + """ + if not self._http: + raise RuntimeError("HTTP client not initialized") + key = parse_aes_key(aes_key_b64) + if full_url: + url = full_url + else: + url = ( + f"{self._cdn_base_url}/download" + f"?encrypted_query_param={urllib.parse.quote(encrypted_query_param, safe='')}" + ) + resp = await self._http.get(url, timeout=60) + resp.raise_for_status() + return aes_ecb_decrypt(resp.content, key) + + async def cdn_download_plain(self, encrypted_query_param: str, full_url: str = "") -> bytes: + """Download from WeChat CDN without decryption.""" + if not self._http: + raise RuntimeError("HTTP client not initialized") + if full_url: + url = full_url + else: + url = ( + f"{self._cdn_base_url}/download" + f"?encrypted_query_param={urllib.parse.quote(encrypted_query_param, safe='')}" + ) + resp = await self._http.get(url, timeout=60) + resp.raise_for_status() + return resp.content + + # -- CDN upload --------------------------------------------------------- + + async def cdn_upload( + self, + file_path: str, + to_user_id: str, + media_type: int, + ) -> Dict[str, Any]: + """Encrypt and upload a file to WeChat CDN. + + Returns dict with: filekey, download_param, aeskey, plaintext_size, + ciphertext_size, raw_md5. + + Improvements over v0.1: + - upload_full_url takes precedence over upload_param (SDK 2.1.x) + - Exponential backoff between retry attempts + - Proper x-encrypted-param header extraction + """ + if not self._http: + raise RuntimeError("HTTP client not initialized") + + file_size_check = Path(file_path).stat().st_size + if file_size_check > MEDIA_MAX_BYTES: + raise ValueError(f"File too large: {file_size_check} bytes (max {MEDIA_MAX_BYTES})") + + plaintext = Path(file_path).read_bytes() + raw_size = len(plaintext) + raw_md5 = hashlib.md5(plaintext).hexdigest() + file_size = aes_ecb_padded_size(raw_size) + filekey = os.urandom(16).hex() + aes_key = os.urandom(16) + + # Get upload URL + upload_resp = await self.api_fetch("ilink/bot/getuploadurl", { + "filekey": filekey, + "media_type": media_type, + "to_user_id": to_user_id, + "rawsize": raw_size, + "rawfilemd5": raw_md5, + "filesize": file_size, + "no_need_thumb": True, + "aeskey": aes_key.hex(), + }) + + # upload_full_url takes precedence (SDK 2.1.x change) + upload_full_url = (upload_resp.get("upload_full_url") or "").strip() + upload_param = upload_resp.get("upload_param") + + if upload_full_url: + cdn_url = upload_full_url + elif upload_param: + cdn_url = ( + f"{self._cdn_base_url}/upload" + f"?encrypted_query_param={urllib.parse.quote(upload_param, safe='')}" + f"&filekey={urllib.parse.quote(filekey, safe='')}" + ) + else: + raise RuntimeError("getUploadUrl returned no upload URL (need upload_full_url or upload_param)") + + # Encrypt + ciphertext = aes_ecb_encrypt(plaintext, aes_key) + + # Upload with retry + exponential backoff + download_param = None + last_error = None + for attempt in range(1, CDN_UPLOAD_MAX_RETRIES + 1): + try: + resp = await self._http.post( + cdn_url, + content=ciphertext, + headers={"Content-Type": "application/octet-stream"}, + timeout=60, + ) + if 400 <= resp.status_code < 500: + cdn_err = resp.headers.get("x-error-message", resp.text[:200]) + raise RuntimeError(f"CDN upload client error {resp.status_code}: {cdn_err}") + if resp.status_code != 200: + cdn_err = resp.headers.get("x-error-message", f"status {resp.status_code}") + size_kb = raw_size // 1024 + if "timeout" in cdn_err.lower(): + raise RuntimeError( + f"CDN upload timeout for {size_kb}KB file. " + f"Try compressing the file." + ) + raise RuntimeError(f"CDN upload server error: {cdn_err}") + + download_param = resp.headers.get("x-encrypted-param") + if not download_param: + raise RuntimeError("CDN response missing x-encrypted-param header") + break + except Exception as e: + last_error = e + if "client error" in str(e): + raise + if attempt < CDN_UPLOAD_MAX_RETRIES: + delay = min(2 ** attempt, 10) + logger.warning("[WeChat] CDN upload attempt %d failed, retrying in %ds: %s", attempt, delay, e) + await asyncio.sleep(delay) + else: + logger.error("[WeChat] CDN upload failed after %d attempts: %s", CDN_UPLOAD_MAX_RETRIES, e) + + if not download_param: + raise last_error or RuntimeError(f"CDN upload failed after {CDN_UPLOAD_MAX_RETRIES} attempts") + + return { + "filekey": filekey, + "download_param": download_param, + "aeskey": aes_key.hex(), + "plaintext_size": raw_size, + "ciphertext_size": file_size, + "raw_md5": raw_md5, + } + + # -- File download helper ----------------------------------------------- + + async def download_remote_file(self, url: str, cache_dir: Path) -> Tuple[str, str]: + """Download a remote URL to cache_dir. Returns (local_path, extension).""" + if not self._http: + raise RuntimeError("HTTP client not initialized") + resp = await self._http.get(url, timeout=60, follow_redirects=True) + resp.raise_for_status() + + ct = resp.headers.get("content-type", "") + ext = ".bin" + if "jpeg" in ct or "jpg" in ct: + ext = ".jpg" + elif "png" in ct: + ext = ".png" + elif "gif" in ct: + ext = ".gif" + elif "webp" in ct: + ext = ".webp" + elif "mp4" in ct: + ext = ".mp4" + else: + url_ext = Path(url.split("?")[0]).suffix.lower() + if url_ext in _MIME_MAP: + ext = url_ext + + filename = f"wx_dl_{uuid.uuid4().hex[:12]}{ext}" + filepath = cache_dir / filename + filepath.write_bytes(resp.content) + return str(filepath), ext diff --git a/gateway/run.py b/gateway/run.py index f6fb563cac9c..3f5eb60b726d 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -1076,6 +1076,7 @@ async def start(self) -> bool: "MATRIX_ALLOWED_USERS", "DINGTALK_ALLOWED_USERS", "FEISHU_ALLOWED_USERS", "WECOM_ALLOWED_USERS", + "WECHAT_ALLOWED_USERS", "GATEWAY_ALLOWED_USERS") ) _allow_all = os.getenv("GATEWAY_ALLOW_ALL_USERS", "").lower() in ("true", "1", "yes") or any( @@ -1086,7 +1087,8 @@ async def start(self) -> bool: "SMS_ALLOW_ALL_USERS", "MATTERMOST_ALLOW_ALL_USERS", "MATRIX_ALLOW_ALL_USERS", "DINGTALK_ALLOW_ALL_USERS", "FEISHU_ALLOW_ALL_USERS", - "WECOM_ALLOW_ALL_USERS") + "WECOM_ALLOW_ALL_USERS", + "WECHAT_ALLOW_ALL_USERS") ) if not _any_allowlist and not _allow_all: logger.warning( @@ -1619,6 +1621,14 @@ def _create_adapter( return None return WeComAdapter(config) + elif platform == Platform.WECHAT: + from gateway.platforms.wechat import WeChatAdapter + from gateway.platforms.wechat_transport import check_wechat_requirements + if not check_wechat_requirements(): + logger.warning("WeChat: httpx or cryptography not installed") + return None + return WeChatAdapter(config) + elif platform == Platform.MATTERMOST: from gateway.platforms.mattermost import MattermostAdapter, check_mattermost_requirements if not check_mattermost_requirements(): @@ -1687,6 +1697,7 @@ def _is_user_authorized(self, source: SessionSource) -> bool: Platform.DINGTALK: "DINGTALK_ALLOWED_USERS", Platform.FEISHU: "FEISHU_ALLOWED_USERS", Platform.WECOM: "WECOM_ALLOWED_USERS", + Platform.WECHAT: "WECHAT_ALLOWED_USERS", } platform_allow_all_map = { Platform.TELEGRAM: "TELEGRAM_ALLOW_ALL_USERS", @@ -1701,6 +1712,7 @@ def _is_user_authorized(self, source: SessionSource) -> bool: Platform.DINGTALK: "DINGTALK_ALLOW_ALL_USERS", Platform.FEISHU: "FEISHU_ALLOW_ALL_USERS", Platform.WECOM: "WECOM_ALLOW_ALL_USERS", + Platform.WECHAT: "WECHAT_ALLOW_ALL_USERS", } # Per-platform allow-all flag (e.g., DISCORD_ALLOW_ALL_USERS=true) @@ -5476,7 +5488,7 @@ async def _handle_deny_command(self, event: MessageEvent) -> str: Platform.TELEGRAM, Platform.DISCORD, Platform.SLACK, Platform.WHATSAPP, Platform.SIGNAL, Platform.MATTERMOST, Platform.MATRIX, Platform.HOMEASSISTANT, Platform.EMAIL, Platform.SMS, Platform.DINGTALK, - Platform.FEISHU, Platform.WECOM, Platform.LOCAL, + Platform.FEISHU, Platform.WECOM, Platform.WECHAT, Platform.LOCAL, }) async def _handle_update_command(self, event: MessageEvent) -> str: diff --git a/hermes_cli/config.py b/hermes_cli/config.py index d90fc2155cd5..785040fd00ca 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -39,6 +39,9 @@ "DINGTALK_CLIENT_ID", "DINGTALK_CLIENT_SECRET", "FEISHU_APP_ID", "FEISHU_APP_SECRET", "FEISHU_ENCRYPT_KEY", "FEISHU_VERIFICATION_TOKEN", "WECOM_BOT_ID", "WECOM_SECRET", + "WECHAT_BOT_TOKEN", "WECHAT_ACCOUNT_ID", "WECHAT_ALLOWED_USERS", + "WECHAT_ALLOW_ALL_USERS", "WECHAT_HOME_CHANNEL", "WECHAT_HOME_CHANNEL_NAME", + "WECHAT_ILINK_APP_ID", "WECHAT_ILINK_CLIENT_VERSION", "TERMINAL_ENV", "TERMINAL_SSH_KEY", "TERMINAL_SSH_PORT", "WHATSAPP_MODE", "WHATSAPP_ENABLED", "MATTERMOST_HOME_CHANNEL", "MATTERMOST_REPLY_MODE", diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index 4a12a34bb0e1..64e7192154f4 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -1546,6 +1546,30 @@ def run_gateway(verbose: int = 0, quiet: bool = False, replace: bool = False): "help": "Chat ID for scheduled results and notifications."}, ], }, + { + "key": "wechat", + "label": "WeChat", + "emoji": "💬", + "token_var": "WECHAT_BOT_TOKEN", + "setup_instructions": [ + "1. Run the WeChat QR login flow to get a bot token from the iLink Bot API", + "2. Scan the QR code in personal WeChat and complete the login confirmation", + "3. Copy the returned token and account ID into Hermes", + "4. Message the bot account from WeChat to establish a context token before sending outbound replies", + "5. Restrict access with WECHAT_ALLOWED_USERS for production use", + ], + "vars": [ + {"name": "WECHAT_BOT_TOKEN", "prompt": "Bot token", "password": True, + "help": "Bearer token returned by the WeChat QR login flow."}, + {"name": "WECHAT_ACCOUNT_ID", "prompt": "Account ID", "password": False, + "help": "The bot account ID returned by the WeChat login flow."}, + {"name": "WECHAT_ALLOWED_USERS", "prompt": "Allowed user IDs (comma-separated, or empty)", "password": False, + "is_allowlist": True, + "help": "Restrict which WeChat users can interact with the bot."}, + {"name": "WECHAT_HOME_CHANNEL", "prompt": "Home user ID (optional, for cron/notifications)", "password": False, + "help": "WeChat user ID for scheduled results and notifications."}, + ], + }, ] diff --git a/hermes_cli/skills_config.py b/hermes_cli/skills_config.py index 7b44014ea598..53c10a71fb09 100644 --- a/hermes_cli/skills_config.py +++ b/hermes_cli/skills_config.py @@ -30,6 +30,7 @@ "dingtalk": "💬 DingTalk", "feishu": "🪽 Feishu", "wecom": "💬 WeCom", + "wechat": "💬 WeChat", "webhook": "🔗 Webhook", } diff --git a/hermes_cli/status.py b/hermes_cli/status.py index 77a3e0ef07fe..0aab7ccf0928 100644 --- a/hermes_cli/status.py +++ b/hermes_cli/status.py @@ -285,6 +285,7 @@ def show_status(args): "DingTalk": ("DINGTALK_CLIENT_ID", None), "Feishu": ("FEISHU_APP_ID", "FEISHU_HOME_CHANNEL"), "WeCom": ("WECOM_BOT_ID", "WECOM_HOME_CHANNEL"), + "WeChat": ("WECHAT_BOT_TOKEN", "WECHAT_HOME_CHANNEL"), } for name, (token_var, home_var) in platforms.items(): diff --git a/hermes_cli/tools_config.py b/hermes_cli/tools_config.py index 804a7a4f115a..af471c098c6e 100644 --- a/hermes_cli/tools_config.py +++ b/hermes_cli/tools_config.py @@ -148,6 +148,7 @@ def _get_plugin_toolset_keys() -> set: "dingtalk": {"label": "💬 DingTalk", "default_toolset": "hermes-dingtalk"}, "feishu": {"label": "🪽 Feishu", "default_toolset": "hermes-feishu"}, "wecom": {"label": "💬 WeCom", "default_toolset": "hermes-wecom"}, + "wechat": {"label": "💬 WeChat", "default_toolset": "hermes-wechat"}, "api_server": {"label": "🌐 API Server", "default_toolset": "hermes-api-server"}, "mattermost": {"label": "💬 Mattermost", "default_toolset": "hermes-mattermost"}, "webhook": {"label": "🔗 Webhook", "default_toolset": "hermes-webhook"}, diff --git a/scripts/wechat_login.py b/scripts/wechat_login.py new file mode 100644 index 000000000000..57264642c0f7 --- /dev/null +++ b/scripts/wechat_login.py @@ -0,0 +1,259 @@ +#!/usr/bin/env python3 +""" +Standalone WeChat QR login script for Hermes Agent. + +Run this on the Pi to authenticate the bot with WeChat: + python3 scripts/wechat_login.py + +After scanning the QR code with WeChat, credentials are saved to +~/.hermes/wechat/accounts/.json and can be loaded by +the Hermes gateway's WeChat adapter. + +Protocol compliance (openclaw-weixin SDK 2.1.x): + - iLink-App-Id and iLink-App-ClientVersion headers on all requests + - IDC redirect (scaned_but_redirect) handling + - QR auto-refresh on expiry (up to 3 times) + - Fixed base URL for QR requests (ilinkai.weixin.qq.com) +""" + +import asyncio +import base64 +import json +import os +import struct +import sys +import time +from pathlib import Path + +try: + import httpx +except ImportError: + print("Error: httpx not installed. Run: pip install httpx") + sys.exit(1) + +# Fixed base URL for all QR code requests (matches SDK login-qr.ts) +FIXED_BASE_URL = "https://ilinkai.weixin.qq.com" +DEFAULT_BOT_TYPE = "3" +ADAPTER_VERSION = "0.2.0" +MAX_QR_REFRESH_COUNT = 3 +QR_LONG_POLL_TIMEOUT_S = 35 + + +def _build_client_version(version: str) -> int: + """Encode version as uint32: 0x00MMNNPP.""" + parts = version.split(".") + major = int(parts[0]) if len(parts) > 0 else 0 + minor = int(parts[1]) if len(parts) > 1 else 0 + patch = int(parts[2]) if len(parts) > 2 else 0 + return ((major & 0xFF) << 16) | ((minor & 0xFF) << 8) | (patch & 0xFF) + + +def _random_uin_header() -> str: + rand_uint32 = struct.unpack(">I", os.urandom(4))[0] + return base64.b64encode(str(rand_uint32).encode()).decode() + + +def _headers() -> dict: + """Build headers with iLink protocol requirements (SDK 2.1.1+).""" + return { + "Content-Type": "application/json", + "X-WECHAT-UIN": _random_uin_header(), + "iLink-App-Id": os.getenv("WECHAT_ILINK_APP_ID", "bot"), + "iLink-App-ClientVersion": str(_build_client_version(ADAPTER_VERSION)), + } + + +async def fetch_qr_code(client: httpx.AsyncClient) -> dict: + """Request a QR code from the WeChat iLink API (always from fixed base URL).""" + url = f"{FIXED_BASE_URL}/ilink/bot/get_bot_qrcode?bot_type={DEFAULT_BOT_TYPE}" + resp = await client.get(url, headers=_headers()) + resp.raise_for_status() + return resp.json() + + +async def poll_qr_status(client: httpx.AsyncClient, base_url: str, qrcode: str) -> dict: + """Long-poll for QR code scan status.""" + url = f"{base_url}/ilink/bot/get_qrcode_status?qrcode={qrcode}" + try: + resp = await client.get(url, headers=_headers(), timeout=QR_LONG_POLL_TIMEOUT_S) + resp.raise_for_status() + return resp.json() + except httpx.TimeoutException: + return {"status": "wait"} + except Exception as e: + # Network/gateway errors: treat as wait and retry + print(f"\n (network error, retrying: {e})") + return {"status": "wait"} + + +def save_credentials(account_id: str, token: str, base_url: str, user_id: str = "") -> Path: + """Save credentials to $HERMES_HOME/wechat/accounts/.json""" + hermes_home = Path(os.getenv("HERMES_HOME", Path.home() / ".hermes")) + accounts_dir = hermes_home / "wechat" / "accounts" + accounts_dir.mkdir(parents=True, exist_ok=True) + + normalized = account_id.strip().lower().replace("@", "-").replace(".", "-") + + data = { + "token": token, + "baseUrl": base_url, + "savedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + } + if user_id: + data["userId"] = user_id + + filepath = accounts_dir / f"{normalized}.json" + filepath.write_text(json.dumps(data, indent=2)) + filepath.chmod(0o600) + + index_path = hermes_home / "wechat" / "accounts.json" + index_path.write_text(json.dumps([normalized], indent=2)) + + return filepath + + +def display_qr(qrcode_url: str) -> None: + """Display QR code in terminal, with fallback to URL.""" + try: + import qrcode as qr_lib + qr = qr_lib.QRCode(box_size=1, border=1) + qr.add_data(qrcode_url) + qr.make() + qr.print_ascii(invert=True) + except ImportError: + pass + print(f"\nQR Code URL: {qrcode_url}") + print("\nScan this QR code with WeChat to connect.\n") + + +async def main(): + print("WeChat Login for Hermes Agent") + print(f"API: {FIXED_BASE_URL}") + print(f"Protocol: iLink 2.1.x (App-Id + ClientVersion headers)\n") + + async with httpx.AsyncClient(follow_redirects=True) as client: + # Step 1: Get QR code + print("Fetching QR code...") + qr_data = await fetch_qr_code(client) + qrcode = qr_data.get("qrcode", "") + qrcode_url = qr_data.get("qrcode_img_content", "") + + if not qrcode: + print("Error: Failed to get QR code from server") + sys.exit(1) + + display_qr(qrcode_url) + + # Step 2: Poll for scan with IDC redirect + QR refresh support + scanned_printed = False + qr_refresh_count = 1 + polling_base_url = FIXED_BASE_URL # May change on IDC redirect + max_attempts = 120 # ~10 minutes with long-poll + + for attempt in range(max_attempts): + status = await poll_qr_status(client, polling_base_url, qrcode) + state = status.get("status", "wait") + + if state == "wait": + if not scanned_printed: + print(".", end="", flush=True) + + elif state == "scaned": + if not scanned_printed: + print("\n\n QR code scanned! Confirm on your phone...") + scanned_printed = True + + elif state == "scaned_but_redirect": + # IDC redirect: switch polling to a different datacenter (SDK 2.1.1+) + redirect_host = str(status.get("redirect_host", "")).strip() + if redirect_host: + polling_base_url = f"https://{redirect_host}" + print(f"\n IDC redirect -> {redirect_host}") + else: + print("\n IDC redirect received but no redirect_host, continuing...") + + elif state == "expired": + qr_refresh_count += 1 + if qr_refresh_count > MAX_QR_REFRESH_COUNT: + print(f"\n\nQR code expired {MAX_QR_REFRESH_COUNT} times. Please try again.") + sys.exit(1) + + print(f"\n\n QR expired, refreshing ({qr_refresh_count}/{MAX_QR_REFRESH_COUNT})...") + try: + qr_data = await fetch_qr_code(client) + qrcode = qr_data.get("qrcode", "") + qrcode_url = qr_data.get("qrcode_img_content", "") + if not qrcode: + print(" Failed to refresh QR code") + sys.exit(1) + scanned_printed = False + polling_base_url = FIXED_BASE_URL # Reset polling URL + display_qr(qrcode_url) + except Exception as e: + print(f" Failed to refresh QR: {e}") + sys.exit(1) + + elif state == "confirmed": + bot_token = status.get("bot_token", "") + account_id = status.get("ilink_bot_id", "") + response_base_url = status.get("baseurl", FIXED_BASE_URL) + user_id = status.get("ilink_user_id", "") + + if not bot_token or not account_id: + print("\n\nLogin confirmed but missing credentials. Response:") + print(json.dumps(status, indent=2)) + sys.exit(1) + + filepath = save_credentials(account_id, bot_token, response_base_url, user_id) + + print(f"\n\nConnected successfully!") + print(f"\nCredentials saved to: {filepath}") + print(f"\nAccount ID: {account_id}") + if user_id: + print(f"User ID: {user_id}") + + print(f"\n--- Add these to ~/.hermes/.env ---") + print(f"WECHAT_BOT_TOKEN={bot_token}") + print(f"WECHAT_ACCOUNT_ID={account_id}") + if response_base_url != FIXED_BASE_URL: + print(f"WECHAT_API_BASE_URL={response_base_url}") + if user_id: + print(f"WECHAT_ALLOWED_USERS={user_id}") + print(f"-----------------------------------") + + # Quick connection test + print(f"\nTesting connection...") + try: + test_headers = { + **_headers(), + "AuthorizationType": "ilink_bot_token", + "Authorization": f"Bearer {bot_token}", + } + test_body = { + "get_updates_buf": "", + "base_info": {"channel_version": f"hermes-wechat/{ADAPTER_VERSION}"}, + } + test_resp = await client.post( + f"{response_base_url}/ilink/bot/getupdates", + json=test_body, + headers=test_headers, + timeout=10, + ) + if test_resp.status_code == 200: + print("Connection test passed!") + else: + print(f"Connection test returned HTTP {test_resp.status_code}") + except Exception as e: + print(f"Connection test failed: {e}") + print("(This may be normal if the server holds the long-poll)") + + return + + await asyncio.sleep(1) + + print("\n\nLogin timed out. Please try again.") + sys.exit(1) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tools/cronjob_tools.py b/tools/cronjob_tools.py index caedaca728d0..98180b886e18 100644 --- a/tools/cronjob_tools.py +++ b/tools/cronjob_tools.py @@ -454,7 +454,7 @@ def remove_cronjob(job_id: str, task_id: str = None) -> str: }, "deliver": { "type": "string", - "description": "Delivery target: origin, local, telegram, discord, slack, whatsapp, signal, matrix, mattermost, homeassistant, dingtalk, feishu, wecom, email, sms, or platform:chat_id or platform:chat_id:thread_id for Telegram topics. Examples: 'origin', 'local', 'telegram', 'telegram:-1001234567890:17585', 'discord:#engineering'" + "description": "Delivery target: origin, local, telegram, discord, slack, whatsapp, signal, matrix, mattermost, homeassistant, dingtalk, feishu, wecom, wechat, email, sms, or platform:chat_id or platform:chat_id:thread_id for Telegram topics. Examples: 'origin', 'local', 'telegram', 'telegram:-1001234567890:17585', 'discord:#engineering'" }, "skills": { "type": "array", diff --git a/tools/send_message_tool.py b/tools/send_message_tool.py index 4e500e694e84..ae7341d2d855 100644 --- a/tools/send_message_tool.py +++ b/tools/send_message_tool.py @@ -154,6 +154,7 @@ def _handle_send(args): "dingtalk": Platform.DINGTALK, "feishu": Platform.FEISHU, "wecom": Platform.WECOM, + "wechat": Platform.WECHAT, "email": Platform.EMAIL, "sms": Platform.SMS, } @@ -396,6 +397,8 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, result = await _send_feishu(pconfig, chat_id, chunk, thread_id=thread_id) elif platform == Platform.WECOM: result = await _send_wecom(pconfig.extra, chat_id, chunk) + elif platform == Platform.WECHAT: + result = await _send_wechat(pconfig, chat_id, chunk) else: result = {"error": f"Direct sending not yet implemented for {platform.value}"} @@ -870,6 +873,36 @@ async def _send_wecom(extra, chat_id, message): return _error(f"WeCom send failed: {e}") +async def _send_wechat(pconfig, chat_id, message): + """Send via WeChat using the adapter's long-poll send pipeline. + + Loads context tokens from disk — the target user must have messaged + the bot at least once for a context_token to exist. + """ + try: + from gateway.platforms.wechat import WeChatAdapter + from gateway.platforms.wechat_transport import check_wechat_requirements + if not check_wechat_requirements(): + return {"error": "WeChat requirements not met. Need httpx + cryptography."} + except ImportError: + return {"error": "WeChat adapter not available."} + + try: + adapter = WeChatAdapter(pconfig) + connected = await adapter.connect() + if not connected: + return _error(f"WeChat: failed to connect - {adapter.fatal_error_message or 'unknown error'}") + try: + result = await adapter.send(chat_id, message) + if not result.success: + return _error(f"WeChat send failed: {result.error}") + return {"success": True, "platform": "wechat", "chat_id": chat_id, "message_id": result.message_id} + finally: + await adapter.disconnect() + except Exception as e: + return _error(f"WeChat send failed: {e}") + + async def _send_feishu(pconfig, chat_id, message, media_files=None, thread_id=None): """Send via Feishu/Lark using the adapter's send pipeline.""" try: diff --git a/toolsets.py b/toolsets.py index 2a359b60a755..40c5e7c7c45b 100644 --- a/toolsets.py +++ b/toolsets.py @@ -353,6 +353,12 @@ "includes": [] }, + "hermes-wechat": { + "description": "WeChat bot toolset - personal WeChat messaging via iLink Bot API (full access)", + "tools": _HERMES_CORE_TOOLS, + "includes": [] + }, + "hermes-sms": { "description": "SMS bot toolset - interact with Hermes via SMS (Twilio)", "tools": _HERMES_CORE_TOOLS, @@ -368,7 +374,7 @@ "hermes-gateway": { "description": "Gateway toolset - union of all messaging platform tools", "tools": [], - "includes": ["hermes-telegram", "hermes-discord", "hermes-whatsapp", "hermes-slack", "hermes-signal", "hermes-homeassistant", "hermes-email", "hermes-sms", "hermes-mattermost", "hermes-matrix", "hermes-dingtalk", "hermes-feishu", "hermes-wecom", "hermes-webhook"] + "includes": ["hermes-telegram", "hermes-discord", "hermes-whatsapp", "hermes-slack", "hermes-signal", "hermes-homeassistant", "hermes-email", "hermes-sms", "hermes-mattermost", "hermes-matrix", "hermes-dingtalk", "hermes-feishu", "hermes-wecom", "hermes-wechat", "hermes-webhook"] } } From 5da9f06f0ff7ef07ef16a021bbe11358e0b19969 Mon Sep 17 00:00:00 2001 From: Xule Lin <43122877+linxule@users.noreply.github.com> Date: Wed, 8 Apr 2026 05:03:51 +0100 Subject: [PATCH 2/3] test(gateway): add WeChat adapter test suite (85 tests) Comprehensive coverage of adapter, transport, and state modules: - Platform enum and config loading from env vars - iLink header encoding and presence verification - AES-128-ECB encrypt/decrypt round-trip with key parsing - Context token persistence (save/load/clear lifecycle) - Sync buffer persistence - Inbound message extraction (text, voice, media, ref_msg) - Media selection priority and CDN upload size guard - Markdown stripping, dedup, typing ticket TTL - Send behavior (context tokens, chunking, self-message filtering) - Source-level registration assertions (config, run, toolsets) Co-Authored-By: Claude Opus 4.6 (1M context) Co-Authored-By: OpenAI Codex --- tests/gateway/test_wechat.py | 855 +++++++++++++++++++++++++++++++++++ 1 file changed, 855 insertions(+) create mode 100644 tests/gateway/test_wechat.py diff --git a/tests/gateway/test_wechat.py b/tests/gateway/test_wechat.py new file mode 100644 index 000000000000..5e97affc3c5c --- /dev/null +++ b/tests/gateway/test_wechat.py @@ -0,0 +1,855 @@ +"""Tests for the WeChat gateway adapter, transport, and state helpers.""" + +import base64 +import importlib +import json +import sys +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest + +from gateway.config import GatewayConfig, Platform, PlatformConfig, _apply_env_overrides + + +REPO_ROOT = Path(__file__).resolve().parents[2] + + +def _read_source(*parts: str) -> str: + return (REPO_ROOT.joinpath(*parts)).read_text(encoding="utf-8") + + +def _media_item(wechat_mod, item_type: int, *, stt_text: str = "") -> dict: + media = {"encrypt_query_param": "eqp", "aes_key": base64.b64encode(b"0123456789abcdef").decode()} + if item_type == wechat_mod.WX_ITEM_IMAGE: + return {"type": item_type, "image_item": {"media": media}} + if item_type == wechat_mod.WX_ITEM_VIDEO: + return {"type": item_type, "video_item": {"media": media}} + if item_type == wechat_mod.WX_ITEM_FILE: + return {"type": item_type, "file_item": {"media": media}} + return {"type": item_type, "voice_item": {"media": media, "text": stt_text}} + + +class _ClosedTask: + def __init__(self, coro): + coro.close() + + def cancel(self): + return None + + +class TestPlatformEnum: + def test_wechat_in_platform_enum(self): + assert Platform.WECHAT.value == "wechat" + + +class TestConfigLoading: + def test_apply_env_overrides_registers_wechat(self, monkeypatch): + monkeypatch.setenv("WECHAT_BOT_TOKEN", "token-123") + monkeypatch.setenv("WECHAT_ACCOUNT_ID", "bot-account") + monkeypatch.setenv("WECHAT_API_BASE_URL", "https://wx.example") + monkeypatch.setenv("WECHAT_CDN_BASE_URL", "https://cdn.example") + monkeypatch.setenv("WECHAT_ILINK_APP_ID", "my-bot") + monkeypatch.setenv("WECHAT_ILINK_CLIENT_VERSION", "12345") + monkeypatch.setenv("WECHAT_HOME_CHANNEL", "user-123") + + config = GatewayConfig() + _apply_env_overrides(config) + + assert Platform.WECHAT in config.platforms + platform_config = config.platforms[Platform.WECHAT] + assert platform_config.enabled is True + assert platform_config.token == "token-123" + assert platform_config.extra["account_id"] == "bot-account" + assert platform_config.extra["base_url"] == "https://wx.example" + assert platform_config.extra["cdn_base_url"] == "https://cdn.example" + assert platform_config.extra["ilink_app_id"] == "my-bot" + assert platform_config.extra["ilink_client_version"] == "12345" + assert platform_config.home_channel is not None + assert platform_config.home_channel.chat_id == "user-123" + + +class TestWeChatAdapterInit: + def test_reads_config_values(self): + from gateway.platforms.wechat import WeChatAdapter + + config = PlatformConfig( + enabled=True, + token="cfg-token", + extra={ + "account_id": "acct-1", + "base_url": "https://wx.example", + "cdn_base_url": "https://cdn.example", + "ilink_app_id": "cfg-app", + "ilink_client_version": "123", + }, + ) + + adapter = WeChatAdapter(config) + + assert adapter._transport._token == "cfg-token" + assert adapter._account_id == "acct-1" + assert adapter._transport._base_url == "https://wx.example" + assert adapter._transport._cdn_base_url == "https://cdn.example" + assert adapter._transport._ilink_app_id == "cfg-app" + assert adapter._transport._ilink_client_version == 123 + + def test_uses_default_base_urls_when_not_configured(self, monkeypatch): + import gateway.platforms.wechat as wechat + + monkeypatch.delenv("WECHAT_API_BASE_URL", raising=False) + monkeypatch.delenv("WECHAT_CDN_BASE_URL", raising=False) + + adapter = wechat.WeChatAdapter(PlatformConfig(enabled=True, token="cfg-token")) + + assert adapter._transport._base_url == wechat.DEFAULT_BASE_URL + assert adapter._transport._cdn_base_url == wechat.CDN_BASE_URL + assert adapter._account_id == "" + + def test_falls_back_to_env_vars(self, monkeypatch): + monkeypatch.setenv("WECHAT_BOT_TOKEN", "env-token") + monkeypatch.setenv("WECHAT_ACCOUNT_ID", "env-account") + monkeypatch.setenv("WECHAT_API_BASE_URL", "https://env.example") + monkeypatch.setenv("WECHAT_CDN_BASE_URL", "https://cdn.env.example") + monkeypatch.setenv("WECHAT_ILINK_APP_ID", "env-app") + monkeypatch.setenv("WECHAT_ILINK_CLIENT_VERSION", "777") + + from gateway.platforms.wechat import WeChatAdapter + + adapter = WeChatAdapter(PlatformConfig(enabled=True)) + + assert adapter._transport._token == "env-token" + assert adapter._account_id == "env-account" + assert adapter._transport._base_url == "https://env.example" + assert adapter._transport._cdn_base_url == "https://cdn.env.example" + assert adapter._transport._ilink_app_id == "env-app" + assert adapter._transport._ilink_client_version == 777 + + +class TestExtractText: + def test_extracts_plain_text(self): + from gateway.platforms.wechat import WeChatAdapter, WX_ITEM_TEXT + + items = [{"type": WX_ITEM_TEXT, "text_item": {"text": "hello"}}] + + assert WeChatAdapter._extract_text(items) == "hello" + + def test_extracts_empty_items(self): + from gateway.platforms.wechat import WeChatAdapter + + assert WeChatAdapter._extract_text([]) == "" + + def test_extracts_voice_stt(self): + from gateway.platforms.wechat import WeChatAdapter, WX_ITEM_VOICE + + items = [{"type": WX_ITEM_VOICE, "voice_item": {"text": "voice transcript"}}] + + assert WeChatAdapter._extract_text(items) == "voice transcript" + + def test_extracts_ref_with_title(self): + from gateway.platforms.wechat import WeChatAdapter, WX_ITEM_TEXT + + items = [ + { + "type": WX_ITEM_TEXT, + "text_item": {"text": "reply"}, + "ref_msg": { + "title": "Alice", + "message_item": {"type": WX_ITEM_TEXT, "text_item": {"text": "quoted message"}}, + }, + } + ] + + assert WeChatAdapter._extract_text(items) == "[Quote: Alice | quoted message]\nreply" + + def test_extracts_ref_with_message_item_only(self): + from gateway.platforms.wechat import WeChatAdapter, WX_ITEM_TEXT + + items = [ + { + "type": WX_ITEM_TEXT, + "text_item": {"text": "reply"}, + "ref_msg": { + "message_item": {"type": WX_ITEM_TEXT, "text_item": {"text": "quoted only"}}, + }, + } + ] + + assert WeChatAdapter._extract_text(items) == "[Quote: quoted only]\nreply" + + def test_ref_msg_with_title_and_content(self): + from gateway.platforms.wechat import WeChatAdapter, WX_ITEM_TEXT + + items = [ + { + "type": WX_ITEM_TEXT, + "text_item": {"text": "fresh text"}, + "ref_msg": { + "title": "Thread title", + "message_item": {"type": WX_ITEM_TEXT, "text_item": {"text": "quoted body"}}, + }, + } + ] + + assert WeChatAdapter._extract_text(items) == "[Quote: Thread title | quoted body]\nfresh text" + + def test_ref_media_returns_text_only(self): + from gateway.platforms.wechat import WeChatAdapter, WX_ITEM_IMAGE, WX_ITEM_TEXT + + items = [ + { + "type": WX_ITEM_TEXT, + "text_item": {"text": "reply only"}, + "ref_msg": { + "title": "Alice", + "message_item": { + "type": WX_ITEM_IMAGE, + "image_item": {"media": {"encrypt_query_param": "eqp"}}, + }, + }, + } + ] + + assert WeChatAdapter._extract_text(items) == "reply only" + + def test_ref_msg_media_returns_text_only(self): + from gateway.platforms.wechat import WeChatAdapter, WX_ITEM_VIDEO, WX_ITEM_TEXT + + items = [ + { + "type": WX_ITEM_TEXT, + "text_item": {"text": "caption"}, + "ref_msg": { + "message_item": { + "type": WX_ITEM_VIDEO, + "video_item": {"media": {"encrypt_query_param": "eqp"}}, + }, + }, + } + ] + + assert WeChatAdapter._extract_text(items) == "caption" + + +class TestMediaSelection: + def test_find_media_item_skips_voice_with_stt(self): + import gateway.platforms.wechat as wechat + + item = { + "type": wechat.WX_ITEM_VOICE, + "voice_item": { + "text": "already transcribed", + "media": {"encrypt_query_param": "eqp"}, + }, + } + + assert wechat.WeChatAdapter._find_media_item([item]) is None + + @pytest.mark.parametrize( + ("items_factory", "expected_type"), + [ + ( + lambda m: [ + _media_item(m, m.WX_ITEM_VOICE), + _media_item(m, m.WX_ITEM_FILE), + _media_item(m, m.WX_ITEM_IMAGE), + _media_item(m, m.WX_ITEM_VIDEO), + ], + "WX_ITEM_IMAGE", + ), + ( + lambda m: [ + _media_item(m, m.WX_ITEM_VOICE), + _media_item(m, m.WX_ITEM_FILE), + _media_item(m, m.WX_ITEM_VIDEO), + ], + "WX_ITEM_VIDEO", + ), + ( + lambda m: [ + _media_item(m, m.WX_ITEM_VOICE), + _media_item(m, m.WX_ITEM_FILE), + ], + "WX_ITEM_FILE", + ), + ( + lambda m: [ + _media_item(m, m.WX_ITEM_VOICE), + ], + "WX_ITEM_VOICE", + ), + ], + ) + def test_media_priority_is_image_then_video_then_file_then_voice(self, items_factory, expected_type): + import gateway.platforms.wechat as wechat + + item = wechat.WeChatAdapter._find_media_item(items_factory(wechat)) + + assert item is not None + assert item["type"] == getattr(wechat, expected_type) + + def test_find_media_item_uses_quoted_media(self): + import gateway.platforms.wechat as wechat + + items = [ + { + "type": wechat.WX_ITEM_TEXT, + "text_item": {"text": "reply"}, + "ref_msg": { + "message_item": _media_item(wechat, wechat.WX_ITEM_FILE), + }, + } + ] + + item = wechat.WeChatAdapter._find_media_item(items) + + assert item is not None + assert item["type"] == wechat.WX_ITEM_FILE + + def test_find_media_item_returns_none_when_missing_encrypt_query_param(self): + from gateway.platforms.wechat import WeChatAdapter, WX_ITEM_IMAGE + + items = [{"type": WX_ITEM_IMAGE, "image_item": {"media": {}}}] + + assert WeChatAdapter._find_media_item(items) is None + + +class TestCryptoHelpers: + def test_aes_ecb_encrypt_decrypt_round_trip(self): + from gateway.platforms.wechat_transport import aes_ecb_decrypt, aes_ecb_encrypt + + key = b"0123456789abcdef" + plaintext = b"hello wechat media payload" + + ciphertext = aes_ecb_encrypt(plaintext, key) + + assert ciphertext != plaintext + assert aes_ecb_decrypt(ciphertext, key) == plaintext + + def test_parse_aes_key_accepts_raw_bytes_base64(self): + from gateway.platforms.wechat_transport import parse_aes_key + + key = b"0123456789abcdef" + encoded = base64.b64encode(key).decode() + + assert parse_aes_key(encoded) == key + + def test_parse_aes_key_accepts_hex_string_base64(self): + from gateway.platforms.wechat_transport import parse_aes_key + + key = b"0123456789abcdef" + encoded = base64.b64encode(key.hex().encode("ascii")).decode() + + assert parse_aes_key(encoded) == key + + +class TestMimeHelpers: + @pytest.mark.parametrize( + ("file_path", "expected"), + [ + ("photo.jpg", "image/jpeg"), + ("video.mp4", "video/mp4"), + ("doc.pdf", "application/pdf"), + ("voice.ogg", "audio/ogg"), + ], + ) + def test_mime_from_path(self, file_path, expected): + from gateway.platforms.wechat_transport import mime_from_path + + assert mime_from_path(file_path) == expected + + +class TestMarkdownToPlain: + @pytest.mark.parametrize( + ("text", "expected"), + [ + ("**bold**", "bold"), + ("*italic*", "italic"), + ("__strong__", "strong"), + ("_emphasis_", "emphasis"), + ("~~strike~~", "strike"), + ("`inline`", "inline"), + ("# Heading", "Heading"), + ("## Heading", "Heading"), + ("### Heading", "Heading"), + ("#### Heading", "Heading"), + ("##### Heading", "Heading"), + ("###### Heading", "Heading"), + ("[link](https://example.com)", "link"), + ("![img](https://example.com/x.png)", ""), + ("| a | b |", "a b"), + ("| --- | --- |", ""), + ("```python\nprint('x')\n```", "print('x')"), + ("plain text", "plain text"), + ("before **bold** after", "before bold after"), + ("mix [site](https://example.com) and `code`", "mix site and code"), + ], + ) + def test_strips_supported_markdown(self, text, expected): + from gateway.platforms.wechat import _markdown_to_plain + + assert _markdown_to_plain(text) == expected + + +class TestDeduplication: + def test_same_key_is_rejected(self): + from gateway.platforms.wechat import WeChatAdapter + + adapter = WeChatAdapter(PlatformConfig(enabled=True, token="token")) + + assert adapter._is_duplicate("msg-1") is False + assert adapter._is_duplicate("msg-1") is True + + def test_different_key_is_accepted(self): + from gateway.platforms.wechat import WeChatAdapter + + adapter = WeChatAdapter(PlatformConfig(enabled=True, token="token")) + + assert adapter._is_duplicate("msg-1") is False + assert adapter._is_duplicate("msg-2") is False + + def test_window_expiry_prunes_old_entries(self, monkeypatch): + import gateway.platforms.wechat as wechat + + adapter = wechat.WeChatAdapter(PlatformConfig(enabled=True, token="token")) + now = 1_000.0 + monkeypatch.setattr(wechat.time, "time", lambda: now) + adapter._seen_messages = { + f"old-{i}": now - wechat.DEDUP_WINDOW_S - 10 for i in range(wechat.DEDUP_MAX_SIZE + 1) + } + + assert adapter._is_duplicate("fresh") is False + assert list(adapter._seen_messages) == ["fresh"] + + +class TestTypingTickets: + @pytest.mark.asyncio + async def test_fresh_ticket_not_refetched(self, monkeypatch): + import gateway.platforms.wechat as wechat + + adapter = wechat.WeChatAdapter(PlatformConfig(enabled=True, token="token")) + now = 1_000.0 + monkeypatch.setattr(wechat.time, "time", lambda: now) + adapter._typing_tickets["user-1"] = ("cached-ticket", now) + adapter._typing_ticket_ttl_s = 60 + adapter._transport.get_config = AsyncMock(return_value={"typing_ticket": "new-ticket"}) + + await adapter._cache_typing_ticket("user-1", "ctx-123") + + adapter._transport.get_config.assert_not_called() + assert adapter._typing_tickets["user-1"] == ("cached-ticket", now) + + @pytest.mark.asyncio + async def test_expired_ticket_is_refetched(self, monkeypatch): + import gateway.platforms.wechat as wechat + + adapter = wechat.WeChatAdapter(PlatformConfig(enabled=True, token="token")) + now = 1_000.0 + monkeypatch.setattr(wechat.time, "time", lambda: now) + adapter._typing_tickets["user-1"] = ("stale-ticket", now - 120) + adapter._typing_ticket_ttl_s = 60 + adapter._transport.get_config = AsyncMock(return_value={"typing_ticket": "fresh-ticket"}) + + await adapter._cache_typing_ticket("user-1", "ctx-123") + + adapter._transport.get_config.assert_awaited_once_with("user-1", "ctx-123") + assert adapter._typing_tickets["user-1"] == ("fresh-ticket", now) + + +class TestSend: + @pytest.mark.asyncio + async def test_send_requires_context_token(self): + from gateway.platforms.wechat import WeChatAdapter + + adapter = WeChatAdapter(PlatformConfig(enabled=True, token="token")) + + result = await adapter.send("user-1", "hello") + + assert result.success is False + assert "context_token" in (result.error or "") + + @pytest.mark.asyncio + async def test_send_skips_empty_text_after_markdown_stripping(self): + from gateway.platforms.wechat import WeChatAdapter + + adapter = WeChatAdapter(PlatformConfig(enabled=True, token="token")) + adapter._context_tokens["user-1"] = "ctx-123" + adapter._transport.send_message = AsyncMock(return_value={"ret": 0}) + + result = await adapter.send("user-1", "![img](https://example.com/a.png)") + + assert result.success is True + assert result.message_id == "skipped-empty" + adapter._transport.send_message.assert_not_called() + + @pytest.mark.asyncio + async def test_send_posts_plain_text_when_context_token_present(self): + from gateway.platforms.wechat import WeChatAdapter + + adapter = WeChatAdapter(PlatformConfig(enabled=True, token="token")) + adapter._context_tokens["user-1"] = "ctx-123" + adapter._transport.send_message = AsyncMock(return_value={"ret": 0}) + + result = await adapter.send("user-1", "**Hello** [there](https://example.com)") + + assert result.success is True + adapter._transport.send_message.assert_awaited_once() + body = adapter._transport.send_message.await_args.args[0] + assert body["msg"]["to_user_id"] == "user-1" + assert body["msg"]["context_token"] == "ctx-123" + assert body["msg"]["item_list"][0]["text_item"]["text"] == "Hello there" + + @pytest.mark.asyncio + async def test_send_chunks_at_4096(self): + from gateway.platforms.wechat import MAX_MESSAGE_LENGTH, WeChatAdapter + + adapter = WeChatAdapter(PlatformConfig(enabled=True, token="token")) + adapter._context_tokens["user-1"] = "ctx-123" + adapter._transport.send_message = AsyncMock(return_value={"ret": 0}) + + result = await adapter.send("user-1", "x" * (MAX_MESSAGE_LENGTH + 200)) + + assert result.success is True + assert adapter._transport.send_message.await_count == 2 + for call in adapter._transport.send_message.await_args_list: + chunk = call.args[0]["msg"]["item_list"][0]["text_item"]["text"] + assert len(chunk) <= MAX_MESSAGE_LENGTH + + +class TestInboundFiltering: + @pytest.mark.asyncio + async def test_non_user_message_is_skipped(self): + from gateway.platforms.wechat import WeChatAdapter + + adapter = WeChatAdapter(PlatformConfig(enabled=True, token="token")) + adapter.handle_message = AsyncMock() + + await adapter._on_message( + { + "from_user_id": "user-1", + "message_type": 2, + "message_id": "m1", + "seq": "1", + "item_list": [{"type": 1, "text_item": {"text": "hi"}}], + } + ) + + adapter.handle_message.assert_not_awaited() + + @pytest.mark.asyncio + async def test_own_account_id_is_skipped(self): + from gateway.platforms.wechat import WeChatAdapter, WX_MSG_TYPE_USER + + adapter = WeChatAdapter( + PlatformConfig(enabled=True, token="token", extra={"account_id": "bot-account"}) + ) + adapter.handle_message = AsyncMock() + + await adapter._on_message( + { + "from_user_id": "bot-account", + "message_type": WX_MSG_TYPE_USER, + "message_id": "m1", + "seq": "1", + "item_list": [{"type": 1, "text_item": {"text": "hi"}}], + } + ) + + adapter.handle_message.assert_not_awaited() + + +class TestCdnUploadGuards: + @pytest.mark.asyncio + async def test_cdn_upload_rejects_files_over_limit(self, tmp_path, monkeypatch): + import gateway.platforms.wechat_transport as wechat_transport + + transport = wechat_transport.WeChatTransport(token="token") + transport._http = AsyncMock() + + file_path = tmp_path / "too-big.bin" + file_path.write_bytes(b"x") + real_stat = wechat_transport.Path.stat + + def fake_stat(path_obj): + if path_obj == file_path: + return SimpleNamespace(st_size=wechat_transport.MEDIA_MAX_BYTES + 1) + return real_stat(path_obj) + + monkeypatch.setattr(wechat_transport.Path, "stat", fake_stat) + + with pytest.raises(ValueError, match="File too large"): + await transport.cdn_upload(str(file_path), "user-1", wechat_transport.UPLOAD_MEDIA_FILE) + + +class TestOutboundAesKeyEncoding: + @pytest.mark.asyncio + @pytest.mark.parametrize( + ("method_name", "kwargs", "type_key"), + [ + ("send_image_file", {"image_path": "/tmp/demo.png"}, "image_item"), + ("send_video", {"video_path": "/tmp/demo.mp4"}, "video_item"), + ("send_document", {"file_path": "/tmp/demo.pdf"}, "file_item"), + ("send_voice", {"audio_path": "/tmp/demo.wav"}, "file_item"), + ], + ) + async def test_outbound_aes_key_is_base64_of_hex_string(self, method_name, kwargs, type_key, monkeypatch): + from gateway.platforms.wechat import WeChatAdapter + + adapter = WeChatAdapter(PlatformConfig(enabled=True, token="token")) + adapter._context_tokens["user-1"] = "ctx-123" + adapter._transport.cdn_upload = AsyncMock( + return_value={ + "download_param": "enc-param", + "aeskey": "683c5d59ca1efb1b93989745deae27c4", + "ciphertext_size": 128, + "plaintext_size": 100, + } + ) + adapter._transport.send_message = AsyncMock(return_value={"ret": 0}) + + if method_name == "send_voice": + monkeypatch.setattr("pathlib.Path.stat", lambda _path, **_kw: SimpleNamespace(st_size=10)) + + result = await getattr(adapter, method_name)("user-1", **kwargs) + + assert result.success is True + body = adapter._transport.send_message.await_args_list[-1].args[0] + media = body["msg"]["item_list"][0][type_key]["media"] + assert base64.b64decode(media["aes_key"]).decode("ascii") == "683c5d59ca1efb1b93989745deae27c4" + + +class TestILinkHeaders: + def test_ilink_headers_present(self): + from gateway.platforms.wechat_transport import WeChatTransport + + transport = WeChatTransport(token="token", ilink_app_id="bot", ilink_client_version=123) + headers = transport._build_headers() + + assert "iLink-App-Id" in headers + assert "iLink-App-ClientVersion" in headers + + @pytest.mark.parametrize( + ("version", "expected"), + [ + ("2.1.7", 0x00020107), + ("0.2.0", 512), + ], + ) + def test_ilink_client_version_encoding(self, version, expected): + from gateway.platforms.wechat_transport import _build_client_version + + assert _build_client_version(version) == expected + + def test_build_common_headers(self): + from gateway.platforms.wechat_transport import WeChatTransport + + transport = WeChatTransport(token="token", ilink_app_id="my-app", ilink_client_version=131335) + + assert transport._build_common_headers() == { + "iLink-App-Id": "my-app", + "iLink-App-ClientVersion": "131335", + } + + def test_build_channel_version(self): + from gateway.platforms.wechat_transport import _build_channel_version + + assert _build_channel_version().startswith("hermes-wechat/") + + +class TestWeChatState: + def test_context_token_load(self, tmp_path, monkeypatch): + import gateway.platforms.wechat_state as wechat_state + + monkeypatch.setattr(wechat_state, "get_hermes_home", lambda: tmp_path) + state_dir = tmp_path / "wechat" + state_dir.mkdir() + (state_dir / "context_tokens.json").write_text('{"user-1":"ctx-123"}', encoding="utf-8") + + assert wechat_state.load_context_tokens() == {"user-1": "ctx-123"} + + def test_context_token_save_load_roundtrip(self, tmp_path, monkeypatch): + import gateway.platforms.wechat_state as wechat_state + + monkeypatch.setattr(wechat_state, "get_hermes_home", lambda: tmp_path) + + wechat_state.save_context_tokens({"user-1": "ctx-123", "user-2": "ctx-456"}) + + assert wechat_state.load_context_tokens() == {"user-1": "ctx-123", "user-2": "ctx-456"} + + def test_context_token_clear(self, tmp_path, monkeypatch): + import gateway.platforms.wechat_state as wechat_state + + monkeypatch.setattr(wechat_state, "get_hermes_home", lambda: tmp_path) + wechat_state.save_context_tokens({"user-1": "ctx-123"}) + + wechat_state.clear_context_tokens() + + assert not (tmp_path / "wechat" / "context_tokens.json").exists() + + def test_context_token_load_empty(self, tmp_path, monkeypatch): + import gateway.platforms.wechat_state as wechat_state + + monkeypatch.setattr(wechat_state, "get_hermes_home", lambda: tmp_path) + + assert wechat_state.load_context_tokens() == {} + + def test_sync_buf_save_load_roundtrip(self, tmp_path, monkeypatch): + import gateway.platforms.wechat_state as wechat_state + + monkeypatch.setattr(wechat_state, "get_hermes_home", lambda: tmp_path) + + wechat_state.save_sync_buf("acct-1", "cursor-123") + + assert wechat_state.load_sync_buf("acct-1") == "cursor-123" + + def test_sync_buf_empty_account_is_blank(self, tmp_path, monkeypatch): + import gateway.platforms.wechat_state as wechat_state + + monkeypatch.setattr(wechat_state, "get_hermes_home", lambda: tmp_path) + + assert wechat_state.load_sync_buf("") == "" + + +class TestWeChatConnectAndPolling: + @pytest.mark.asyncio + async def test_context_token_reload_at_connect(self, monkeypatch): + import gateway.platforms.wechat as wechat + + adapter = wechat.WeChatAdapter(PlatformConfig(enabled=True, token="token")) + adapter._transport.open = AsyncMock() + monkeypatch.setattr(wechat, "check_wechat_requirements", lambda: True) + monkeypatch.setattr(wechat, "load_context_tokens", lambda: {"user-1": "ctx-123"}) + monkeypatch.setattr(wechat.asyncio, "create_task", lambda coro: _ClosedTask(coro)) + + connected = await adapter.connect() + + assert connected is True + assert adapter._context_tokens == {"user-1": "ctx-123"} + adapter._transport.open.assert_awaited_once() + + @pytest.mark.asyncio + async def test_context_token_clear_on_session_expiry(self, monkeypatch): + import gateway.platforms.wechat as wechat + + adapter = wechat.WeChatAdapter(PlatformConfig(enabled=True, token="token")) + adapter._running = True + adapter._context_tokens = {"user-1": "ctx-123"} + adapter._transport.get_updates = AsyncMock(return_value={"ret": 0, "errcode": -14}) + clear_mock = Mock() + + monkeypatch.setattr(wechat, "load_sync_buf", lambda _account_id: "") + monkeypatch.setattr(wechat, "clear_context_tokens", clear_mock) + monkeypatch.setattr(wechat.time, "time", lambda: 0.0) + + async def fake_sleep(_seconds): + adapter._running = False + + monkeypatch.setattr(wechat.asyncio, "sleep", fake_sleep) + + await adapter._poll_loop() + + assert adapter._context_tokens == {} + clear_mock.assert_called_once() + + +class TestTransportUploadBehavior: + @pytest.mark.asyncio + async def test_cdn_upload_full_url_priority(self, tmp_path): + from gateway.platforms.wechat_transport import UPLOAD_MEDIA_IMAGE, WeChatTransport + + file_path = tmp_path / "demo.bin" + file_path.write_bytes(b"hello world") + + post_mock = AsyncMock(return_value=SimpleNamespace(status_code=200, headers={"x-encrypted-param": "enc"}, text="OK")) + transport = WeChatTransport(token="token") + transport._http = SimpleNamespace(post=post_mock) + transport.api_fetch = AsyncMock( + return_value={ + "upload_full_url": "https://upload.example/full", + "upload_param": "legacy-param", + } + ) + + result = await transport.cdn_upload(str(file_path), "user-1", UPLOAD_MEDIA_IMAGE) + + assert result["download_param"] == "enc" + assert post_mock.await_args.args[0] == "https://upload.example/full" + + @pytest.mark.asyncio + async def test_cdn_retry_with_backoff(self, tmp_path, monkeypatch): + from gateway.platforms.wechat_transport import UPLOAD_MEDIA_FILE, WeChatTransport + + file_path = tmp_path / "demo.bin" + file_path.write_bytes(b"hello world") + + responses = [ + SimpleNamespace(status_code=500, headers={"x-error-message": "server error"}, text="bad"), + SimpleNamespace(status_code=500, headers={"x-error-message": "server error"}, text="bad"), + SimpleNamespace(status_code=200, headers={"x-encrypted-param": "enc"}, text="OK"), + ] + post_mock = AsyncMock(side_effect=responses) + sleep_mock = AsyncMock() + + transport = WeChatTransport(token="token") + transport._http = SimpleNamespace(post=post_mock) + transport.api_fetch = AsyncMock(return_value={"upload_param": "param"}) + monkeypatch.setattr("gateway.platforms.wechat_transport.asyncio.sleep", sleep_mock) + + result = await transport.cdn_upload(str(file_path), "user-1", UPLOAD_MEDIA_FILE) + + assert result["download_param"] == "enc" + assert post_mock.await_count == 3 + assert sleep_mock.await_count == 2 + + +class TestSilkFallback: + def test_silk_to_wav_fallback(self, monkeypatch): + from gateway.platforms.wechat import _silk_to_wav + + def raise_missing(*args, **kwargs): + raise FileNotFoundError + + monkeypatch.setattr("subprocess.run", raise_missing) + + assert _silk_to_wav(b"fake-silk") is None + + +class TestModuleImports: + def test_transport_imports_clean(self): + sys.modules.pop("gateway.platforms.wechat", None) + sys.modules.pop("gateway.platforms.wechat_transport", None) + + mod = importlib.import_module("gateway.platforms.wechat_transport") + + assert mod.__name__ == "gateway.platforms.wechat_transport" + assert "gateway.platforms.wechat" not in sys.modules + + def test_state_imports_clean(self): + sys.modules.pop("gateway.platforms.wechat", None) + sys.modules.pop("gateway.platforms.wechat_state", None) + + mod = importlib.import_module("gateway.platforms.wechat_state") + + assert mod.__name__ == "gateway.platforms.wechat_state" + assert "gateway.platforms.wechat" not in sys.modules + + +class TestSourceRegistration: + def test_platform_enum_registered(self): + assert "WECHAT" in _read_source("gateway", "config.py") + + def test_toolset_registered(self): + assert "hermes-wechat" in _read_source("toolsets.py") + + def test_gateway_run_adapter_factory(self): + source = _read_source("gateway", "run.py") + + assert "Platform.WECHAT" in source + assert "WeChatAdapter" in source + + def test_gateway_run_auth_maps(self): + source = _read_source("gateway", "run.py") + + assert "WECHAT_ALLOWED_USERS" in source + assert "WECHAT_ALLOW_ALL_USERS" in source From 6fb530773c4c615a117f614828018d4ee53241e0 Mon Sep 17 00:00:00 2001 From: Xule Lin <43122877+linxule@users.noreply.github.com> Date: Wed, 8 Apr 2026 05:04:00 +0100 Subject: [PATCH 3/3] docs(gateway): add WeChat setup guide and platform references - New: website/docs/user-guide/messaging/wechat.md (full setup guide with env vars, feature matrix, architecture overview, troubleshooting) - Update messaging index with WeChat in architecture diagram, toolset table, and platform links - Add WECHAT_* env vars to environment-variables.md reference - Add WeChat to README.md platform mentions and AGENTS.md directory listing - Add wechat to website sidebar navigation Co-Authored-By: Claude Opus 4.6 (1M context) Co-Authored-By: OpenAI Codex --- AGENTS.md | 2 +- README.md | 6 +- .../docs/reference/environment-variables.md | 10 + website/docs/user-guide/messaging/index.md | 6 +- website/docs/user-guide/messaging/wechat.md | 192 ++++++++++++++++++ website/sidebars.ts | 1 + 6 files changed, 212 insertions(+), 5 deletions(-) create mode 100644 website/docs/user-guide/messaging/wechat.md diff --git a/AGENTS.md b/AGENTS.md index 8045c3d213df..78942be31ce1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -55,7 +55,7 @@ hermes-agent/ ├── gateway/ # Messaging platform gateway │ ├── run.py # Main loop, slash commands, message dispatch │ ├── session.py # SessionStore — conversation persistence -│ └── platforms/ # Adapters: telegram, discord, slack, whatsapp, homeassistant, signal +│ └── platforms/ # Adapters: telegram, discord, slack, whatsapp, wechat, homeassistant, signal ├── acp_adapter/ # ACP server (VS Code / Zed / JetBrains integration) ├── cron/ # Scheduler (jobs.py, scheduler.py) ├── environments/ # RL training environments (Atropos) diff --git a/README.md b/README.md index fde4cae334a8..6c632d5ce192 100644 --- a/README.md +++ b/README.md @@ -17,7 +17,7 @@ Use any model you want — [Nous Portal](https://portal.nousresearch.com), [Open - + @@ -64,7 +64,7 @@ hermes doctor # Diagnose any issues ## CLI vs Messaging Quick Reference -Hermes has two entry points: start the terminal UI with `hermes`, or run the gateway and talk to it from Telegram, Discord, Slack, WhatsApp, Signal, or Email. Once you're in a conversation, many slash commands are shared across both interfaces. +Hermes has two entry points: start the terminal UI with `hermes`, or run the gateway and talk to it from Telegram, Discord, Slack, WhatsApp, Signal, WeChat, or Email. Once you're in a conversation, many slash commands are shared across both interfaces. | Action | CLI | Messaging platforms | |---------|-----|---------------------| @@ -91,7 +91,7 @@ All documentation lives at **[hermes-agent.nousresearch.com/docs](https://hermes | [Quickstart](https://hermes-agent.nousresearch.com/docs/getting-started/quickstart) | Install → setup → first conversation in 2 minutes | | [CLI Usage](https://hermes-agent.nousresearch.com/docs/user-guide/cli) | Commands, keybindings, personalities, sessions | | [Configuration](https://hermes-agent.nousresearch.com/docs/user-guide/configuration) | Config file, providers, models, all options | -| [Messaging Gateway](https://hermes-agent.nousresearch.com/docs/user-guide/messaging) | Telegram, Discord, Slack, WhatsApp, Signal, Home Assistant | +| [Messaging Gateway](https://hermes-agent.nousresearch.com/docs/user-guide/messaging) | Telegram, Discord, Slack, WhatsApp, Signal, WeChat, Home Assistant | | [Security](https://hermes-agent.nousresearch.com/docs/user-guide/security) | Command approval, DM pairing, container isolation | | [Tools & Toolsets](https://hermes-agent.nousresearch.com/docs/user-guide/features/tools) | 40+ tools, toolset system, terminal backends | | [Skills System](https://hermes-agent.nousresearch.com/docs/user-guide/features/skills) | Procedural memory, Skills Hub, creating skills | diff --git a/website/docs/reference/environment-variables.md b/website/docs/reference/environment-variables.md index 7d40546c39a7..bf657d470883 100644 --- a/website/docs/reference/environment-variables.md +++ b/website/docs/reference/environment-variables.md @@ -223,6 +223,16 @@ For cloud sandbox backends, persistence is filesystem-oriented. `TERMINAL_LIFETI | `WECOM_WEBSOCKET_URL` | Custom WebSocket URL (default: `wss://openws.work.weixin.qq.com`) | | `WECOM_ALLOWED_USERS` | Comma-separated WeCom user IDs allowed to message the bot | | `WECOM_HOME_CHANNEL` | WeCom chat ID for cron delivery and notifications | +| `WECHAT_BOT_TOKEN` | WeChat iLink bot token (from `wechat_login.py`) | +| `WECHAT_API_BASE_URL` | WeChat iLink API base URL (default: `https://ilinkai.weixin.qq.com`) | +| `WECHAT_CDN_BASE_URL` | WeChat CDN base URL (default: `https://novac2c.cdn.weixin.qq.com/c2c`) | +| `WECHAT_ACCOUNT_ID` | WeChat account ID (wxid) for self-message filtering | +| `WECHAT_ILINK_APP_ID` | iLink app identifier (default: `bot`) | +| `WECHAT_ILINK_CLIENT_VERSION` | iLink client version string (default: adapter version) | +| `WECHAT_ALLOWED_USERS` | Comma-separated WeChat user IDs allowed to message the bot | +| `WECHAT_ALLOW_ALL_USERS` | Set to `true` to allow all users (default: `false`) | +| `WECHAT_HOME_CHANNEL` | WeChat chat ID for cron delivery and notifications | +| `WECHAT_HOME_CHANNEL_NAME` | Display name for the home channel | | `MATTERMOST_URL` | Mattermost server URL (e.g. `https://mm.example.com`) | | `MATTERMOST_TOKEN` | Bot token or personal access token for Mattermost | | `MATTERMOST_ALLOWED_USERS` | Comma-separated Mattermost user IDs allowed to message the bot | diff --git a/website/docs/user-guide/messaging/index.md b/website/docs/user-guide/messaging/index.md index fa662305bef6..b70eaf14671f 100644 --- a/website/docs/user-guide/messaging/index.md +++ b/website/docs/user-guide/messaging/index.md @@ -6,7 +6,7 @@ description: "Chat with Hermes from Telegram, Discord, Slack, WhatsApp, Signal, # Messaging Gateway -Chat with Hermes from Telegram, Discord, Slack, WhatsApp, Signal, SMS, Email, Home Assistant, Mattermost, Matrix, DingTalk, Feishu/Lark, WeCom, or your browser. The gateway is a single background process that connects to all your configured platforms, handles sessions, runs cron jobs, and delivers voice messages. +Chat with Hermes from Telegram, Discord, Slack, WhatsApp, Signal, SMS, Email, Home Assistant, Mattermost, Matrix, DingTalk, Feishu/Lark, WeCom, WeChat, or your browser. The gateway is a single background process that connects to all your configured platforms, handles sessions, runs cron jobs, and delivers voice messages. For the full voice feature set — including CLI microphone mode, spoken replies in messaging, and Discord voice-channel conversations — see [Voice Mode](/docs/user-guide/features/voice-mode) and [Use Voice Mode with Hermes](/docs/guides/use-voice-mode-with-hermes). @@ -27,6 +27,7 @@ For the full voice feature set — including CLI microphone mode, spoken replies | DingTalk | — | — | — | — | — | ✅ | ✅ | | Feishu/Lark | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | | WeCom | ✅ | ✅ | ✅ | — | — | ✅ | ✅ | +| WeChat | ✅ | ✅ | ✅ | — | — | ✅ | — | **Voice** = TTS audio replies and/or voice message transcription. **Images** = send/receive images. **Files** = send/receive file attachments. **Threads** = threaded conversations. **Reactions** = emoji reactions on messages. **Typing** = typing indicator while processing. **Streaming** = progressive message updates via editing. @@ -49,6 +50,7 @@ flowchart TB dt[DingTalk] fs[Feishu/Lark] wc[WeCom] + wx[WeChat] api["API Server
(OpenAI-compatible)"] wh[Webhooks] end @@ -352,6 +354,7 @@ Each platform has its own toolset: | DingTalk | `hermes-dingtalk` | Full tools including terminal | | Feishu/Lark | `hermes-feishu` | Full tools including terminal | | WeCom | `hermes-wecom` | Full tools including terminal | +| WeChat | `hermes-wechat` | Full tools including terminal | | API Server | `hermes` (default) | Full tools including terminal | | Webhooks | `hermes-webhook` | Full tools including terminal | @@ -370,5 +373,6 @@ Each platform has its own toolset: - [DingTalk Setup](dingtalk.md) - [Feishu/Lark Setup](feishu.md) - [WeCom Setup](wecom.md) +- [WeChat Setup](wechat.md) - [Open WebUI + API Server](open-webui.md) - [Webhooks](webhooks.md) diff --git a/website/docs/user-guide/messaging/wechat.md b/website/docs/user-guide/messaging/wechat.md new file mode 100644 index 000000000000..6a7dbda7258c --- /dev/null +++ b/website/docs/user-guide/messaging/wechat.md @@ -0,0 +1,192 @@ +--- +sidebar_position: 11 +title: "WeChat" +description: "Set up Hermes Agent as a WeChat bot via the iLink Bot API" +--- + +# WeChat Setup + +Hermes Agent can connect to WeChat (personal) through the iLink Bot API. The WeChat adapter runs as part of the normal Hermes gateway process: it long-polls for new direct messages, downloads and decrypts CDN media, and sends replies back through the same API using the required `context_token` from the inbound message. + +This integration is best for 1:1 bot conversations on WeChat where you want Hermes available from a phone-friendly messaging client with native support for images, videos, files, and typing indicators. + +:::note +This is the **personal WeChat** adapter (via iLink Bot). For **Enterprise WeChat (WeCom/Qiwei)**, see the [WeCom](./wecom.md) adapter. +::: + +## Overview + +The WeChat integration provides: + +- Plain-text chat replies with automatic markdown stripping +- Native image, video, and file delivery via CDN with AES-128-ECB encryption +- Inbound voice handling with SILK transcoding and speech-to-text fallback +- Referenced message (quote) context extraction +- Typing indicators via `getconfig` + `sendtyping` +- Context token persistence across gateway restarts +- Session resume from saved `get_updates_buf` +- iLink 2.1.x protocol compliance (App-Id, ClientVersion headers, IDC redirect) + +## Prerequisites + +Install the required Python packages: + +```bash +pip install httpx cryptography +``` + +Optional: + +```bash +pip install qrcode # Render QR code in terminal during login +pip install silk-python # Transcode SILK voice messages to WAV +``` + +## Setup + +### Option A: QR Login Script + +Run the login helper: + +```bash +python3 scripts/wechat_login.py +``` + +The script will: + +1. Request a QR code from the iLink Bot API +2. Render the QR code in your terminal (if `qrcode` is installed) +3. Handle IDC redirects if your WeChat account is on a different datacenter +4. Auto-refresh expired QR codes (up to 3 times) +5. Save credentials to `~/.hermes/wechat/accounts/` +6. Print the environment variables to add to `~/.hermes/.env` + +### Option B: Hermes Gateway Setup Wizard + +```bash +hermes gateway setup +``` + +Choose **WeChat** and paste the token and account ID returned by the QR login step. + +### Manual Configuration + +Add the required settings to `~/.hermes/.env`: + +```bash +WECHAT_BOT_TOKEN=your-bot-token +WECHAT_ACCOUNT_ID=your-ilink-bot-id + +# Optional overrides +# WECHAT_API_BASE_URL=https://ilinkai.weixin.qq.com +# WECHAT_CDN_BASE_URL=https://novac2c.cdn.weixin.qq.com/c2c + +# iLink protocol headers (defaults are usually fine) +# WECHAT_ILINK_APP_ID=bot +# WECHAT_ILINK_CLIENT_VERSION=512 + +# Security +# WECHAT_ALLOWED_USERS=user-id-1,user-id-2 +# WECHAT_ALLOW_ALL_USERS=true + +# Optional home channel for cron / send_message +# WECHAT_HOME_CHANNEL=user-id-1 +``` + +Start the gateway: + +```bash +hermes gateway +``` + +## Environment Variables + +| Variable | Required | Default | Description | +|----------|----------|---------|-------------| +| `WECHAT_BOT_TOKEN` | Yes | — | Bot token from QR login | +| `WECHAT_ACCOUNT_ID` | Yes | — | iLink bot account ID from QR login | +| `WECHAT_API_BASE_URL` | No | `https://ilinkai.weixin.qq.com` | Override the API base URL | +| `WECHAT_CDN_BASE_URL` | No | `https://novac2c.cdn.weixin.qq.com/c2c` | Override the CDN base URL | +| `WECHAT_ILINK_APP_ID` | No | `bot` | iLink application identifier (SDK 2.1.1+) | +| `WECHAT_ILINK_CLIENT_VERSION` | No | Auto-computed | iLink client version as uint32 (SDK 2.1.1+) | +| `WECHAT_ALLOWED_USERS` | No | — | Comma-separated user IDs allowed to message the bot | +| `WECHAT_ALLOW_ALL_USERS` | No | `false` | Allow all users without an allowlist | +| `WECHAT_HOME_CHANNEL` | No | — | Default user/chat ID for cron delivery and `send_message` | +| `WECHAT_HOME_CHANNEL_NAME` | No | `Home` | Display name for the home channel | + +## Supported Features + +| Feature | Status | Notes | +|---------|--------|-------| +| Text replies | Supported | Markdown auto-stripped for WeChat compatibility | +| Image send/receive | Supported | AES-128-ECB encrypted via CDN | +| File attachments | Supported | PDF, DOC, ZIP, etc. | +| Video send/receive | Supported | MP4 via CDN | +| Voice messages | Supported | SILK transcoding with fallback to file attachment | +| Referenced messages | Supported | Quoted reply context extracted as `[Quote: ...]` | +| Typing indicators | Supported | Via `getconfig` + `sendtyping` | +| Context token persistence | Supported | Survives gateway restarts | +| Session resume | Supported | Picks up from saved `get_updates_buf` | +| IDC redirect | Supported | Login handles `scaned_but_redirect` (SDK 2.1.1+) | + +## Architecture + +The WeChat adapter uses a 3-file architecture: + +| File | Purpose | +|------|---------| +| `wechat.py` | Adapter lifecycle, message routing, platform API | +| `wechat_transport.py` | HTTP layer, CDN upload/download, AES crypto, iLink headers | +| `wechat_state.py` | Context token and sync buffer persistence | + +## Known Limitations + +- **No streaming**: WeChat does not support message editing. Streaming mode sends raw `MEDIA:` tags as text. Disable streaming for WeChat or accept text-only output. +- **Audio delivery**: Outbound audio is sent as a file attachment, not a native voice bubble. Bot-originated voice playback is not reliably supported by the WeChat client. This matches the official SDK behavior. +- **Context token required**: Outbound replies require a valid `context_token`. A user must message the bot first before Hermes can reply or proactively send. +- **DM only**: The adapter targets direct-message conversations. Group chat is not supported. +- **Session expiry**: If WeChat returns `errcode -14`, the adapter pauses for one hour and clears context tokens. Users need to re-message after the pause. +- **Markdown**: WeChat does not render markdown. The adapter strips all markdown formatting before delivery. + +## Troubleshooting + +### "No token configured" + +The gateway cannot see `WECHAT_BOT_TOKEN`. + +- Run `python3 scripts/wechat_login.py` again +- Add the printed token to `~/.hermes/.env` +- Restart the gateway + +### Bot connects but cannot reply + +Outbound sends require a fresh `context_token`. + +- Send the bot a new message from WeChat first +- The token persists across restarts, so this is usually a one-time issue per user + +### Media upload fails + +Common causes: + +- `cryptography` is missing — run `pip install cryptography` +- File exceeds the 100 MB limit +- CDN returned no `x-encrypted-param` header — check gateway logs + +The adapter retries CDN uploads up to 3 times with exponential backoff. If uploads consistently fail, check your network connectivity to the WeChat CDN. + +### Voice messages not transcribed + +- Install `silk-python` or ensure `ffmpeg` is available for SILK→WAV transcoding +- If neither is available, voice is stored as raw SILK and WeChat's built-in STT text is used when present + +### IDC redirect during login + +If your WeChat account is registered in a different datacenter region, the login script handles this automatically via `scaned_but_redirect`. If login hangs, check the script output for redirect messages and ensure the redirect host is reachable. + +### Session pauses for one hour + +WeChat returned session-expiry `errcode -14`. Context tokens are cleared. + +- Wait for the 1-hour cooldown, or +- Re-run `python3 scripts/wechat_login.py` to get a fresh token diff --git a/website/sidebars.ts b/website/sidebars.ts index 5e1ebf2d6bd2..5769fa0e2804 100644 --- a/website/sidebars.ts +++ b/website/sidebars.ts @@ -107,6 +107,7 @@ const sidebars: SidebarsConfig = { 'user-guide/messaging/dingtalk', 'user-guide/messaging/feishu', 'user-guide/messaging/wecom', + 'user-guide/messaging/wechat', 'user-guide/messaging/open-webui', 'user-guide/messaging/webhooks', ],
A real terminal interfaceFull TUI with multiline editing, slash-command autocomplete, conversation history, interrupt-and-redirect, and streaming tool output.
Lives where you doTelegram, Discord, Slack, WhatsApp, Signal, and CLI — all from a single gateway process. Voice memo transcription, cross-platform conversation continuity.
Lives where you doTelegram, Discord, Slack, WhatsApp, Signal, WeChat, and CLI — all from a single gateway process. Voice memo transcription, cross-platform conversation continuity.
A closed learning loopAgent-curated memory with periodic nudges. Autonomous skill creation after complex tasks. Skills self-improve during use. FTS5 session search with LLM summarization for cross-session recall. Honcho dialectic user modeling. Compatible with the agentskills.io open standard.
Scheduled automationsBuilt-in cron scheduler with delivery to any platform. Daily reports, nightly backups, weekly audits — all in natural language, running unattended.
Delegates and parallelizesSpawn isolated subagents for parallel workstreams. Write Python scripts that call tools via RPC, collapsing multi-step pipelines into zero-context-cost turns.