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"]