diff --git a/.env.example b/.env.example index 35e22d4..c307413 100644 --- a/.env.example +++ b/.env.example @@ -10,9 +10,21 @@ MODEL_ID=claude-sonnet-4-20250514 # TELEGRAM_BOT_TOKEN=123456:ABC-DEF # Feishu / Lark (optional, for s04_channels.py) +# Create a self-built app at https://open.feishu.cn (or https://open.larksuite.com +# for international Lark), grab App ID + App Secret, and enable "Long Connection" +# (长连接) mode under Event Subscription. Install: pip install lark-oapi # FEISHU_APP_ID=cli_xxxxxxxx # FEISHU_APP_SECRET=xxxxxxxx -# FEISHU_DOMAIN=feishu # "feishu" for domestic, "lark" for international +# Inbound mode: ws (long connection, default, recommended) | webhook +# In ws mode the SDK pushes events into the agent loop automatically; in webhook +# mode you expose parse_event behind your own HTTP endpoint. +# FEISHU_MODE=ws +# International Lark? Set to true to use open.larksuite.com instead of open.feishu.cn +# FEISHU_IS_LARK=false +# Optional, webhook mode only: encrypt key for signature verification +# FEISHU_ENCRYPT_KEY= +# Optional: bot open_id, used to filter group messages to only those @-mentioning the bot +# FEISHU_BOT_OPEN_ID=ou_xxxxxxxx # Heartbeat (optional, for s07_heartbeat_cron.py) # HEARTBEAT_INTERVAL=1800 diff --git a/requirements.txt b/requirements.txt index 3b5bf30..ad00a40 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,3 +4,4 @@ websockets>=12.0 croniter>=2.0.0 python-telegram-bot>=21.0 httpx>=0.27.0 +lark-oapi>=1.7.0 diff --git a/sessions/en/s04_channels.md b/sessions/en/s04_channels.md index d01d3a7..3a11322 100644 --- a/sessions/en/s04_channels.md +++ b/sessions/en/s04_channels.md @@ -27,7 +27,7 @@ - **InboundMessage**: a dataclass that normalizes all platform payloads into one format. - **Channel ABC**: `receive()` + `send()` is the entire contract. - **TelegramChannel**: long-polling, offset persistence, media group buffering, text coalescing. -- **FeishuChannel**: webhook-based, token auth, mention detection, multi-type message parsing. +- **FeishuChannel**: two inbound modes -- long connection (WebSocket, default; the lark-oapi SDK dials out to Feishu, so no public URL needed) and webhook (you expose parse_event behind your own HTTP endpoint). Outbound always goes through `im/v1/messages` + tenant token; mention detection, multi-type message parsing. - **ChannelManager**: registry that holds all active channels. ## Key Code Walkthrough @@ -146,6 +146,11 @@ python en/s04_channels.py # With Feishu -- add to .env: # FEISHU_APP_ID=cli_xxxxx # FEISHU_APP_SECRET=xxxxx +# FEISHU_MODE=ws # ws=long connection (default, recommended) | webhook +# FEISHU_IS_LARK=false # true=international Lark, else domestic Feishu +# FEISHU_BOT_OPEN_ID=ou_xx # optional, group chats only respond when @-mentioning the bot +# FEISHU_ENCRYPT_KEY= # optional, webhook-mode signature verification only +# Long-connection mode requires: pip install lark-oapi # REPL commands # You > /channels (list registered channels) diff --git a/sessions/en/s04_channels.py b/sessions/en/s04_channels.py index e14fa7e..42af6b0 100644 --- a/sessions/en/s04_channels.py +++ b/sessions/en/s04_channels.py @@ -18,6 +18,8 @@ ANTHROPIC_API_KEY=sk-ant-xxxxx MODEL_ID=claude-sonnet-4-20250514 # Optional: TELEGRAM_BOT_TOKEN, FEISHU_APP_ID, FEISHU_APP_SECRET + # Feishu inbound mode: FEISHU_MODE=ws (long connection, recommended) + # or webhook (you expose parse_event behind your own HTTP endpoint). """ import json, os, sys, time, threading @@ -35,6 +37,17 @@ except ImportError: HAS_HTTPX = False +# Feishu long-connection (WebSocket) client -- optional. The lark-oapi SDK +# maintains an outbound WebSocket to Feishu, so no public callback URL is +# needed. Install with: pip install lark-oapi +try: + import lark_oapi as lark + from lark_oapi.api.im.v1 import P2ImMessageReceiveV1 # noqa: F401 (type hint) + HAS_LARK = True +except ImportError: + lark = None # type: ignore[assignment] + HAS_LARK = False + # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- @@ -376,8 +389,8 @@ def __init__(self, account: ChannelAccount) -> None: self.app_secret = account.config.get("app_secret", "") self._encrypt_key = account.config.get("encrypt_key", "") self._bot_open_id = account.config.get("bot_open_id", "") - is_lark = account.config.get("is_lark", False) - self.api_base = ("https://open.larksuite.com/open-apis" if is_lark + self._is_lark = account.config.get("is_lark", False) + self.api_base = ("https://open.larksuite.com/open-apis" if self._is_lark else "https://open.feishu.cn/open-apis") self._tenant_token: str = "" self._token_expires_at: float = 0.0 @@ -478,6 +491,122 @@ def parse_event(self, payload: dict, token: str = "") -> InboundMessage | None: media=media, is_group=is_group, raw=payload, ) + def _ws_bot_mentioned(self, message: Any) -> bool: + """Whether the bot is @-mentioned in a long-connection event message. + + Webhook payloads give mentions as dicts; long-connection events give + typed MentionEvent objects whose .id is a UserId. We check both shapes. + """ + for m in (message.mentions or []): + mid = getattr(m, "id", None) + if mid is not None and getattr(mid, "open_id", None) == self._bot_open_id: + return True + if getattr(m, "key", None) == self._bot_open_id: + return True + return False + + def parse_ws_event(self, data: Any) -> InboundMessage | None: + """Parse a long-connection (WebSocket) P2ImMessageReceiveV1 event. + + This is the inbound twin of parse_event: same InboundMessage shape, + different source -- the lark-oapi SDK delivers typed event objects + instead of raw webhook JSON. We reuse _parse_content for the message + body (note: ws events name the field message_type, not msg_type). + """ + try: + event = getattr(data, "event", None) + if event is None: + return None + message = event.message + sender = event.sender + if message is None or sender is None: + return None + + user_id = "" + sid = sender.sender_id + if sid is not None: + user_id = sid.open_id or sid.user_id or sid.union_id or "" + chat_id = message.chat_id or "" + chat_type = message.chat_type or "" + is_group = chat_type == "group" + + if is_group and self._bot_open_id and not self._ws_bot_mentioned(message): + return None + + # Reuse the webhook content parser; ws events use message_type. + text, media = self._parse_content( + {"msg_type": message.message_type, "content": message.content or "{}"} + ) + if not text: + return None + + raw: dict = {} + if lark is not None: + try: + raw = lark.JSON.marshal(data) + except Exception: + raw = {} + return InboundMessage( + text=text, sender_id=user_id, channel="feishu", + account_id=self.account_id, + peer_id=user_id if chat_type == "p2p" else chat_id, + media=media, is_group=is_group, raw=raw, + ) + except Exception as exc: + print(f" {RED}[feishu] ws parse error: {exc}{RESET}") + return None + + def start_long_connection( + self, msg_queue: list, q_lock: threading.Lock, + ) -> threading.Thread | None: + """Start the WebSocket long-connection client in a daemon thread. + + The SDK maintains an outbound WebSocket to Feishu and auto-reconnects; + no public callback URL, encrypt key, or challenge handshake is needed. + Inbound events are parsed into InboundMessage and pushed into the + shared msg_queue -- the same pipeline Telegram and CLI feed into, so + the agent loop needs no changes. Requires lark-oapi. + """ + if not HAS_LARK: + print(f" {RED}[feishu] Long-connection mode needs lark-oapi: " + f"pip install lark-oapi{RESET}") + return None + if not (self.app_id and self.app_secret): + print(f" {RED}[feishu] Long-connection needs FEISHU_APP_ID + " + f"FEISHU_APP_SECRET{RESET}") + return None + + def _on_msg(data: Any) -> None: + inbound = self.parse_ws_event(data) + if inbound is not None: + with q_lock: + msg_queue.append(inbound) + print_channel(f" [feishu/ws] {inbound.sender_id}: " + f"{inbound.text[:80]}") + + dispatcher = (lark.EventDispatcherHandler.builder("", "") + .register_p2_im_message_receive_v1(_on_msg).build()) + domain = ("https://open.larksuite.com" if self._is_lark + else "https://open.feishu.cn") + ws_client = lark.ws.Client( + self.app_id, self.app_secret, + event_handler=dispatcher, + log_level=lark.LogLevel.INFO, + domain=domain, + ) + + def _run() -> None: + print_channel(f" [feishu/ws] Long connection started for " + f"{self.account_id}") + try: + ws_client.start() # blocking; auto-reconnect handled by SDK + except Exception as exc: + print(f" {RED}[feishu/ws] connection error: {exc}{RESET}") + + t = threading.Thread(target=_run, daemon=True, name="feishu-ws") + t.start() + return t + def receive(self) -> InboundMessage | None: return None @@ -711,6 +840,7 @@ def agent_loop() -> None: fs_id = os.getenv("FEISHU_APP_ID", "").strip() fs_secret = os.getenv("FEISHU_APP_SECRET", "").strip() + fs_channel: FeishuChannel | None = None if fs_id and fs_secret and HAS_HTTPX: fs_acc = ChannelAccount( channel="feishu", account_id="feishu-primary", @@ -722,7 +852,15 @@ def agent_loop() -> None: }, ) mgr.accounts.append(fs_acc) - mgr.register(FeishuChannel(fs_acc)) + fs_channel = FeishuChannel(fs_acc) + mgr.register(fs_channel) + + # Feishu inbound mode: long connection (default) vs webhook. In ws mode the + # SDK pushes events into msg_queue through a background thread; in webhook + # mode you must expose parse_event behind your own HTTP endpoint yourself. + fs_mode = os.getenv("FEISHU_MODE", "ws").strip().lower() + if fs_channel is not None and fs_mode == "ws": + fs_channel.start_long_connection(msg_queue, q_lock) print_info("=" * 60) print_info(" claw0 | Section 04: Channels") @@ -735,12 +873,12 @@ def agent_loop() -> None: conversations: dict[str, list[dict]] = {} while True: - # Drain Telegram queue + # Drain the inbound queue (Telegram polling + Feishu long-connection). with q_lock: tg_msgs = msg_queue[:] msg_queue.clear() for m in tg_msgs: - print_channel(f"\n [telegram] {m.sender_id}: {m.text[:80]}") + print_channel(f"\n [{m.channel}] {m.sender_id}: {m.text[:80]}") run_agent_turn(m, conversations, mgr) # CLI input (non-blocking when Telegram is active) diff --git a/sessions/ja/s04_channels.md b/sessions/ja/s04_channels.md index 7822de4..82cac3e 100644 --- a/sessions/ja/s04_channels.md +++ b/sessions/ja/s04_channels.md @@ -27,7 +27,7 @@ - **InboundMessage**: 全プラットフォームのペイロードを1つのフォーマットに正規化するデータクラス。 - **Channel ABC**: `receive()` + `send()` がコントラクトの全て。 - **TelegramChannel**: ロングポーリング、オフセット永続化、メディアグループバッファリング、テキスト結合。 -- **FeishuChannel**: Webhookベース、トークン認証、メンション検出、複数タイプのメッセージ解析。 +- **FeishuChannel**: 2つの受信モード -- 長接続 (WebSocket, デフォルト; lark-oapi SDK が飛書へ発信接続するため公開URL不要) と webhook (parse_event を自身のHTTPエンドポイントに公開). 送信は常に `im/v1/messages` + テナントトークン; メンション検出、複数タイプのメッセージ解析。 - **ChannelManager**: アクティブな全チャネルを保持するレジストリ。 ## コードウォークスルー @@ -143,6 +143,11 @@ python ja/s04_channels.py # Feishu を使う場合 -- .env に追加: # FEISHU_APP_ID=cli_xxxxx # FEISHU_APP_SECRET=xxxxx +# FEISHU_MODE=ws # ws=長接続 (デフォルト, 推奨) | webhook +# FEISHU_IS_LARK=false # true=国際版 Lark, それ以外は国内飛書 +# FEISHU_BOT_OPEN_ID=ou_xx # 任意, グループチャットはボットへのメンション時のみ応答 +# FEISHU_ENCRYPT_KEY= # 任意, webhook モードの署名検証のみ +# 長接続モードには要インストール: pip install lark-oapi # REPL コマンド # You > /channels (登録済みチャネルの一覧) diff --git a/sessions/ja/s04_channels.py b/sessions/ja/s04_channels.py index 9e41469..711e52c 100644 --- a/sessions/ja/s04_channels.py +++ b/sessions/ja/s04_channels.py @@ -15,6 +15,8 @@ ANTHROPIC_API_KEY=sk-ant-xxxxx MODEL_ID=claude-sonnet-4-20250514 # 任意: TELEGRAM_BOT_TOKEN, FEISHU_APP_ID, FEISHU_APP_SECRET, FEISHU_ENCRYPT_KEY + # 飛書の受信モード: FEISHU_MODE=ws (長接続, 推奨) + # または webhook (parse_event を自身のHTTPエンドポイントに公開). """ import json, os, sys, time, threading @@ -32,6 +34,16 @@ except ImportError: HAS_HTTPX = False +# 飛書の長接続 (WebSocket) クライアント -- 任意. lark-oapi SDK が飛書への +# 送信側 WebSocket を維持するため、公開コールバックURLは不要. インストール: pip install lark-oapi +try: + import lark_oapi as lark + from lark_oapi.api.im.v1 import P2ImMessageReceiveV1 # noqa: F401 (型ヒント) + HAS_LARK = True +except ImportError: + lark = None # type: ignore[assignment] + HAS_LARK = False + # --------------------------------------------------------------------------- # 設定 # --------------------------------------------------------------------------- @@ -365,8 +377,8 @@ def __init__(self, account: ChannelAccount) -> None: self.app_secret = account.config.get("app_secret", "") self._encrypt_key = account.config.get("encrypt_key", "") self._bot_open_id = account.config.get("bot_open_id", "") - is_lark = account.config.get("is_lark", False) - self.api_base = ("https://open.larksuite.com/open-apis" if is_lark + self._is_lark = account.config.get("is_lark", False) + self.api_base = ("https://open.larksuite.com/open-apis" if self._is_lark else "https://open.feishu.cn/open-apis") self._tenant_token: str = "" self._token_expires_at: float = 0.0 @@ -467,6 +479,119 @@ def parse_event(self, payload: dict, token: str = "") -> InboundMessage | None: media=media, is_group=is_group, raw=payload, ) + def _ws_bot_mentioned(self, message: Any) -> bool: + """長接続イベントメッセージでボットがメンションされているか. + + webhook ペイロードの mention は dict だが、長接続イベントは型付きの + MentionEvent オブジェクトを渡し、その .id は UserId. 両方の形を検査する. + """ + for m in (message.mentions or []): + mid = getattr(m, "id", None) + if mid is not None and getattr(mid, "open_id", None) == self._bot_open_id: + return True + if getattr(m, "key", None) == self._bot_open_id: + return True + return False + + def parse_ws_event(self, data: Any) -> InboundMessage | None: + """長接続 (WebSocket) P2ImMessageReceiveV1 イベントを解析する. + + これは parse_event の受信側の双生: 同じ InboundMessage 形式だがソースが異なる -- + lark-oapi SDK は生の webhook JSON ではなく型付きイベントオブジェクトを渡す. + メッセージ本文は _parse_content を再利用 (注意: ws イベントのフィールドは msg_type でなく message_type). + """ + try: + event = getattr(data, "event", None) + if event is None: + return None + message = event.message + sender = event.sender + if message is None or sender is None: + return None + + user_id = "" + sid = sender.sender_id + if sid is not None: + user_id = sid.open_id or sid.user_id or sid.union_id or "" + chat_id = message.chat_id or "" + chat_type = message.chat_type or "" + is_group = chat_type == "group" + + if is_group and self._bot_open_id and not self._ws_bot_mentioned(message): + return None + + # webhook の内容パーサを再利用; ws イベントは message_type を使用. + text, media = self._parse_content( + {"msg_type": message.message_type, "content": message.content or "{}"} + ) + if not text: + return None + + raw: dict = {} + if lark is not None: + try: + raw = lark.JSON.marshal(data) + except Exception: + raw = {} + return InboundMessage( + text=text, sender_id=user_id, channel="feishu", + account_id=self.account_id, + peer_id=user_id if chat_type == "p2p" else chat_id, + media=media, is_group=is_group, raw=raw, + ) + except Exception as exc: + print(f" {RED}[feishu] ws parse error: {exc}{RESET}") + return None + + def start_long_connection( + self, msg_queue: list, q_lock: threading.Lock, + ) -> threading.Thread | None: + """デーモンスレッドで WebSocket 長接続クライアントを起動する. + + SDK は飛書への送信側 WebSocket を維持し自動再接続する; 公開コールバックURL、 + 暗号化キー、challenge ハンドシェイクは不要. 受信イベントは InboundMessage に + 解析され共有の msg_queue に投入される -- Telegram や CLI と同じパイプラインなので、 + エージェントループの変更は不要. lark-oapi が必要. + """ + if not HAS_LARK: + print(f" {RED}[feishu] 長接続モードには lark-oapi が必要です: " + f"pip install lark-oapi{RESET}") + return None + if not (self.app_id and self.app_secret): + print(f" {RED}[feishu] 長接続には FEISHU_APP_ID + " + f"FEISHU_APP_SECRET が必要です{RESET}") + return None + + def _on_msg(data: Any) -> None: + inbound = self.parse_ws_event(data) + if inbound is not None: + with q_lock: + msg_queue.append(inbound) + print_channel(f" [feishu/ws] {inbound.sender_id}: " + f"{inbound.text[:80]}") + + dispatcher = (lark.EventDispatcherHandler.builder("", "") + .register_p2_im_message_receive_v1(_on_msg).build()) + domain = ("https://open.larksuite.com" if self._is_lark + else "https://open.feishu.cn") + ws_client = lark.ws.Client( + self.app_id, self.app_secret, + event_handler=dispatcher, + log_level=lark.LogLevel.INFO, + domain=domain, + ) + + def _run() -> None: + print_channel(f" [feishu/ws] {self.account_id} の長接続を開始しました") + try: + ws_client.start() # ブロッキング; 自動再接続はSDKが処理 + except Exception as exc: + print(f" {RED}[feishu/ws] 接続エラー: {exc}{RESET}") + + t = threading.Thread(target=_run, daemon=True, name="feishu-ws") + t.start() + return t + def receive(self) -> InboundMessage | None: return None @@ -700,6 +825,7 @@ def agent_loop() -> None: fs_id = os.getenv("FEISHU_APP_ID", "").strip() fs_secret = os.getenv("FEISHU_APP_SECRET", "").strip() + fs_channel: FeishuChannel | None = None if fs_id and fs_secret and HAS_HTTPX: fs_acc = ChannelAccount( channel="feishu", account_id="feishu-primary", @@ -711,7 +837,15 @@ def agent_loop() -> None: }, ) mgr.accounts.append(fs_acc) - mgr.register(FeishuChannel(fs_acc)) + fs_channel = FeishuChannel(fs_acc) + mgr.register(fs_channel) + + # 飛書の受信モード: 長接続 (デフォルト) vs webhook. ws モードでは SDK が + # バックグラウンドスレッド経由でイベントを msg_queue に投入する; webhook モードでは + # parse_event を自身のHTTPエンドポイントに公開する必要がある. + fs_mode = os.getenv("FEISHU_MODE", "ws").strip().lower() + if fs_channel is not None and fs_mode == "ws": + fs_channel.start_long_connection(msg_queue, q_lock) print_info("=" * 60) print_info(" claw0 | Section 04: Channels") @@ -724,12 +858,12 @@ def agent_loop() -> None: conversations: dict[str, list[dict]] = {} while True: - # Telegramキューを排出 + # 受信キューを排出 (Telegram ポーリング + 飛書の長接続). with q_lock: tg_msgs = msg_queue[:] msg_queue.clear() for m in tg_msgs: - print_channel(f"\n [telegram] {m.sender_id}: {m.text[:80]}") + print_channel(f"\n [{m.channel}] {m.sender_id}: {m.text[:80]}") run_agent_turn(m, conversations, mgr) # CLI入力 (Telegramがアクティブな場合はノンブロッキング) diff --git a/sessions/zh/s04_channels.md b/sessions/zh/s04_channels.md index 25cbdd3..927d406 100644 --- a/sessions/zh/s04_channels.md +++ b/sessions/zh/s04_channels.md @@ -27,7 +27,7 @@ - **InboundMessage**: 一个 dataclass, 将所有平台的消息负载统一为同一格式. - **Channel ABC**: `receive()` + `send()` 就是全部接口契约. - **TelegramChannel**: 长轮询, offset 持久化, 媒体组缓冲, 文本合并. -- **FeishuChannel**: 基于 webhook, token 认证, @提及检测, 多类型消息解析. +- **FeishuChannel**: 支持两种入站模式 -- 长连接 (WebSocket, 默认, 经 lark-oapi SDK 主动连出, 无需公网) 与 webhook (由你暴露 HTTP 端点); 出站统一走 `im/v1/messages` + tenant token; @提及检测, 多类型消息解析. - **ChannelManager**: 持有所有活跃通道的注册中心. ## 核心代码走读 @@ -146,6 +146,11 @@ python zh/s04_channels.py # 启用飞书 -- 在 .env 中添加: # FEISHU_APP_ID=cli_xxxxx # FEISHU_APP_SECRET=xxxxx +# FEISHU_MODE=ws # ws=长连接(默认, 推荐) | webhook +# FEISHU_IS_LARK=false # true=国际版 Lark, 否则国内飞书 +# FEISHU_BOT_OPEN_ID=ou_xx # 可选, 群聊只响应 @机器人 的消息 +# FEISHU_ENCRYPT_KEY= # 可选, 仅 webhook 模式签名校验 +# 长连接需安装: pip install lark-oapi # REPL 命令 # You > /channels (列出已注册的通道) diff --git a/sessions/zh/s04_channels.py b/sessions/zh/s04_channels.py index e595921..1ffb47c 100644 --- a/sessions/zh/s04_channels.py +++ b/sessions/zh/s04_channels.py @@ -14,6 +14,8 @@ ANTHROPIC_API_KEY=sk-ant-xxxxx MODEL_ID=claude-sonnet-4-20250514 # 可选: TELEGRAM_BOT_TOKEN, FEISHU_APP_ID, FEISHU_APP_SECRET, FEISHU_ENCRYPT_KEY + # 飞书入站模式: FEISHU_MODE=ws (长连接, 推荐) + # 或 webhook (自行将 parse_event 暴露在 HTTP 端点后). """ import json, os, sys, time, threading @@ -31,6 +33,16 @@ except ImportError: HAS_HTTPX = False +# 飞书长连接 (WebSocket) 客户端 -- 可选. lark-oapi SDK 维护一条到飞书的出站 WebSocket, +# 因此无需公网回调地址. 安装: pip install lark-oapi +try: + import lark_oapi as lark + from lark_oapi.api.im.v1 import P2ImMessageReceiveV1 # noqa: F401 (类型标注) + HAS_LARK = True +except ImportError: + lark = None # type: ignore[assignment] + HAS_LARK = False + # --------------------------------------------------------------------------- # 配置 # --------------------------------------------------------------------------- @@ -364,8 +376,8 @@ def __init__(self, account: ChannelAccount) -> None: self.app_secret = account.config.get("app_secret", "") self._encrypt_key = account.config.get("encrypt_key", "") self._bot_open_id = account.config.get("bot_open_id", "") - is_lark = account.config.get("is_lark", False) - self.api_base = ("https://open.larksuite.com/open-apis" if is_lark + self._is_lark = account.config.get("is_lark", False) + self.api_base = ("https://open.larksuite.com/open-apis" if self._is_lark else "https://open.feishu.cn/open-apis") self._tenant_token: str = "" self._token_expires_at: float = 0.0 @@ -466,6 +478,119 @@ def parse_event(self, payload: dict, token: str = "") -> InboundMessage | None: media=media, is_group=is_group, raw=payload, ) + def _ws_bot_mentioned(self, message: Any) -> bool: + """长连接事件消息中机器人是否被 @. + + webhook 回调里的 mention 是 dict; 长连接事件给出带类型的 MentionEvent + 对象, 其 .id 是 UserId. 这里两种形态都检查. + """ + for m in (message.mentions or []): + mid = getattr(m, "id", None) + if mid is not None and getattr(mid, "open_id", None) == self._bot_open_id: + return True + if getattr(m, "key", None) == self._bot_open_id: + return True + return False + + def parse_ws_event(self, data: Any) -> InboundMessage | None: + """解析长连接 (WebSocket) P2ImMessageReceiveV1 事件. + + 这是 parse_event 的入站双胞胎: 同样的 InboundMessage 结构, 来源不同 -- + lark-oapi SDK 投递带类型的事件对象, 而非原始 webhook JSON. + 消息体复用 _parse_content (注意: ws 事件的字段叫 message_type, 非 msg_type). + """ + try: + event = getattr(data, "event", None) + if event is None: + return None + message = event.message + sender = event.sender + if message is None or sender is None: + return None + + user_id = "" + sid = sender.sender_id + if sid is not None: + user_id = sid.open_id or sid.user_id or sid.union_id or "" + chat_id = message.chat_id or "" + chat_type = message.chat_type or "" + is_group = chat_type == "group" + + if is_group and self._bot_open_id and not self._ws_bot_mentioned(message): + return None + + # 复用 webhook 的内容解析器; ws 事件用 message_type. + text, media = self._parse_content( + {"msg_type": message.message_type, "content": message.content or "{}"} + ) + if not text: + return None + + raw: dict = {} + if lark is not None: + try: + raw = lark.JSON.marshal(data) + except Exception: + raw = {} + return InboundMessage( + text=text, sender_id=user_id, channel="feishu", + account_id=self.account_id, + peer_id=user_id if chat_type == "p2p" else chat_id, + media=media, is_group=is_group, raw=raw, + ) + except Exception as exc: + print(f" {RED}[feishu] ws parse error: {exc}{RESET}") + return None + + def start_long_connection( + self, msg_queue: list, q_lock: threading.Lock, + ) -> threading.Thread | None: + """在守护线程中启动 WebSocket 长连接客户端. + + SDK 维护一条到飞书的出站 WebSocket 并自动重连; 无需公网回调地址、 + 加密策略或 challenge 握手. 入站事件被解析成 InboundMessage 推入共享的 + msg_queue -- 与 Telegram、CLI 喂进的是同一条管道, 故 agent 循环无需改动. + 需要 lark-oapi. + """ + if not HAS_LARK: + print(f" {RED}[feishu] 长连接模式需要 lark-oapi: " + f"pip install lark-oapi{RESET}") + return None + if not (self.app_id and self.app_secret): + print(f" {RED}[feishu] 长连接需要 FEISHU_APP_ID + " + f"FEISHU_APP_SECRET{RESET}") + return None + + def _on_msg(data: Any) -> None: + inbound = self.parse_ws_event(data) + if inbound is not None: + with q_lock: + msg_queue.append(inbound) + print_channel(f" [feishu/ws] {inbound.sender_id}: " + f"{inbound.text[:80]}") + + dispatcher = (lark.EventDispatcherHandler.builder("", "") + .register_p2_im_message_receive_v1(_on_msg).build()) + domain = ("https://open.larksuite.com" if self._is_lark + else "https://open.feishu.cn") + ws_client = lark.ws.Client( + self.app_id, self.app_secret, + event_handler=dispatcher, + log_level=lark.LogLevel.INFO, + domain=domain, + ) + + def _run() -> None: + print_channel(f" [feishu/ws] 已为 {self.account_id} 启动长连接") + try: + ws_client.start() # 阻塞; 自动重连由 SDK 处理 + except Exception as exc: + print(f" {RED}[feishu/ws] 连接错误: {exc}{RESET}") + + t = threading.Thread(target=_run, daemon=True, name="feishu-ws") + t.start() + return t + def receive(self) -> InboundMessage | None: return None @@ -699,6 +824,7 @@ def agent_loop() -> None: fs_id = os.getenv("FEISHU_APP_ID", "").strip() fs_secret = os.getenv("FEISHU_APP_SECRET", "").strip() + fs_channel: FeishuChannel | None = None if fs_id and fs_secret and HAS_HTTPX: fs_acc = ChannelAccount( channel="feishu", account_id="feishu-primary", @@ -710,7 +836,14 @@ def agent_loop() -> None: }, ) mgr.accounts.append(fs_acc) - mgr.register(FeishuChannel(fs_acc)) + fs_channel = FeishuChannel(fs_acc) + mgr.register(fs_channel) + + # 飞书入站模式: 长连接 (默认) vs webhook. ws 模式下 SDK 经后台线程把事件推入 + # msg_queue; webhook 模式需自行将 parse_event 暴露在 HTTP 端点后. + fs_mode = os.getenv("FEISHU_MODE", "ws").strip().lower() + if fs_channel is not None and fs_mode == "ws": + fs_channel.start_long_connection(msg_queue, q_lock) print_info("=" * 60) print_info(" claw0 | Section 04: Channels") @@ -723,12 +856,12 @@ def agent_loop() -> None: conversations: dict[str, list[dict]] = {} while True: - # 排空 Telegram 队列 + # 排空入站队列 (Telegram 轮询 + 飞书长连接). with q_lock: tg_msgs = msg_queue[:] msg_queue.clear() for m in tg_msgs: - print_channel(f"\n [telegram] {m.sender_id}: {m.text[:80]}") + print_channel(f"\n [{m.channel}] {m.sender_id}: {m.text[:80]}") run_agent_turn(m, conversations, mgr) # CLI 输入 (当 Telegram 活跃时使用非阻塞模式)