From 9d8c6557ae10f8e3a399dc75c7c5ff437f5a8399 Mon Sep 17 00:00:00 2001 From: AaronZ345 <34849476+AaronZ345@users.noreply.github.com> Date: Fri, 17 Jul 2026 12:54:26 +0800 Subject: [PATCH] feat(feishu): add scheduler action cards --- channel/feishu/feishu_channel.py | 151 +++++++++++++++++-- channel/feishu/feishu_scheduler_card.py | 185 ++++++++++++++++++++++++ docs/channels/feishu.mdx | 7 +- docs/ja/channels/feishu.mdx | 7 +- docs/zh/channels/feishu.mdx | 7 +- tests/test_feishu_scheduler_card.py | 140 ++++++++++++++++++ 6 files changed, 485 insertions(+), 12 deletions(-) create mode 100644 channel/feishu/feishu_scheduler_card.py create mode 100644 tests/test_feishu_scheduler_card.py diff --git a/channel/feishu/feishu_channel.py b/channel/feishu/feishu_channel.py index 9a9f3307..8d17f2d5 100644 --- a/channel/feishu/feishu_channel.py +++ b/channel/feishu/feishu_channel.py @@ -28,6 +28,11 @@ from bridge.context import ContextType from bridge.reply import Reply, ReplyType from channel.chat_channel import ChatChannel, check_prefix from channel.feishu.feishu_message import FeishuMessage +from channel.feishu.feishu_scheduler_card import ( + build_scheduler_card, + handle_scheduler_action, + tasks_for_receivers, +) from common import utils from common.expired_dict import ExpiredDict from common.log import logger @@ -350,6 +355,10 @@ class FeiShuChanel(ChatChannel): def _startup_websocket(self): """启动长连接接收事件(websocket模式)""" _ensure_lark_imported() + from lark_oapi.event.callback.model.p2_card_action_trigger import ( + P2CardActionTriggerResponse, + ) + logger.debug("[FeiShu] Starting in websocket mode...") # 创建事件处理器 @@ -372,10 +381,25 @@ class FeiShuChanel(ChatChannel): except Exception as e: logger.error(f"[FeiShu] websocket handle message error: {e}", exc_info=True) + def handle_card_action(data): + """Handle Card 2.0 button callbacks and update the card in place.""" + try: + event_dict = json.loads(lark.JSON.marshal(data)) + response = self._handle_card_action_event(event_dict.get("event", {})) + return P2CardActionTriggerResponse(response) + except Exception as e: + logger.error(f"[FeiShu] websocket handle card action error: {e}", exc_info=True) + return P2CardActionTriggerResponse( + {"toast": {"type": "error", "content": "Task update failed"}} + ) + # 构建事件分发器 - event_handler = lark.EventDispatcherHandler.builder("", "") \ - .register_p2_im_message_receive_v1(handle_message_event) \ + event_handler = ( + lark.EventDispatcherHandler.builder("", "") + .register_p2_im_message_receive_v1(handle_message_event) + .register_p2_card_action_trigger(handle_card_action) .build() + ) def start_client_with_retry(): """Run ws client in this thread with its own event loop to avoid conflicts.""" @@ -470,6 +494,97 @@ class FeiShuChanel(ChatChannel): # so reaching here means the bot was indeed mentioned. return True + def _get_scheduler_task_store(self): + """Reuse the live scheduler store, with a path-compatible fallback.""" + from agent.tools.scheduler.integration import get_task_store + + task_store = get_task_store() + if task_store is not None: + return task_store + + from agent.tools.scheduler.task_store import TaskStore + + workspace_root = utils.expand_path(conf().get("agent_workspace", "~/cow")) + return TaskStore(os.path.join(workspace_root, "scheduler", "tasks.json")) + + def _send_scheduler_card(self, feishu_msg, is_group: bool, receive_id_type: str) -> bool: + """Reply to ``/tasks`` with tasks scoped to the current chat.""" + task_store = self._get_scheduler_task_store() + receivers = {feishu_msg.other_user_id} + tasks = tasks_for_receivers(task_store.list_tasks(), receivers) + card = build_scheduler_card(tasks) + headers = { + "Authorization": "Bearer " + feishu_msg.access_token, + "Content-Type": "application/json", + } + content = json.dumps(card, ensure_ascii=False) + + if is_group and feishu_msg.msg_id: + url = ( + "https://open.feishu.cn/open-apis/im/v1/messages/" + f"{feishu_msg.msg_id}/reply" + ) + response = requests.post( + url, + headers=headers, + json={"msg_type": "interactive", "content": content}, + timeout=(5, 10), + ) + else: + url = "https://open.feishu.cn/open-apis/im/v1/messages" + response = requests.post( + url, + headers=headers, + params={"receive_id_type": receive_id_type}, + json={ + "receive_id": feishu_msg.other_user_id, + "msg_type": "interactive", + "content": content, + }, + timeout=(5, 10), + ) + + result = response.json() + if result.get("code") == 0: + logger.info("[FeiShu] scheduler card sent") + return True + logger.error( + "[FeiShu] scheduler card failed, " + f"code={result.get('code')}, msg={result.get('msg')}" + ) + return False + + def _handle_card_action_event(self, event: dict) -> dict: + """Apply a scheduler card action within its chat/operator ownership scope.""" + action = event.get("action") or {} + value = action.get("value") or {} + if value.get("cowagent") != "scheduler": + return {} + + context = event.get("context") or {} + operator = event.get("operator") or {} + callback_receivers = { + receiver + for receiver in (context.get("open_chat_id"), operator.get("open_id")) + if receiver + } + target_receiver = value.get("receiver") + allowed_receivers = ( + {target_receiver} + if target_receiver and target_receiver in callback_receivers + else set() + ) + response = handle_scheduler_action( + value, + self._get_scheduler_task_store(), + allowed_receivers, + ) + logger.info( + "[FeiShu] scheduler card action handled, " + f"action={value.get('action')}, task_id={value.get('task_id')}" + ) + return response + def _handle_message_event(self, event: dict): """ 处理消息事件的核心逻辑 @@ -521,6 +636,11 @@ class FeiShuChanel(ChatChannel): if not feishu_msg: return + if feishu_msg.ctype == ContextType.TEXT and feishu_msg.content.strip().lower() == "/tasks": + if not self._send_scheduler_card(feishu_msg, is_group, receive_id_type): + logger.warning("[FeiShu] /tasks card delivery failed") + return + # 处理文件缓存逻辑 from channel.file_cache import get_file_cache file_cache = get_file_cache() @@ -1565,6 +1685,7 @@ class FeishuController: FAILED_MSG = '{"success": false}' SUCCESS_MSG = '{"success": true}' MESSAGE_RECEIVE_TYPE = "im.message.receive_v1" + CARD_ACTION_TYPE = "card.action.trigger" def GET(self): return "Feishu service start success!" @@ -1581,15 +1702,27 @@ class FeishuController: varify_res = {"challenge": request.get("challenge")} return json.dumps(varify_res) - # 2.消息接收处理 - # token 校验 - header = request.get("header") - if not header or header.get("token") != channel.feishu_token: + # 2. Verify callbacks. Card callbacks may carry the verification + # token in event.token while message events carry it in the header. + header = request.get("header") or {} + event = request.get("event") or {} + event_type = header.get("event_type") or request.get("type") + callback_token = ( + header.get("token") + or event.get("token") + or request.get("token") + ) + if callback_token != channel.feishu_token: return self.FAILED_MSG - # 处理消息事件 - event = request.get("event") - if header.get("event_type") == self.MESSAGE_RECEIVE_TYPE and event: + if event_type == self.CARD_ACTION_TYPE and event: + return json.dumps( + channel._handle_card_action_event(event), + ensure_ascii=False, + ) + + # 3. Handle message events. + if event_type == self.MESSAGE_RECEIVE_TYPE and event: channel._handle_message_event(event) return self.SUCCESS_MSG diff --git a/channel/feishu/feishu_scheduler_card.py b/channel/feishu/feishu_scheduler_card.py new file mode 100644 index 00000000..9a2485f7 --- /dev/null +++ b/channel/feishu/feishu_scheduler_card.py @@ -0,0 +1,185 @@ +"""Feishu scheduler card rendering and callback handling.""" + +from __future__ import annotations + +from typing import Any, Dict, Iterable, List, Set + + +_MAX_TASKS = 20 + + +def tasks_for_receivers(tasks: Iterable[dict], receivers: Set[str]) -> List[dict]: + """Return Feishu tasks owned by one of the callback's trusted receivers.""" + visible = [] + for task in tasks: + action = task.get("action") or {} + if action.get("channel_type") != "feishu": + continue + if action.get("receiver") in receivers: + visible.append(task) + return visible + + +def build_scheduler_card(tasks: Iterable[dict]) -> Dict[str, Any]: + """Build a Card 2.0 task list with explicit, idempotent actions.""" + task_list = list(tasks) + elements: List[Dict[str, Any]] = [] + + if not task_list: + elements.append({"tag": "markdown", "content": "No scheduled tasks in this chat."}) + else: + for index, task in enumerate(task_list[:_MAX_TASKS]): + if index: + elements.append({"tag": "hr"}) + task_id = str(task.get("id") or "") + receiver = str((task.get("action") or {}).get("receiver") or "") + enabled = task.get("enabled", True) + status = "Enabled" if enabled else "Disabled" + next_run = str(task.get("next_run_at") or "Unknown").replace("T", " ") + elements.append( + { + "tag": "markdown", + "content": "**{}** · {}\n`{}` · {}\nNext: {}".format( + task.get("name") or "Unnamed task", + status, + task_id, + _format_schedule(task.get("schedule") or {}), + next_run, + ), + } + ) + toggle_action = "disable" if enabled else "enable" + toggle_text = "Disable" if enabled else "Enable" + toggle_type = "default" if enabled else "primary" + elements.append( + { + "tag": "column_set", + "columns": [ + { + "tag": "column", + "elements": [ + _button( + toggle_text, + toggle_type, + toggle_action, + task_id, + receiver, + ) + ], + }, + { + "tag": "column", + "elements": [_button("Delete", "danger", "delete", task_id, receiver)], + }, + ], + } + ) + + hidden = len(task_list) - _MAX_TASKS + if hidden > 0: + elements.extend( + [ + {"tag": "hr"}, + { + "tag": "markdown", + "content": "{} more tasks are hidden.".format(hidden), + "text_size": "notation", + }, + ] + ) + + return { + "schema": "2.0", + "config": {"update_multi": True, "enable_forward_interaction": False}, + "header": { + "template": "blue", + "title": {"tag": "plain_text", "content": "Scheduled tasks"}, + }, + "body": {"elements": elements}, + } + + +def handle_scheduler_action( + value: Dict[str, Any], task_store: Any, allowed_receivers: Set[str] +) -> Dict[str, Any]: + """Apply an owned scheduler action and return a Feishu callback response.""" + if value.get("cowagent") != "scheduler": + return {} + + task_id = str(value.get("task_id") or "") + action = value.get("action") + if not task_id or action not in {"enable", "disable", "delete"}: + return _toast("error", "Invalid scheduler action") + + task = task_store.get_task(task_id) + if not task: + return _response( + "info", + "Task no longer exists", + build_scheduler_card(tasks_for_receivers(task_store.list_tasks(), allowed_receivers)), + ) + + task_receiver = (task.get("action") or {}).get("receiver") + task_channel = (task.get("action") or {}).get("channel_type") + value_receiver = value.get("receiver") + if ( + task_channel != "feishu" + or task_receiver not in allowed_receivers + or (value_receiver and value_receiver != task_receiver) + ): + return _toast("error", "Task is not available in this chat") + + try: + if action == "delete": + task_store.delete_task(task_id) + message = "Task deleted" + else: + enabled = action == "enable" + task_store.enable_task(task_id, enabled) + message = "Task enabled" if enabled else "Task disabled" + except (OSError, ValueError) as exc: + return _toast("error", "Task update failed: {}".format(exc)) + + visible = tasks_for_receivers(task_store.list_tasks(), allowed_receivers) + return _response("success", message, build_scheduler_card(visible)) + + +def _response(toast_type: str, content: str, card: Dict[str, Any]) -> Dict[str, Any]: + response = _toast(toast_type, content) + response["card"] = {"type": "raw", "data": card} + return response + + +def _toast(toast_type: str, content: str) -> Dict[str, Any]: + return {"toast": {"type": toast_type, "content": content}} + + +def _button( + text: str, + button_type: str, + action: str, + task_id: str, + receiver: str, +) -> Dict[str, Any]: + return { + "tag": "button", + "text": {"tag": "plain_text", "content": text}, + "type": button_type, + "value": { + "cowagent": "scheduler", + "action": action, + "task_id": task_id, + "receiver": receiver, + }, + } + + +def _format_schedule(schedule: Dict[str, Any]) -> str: + schedule_type = schedule.get("type") + if schedule_type == "cron": + return "cron {}".format(schedule.get("expression") or "?") + if schedule_type == "interval": + return "every {}s".format(schedule.get("seconds") or "?") + if schedule_type == "once": + return "once at {}".format(schedule.get("run_at") or "?") + return str(schedule_type or "unknown schedule") diff --git a/docs/channels/feishu.mdx b/docs/channels/feishu.mdx index 1283d0c1..78e12cc3 100644 --- a/docs/channels/feishu.mdx +++ b/docs/channels/feishu.mdx @@ -84,7 +84,9 @@ im:message,im:message.group_at_msg,im:message.group_at_msg:readonly,im:message.p 2. Click **Add Event**, search for "Receive Message" and choose **Receive Message v2.0**. -3. Click **Version Management & Release**, create a version and apply for **Production Release**. Approve the request in the Feishu client: +3. Under **Callbacks**, add **Card Action Trigger** (`card.action.trigger`) to use the `/tasks` controls. + +4. Click **Version Management & Release**, create a version and apply for **Production Release**. Approve the request in the Feishu client: @@ -98,6 +100,7 @@ im:message,im:message.group_at_msg,im:message.group_at_msg:readonly,im:message.p | Image messages | ✅ send/receive | | Voice messages | ✅ send/receive | | Streaming reply | ✅ (powered by Feishu cardkit streaming card) | +| Scheduler controls | ✅ `/tasks` list with enable, disable and delete buttons | Streaming reply requires the `cardkit:card:write` permission (already enabled by one-click creation) and Feishu client version ≥ 7.20. Older clients see an upgrade prompt; if the permission or version is not satisfied, replies fall back to plain text automatically. @@ -108,3 +111,5 @@ im:message,im:message.group_at_msg,im:message.group_at_msg:readonly,im:message.p After connection, search for the bot name in Feishu to start a chat. To use in groups, add the bot to a group and @-mention it. + +Send `/tasks` in a private chat, or @-mention the bot with `/tasks` in a group, to manage tasks belonging to that chat. diff --git a/docs/ja/channels/feishu.mdx b/docs/ja/channels/feishu.mdx index d703954a..c7107643 100644 --- a/docs/ja/channels/feishu.mdx +++ b/docs/ja/channels/feishu.mdx @@ -81,7 +81,9 @@ im:message,im:message.group_at_msg,im:message.group_at_msg:readonly,im:message.p 2. **イベントを追加** で「メッセージ受信」を検索し、**メッセージ受信 v2.0** を選択。 -3. **バージョン管理とリリース** で新バージョンを作成し **本番リリース** を申請、Feishu クライアントで承認: +3. **コールバック** で **カードアクショントリガー**(`card.action.trigger`)を追加すると `/tasks` の操作ボタンを使用できます。 + +4. **バージョン管理とリリース** で新バージョンを作成し **本番リリース** を申請、Feishu クライアントで承認: @@ -95,6 +97,7 @@ im:message,im:message.group_at_msg,im:message.group_at_msg:readonly,im:message.p | 画像メッセージ | ✅ 送受信 | | 音声メッセージ | ✅ 送受信 | | ストリーミング応答 | ✅(Feishu cardkit ストリーミングカードベース) | +| スケジューラー操作 | ✅ `/tasks` で一覧、有効化、無効化、削除 | ストリーミング応答には `cardkit:card:write` 権限(ワンクリック作成では自動付与)と Feishu クライアント 7.20 以上が必要です。古いクライアントではアップグレード案内が表示され、権限/バージョン未充足時は通常テキスト応答に自動フォールバックします。 @@ -105,3 +108,5 @@ im:message,im:message.group_at_msg,im:message.group_at_msg:readonly,im:message.p 接続完了後、Feishu で Bot 名を検索してチャットを開始できます。 グループで使う場合は Bot をグループに追加し、@メンションでメッセージを送ってください。 + +個人チャットでは `/tasks`、グループでは Bot に @メンションして `/tasks` を送ると、そのチャットのタスクを管理できます。 diff --git a/docs/zh/channels/feishu.mdx b/docs/zh/channels/feishu.mdx index 5cb8fe80..ada0058c 100644 --- a/docs/zh/channels/feishu.mdx +++ b/docs/zh/channels/feishu.mdx @@ -85,7 +85,9 @@ im:message,im:message.group_at_msg,im:message.group_at_msg:readonly,im:message.p 2. 点击 **添加事件**,搜索 "接收消息",选择 **接收消息 v2.0** 并确认。 -3. 点击 **版本管理与发布**,创建版本并申请 **线上发布**,在飞书客户端审核通过: +3. 在 **回调** 中添加 **卡片回传交互**(`card.action.trigger`),以启用 `/tasks` 操作按钮。 + +4. 点击 **版本管理与发布**,创建版本并申请 **线上发布**,在飞书客户端审核通过: @@ -99,6 +101,7 @@ im:message,im:message.group_at_msg,im:message.group_at_msg:readonly,im:message.p | 图片消息 | ✅ 收发 | | 语音消息 | ✅ 收发 | | 流式回复 | ✅(通过 `feishu_stream_reply` 配置控制,默认开启) | +| 定时任务操作卡 | ✅ `/tasks` 查看、启用、停用与删除 | 流式回复需要机器人具备 `cardkit:card:write` 权限(一键创建已默认开通),且接收方飞书客户端版本 ≥ 7.20。低版本客户端会显示升级提示,权限或版本不满足时自动降级为普通文本回复。 @@ -109,3 +112,5 @@ im:message,im:message.group_at_msg,im:message.group_at_msg:readonly,im:message.p 完成接入后,在飞书中搜索机器人名称即可开始单聊对话。 如需在群聊中使用,将机器人添加到群中,@机器人发送消息即可。 + +私聊发送 `/tasks`,或在群聊中 @机器人并发送 `/tasks`,即可管理属于当前会话的定时任务。 diff --git a/tests/test_feishu_scheduler_card.py b/tests/test_feishu_scheduler_card.py new file mode 100644 index 00000000..13913887 --- /dev/null +++ b/tests/test_feishu_scheduler_card.py @@ -0,0 +1,140 @@ +from copy import deepcopy + +from channel.feishu.feishu_scheduler_card import ( + build_scheduler_card, + handle_scheduler_action, + tasks_for_receivers, +) + + +class FakeTaskStore: + def __init__(self, tasks): + self.tasks = {task["id"]: deepcopy(task) for task in tasks} + self.enable_calls = [] + self.delete_calls = [] + + def list_tasks(self): + return list(self.tasks.values()) + + def get_task(self, task_id): + return self.tasks.get(task_id) + + def enable_task(self, task_id, enabled=True): + self.enable_calls.append((task_id, enabled)) + self.tasks[task_id]["enabled"] = enabled + + def delete_task(self, task_id): + self.delete_calls.append(task_id) + del self.tasks[task_id] + + +def _task(task_id, receiver, enabled=True): + return { + "id": task_id, + "name": "Daily report", + "enabled": enabled, + "schedule": {"type": "cron", "expression": "0 9 * * *"}, + "next_run_at": "2026-07-18T09:00:00", + "action": {"receiver": receiver, "channel_type": "feishu"}, + } + + +def test_scheduler_card_has_explicit_idempotent_actions(): + card = build_scheduler_card([_task("task-1", "chat-1")]) + + assert card["schema"] == "2.0" + assert card["config"]["enable_forward_interaction"] is False + assert card["header"]["title"]["content"] == "Scheduled tasks" + action_row = next( + element for element in card["body"]["elements"] if element.get("tag") == "column_set" + ) + values = [column["elements"][0]["value"] for column in action_row["columns"]] + assert values == [ + { + "cowagent": "scheduler", + "action": "disable", + "task_id": "task-1", + "receiver": "chat-1", + }, + { + "cowagent": "scheduler", + "action": "delete", + "task_id": "task-1", + "receiver": "chat-1", + }, + ] + + +def test_tasks_are_scoped_to_callback_chat_or_operator(): + tasks = [ + _task("group", "chat-1"), + _task("private", "user-1"), + _task("other", "chat-2"), + ] + + visible = tasks_for_receivers(tasks, {"chat-1", "user-1"}) + + assert [task["id"] for task in visible] == ["group", "private"] + + +def test_disable_action_is_idempotent_and_returns_refreshed_card(): + store = FakeTaskStore([_task("task-1", "chat-1")]) + value = {"cowagent": "scheduler", "action": "disable", "task_id": "task-1"} + + first = handle_scheduler_action(value, store, {"chat-1"}) + second = handle_scheduler_action(value, store, {"chat-1"}) + + assert store.enable_calls == [("task-1", False), ("task-1", False)] + assert first["toast"] == {"type": "success", "content": "Task disabled"} + assert second["card"]["type"] == "raw" + refreshed_columns = next( + element + for element in second["card"]["data"]["body"]["elements"] + if element.get("tag") == "column_set" + )["columns"] + assert refreshed_columns[0]["elements"][0]["value"]["action"] == "enable" + + +def test_action_rejects_task_from_another_receiver(): + store = FakeTaskStore([_task("task-1", "chat-2")]) + + response = handle_scheduler_action( + {"cowagent": "scheduler", "action": "delete", "task_id": "task-1"}, + store, + {"chat-1", "user-1"}, + ) + + assert response["toast"] == {"type": "error", "content": "Task is not available in this chat"} + assert store.delete_calls == [] + + +def test_action_rejects_mismatched_receiver_embedded_in_card_value(): + store = FakeTaskStore([_task("task-1", "user-1")]) + + response = handle_scheduler_action( + { + "cowagent": "scheduler", + "action": "delete", + "task_id": "task-1", + "receiver": "chat-1", + }, + store, + {"chat-1"}, + ) + + assert response["toast"] == {"type": "error", "content": "Task is not available in this chat"} + assert store.delete_calls == [] + + +def test_delete_action_removes_owned_task(): + store = FakeTaskStore([_task("task-1", "user-1")]) + + response = handle_scheduler_action( + {"cowagent": "scheduler", "action": "delete", "task_id": "task-1"}, + store, + {"user-1"}, + ) + + assert store.delete_calls == ["task-1"] + assert response["toast"] == {"type": "success", "content": "Task deleted"} + assert "No scheduled tasks" in response["card"]["data"]["body"]["elements"][0]["content"]