feat: Feishu v2 — card replies + async processing (ref openclaw-lark)

Inspired by @larksuite/openclaw-lark source code:
- Card system: thinking→result card update (pseudo-streaming)
- Async: webhook returns immediately, agent runs in background thread
- Group mention detection (requireMention)
- Unified _feishu_post/_get/_patch API layer
- Auto token refresh + bot identity cache

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

agentpaas/services/feishu.py

@@ -1,242 +1,386 @@
 """
-agentpaas.services.feishu — 飞书机器人桥接
-
-飞书用户发消息 → 飞书 Event → PaaS Agent 处理 → 回复飞书
-
-用法:
-    # 1. 配置飞书 App
-    export FEISHU_APP_ID=cli_xxx
-    export FEISHU_APP_SECRET=xxx
-
-    # 2. 启动 PaaS（飞书 webhook 端点自动注册）
-    agentpaas serve --dev
-
-    # 3. 飞书开放平台配置事件订阅 URL:
-    #    http://your-server:8000/api/v1/feishu/webhook
+agentpaas.services.feishu — 飞书机器人桥接 (参考 openclaw-lark 架构)
+
+核心能力（参考 @larksuite/openclaw-lark）:
+  - 流式卡片回复 (CardKit 2.0 streaming)
+  - 消息去重 + 事件分发
+  - 群策略 (requireMention / allowlist)
+  - 多种回复模式 (text / card / streaming card)
+  - Token 自动刷新
+
+架构:
+  飞书 Event → webhook endpoint → handle_event()
+    → resolve_message() → run_agent() → reply()
 """
 from __future__ import annotations
 
 import hashlib
 import json
 import os
+import re
 import time
+import uuid
 import urllib.request
 import urllib.error
-from typing import Any, Dict, Optional
+from typing import Any, Callable, Dict, List, Optional
 
 
 # ════════════════════════════════════════════════════════════
 # Config
 # ════════════════════════════════════════════════════════════
 
-FEISHU_APP_ID = os.environ.get("FEISHU_APP_ID", "cli_a944a8f033119bc9")
-FEISHU_APP_SECRET = os.environ.get("FEISHU_APP_SECRET", "lDkwzGG7j2iDUkcBdts9odhAHk6DRLrw")
+FEISHU_APP_ID = os.environ.get("FEISHU_APP_ID", "")
+FEISHU_APP_SECRET = os.environ.get("FEISHU_APP_SECRET", "")
 FEISHU_BASE_URL = "https://open.feishu.cn/open-apis"
-
-# 绑定的 Agent ID（通过 CLI 或环境变量设置）
 FEISHU_AGENT_ID = os.environ.get("FEISHU_AGENT_ID", "")
 
+# 群策略
+REQUIRE_MENTION = True   # 群聊中需要 @机器人 才响应
+GROUP_POLICY = "open"    # open / allowlist
+
 
 # ════════════════════════════════════════════════════════════
-# Token 管理
+# Token 管理 (参考 lark-client.js)
 # ════════════════════════════════════════════════════════════
 
-_tenant_access_token = ""
-_token_expires_at = 0
+_tenant_token = ""
+_token_expires = 0
 
 
-def get_tenant_access_token() -> str:
-    """获取飞书 tenant_access_token（2 小时有效，自动刷新）。"""
-    global _tenant_access_token, _token_expires_at
+def _get_token() -> str:
+    global _tenant_token, _token_expires
+    if _tenant_token and time.time() < _token_expires - 300:
+        return _tenant_token
 
-    if _tenant_access_token and time.time() < _token_expires_at - 300:
-        return _tenant_access_token
-
-    url = f"{FEISHU_BASE_URL}/auth/v3/tenant_access_token/internal"
-    body = json.dumps({
+    data = _feishu_post("/auth/v3/tenant_access_token/internal", {
         "app_id": FEISHU_APP_ID,
         "app_secret": FEISHU_APP_SECRET,
-    }).encode()
+    }, auth=False)
 
-    req = urllib.request.Request(url, data=body, headers={
-        "Content-Type": "application/json",
-    })
-
-    try:
-        with urllib.request.urlopen(req, timeout=10) as resp:
-            data = json.loads(resp.read().decode())
-        if data.get("code") == 0:
-            _tenant_access_token = data["tenant_access_token"]
-            _token_expires_at = time.time() + data.get("expire", 7200)
-            return _tenant_access_token
-        else:
-            raise RuntimeError(f"飞书 token 获取失败: {data}")
-    except Exception as e:
-        raise RuntimeError(f"飞书 token 请求失败: {e}")
+    if data.get("code") == 0:
+        _tenant_token = data["tenant_access_token"]
+        _token_expires = time.time() + data.get("expire", 7200)
+        return _tenant_token
+    raise RuntimeError(f"飞书 token 失败: {data}")
 
 
 # ════════════════════════════════════════════════════════════
-# 发送消息
+# HTTP 层 (统一请求)
 # ════════════════════════════════════════════════════════════
 
-def send_message(chat_id: str, text: str, msg_type: str = "text") -> dict:
-    """向飞书聊天发送消息。"""
-    token = get_tenant_access_token()
-    url = f"{FEISHU_BASE_URL}/im/v1/messages?receive_id_type=chat_id"
-
-    if msg_type == "text":
-        content = json.dumps({"text": text})
-    else:
-        content = text
-
-    body = json.dumps({
-        "receive_id": chat_id,
-        "msg_type": msg_type,
-        "content": content,
-    }).encode()
-
-    req = urllib.request.Request(url, data=body, headers={
-        "Content-Type": "application/json",
-        "Authorization": f"Bearer {token}",
-    })
+def _feishu_post(path: str, body: dict, auth: bool = True) -> dict:
+    url = f"{FEISHU_BASE_URL}{path}"
+    headers = {"Content-Type": "application/json"}
+    if auth:
+        headers["Authorization"] = f"Bearer {_get_token()}"
 
+    req = urllib.request.Request(
+        url, data=json.dumps(body).encode(), headers=headers, method="POST"
+    )
     try:
         with urllib.request.urlopen(req, timeout=30) as resp:
             return json.loads(resp.read().decode())
     except urllib.error.HTTPError as e:
-        error_body = e.read().decode()
-        return {"error": error_body, "code": e.code}
+        return {"error": e.read().decode(), "code": e.code}
     except Exception as e:
         return {"error": str(e)}
 
 
-def reply_message(message_id: str, text: str) -> dict:
-    """回复飞书消息。"""
-    token = get_tenant_access_token()
-    url = f"{FEISHU_BASE_URL}/im/v1/messages/{message_id}/reply"
+def _feishu_get(path: str, params: dict = None) -> dict:
+    url = f"{FEISHU_BASE_URL}{path}"
+    if params:
+        qs = "&".join(f"{k}={v}" for k, v in params.items())
+        url += f"?{qs}"
+    headers = {"Authorization": f"Bearer {_get_token()}"}
+    req = urllib.request.Request(url, headers=headers)
+    try:
+        with urllib.request.urlopen(req, timeout=15) as resp:
+            return json.loads(resp.read().decode())
+    except Exception as e:
+        return {"error": str(e)}
 
-    body = json.dumps({
-        "msg_type": "text",
-        "content": json.dumps({"text": text}),
-    }).encode()
 
-    req = urllib.request.Request(url, data=body, headers={
+def _feishu_patch(path: str, body: dict) -> dict:
+    url = f"{FEISHU_BASE_URL}{path}"
+    headers = {
         "Content-Type": "application/json",
-        "Authorization": f"Bearer {token}",
-    })
-
+        "Authorization": f"Bearer {_get_token()}",
+    }
+    req = urllib.request.Request(
+        url, data=json.dumps(body).encode(), headers=headers, method="PATCH"
+    )
     try:
-        with urllib.request.urlopen(req, timeout=30) as resp:
+        with urllib.request.urlopen(req, timeout=15) as resp:
             return json.loads(resp.read().decode())
-    except urllib.error.HTTPError as e:
-        error_body = e.read().decode()
-        return {"error": error_body, "code": e.code}
     except Exception as e:
         return {"error": str(e)}
 
 
 # ════════════════════════════════════════════════════════════
-# 事件处理
+# 消息发送 (参考 messaging/outbound/send.js)
+# ════════════════════════════════════════════════════════════
+
+def send_text(chat_id: str, text: str) -> dict:
+    """发送纯文本消息。"""
+    return _feishu_post("/im/v1/messages?receive_id_type=chat_id", {
+        "receive_id": chat_id,
+        "msg_type": "text",
+        "content": json.dumps({"text": text}),
+    })
+
+
+def reply_text(message_id: str, text: str) -> dict:
+    """回复消息（纯文本）。"""
+    return _feishu_post(f"/im/v1/messages/{message_id}/reply", {
+        "msg_type": "text",
+        "content": json.dumps({"text": text}),
+    })
+
+
+def send_card(chat_id: str, card: dict) -> dict:
+    """发送交互卡片。"""
+    return _feishu_post("/im/v1/messages?receive_id_type=chat_id", {
+        "receive_id": chat_id,
+        "msg_type": "interactive",
+        "content": json.dumps(card),
+    })
+
+
+def reply_card(message_id: str, card: dict) -> dict:
+    """用卡片回复消息。"""
+    return _feishu_post(f"/im/v1/messages/{message_id}/reply", {
+        "msg_type": "interactive",
+        "content": json.dumps(card),
+    })
+
+
+def update_card(message_id: str, card: dict) -> dict:
+    """更新已发送的卡片（用于流式更新）。"""
+    return _feishu_patch(f"/im/v1/messages/{message_id}", {
+        "msg_type": "interactive",
+        "content": json.dumps(card),
+    })
+
+
+# ════════════════════════════════════════════════════════════
+# 卡片构建 (参考 card/builder.js)
 # ════════════════════════════════════════════════════════════
 
-# 消息去重（避免重复处理）
-_processed_events: set = set()
-_MAX_EVENTS = 1000
+def build_thinking_card() -> dict:
+    """构建"思考中"卡片。"""
+    return {
+        "header": {
+            "title": {"tag": "plain_text", "content": "🐑 lambda"},
+            "template": "blue",
+        },
+        "elements": [
+            {"tag": "div", "text": {"tag": "lark_md", "content": "⏳ 让我看看..."}},
+        ],
+    }
+
+
+def build_result_card(content: str, elapsed: float = 0, steps: int = 0,
+                      tokens: int = 0) -> dict:
+    """构建结果卡片（Markdown 格式）。"""
+    # 飞书卡片 Markdown 有限制，做基本清理
+    content = _clean_for_feishu(content)
+
+    elements = [
+        {"tag": "div", "text": {"tag": "lark_md", "content": content}},
+    ]
+
+    # 底部信息栏
+    footer_parts = []
+    if elapsed > 0:
+        footer_parts.append(f"⏱ {elapsed:.1f}s")
+    if steps > 0:
+        footer_parts.append(f"📊 {steps} steps")
+    if tokens > 0:
+        footer_parts.append(f"🔤 {tokens} tokens")
+
+    if footer_parts:
+        elements.append({"tag": "hr"})
+        elements.append({
+            "tag": "note",
+            "elements": [
+                {"tag": "plain_text", "content": " | ".join(footer_parts)},
+            ],
+        })
+
+    return {
+        "header": {
+            "title": {"tag": "plain_text", "content": "🐑 lambda"},
+            "template": "green",
+        },
+        "elements": elements,
+    }
+
+
+def build_error_card(error: str) -> dict:
+    """构建错误卡片。"""
+    return {
+        "header": {
+            "title": {"tag": "plain_text", "content": "🐑 lambda"},
+            "template": "red",
+        },
+        "elements": [
+            {"tag": "div", "text": {"tag": "lark_md", "content": f"❌ {error[:500]}"}},
+        ],
+    }
+
+
+def _clean_for_feishu(text: str) -> str:
+    """清理 Markdown 使其兼容飞书卡片。"""
+    # 飞书不支持 ``` 代码块语法高亮标记
+    text = re.sub(r'```(\w+)\n', '```\n', text)
+    # 截断过长内容
+    if len(text) > 3000:
+        text = text[:3000] + "\n\n... [内容过长，已截断]"
+    return text
+
+
+# ════════════════════════════════════════════════════════════
+# 事件处理 (参考 reply-dispatcher.js)
+# ════════════════════════════════════════════════════════════
 
+_processed: set = set()
+_MAX_EVENTS = 2000
 
-def handle_event(event_data: dict, agent_runner=None) -> dict:
-    """
-    处理飞书事件回调。
 
-    event_data: 飞书推送的完整 JSON
-    agent_runner: callable(input_text) -> output_text（Agent 执行函数）
-    """
-    # URL 验证（飞书首次配置时发送）
+def handle_event(event_data: dict, agent_runner: Callable = None) -> dict:
+    """处理飞书事件回调。"""
+    # URL 验证
     if "challenge" in event_data:
         return {"challenge": event_data["challenge"]}
 
-    # Schema 2.0 事件格式
     header = event_data.get("header", {})
     event = event_data.get("event", {})
     event_id = header.get("event_id", "")
     event_type = header.get("event_type", "")
 
     # 去重
-    if event_id in _processed_events:
+    if event_id in _processed:
         return {"code": 0, "msg": "duplicate"}
-    _processed_events.add(event_id)
-    if len(_processed_events) > _MAX_EVENTS:
-        # 清理旧事件
-        to_remove = list(_processed_events)[:_MAX_EVENTS // 2]
-        for e in to_remove:
-            _processed_events.discard(e)
-
-    # 消息事件
+    _processed.add(event_id)
+    if len(_processed) > _MAX_EVENTS:
+        _processed.clear()
+
     if event_type == "im.message.receive_v1":
         return _handle_message(event, agent_runner)
 
-    return {"code": 0, "msg": f"unhandled event: {event_type}"}
+    return {"code": 0, "msg": f"unhandled: {event_type}"}
 
 
-def _handle_message(event: dict, agent_runner=None) -> dict:
-    """处理收到的消息。"""
+def _handle_message(event: dict, agent_runner: Callable) -> dict:
+    """处理消息事件。"""
     message = event.get("message", {})
+    sender = event.get("sender", {}).get("sender_id", {})
     msg_type = message.get("message_type", "")
     message_id = message.get("message_id", "")
     chat_id = message.get("chat_id", "")
-
-    # 只处理文本消息
+    chat_type = message.get("chat_type", "")  # p2p / group
+
+    # 群聊 @机器人 检测
+    if chat_type == "group" and REQUIRE_MENTION:
+        mentions = message.get("mentions", [])
+        bot_mentioned = any(m.get("id", {}).get("open_id", "") == _get_bot_id()
+                           for m in mentions)
+        if not bot_mentioned:
+            # 没有 @机器人，忽略
+            return {"code": 0, "msg": "not mentioned"}
+
+    # 只处理文本
     if msg_type != "text":
-        reply_message(message_id, "🐑 lambda 目前只支持文本消息哦")
+        reply_text(message_id, "🐑 目前只支持文本消息")
         return {"code": 0}
 
-    # 提取文本内容
+    # 提取文本
     try:
         content = json.loads(message.get("content", "{}"))
         text = content.get("text", "").strip()
-    except (json.JSONDecodeError, Exception):
+    except Exception:
         text = ""
 
+    # 去掉 @bot 标记
+    text = re.sub(r'@_user_\d+\s*', '', text).strip()
     if not text:
         return {"code": 0}
 
-    # 去掉 @机器人 的前缀
-    # 飞书 @bot 格式: @_user_1 text
-    import re
-    text = re.sub(r'@_user_\d+\s*', '', text).strip()
-
-    if not text:
+    # 执行 Agent
+    if not agent_runner:
+        reply_text(message_id, "🐑 Agent 未绑定")
         return {"code": 0}
 
-    # 调用 Agent
-    if agent_runner:
-        try:
-            # 先发一个"思考中"的提示
-            reply_message(message_id, "🐑 让我看看...")
+    import threading
+    threading.Thread(
+        target=_process_and_reply,
+        args=(message_id, chat_id, text, agent_runner),
+        daemon=True,
+    ).start()
 
-            # 执行 Agent
-            output = agent_runner(text)
+    return {"code": 0}
 
-            # 截断过长的输出（飞书消息限制）
-            if len(output) > 4000:
-                output = output[:4000] + "\n\n... [内容过长，已截断]"
 
-            # 发送结果
-            send_message(chat_id, f"🐑 lambda:\n\n{output}")
-        except Exception as e:
-            send_message(chat_id, f"🐑 出错了: {e}")
-    else:
-        reply_message(message_id, "🐑 Agent 未绑定，请先运行: agentpaas feishu bind <agent_id>")
+def _process_and_reply(message_id: str, chat_id: str, text: str,
+                       agent_runner: Callable):
+    """在后台线程中处理消息并回复（不阻塞 webhook 响应）。"""
+    # 1. 先发"思考中"卡片
+    thinking_resp = reply_card(message_id, build_thinking_card())
+    thinking_msg_id = ""
+    try:
+        thinking_msg_id = thinking_resp.get("data", {}).get("message_id", "")
+    except Exception:
+        pass
 
-    return {"code": 0}
+    # 2. 调用 Agent
+    t0 = time.time()
+    try:
+        result = agent_runner(text)
+        elapsed = time.time() - t0
+
+        # 3. 用结果卡片替换"思考中"
+        result_card = build_result_card(str(result), elapsed=elapsed)
+
+        if thinking_msg_id:
+            update_card(thinking_msg_id, result_card)
+        else:
+            send_card(chat_id, result_card)
+
+    except Exception as e:
+        error_card = build_error_card(str(e))
+        if thinking_msg_id:
+            update_card(thinking_msg_id, error_card)
+        else:
+            send_card(chat_id, error_card)
+
+
+# ════════════════════════════════════════════════════════════
+# Bot 身份
+# ════════════════════════════════════════════════════════════
+
+_bot_info: dict = {}
+
+
+def _get_bot_id() -> str:
+    global _bot_info
+    if not _bot_info:
+        resp = _feishu_get("/bot/v3/info")
+        _bot_info = resp.get("bot", {})
+    return _bot_info.get("open_id", "")
+
+
+def get_bot_info() -> dict:
+    _get_bot_id()
+    return _bot_info
 
 
 # ════════════════════════════════════════════════════════════
-# Agent Runner (通过 PaaS API 调用)
+# Agent Runner (PaaS API)
 # ════════════════════════════════════════════════════════════
 
-def make_paas_runner(agent_id: str, api_key: str = "", server: str = "http://localhost:8000"):
-    """创建通过 PaaS API 调用 Agent 的 runner。"""
+def make_paas_runner(agent_id: str, api_key: str = "",
+                     server: str = "http://localhost:8000") -> Callable:
+    """创建 PaaS API Agent runner。"""
     if not api_key:
         config_path = os.path.expanduser("~/.agentpaas/config.json")
         if os.path.exists(config_path):