github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-04-30 22:08:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(channels): add DingTalk channel integration (#2628)

Browse Source 
* feat(channels): add DingTalk channel integration

Add a new DingTalk messaging channel using the dingtalk-stream SDK
with Stream Push (WebSocket), requiring no public IP. Supports both
plain sampleMarkdown replies and optional AI Card streaming for a
typewriter effect when card_template_id is configured.

- Add DingTalkChannel implementation with token management, message
  routing, allowed_users filtering, and markdown adaptation
- Register dingtalk in channel service registry and capability map
- Propagate inbound metadata to outbound messages in ChannelManager
  for DingTalk sender context (sender_staff_id, conversation_type)
- Add dingtalk-stream dependency to pyproject.toml
- Add configuration examples in config.example.yaml and .env.example
- Update all README translations with setup instructions
- Add comprehensive test suite (test_dingtalk_channel.py) and
  metadata propagation test in test_channels.py
- Update backend CLAUDE.md to document DingTalk channel

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix(channels): address PR review feedback for DingTalk integration

- Replace runtime mutation of CHANNEL_CAPABILITIES with a
  `supports_streaming` property on the Channel base class, overridden
  by DingTalkChannel, FeishuChannel, and WeComChannel
- Store stream client reference and attempt graceful disconnect in
  stop(); guard _on_chatbot_message with _running check to prevent
  post-stop message processing
- Use msg.chat_id as the primary routing key in send/send_file via
  a shared _resolve_routing helper, with metadata as fallback
- Fix process() return type annotation from tuple[str, str] to
  tuple[int, str] to match AckMessage.STATUS_OK
- Protect _incoming_messages with threading.Lock for cross-thread
  safety between the Stream Push thread and the asyncio loop
- Re-add Docker Compose URL guidance removed during DingTalk setup
  docs addition in README.md
- Fix incomplete sentence in README_zh.md (missing verb "启用")

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix(docs): restore plain paragraph format for Docker Compose note

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix(channels): fix isinstance TypeError and add file size guard in DingTalk channel

Use tuple syntax for isinstance() type check to avoid runtime TypeError
with PEP 604 union types. Add upload size limit (20MB) before reading
files into memory. Narrow exception handlers to specific types.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix(channels): propagate markdown fallback errors and validate access token response

- Re-raise exceptions in _send_markdown_fallback to prevent partial
  deliveries (files sent without accompanying text)
- Validate _get_access_token response: reject non-dict bodies, empty
  tokens, and coerce invalid expireIn to a safe default
- Add tests for both fixes

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix(channels): validate upload response and broaden send_file exception handling

- Validate _upload_media JSON response: handle JSONDecodeError and
  non-dict payloads gracefully by returning None
- Broaden send_file exception tuple to include TypeError and
  AttributeError for unexpected JSON shapes

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix(channels): fix streaming race on channel registration and slim outbound metadata

- Register channel in service before calling start() to avoid race
  where background receiver publishes inbound before registration,
  causing manager to fall back to static CHANNEL_CAPABILITIES
- Strip known-large metadata keys (raw_message, ref_msg) from outbound
  messages to prevent memory bloat from propagated inbound payloads

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Update service.py

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update CLAUDE.md

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
He Wang 2026-04-30 11:25:33 +08:00 committed by GitHub
parent 0691c4dda3
commit 08afdcb907
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
18 changed files with 2544 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.env.example
View File

@ -40,6 +40,8 @@ INFOQUEST_API_KEY=your-infoquest-api-key
#
# WECOM_BOT_ID=your-wecom-bot-id
# WECOM_BOT_SECRET=your-wecom-bot-secret
# DINGTALK_CLIENT_ID=your-dingtalk-client-id
# DINGTALK_CLIENT_SECRET=your-dingtalk-client-secret
# Set to "false" to disable Swagger UI, ReDoc, and OpenAPI schema in production
# GATEWAY_ENABLE_DOCS=false

20
README.md
View File

@ -345,6 +345,7 @@ DeerFlow supports receiving tasks from messaging apps. Channels auto-start when
| Feishu / Lark | WebSocket | Moderate |
| WeChat | Tencent iLink (long-polling) | Moderate |
| WeCom | WebSocket | Moderate |
| DingTalk | Stream Push (WebSocket) | Moderate |
**Configuration in `config.yaml`:**
@ -414,6 +415,13 @@ channels:
context:
thinking_enabled: true
subagent_enabled: true
dingtalk:
enabled: true
client_id: $DINGTALK_CLIENT_ID # Client ID of your DingTalk application
client_secret: $DINGTALK_CLIENT_SECRET # Client Secret of your DingTalk application
allowed_users: [] # empty = allow all
card_template_id: "" # Optional: AI Card template ID for streaming typewriter effect
```
Notes:
@ -442,6 +450,10 @@ WECHAT_ILINK_BOT_ID=your_ilink_bot_id
# WeCom
WECOM_BOT_ID=your_bot_id
WECOM_BOT_SECRET=your_bot_secret
# DingTalk
DINGTALK_CLIENT_ID=your_client_id
DINGTALK_CLIENT_SECRET=your_client_secret
```
**Telegram Setup**
@ -480,6 +492,14 @@ WECOM_BOT_SECRET=your_bot_secret
4. Make sure backend dependencies include `wecom-aibot-python-sdk`. The channel uses a WebSocket long connection and does not require a public callback URL.
5. The current integration supports inbound text, image, and file messages. Final images/files generated by the agent are also sent back to the WeCom conversation.
**DingTalk Setup**
1. Create a DingTalk application in the [DingTalk Developer Console](https://open.dingtalk.com/) and enable **Robot** capability.
2. Set the message receiving mode to **Stream Mode** in the robot configuration page.
3. Copy the `Client ID` and `Client Secret`, set `DINGTALK_CLIENT_ID` and `DINGTALK_CLIENT_SECRET` in `.env`, and enable the channel in `config.yaml`.
4. *(Optional)* To enable streaming AI Card replies (typewriter effect), create an **AI Card** template on the [DingTalk Card Platform](https://open.dingtalk.com/document/dingstart/typewriter-effect-streaming-ai-card), then set `card_template_id` in `config.yaml` to the template ID. You also need to apply for the `Card.Streaming.Write` and `Card.Instance.Write` permissions.
When DeerFlow runs in Docker Compose, IM channels execute inside the `gateway` container. In that case, do not point `channels.langgraph_url` or `channels.gateway_url` at `localhost`; use container service names such as `http://gateway:8001/api` and `http://gateway:8001`, or set `DEER_FLOW_CHANNELS_LANGGRAPH_URL` and `DEER_FLOW_CHANNELS_GATEWAY_URL`.
**Commands**

19
README_fr.md
View File

@ -290,6 +290,7 @@ DeerFlow peut recevoir des tâches depuis des applications de messagerie. Les ca
| Telegram | Bot API (long-polling) | Facile |
| Slack | Socket Mode | Modérée |
| Feishu / Lark | WebSocket | Modérée |
| DingTalk | Stream Push (WebSocket) | Modérée |
**Configuration dans `config.yaml` :**
@ -341,6 +342,13 @@ channels:
context:
thinking_enabled: true
subagent_enabled: true
dingtalk:
enabled: true
client_id: $DINGTALK_CLIENT_ID # ClientId depuis DingTalk Open Platform
client_secret: $DINGTALK_CLIENT_SECRET # ClientSecret depuis DingTalk Open Platform
allowed_users: [] # vide = tout le monde autorisé
card_template_id: "" # Optionnel : ID de modèle AI Card pour l'effet machine à écrire en streaming
```
Définissez les clés API correspondantes dans votre fichier `.env` :
@ -356,6 +364,10 @@ SLACK_APP_TOKEN=xapp-...
# Feishu / Lark
FEISHU_APP_ID=cli_xxxx
FEISHU_APP_SECRET=your_app_secret
# DingTalk
DINGTALK_CLIENT_ID=your_client_id
DINGTALK_CLIENT_SECRET=your_client_secret
```
**Configuration Telegram**
@ -378,6 +390,13 @@ FEISHU_APP_SECRET=your_app_secret
3. Dans **Events**, abonnez-vous à `im.message.receive_v1` et sélectionnez le mode **Long Connection**.
4. Copiez l'App ID et l'App Secret. Définissez `FEISHU_APP_ID` et `FEISHU_APP_SECRET` dans `.env` et activez le canal dans `config.yaml`.
**Configuration DingTalk**
1. Créez une application sur [DingTalk Open Platform](https://open.dingtalk.com/) et activez la capacité **Robot**.
2. Dans la page de configuration du robot, définissez le mode de réception des messages sur **Stream**.
3. Copiez le `Client ID` et le `Client Secret`. Définissez `DINGTALK_CLIENT_ID` et `DINGTALK_CLIENT_SECRET` dans `.env` et activez le canal dans `config.yaml`.
4. *(Optionnel)* Pour activer les réponses en streaming AI Card (effet machine à écrire), créez un modèle **AI Card** sur la [plateforme de cartes DingTalk](https://open.dingtalk.com/document/dingstart/typewriter-effect-streaming-ai-card), puis définissez `card_template_id` dans `config.yaml` avec l'ID du modèle. Vous devez également demander les permissions `Card.Streaming.Write` et `Card.Instance.Write`.
**Commandes**
Une fois un canal connecté, vous pouvez interagir avec DeerFlow directement depuis le chat :

19
README_ja.md
View File

@ -243,6 +243,7 @@ DeerFlowはメッセージングアプリからのタスク受信をサポート
| Telegram | Bot API（ロングポーリング） | 簡単 |
| Slack | Socket Mode | 中程度 |
| Feishu / Lark | WebSocket | 中程度 |
| DingTalk | Stream Push（WebSocket） | 中程度 |
**`config.yaml`での設定：**
@ -294,6 +295,13 @@ channels:
context:
thinking_enabled: true
subagent_enabled: true
dingtalk:
enabled: true
client_id: $DINGTALK_CLIENT_ID # DingTalk Open PlatformのClientId
client_secret: $DINGTALK_CLIENT_SECRET # DingTalk Open PlatformのClientSecret
allowed_users: [] # 空 = 全員許可
card_template_id: "" # オプション：ストリーミングタイプライター効果用のAIカードテンプレートID
```
対応するAPIキーを`.env`ファイルに設定します：
@ -309,6 +317,10 @@ SLACK_APP_TOKEN=xapp-...
# Feishu / Lark
FEISHU_APP_ID=cli_xxxx
FEISHU_APP_SECRET=your_app_secret
# DingTalk
DINGTALK_CLIENT_ID=your_client_id
DINGTALK_CLIENT_SECRET=your_client_secret
```
**Telegramのセットアップ**
@ -331,6 +343,13 @@ FEISHU_APP_SECRET=your_app_secret
3. **イベント**で`im.message.receive_v1`を購読し、**ロングコネクション**モードを選択。
4. App IDとApp Secretをコピー。`.env`に`FEISHU_APP_ID`と`FEISHU_APP_SECRET`を設定し、`config.yaml`でチャネルを有効にします。
**DingTalkのセットアップ**
1. [DingTalk Open Platform](https://open.dingtalk.com/)でアプリを作成し、**ロボット**機能を有効化します。
2. ロボット設定ページでメッセージ受信モードを**Streamモード**に設定します。
3. `Client ID`と`Client Secret`をコピー。`.env`に`DINGTALK_CLIENT_ID`と`DINGTALK_CLIENT_SECRET`を設定し、`config.yaml`でチャネルを有効にします。
4. *（オプション）* ストリーミングAIカード返信（タイプライター効果）を有効にするには、[DingTalkカードプラットフォーム](https://open.dingtalk.com/document/dingstart/typewriter-effect-streaming-ai-card)で**AIカード**テンプレートを作成し、`config.yaml`の`card_template_id`にテンプレートIDを設定します。`Card.Streaming.Write` および `Card.Instance.Write` 権限の申請も必要です。
**コマンド**
チャネル接続後、チャットから直接DeerFlowと対話できます：

15
README_ru.md
View File

@ -256,6 +256,7 @@ DeerFlow принимает задачи прямо из мессенджеро
| Telegram | Bot API (long-polling) | Просто |
| Slack | Socket Mode | Средне |
| Feishu / Lark | WebSocket | Средне |
| DingTalk | Stream Push (WebSocket) | Средне |
**Конфигурация в `config.yaml`:**
@ -278,6 +279,13 @@ channels:
enabled: true
bot_token: $TELEGRAM_BOT_TOKEN
allowed_users: []
dingtalk:
enabled: true
client_id: $DINGTALK_CLIENT_ID # ClientId с DingTalk Open Platform
client_secret: $DINGTALK_CLIENT_SECRET # ClientSecret с DingTalk Open Platform
allowed_users: [] # пусто = разрешить всем
card_template_id: "" # Опционально: ID шаблона AI Card для потокового эффекта печатной машинки
```
**Настройка Telegram**
@ -285,6 +293,13 @@ channels:
1. Напишите [@BotFather](https://t.me/BotFather), отправьте `/newbot` и скопируйте HTTP API-токен.
2. Укажите `TELEGRAM_BOT_TOKEN` в `.env` и включите канал в `config.yaml`.
**Настройка DingTalk**
1. Создайте приложение на [DingTalk Open Platform](https://open.dingtalk.com/) и включите возможность **Робот**.
2. На странице настроек робота установите режим приёма сообщений на **Stream**.
3. Скопируйте `Client ID` и `Client Secret`. Укажите `DINGTALK_CLIENT_ID` и `DINGTALK_CLIENT_SECRET` в `.env` и включите канал в `config.yaml`.
4. *(Опционально)* Для включения потоковых ответов AI Card (эффект печатной машинки) создайте шаблон **AI Card** на [платформе карточек DingTalk](https://open.dingtalk.com/document/dingstart/typewriter-effect-streaming-ai-card), затем укажите `card_template_id` в `config.yaml` с ID шаблона. Также необходимо запросить разрешения `Card.Streaming.Write` и `Card.Instance.Write`.
**Доступные команды**
| Команда | Описание |

19
README_zh.md
View File

@ -248,6 +248,7 @@ DeerFlow 支持从即时通讯应用接收任务。只要配置完成，对应
| Slack | Socket Mode | 中等 |
| Feishu / Lark | WebSocket | 中等 |
| 企业微信智能机器人 | WebSocket | 中等 |
| 钉钉 | Stream Push（WebSocket） | 中等 |
**`config.yaml` 中的配置示例：**
@ -304,6 +305,13 @@ channels:
context:
thinking_enabled: true
subagent_enabled: true
dingtalk:
enabled: true
client_id: $DINGTALK_CLIENT_ID # 钉钉开放平台 ClientId
client_secret: $DINGTALK_CLIENT_SECRET # 钉钉开放平台 ClientSecret
allowed_users: [] # 留空表示允许所有人
card_template_id: "" # 可选：AI 卡片模板 ID，用于流式打字机效果
```
说明：
@ -327,6 +335,10 @@ FEISHU_APP_SECRET=your_app_secret
# 企业微信智能机器人
WECOM_BOT_ID=your_bot_id
WECOM_BOT_SECRET=your_bot_secret
# 钉钉
DINGTALK_CLIENT_ID=your_client_id
DINGTALK_CLIENT_SECRET=your_client_secret
```
**Telegram 配置**
@ -357,6 +369,13 @@ WECOM_BOT_SECRET=your_bot_secret
4. 安装后端依赖时确保包含 `wecom-aibot-python-sdk`，渠道会通过 WebSocket 长连接接收消息，无需公网回调地址。
5. 当前支持文本、图片和文件入站消息；agent 生成的最终图片/文件也会回传到企业微信会话中。
**钉钉配置**
1. 在 [钉钉开放平台](https://open.dingtalk.com/) 创建应用，并启用 **机器人** 能力。
2. 在机器人配置页面设置消息接收模式为 **Stream模式**。
3. 复制 `Client ID` 和 `Client Secret`，在 `.env` 中设置 `DINGTALK_CLIENT_ID` 和 `DINGTALK_CLIENT_SECRET`，并在 `config.yaml` 中启用该渠道。
4. *（可选）* 如需开启流式 AI 卡片回复（打字机效果），请在[钉钉卡片平台](https://open.dingtalk.com/document/dingstart/typewriter-effect-streaming-ai-card)创建 **AI 卡片**模板，然后在 `config.yaml` 中将 `card_template_id` 设为该模板 ID。同时需要申请 `Card.Streaming.Write` 和 `Card.Instance.Write` 权限。
**命令**
渠道连接完成后，你可以直接在聊天窗口里和 DeerFlow 交互：

15
backend/CLAUDE.md
View File

@ -112,7 +112,7 @@ CI runs these regression tests for every pull request via [.github/workflows/bac
The backend is split into two layers with a strict dependency direction:
- **Harness** (`packages/harness/deerflow/`): Publishable agent framework package (`deerflow-harness`). Import prefix: `deerflow.*`. Contains agent orchestration, tools, sandbox, models, MCP, skills, config — everything needed to build and run agents.
- **App** (`app/`): Unpublished application code. Import prefix: `app.*`. Contains the FastAPI Gateway API and IM channel integrations (Feishu, Slack, Telegram).
- **App** (`app/`): Unpublished application code. Import prefix: `app.*`. Contains the FastAPI Gateway API and IM channel integrations (Feishu, Slack, Telegram, DingTalk).
**Dependency rule**: App imports deerflow, but deerflow never imports app. This boundary is enforced by `tests/test_harness_boundary.py` which runs in CI.
@ -312,7 +312,8 @@ Proxied through nginx: `/api/langgraph/*` → LangGraph, all other `/api/*` →
### IM Channels System (`app/channels/`)
Bridges external messaging platforms (Feishu, Slack, Telegram) to the DeerFlow agent via Gateway's LangGraph-compatible API.
Bridges external messaging platforms (Feishu, Slack, Telegram, DingTalk) to the DeerFlow agent via the LangGraph Server.
**Architecture**: Channels communicate with Gateway through the `langgraph-sdk` HTTP client (same as the frontend), ensuring threads are created and managed server-side. The internal SDK client injects process-local internal auth plus a matching CSRF cookie/header pair so Gateway accepts state-changing thread/run requests from channel workers without relying on browser session cookies.
@ -322,7 +323,7 @@ Bridges external messaging platforms (Feishu, Slack, Telegram) to the DeerFlow a
- `manager.py` - Core dispatcher: creates threads via `client.threads.create()`, routes commands, keeps Slack/Telegram on `client.runs.wait()`, and uses `client.runs.stream(["messages-tuple", "values"])` for Feishu incremental outbound updates
- `base.py` - Abstract `Channel` base class (start/stop/send lifecycle)
- `service.py` - Manages lifecycle of all configured channels from `config.yaml`
- `slack.py` / `feishu.py` / `telegram.py` - Platform-specific implementations (`feishu.py` tracks the running card `message_id` in memory and patches the same card in place)
- `slack.py` / `feishu.py` / `telegram.py` / `dingtalk.py` - Platform-specific implementations (`feishu.py` tracks the running card `message_id` in memory and patches the same card in place; `dingtalk.py` optionally uses AI Card streaming for in-place updates when `card_template_id` is configured)
**Message Flow**:
1. External platform -> Channel impl -> `MessageBus.publish_inbound()`
@ -331,14 +332,16 @@ Bridges external messaging platforms (Feishu, Slack, Telegram) to the DeerFlow a
4. Feishu chat: `runs.stream()` → accumulate AI text → publish multiple outbound updates (`is_final=False`) → publish final outbound (`is_final=True`)
5. Slack/Telegram chat: `runs.wait()` → extract final response → publish outbound
6. Feishu channel sends one running reply card up front, then patches the same card for each outbound update (card JSON sets `config.update_multi=true` for Feishu's patch API requirement)
7. For commands (`/new`, `/status`, `/models`, `/memory`, `/help`): handle locally or query Gateway API
8. Outbound → channel callbacks → platform reply
7. DingTalk AI Card mode (when `card_template_id` configured): `runs.stream()` → create card with initial text → stream updates via `PUT /v1.0/card/streaming` → finalize on `is_final=True`. Falls back to `sampleMarkdown` if card creation or streaming fails
8. For commands (`/new`, `/status`, `/models`, `/memory`, `/help`): handle locally or query Gateway API
9. Outbound → channel callbacks → platform reply
**Configuration** (`config.yaml` -> `channels`):
- `langgraph_url` - LangGraph-compatible Gateway API base URL (default: `http://localhost:8001/api`)
- `gateway_url` - Gateway API URL for auxiliary commands (default: `http://localhost:8001`)
- In Docker Compose, IM channels run inside the `gateway` container, so `localhost` points back to that container. Use `http://gateway:8001/api` for `langgraph_url` and `http://gateway:8001` for `gateway_url`, or set `DEER_FLOW_CHANNELS_LANGGRAPH_URL` / `DEER_FLOW_CHANNELS_GATEWAY_URL`.
- Per-channel configs: `feishu` (app_id, app_secret), `slack` (bot_token, app_token), `telegram` (bot_token)
- Per-channel configs: `feishu` (app_id, app_secret), `slack` (bot_token, app_token), `telegram` (bot_token), `dingtalk` (client_id, client_secret, optional `card_template_id` for AI Card streaming)
### Memory System (`packages/harness/deerflow/agents/memory/`)

4
backend/app/channels/base.py
View File

@ -31,6 +31,10 @@ class Channel(ABC):
def is_running(self) -> bool:
return self._running
@property
def supports_streaming(self) -> bool:
return False
# -- lifecycle ---------------------------------------------------------
@abstractmethod

740
backend/app/channels/dingtalk.py Normal file
View File

@ -0,0 +1,740 @@
"""DingTalk channel implementation."""
from __future__ import annotations
import asyncio
import json
import logging
import re
import threading
import time
from pathlib import Path
from typing import Any
import httpx
from app.channels.base import Channel
from app.channels.commands import KNOWN_CHANNEL_COMMANDS
from app.channels.message_bus import InboundMessage, InboundMessageType, MessageBus, OutboundMessage, ResolvedAttachment
logger = logging.getLogger(__name__)
DINGTALK_API_BASE = "https://api.dingtalk.com"
_TOKEN_REFRESH_MARGIN_SECONDS = 300
_CONVERSATION_TYPE_P2P = "1"
_CONVERSATION_TYPE_GROUP = "2"
_MAX_UPLOAD_SIZE_BYTES = 20 * 1024 * 1024
def _normalize_conversation_type(raw: Any) -> str:
"""Normalize ``conversationType`` to ``"1"`` (P2P) or ``"2"`` (group).
Stream payloads may send int or string values.
"""
if raw is None:
return _CONVERSATION_TYPE_P2P
s = str(raw).strip()
if s == _CONVERSATION_TYPE_GROUP:
return _CONVERSATION_TYPE_GROUP
return _CONVERSATION_TYPE_P2P
def _normalize_allowed_users(allowed_users: Any) -> set[str]:
if allowed_users is None:
return set()
if isinstance(allowed_users, str):
values = [allowed_users]
elif isinstance(allowed_users, (list, tuple, set)):
values = allowed_users
else:
logger.warning(
"DingTalk allowed_users should be a list of user IDs; treating %s as one string value",
type(allowed_users).__name__,
)
values = [allowed_users]
return {str(uid) for uid in values if str(uid)}
def _is_dingtalk_command(text: str) -> bool:
if not text.startswith("/"):
return False
return text.split(maxsplit=1)[0].lower() in KNOWN_CHANNEL_COMMANDS
def _extract_text_from_rich_text(rich_text_list: list) -> str:
parts: list[str] = []
for item in rich_text_list:
if isinstance(item, dict) and "text" in item:
parts.append(item["text"])
return " ".join(parts)
_FENCED_CODE_BLOCK_RE = re.compile(r"```(\w*)\n(.*?)```", re.DOTALL)
_INLINE_CODE_RE = re.compile(r"`([^`\n]+)`")
_HORIZONTAL_RULE_RE = re.compile(r"^-{3,}$", re.MULTILINE)
_TABLE_SEPARATOR_RE = re.compile(r"^\|[-:| ]+\|$", re.MULTILINE)
def _convert_markdown_table(text: str) -> str:
# DingTalk sampleMarkdown does not render pipe-delimited tables.
lines = text.split("\n")
result: list[str] = []
i = 0
while i < len(lines):
line = lines[i]
# Detect table: header row followed by separator row
if i + 1 < len(lines) and line.strip().startswith("|") and _TABLE_SEPARATOR_RE.match(lines[i + 1].strip()):
headers = [h.strip() for h in line.strip().strip("|").split("|")]
i += 2 # skip header + separator
while i < len(lines) and lines[i].strip().startswith("|"):
cells = [c.strip() for c in lines[i].strip().strip("|").split("|")]
for h, c in zip(headers, cells):
result.append(f"> **{h}**: {c}")
result.append("")
i += 1
else:
result.append(line)
i += 1
return "\n".join(result)
def _adapt_markdown_for_dingtalk(text: str) -> str:
"""Adapt markdown for DingTalk's limited sampleMarkdown renderer."""
def _code_block_to_quote(match: re.Match) -> str:
lang = match.group(1)
code = match.group(2).rstrip("\n")
prefix = f"> **{lang}**\n" if lang else ""
quoted_lines = "\n".join(f"> {line}" for line in code.split("\n"))
return f"{prefix}{quoted_lines}\n"
text = _FENCED_CODE_BLOCK_RE.sub(_code_block_to_quote, text)
text = _INLINE_CODE_RE.sub(r"**\1**", text)
text = _convert_markdown_table(text)
text = _HORIZONTAL_RULE_RE.sub("───────────", text)
return text
class DingTalkChannel(Channel):
"""DingTalk IM channel using Stream Push (WebSocket, no public IP needed)."""
def __init__(self, bus: MessageBus, config: dict[str, Any]) -> None:
super().__init__(name="dingtalk", bus=bus, config=config)
self._thread: threading.Thread | None = None
self._main_loop: asyncio.AbstractEventLoop | None = None
self._client_id: str = ""
self._client_secret: str = ""
self._allowed_users: set[str] = _normalize_allowed_users(config.get("allowed_users"))
self._cached_token: str = ""
self._token_expires_at: float = 0.0
self._token_lock = asyncio.Lock()
self._card_template_id: str = config.get("card_template_id", "")
self._card_track_ids: dict[str, str] = {}
self._dingtalk_client: Any = None
self._stream_client: Any = None
self._incoming_messages: dict[str, Any] = {}
self._incoming_messages_lock = threading.Lock()
self._card_repliers: dict[str, Any] = {}
@property
def supports_streaming(self) -> bool:
return bool(self._card_template_id)
async def start(self) -> None:
if self._running:
return
try:
import dingtalk_stream # noqa: F401
except ImportError:
logger.error("dingtalk-stream is not installed. Install it with: uv add dingtalk-stream")
return
client_id = self.config.get("client_id", "")
client_secret = self.config.get("client_secret", "")
if not client_id or not client_secret:
logger.error("DingTalk channel requires client_id and client_secret")
return
self._client_id = client_id
self._client_secret = client_secret
self._main_loop = asyncio.get_running_loop()
if self._card_template_id:
logger.info("[DingTalk] AI Card mode enabled (template=%s)", self._card_template_id)
self._running = True
self.bus.subscribe_outbound(self._on_outbound)
self._thread = threading.Thread(
target=self._run_stream,
args=(client_id, client_secret),
daemon=True,
)
self._thread.start()
logger.info("DingTalk channel started")
async def stop(self) -> None:
self._running = False
self.bus.unsubscribe_outbound(self._on_outbound)
stream_client = self._stream_client
if stream_client is not None:
try:
if hasattr(stream_client, "disconnect"):
stream_client.disconnect()
except Exception:
logger.debug("[DingTalk] error disconnecting stream client", exc_info=True)
self._dingtalk_client = None
self._stream_client = None
with self._incoming_messages_lock:
self._incoming_messages.clear()
self._card_repliers.clear()
self._card_track_ids.clear()
if self._thread:
self._thread.join(timeout=5)
self._thread = None
logger.info("DingTalk channel stopped")
def _resolve_routing(self, msg: OutboundMessage) -> tuple[str, str, str]:
"""Return (conversation_type, sender_staff_id, conversation_id).
Uses msg.chat_id as the primary routing key; metadata as fallback.
"""
conversation_type = _normalize_conversation_type(msg.metadata.get("conversation_type"))
sender_staff_id = msg.metadata.get("sender_staff_id", "")
conversation_id = msg.metadata.get("conversation_id", "")
if conversation_type == _CONVERSATION_TYPE_GROUP:
conversation_id = msg.chat_id or conversation_id
else:
sender_staff_id = msg.chat_id or sender_staff_id
return conversation_type, sender_staff_id, conversation_id
async def send(self, msg: OutboundMessage, *, _max_retries: int = 3) -> None:
conversation_type, sender_staff_id, conversation_id = self._resolve_routing(msg)
robot_code = self._client_id
# Card mode: stream update to existing AI card
source_key = self._make_card_source_key_from_outbound(msg)
out_track_id = self._card_track_ids.get(source_key)
# ``card_template_id`` enables ``runs.stream`` (non-final + final outbounds).
# If card creation failed, skip non-final chunks to avoid duplicate messages.
if self._card_template_id and not out_track_id and not msg.is_final:
return
if out_track_id:
try:
await self._stream_update_card(
out_track_id,
msg.text,
is_finalize=msg.is_final,
)
except Exception:
logger.warning("[DingTalk] card stream failed, falling back to sampleMarkdown")
if msg.is_final:
self._card_track_ids.pop(source_key, None)
self._card_repliers.pop(out_track_id, None)
await self._send_markdown_fallback(robot_code, conversation_type, sender_staff_id, conversation_id, msg.text)
return
if msg.is_final:
self._card_track_ids.pop(source_key, None)
self._card_repliers.pop(out_track_id, None)
return
# Non-card mode: send sampleMarkdown with retry
last_exc: Exception | None = None
for attempt in range(_max_retries):
try:
if conversation_type == _CONVERSATION_TYPE_GROUP:
await self._send_group_message(robot_code, conversation_id, msg.text, at_user_ids=[sender_staff_id] if sender_staff_id else None)
else:
await self._send_p2p_message(robot_code, sender_staff_id, msg.text)
return
except Exception as exc:
last_exc = exc
if attempt < _max_retries - 1:
delay = 2**attempt
logger.warning(
"[DingTalk] send failed (attempt %d/%d), retrying in %ds: %s",
attempt + 1,
_max_retries,
delay,
exc,
)
await asyncio.sleep(delay)
logger.error("[DingTalk] send failed after %d attempts: %s", _max_retries, last_exc)
if last_exc is None:
raise RuntimeError("DingTalk send failed without an exception from any attempt")
raise last_exc
async def _send_markdown_fallback(
self,
robot_code: str,
conversation_type: str,
sender_staff_id: str,
conversation_id: str,
text: str,
) -> None:
try:
if conversation_type == _CONVERSATION_TYPE_GROUP:
await self._send_group_message(robot_code, conversation_id, text)
else:
await self._send_p2p_message(robot_code, sender_staff_id, text)
except Exception:
logger.exception("[DingTalk] markdown fallback also failed")
raise
async def send_file(self, msg: OutboundMessage, attachment: ResolvedAttachment) -> bool:
if attachment.size > _MAX_UPLOAD_SIZE_BYTES:
logger.warning("[DingTalk] file too large (%d bytes), skipping: %s", attachment.size, attachment.filename)
return False
conversation_type, sender_staff_id, conversation_id = self._resolve_routing(msg)
robot_code = self._client_id
try:
media_id = await self._upload_media(attachment.actual_path, "image" if attachment.is_image else "file")
if not media_id:
return False
if attachment.is_image:
msg_key = "sampleImageMsg"
msg_param = json.dumps({"photoURL": media_id})
else:
msg_key = "sampleFile"
msg_param = json.dumps(
{
"fileUrl": media_id,
"fileName": attachment.filename,
"fileSize": str(attachment.size),
}
)
token = await self._get_access_token()
async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
if conversation_type == _CONVERSATION_TYPE_GROUP:
response = await client.post(
f"{DINGTALK_API_BASE}/v1.0/robot/groupMessages/send",
headers=self._api_headers(token),
json={
"msgKey": msg_key,
"msgParam": msg_param,
"robotCode": robot_code,
"openConversationId": conversation_id,
},
)
else:
response = await client.post(
f"{DINGTALK_API_BASE}/v1.0/robot/oToMessages/batchSend",
headers=self._api_headers(token),
json={
"msgKey": msg_key,
"msgParam": msg_param,
"robotCode": robot_code,
"userIds": [sender_staff_id],
},
)
response.raise_for_status()
logger.info("[DingTalk] file sent: %s", attachment.filename)
return True
except (httpx.HTTPError, OSError, ValueError, TypeError, AttributeError):
logger.exception("[DingTalk] failed to send file: %s", attachment.filename)
return False
# -- stream client (runs in dedicated thread) --------------------------
def _run_stream(self, client_id: str, client_secret: str) -> None:
try:
import dingtalk_stream
credential = dingtalk_stream.Credential(client_id, client_secret)
client = dingtalk_stream.DingTalkStreamClient(credential)
self._stream_client = client
client.register_callback_handler(
dingtalk_stream.chatbot.ChatbotMessage.TOPIC,
_DingTalkMessageHandler(self),
)
client.start_forever()
except Exception:
if self._running:
logger.exception("DingTalk Stream Push error")
finally:
self._stream_client = None
def _on_chatbot_message(self, message: Any) -> None:
if not self._running:
return
try:
sender_staff_id = message.sender_staff_id or ""
conversation_type = _normalize_conversation_type(message.conversation_type)
conversation_id = message.conversation_id or ""
msg_id = message.message_id or ""
sender_nick = message.sender_nick or ""
if self._allowed_users and sender_staff_id not in self._allowed_users:
logger.debug("[DingTalk] ignoring message from non-allowed user: %s", sender_staff_id)
return
text = self._extract_text(message)
if not text:
logger.info("[DingTalk] empty text, ignoring message")
return
logger.info(
"[DingTalk] parsed message: conv_type=%s, msg_id=%s, sender=%s(%s), text=%r",
conversation_type,
msg_id,
sender_staff_id,
sender_nick,
text[:100],
)
if _is_dingtalk_command(text):
msg_type = InboundMessageType.COMMAND
else:
msg_type = InboundMessageType.CHAT
# P2P: topic_id=None (single thread per user, like Telegram private chat)
# Group: topic_id=msg_id (each new message starts a new topic, like Feishu)
topic_id: str | None = msg_id if conversation_type == _CONVERSATION_TYPE_GROUP else None
# chat_id uses conversation_id for groups, sender_staff_id for P2P
chat_id = conversation_id if conversation_type == _CONVERSATION_TYPE_GROUP else sender_staff_id
inbound = self._make_inbound(
chat_id=chat_id,
user_id=sender_staff_id,
text=text,
msg_type=msg_type,
thread_ts=msg_id,
metadata={
"conversation_type": conversation_type,
"conversation_id": conversation_id,
"sender_staff_id": sender_staff_id,
"sender_nick": sender_nick,
"message_id": msg_id,
},
)
inbound.topic_id = topic_id
if self._card_template_id:
source_key = self._make_card_source_key(inbound)
with self._incoming_messages_lock:
self._incoming_messages[source_key] = message
if self._main_loop and self._main_loop.is_running():
logger.info("[DingTalk] publishing inbound message to bus (type=%s, msg_id=%s)", msg_type.value, msg_id)
fut = asyncio.run_coroutine_threadsafe(
self._prepare_inbound(chat_id, inbound),
self._main_loop,
)
fut.add_done_callback(lambda f, mid=msg_id: self._log_future_error(f, "prepare_inbound", mid))
else:
logger.warning("[DingTalk] main loop not running, cannot publish inbound message")
except Exception:
logger.exception("[DingTalk] error processing chatbot message")
@staticmethod
def _extract_text(message: Any) -> str:
msg_type = message.message_type
if msg_type == "text" and message.text:
return message.text.content.strip()
if msg_type == "richText" and message.rich_text_content:
return _extract_text_from_rich_text(message.rich_text_content.rich_text_list).strip()
return ""
async def _prepare_inbound(self, chat_id: str, inbound: InboundMessage) -> None:
# Running reply must finish before publish_inbound so AI card tracks are
# registered before the manager emits streaming outbounds.
await self._send_running_reply(chat_id, inbound)
await self.bus.publish_inbound(inbound)
async def _send_running_reply(self, chat_id: str, inbound: InboundMessage) -> None:
conversation_type = inbound.metadata.get("conversation_type", _CONVERSATION_TYPE_P2P)
sender_staff_id = inbound.metadata.get("sender_staff_id", "")
conversation_id = inbound.metadata.get("conversation_id", "")
text = "\u23f3 Working on it..."
try:
if self._card_template_id:
source_key = self._make_card_source_key(inbound)
with self._incoming_messages_lock:
chatbot_message = self._incoming_messages.pop(source_key, None)
out_track_id = await self._create_and_deliver_card(
text,
chatbot_message=chatbot_message,
)
if out_track_id:
self._card_track_ids[source_key] = out_track_id
logger.info("[DingTalk] AI card running reply sent for chat=%s", chat_id)
return
robot_code = self._client_id
if conversation_type == _CONVERSATION_TYPE_GROUP:
await self._send_text_message_to_group(robot_code, conversation_id, text)
else:
await self._send_text_message_to_user(robot_code, sender_staff_id, text)
logger.info("[DingTalk] 'Working on it...' reply sent for chat=%s", chat_id)
except Exception:
logger.exception("[DingTalk] failed to send running reply for chat=%s", chat_id)
# -- DingTalk API helpers ----------------------------------------------
async def _get_access_token(self) -> str:
if self._cached_token and time.monotonic() < self._token_expires_at:
return self._cached_token
async with self._token_lock:
if self._cached_token and time.monotonic() < self._token_expires_at:
return self._cached_token
async with httpx.AsyncClient(timeout=httpx.Timeout(10.0)) as client:
response = await client.post(
f"{DINGTALK_API_BASE}/v1.0/oauth2/accessToken",
json={"appKey": self._client_id, "appSecret": self._client_secret}, # DingTalk API field names
)
response.raise_for_status()
data = response.json()
if not isinstance(data, dict):
raise ValueError(f"DingTalk access token response must be a JSON object, got {type(data).__name__}")
access_token = data.get("accessToken")
if not isinstance(access_token, str) or not access_token.strip():
raise ValueError("DingTalk access token response did not contain a usable accessToken")
raw_expires_in = data.get("expireIn", 7200)
try:
expires_in = int(raw_expires_in)
except (TypeError, ValueError):
logger.warning("[DingTalk] invalid expireIn value %r, using default 7200s", raw_expires_in)
expires_in = 7200
self._cached_token = access_token.strip()
self._token_expires_at = time.monotonic() + expires_in - _TOKEN_REFRESH_MARGIN_SECONDS
return self._cached_token
@staticmethod
def _api_headers(token: str) -> dict[str, str]:
return {
"x-acs-dingtalk-access-token": token,
"Content-Type": "application/json",
}
async def _send_text_message_to_user(self, robot_code: str, user_id: str, text: str) -> None:
token = await self._get_access_token()
async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
response = await client.post(
f"{DINGTALK_API_BASE}/v1.0/robot/oToMessages/batchSend",
headers=self._api_headers(token),
json={
"msgKey": "sampleText",
"msgParam": json.dumps({"content": text}),
"robotCode": robot_code,
"userIds": [user_id],
},
)
response.raise_for_status()
async def _send_text_message_to_group(self, robot_code: str, conversation_id: str, text: str) -> None:
token = await self._get_access_token()
async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
response = await client.post(
f"{DINGTALK_API_BASE}/v1.0/robot/groupMessages/send",
headers=self._api_headers(token),
json={
"msgKey": "sampleText",
"msgParam": json.dumps({"content": text}),
"robotCode": robot_code,
"openConversationId": conversation_id,
},
)
response.raise_for_status()
async def _send_p2p_message(self, robot_code: str, user_id: str, text: str) -> None:
text = _adapt_markdown_for_dingtalk(text)
token = await self._get_access_token()
async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
response = await client.post(
f"{DINGTALK_API_BASE}/v1.0/robot/oToMessages/batchSend",
headers=self._api_headers(token),
json={
"msgKey": "sampleMarkdown",
"msgParam": json.dumps({"title": "DeerFlow", "text": text}),
"robotCode": robot_code,
"userIds": [user_id],
},
)
response.raise_for_status()
data = response.json()
if data.get("processQueryKey"):
logger.info("[DingTalk] P2P message sent to user=%s", user_id)
else:
logger.warning("[DingTalk] P2P send response: %s", data)
async def _send_group_message(
self,
robot_code: str,
conversation_id: str,
text: str,
*,
at_user_ids: list[str] | None = None, # noqa: ARG002
) -> None:
# at_user_ids accepted for call-site compatibility but not passed to the API
# (sampleMarkdown does not support @mentions).
text = _adapt_markdown_for_dingtalk(text)
token = await self._get_access_token()
async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
response = await client.post(
f"{DINGTALK_API_BASE}/v1.0/robot/groupMessages/send",
headers=self._api_headers(token),
json={
"msgKey": "sampleMarkdown",
"msgParam": json.dumps({"title": "DeerFlow", "text": text}),
"robotCode": robot_code,
"openConversationId": conversation_id,
},
)
response.raise_for_status()
data = response.json()
if data.get("processQueryKey"):
logger.info("[DingTalk] group message sent to conversation=%s", conversation_id)
else:
logger.warning("[DingTalk] group send response: %s", data)
# -- AI Card streaming helpers -------------------------------------------
def _make_card_source_key(self, inbound: InboundMessage) -> str:
m = inbound.metadata
return f"{m.get('conversation_type', '')}:{m.get('sender_staff_id', '')}:{m.get('conversation_id', '')}:{m.get('message_id', '')}"
def _make_card_source_key_from_outbound(self, msg: OutboundMessage) -> str:
m = msg.metadata
correlation_id = m.get("message_id") or msg.thread_ts or ""
return f"{m.get('conversation_type', '')}:{m.get('sender_staff_id', '')}:{m.get('conversation_id', '')}:{correlation_id}"
async def _create_and_deliver_card(
self,
initial_text: str,
*,
chatbot_message: Any = None,
) -> str | None:
if self._dingtalk_client is None or chatbot_message is None:
logger.warning("[DingTalk] SDK client or chatbot_message unavailable, skipping AI card")
return None
try:
from dingtalk_stream.card_replier import AICardReplier
except ImportError:
logger.warning("[DingTalk] dingtalk-stream card_replier not available")
return None
try:
replier = AICardReplier(self._dingtalk_client, chatbot_message)
card_instance_id = await replier.async_create_and_deliver_card(
card_template_id=self._card_template_id,
card_data={"content": initial_text},
)
if not card_instance_id:
return None
self._card_repliers[card_instance_id] = replier
logger.info("[DingTalk] AI card created: outTrackId=%s", card_instance_id)
return card_instance_id
except Exception:
logger.exception("[DingTalk] failed to create AI card")
return None
async def _stream_update_card(
self,
out_track_id: str,
content: str,
*,
is_finalize: bool = False,
is_error: bool = False,
) -> None:
replier = self._card_repliers.get(out_track_id)
if not replier:
raise RuntimeError(f"No AICardReplier found for track ID {out_track_id}")
await replier.async_streaming(
card_instance_id=out_track_id,
content_key="content",
content_value=content,
append=False,
finished=is_finalize,
failed=is_error,
)
# -- media upload --------------------------------------------------------
async def _upload_media(self, file_path: str | Path, media_type: str) -> str | None:
try:
file_bytes = await asyncio.to_thread(Path(file_path).read_bytes)
token = await self._get_access_token()
async with httpx.AsyncClient(timeout=httpx.Timeout(60.0)) as client:
response = await client.post(
f"{DINGTALK_API_BASE}/v1.0/files/upload",
headers={"x-acs-dingtalk-access-token": token},
files={"file": ("upload", file_bytes)},
data={"type": media_type},
)
response.raise_for_status()
try:
payload = response.json()
except json.JSONDecodeError:
logger.exception("[DingTalk] failed to decode upload response JSON: %s", file_path)
return None
if not isinstance(payload, dict):
logger.warning("[DingTalk] unexpected upload response type %s for %s", type(payload).__name__, file_path)
return None
return payload.get("mediaId")
except (httpx.HTTPError, OSError):
logger.exception("[DingTalk] failed to upload media: %s", file_path)
return None
@staticmethod
def _log_future_error(fut: Any, name: str, msg_id: str) -> None:
try:
exc = fut.exception()
if exc:
logger.error("[DingTalk] %s failed for msg_id=%s: %s", name, msg_id, exc)
except (asyncio.CancelledError, asyncio.InvalidStateError):
pass
class _DingTalkMessageHandler:
"""Callback handler registered with dingtalk-stream."""
def __init__(self, channel: DingTalkChannel) -> None:
self._channel = channel
def pre_start(self) -> None:
if hasattr(self, "dingtalk_client") and self.dingtalk_client is not None:
self._channel._dingtalk_client = self.dingtalk_client
async def raw_process(self, callback_message: Any) -> Any:
import dingtalk_stream
from dingtalk_stream.frames import Headers
code, message = await self.process(callback_message)
ack_message = dingtalk_stream.AckMessage()
ack_message.code = code
ack_message.headers.message_id = callback_message.headers.message_id
ack_message.headers.content_type = Headers.CONTENT_TYPE_APPLICATION_JSON
ack_message.data = {"response": message}
return ack_message
async def process(self, callback: Any) -> tuple[int, str]:
import dingtalk_stream
incoming_message = dingtalk_stream.ChatbotMessage.from_dict(callback.data)
self._channel._on_chatbot_message(incoming_message)
return dingtalk_stream.AckMessage.STATUS_OK, "OK"

4
backend/app/channels/feishu.py
View File

@ -63,6 +63,10 @@ class FeishuChannel(Channel):
self._GetMessageResourceRequest = None
self._thread_lock = threading.Lock()
@property
def supports_streaming(self) -> bool:
return True
async def start(self) -> None:
if self._running:
return

20
backend/app/channels/manager.py
View File

@ -38,6 +38,7 @@ STREAM_UPDATE_MIN_INTERVAL_SECONDS = 0.35
THREAD_BUSY_MESSAGE = "This conversation is already processing another request. Please wait for it to finish and try again."
CHANNEL_CAPABILITIES = {
"dingtalk": {"supports_streaming": False},
"discord": {"supports_streaming": False},
"feishu": {"supports_streaming": True},
"slack": {"supports_streaming": False},
@ -48,6 +49,13 @@ CHANNEL_CAPABILITIES = {
InboundFileReader = Callable[[dict[str, Any], httpx.AsyncClient], Awaitable[bytes | None]]
_METADATA_DROP_KEYS = frozenset({"raw_message", "ref_msg"})
def _slim_metadata(meta: dict[str, Any]) -> dict[str, Any]:
"""Return a shallow copy of *meta* with known-large keys removed."""
return {k: v for k, v in meta.items() if k not in _METADATA_DROP_KEYS}
INBOUND_FILE_READERS: dict[str, InboundFileReader] = {}
@ -543,6 +551,13 @@ class ChannelManager:
@staticmethod
def _channel_supports_streaming(channel_name: str) -> bool:
from .service import get_channel_service
service = get_channel_service()
if service:
channel = service.get_channel(channel_name)
if channel is not None:
return channel.supports_streaming
return CHANNEL_CAPABILITIES.get(channel_name, {}).get("supports_streaming", False)
def _resolve_session_layer(self, msg: InboundMessage) -> tuple[dict[str, Any], dict[str, Any]]:
@ -772,6 +787,7 @@ class ChannelManager:
artifacts=artifacts,
attachments=attachments,
thread_ts=msg.thread_ts,
metadata=_slim_metadata(msg.metadata),
)
logger.info("[Manager] publishing outbound message to bus: channel=%s, chat_id=%s", msg.channel_name, msg.chat_id)
await self.bus.publish_outbound(outbound)
@ -833,6 +849,7 @@ class ChannelManager:
text=latest_text,
is_final=False,
thread_ts=msg.thread_ts,
metadata=_slim_metadata(msg.metadata),
)
)
last_published_text = latest_text
@ -877,6 +894,7 @@ class ChannelManager:
attachments=attachments,
is_final=True,
thread_ts=msg.thread_ts,
metadata=_slim_metadata(msg.metadata),
)
)
@ -935,6 +953,7 @@ class ChannelManager:
thread_id=self.store.get_thread_id(msg.channel_name, msg.chat_id) or "",
text=reply,
thread_ts=msg.thread_ts,
metadata=_slim_metadata(msg.metadata),
)
await self.bus.publish_outbound(outbound)
@ -968,5 +987,6 @@ class ChannelManager:
thread_id=self.store.get_thread_id(msg.channel_name, msg.chat_id) or "",
text=error_text,
thread_ts=msg.thread_ts,
metadata=_slim_metadata(msg.metadata),
)
await self.bus.publish_outbound(outbound)

9
backend/app/channels/service.py
View File

@ -18,6 +18,7 @@ if TYPE_CHECKING:
# Channel name → import path for lazy loading
_CHANNEL_REGISTRY: dict[str, str] = {
"dingtalk": "app.channels.dingtalk:DingTalkChannel",
"discord": "app.channels.discord:DiscordChannel",
"feishu": "app.channels.feishu:FeishuChannel",
"slack": "app.channels.slack:SlackChannel",
@ -28,6 +29,7 @@ _CHANNEL_REGISTRY: dict[str, str] = {
# Keys that indicate a user has configured credentials for a channel.
_CHANNEL_CREDENTIAL_KEYS: dict[str, list[str]] = {
"dingtalk": ["client_id", "client_secret"],
"discord": ["bot_token"],
"feishu": ["app_id", "app_secret"],
"slack": ["bot_token", "app_token"],
@ -166,11 +168,16 @@ class ChannelService:
try:
channel = channel_cls(bus=self.bus, config=config)
await channel.start()
self._channels[name] = channel
await channel.start()
if not channel.is_running:
self._channels.pop(name, None)
logger.error("Channel %s did not enter a running state after start()", name)
return False
logger.info("Channel %s started", name)
return True
except Exception:
self._channels.pop(name, None)
logger.exception("Failed to start channel %s", name)
return False

4
backend/app/channels/wecom.py
View File

@ -29,6 +29,10 @@ class WeComChannel(Channel):
self._ws_stream_ids: dict[str, str] = {}
self._working_message = "Working on it..."
@property
def supports_streaming(self) -> bool:
return True
def _clear_ws_context(self, thread_ts: str | None) -> None:
if not thread_ts:
return

1
backend/pyproject.toml
View File

@ -17,6 +17,7 @@ dependencies = [
"langgraph-sdk>=0.1.51",
"markdown-to-mrkdwn>=0.3.1",
"wecom-aibot-python-sdk>=0.1.6",
"dingtalk-stream>=0.24.3",
"bcrypt>=4.0.0",
"pyjwt>=2.9.0",
"email-validator>=2.0.0",

84
backend/tests/test_channels.py
View File

@ -462,6 +462,7 @@ class TestChannelManager:
)
mock_channel = MagicMock()
mock_channel.receive_file = AsyncMock(return_value=modified_msg)
mock_channel.supports_streaming = False
mock_service = MagicMock()
mock_service.get_channel.return_value = mock_channel
monkeypatch.setattr("app.channels.service.get_channel_service", lambda: mock_service)
@ -535,6 +536,89 @@ class TestChannelManager:
_run(go())
def test_handle_chat_outbound_preserves_inbound_metadata(self):
"""DingTalk (and similar) need inbound metadata on outbound sends (e.g. sender_staff_id)."""
from app.channels.manager import ChannelManager
async def go():
bus = MessageBus()
store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json")
manager = ChannelManager(bus=bus, store=store)
outbound_received: list[OutboundMessage] = []
async def capture_outbound(msg: OutboundMessage) -> None:
outbound_received.append(msg)
bus.subscribe_outbound(capture_outbound)
mock_client = _make_mock_langgraph_client()
manager._client = mock_client
await manager.start()
meta = {
"sender_staff_id": "staff_001",
"conversation_type": "1",
"conversation_id": "conv_001",
}
inbound = InboundMessage(
channel_name="test",
chat_id="chat1",
user_id="user1",
text="hi",
metadata=meta,
)
await bus.publish_inbound(inbound)
await _wait_for(lambda: len(outbound_received) >= 1)
await manager.stop()
assert len(outbound_received) == 1
assert outbound_received[0].metadata == meta
_run(go())
def test_handle_chat_outbound_drops_large_metadata_keys(self):
"""Large metadata keys like raw_message should be stripped from outbound messages."""
from app.channels.manager import ChannelManager
async def go():
bus = MessageBus()
store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json")
manager = ChannelManager(bus=bus, store=store)
outbound_received: list[OutboundMessage] = []
async def capture_outbound(msg: OutboundMessage) -> None:
outbound_received.append(msg)
bus.subscribe_outbound(capture_outbound)
mock_client = _make_mock_langgraph_client()
manager._client = mock_client
await manager.start()
meta = {
"sender_staff_id": "staff_001",
"conversation_type": "1",
"raw_message": {"huge": "payload" * 1000},
"ref_msg": {"also": "large"},
}
inbound = InboundMessage(
channel_name="test",
chat_id="chat1",
user_id="user1",
text="hi",
metadata=meta,
)
await bus.publish_inbound(inbound)
await _wait_for(lambda: len(outbound_received) >= 1)
await manager.stop()
assert len(outbound_received) == 1
out_meta = outbound_received[0].metadata
assert "sender_staff_id" in out_meta
assert "conversation_type" in out_meta
assert "raw_message" not in out_meta
assert "ref_msg" not in out_meta
_run(go())
def test_handle_chat_uses_channel_session_overrides(self):
from app.channels.manager import ChannelManager

1554
backend/tests/test_dingtalk_channel.py Normal file
View File

File diff suppressed because it is too large Load Diff

15
backend/uv.lock generated
View File

@ -746,6 +746,7 @@ source = { virtual = "." }
dependencies = [
{ name = "bcrypt" },
{ name = "deerflow-harness" },
{ name = "dingtalk-stream" },
{ name = "email-validator" },
{ name = "fastapi" },
{ name = "httpx" },
@ -779,6 +780,7 @@ requires-dist = [
{ name = "bcrypt", specifier = ">=4.0.0" },
{ name = "deerflow-harness", editable = "packages/harness" },
{ name = "deerflow-harness", extras = ["postgres"], marker = "extra == 'postgres'", editable = "packages/harness" },
{ name = "dingtalk-stream", specifier = ">=0.24.3" },
{ name = "email-validator", specifier = ">=2.0.0" },
{ name = "fastapi", specifier = ">=0.115.0" },
{ name = "httpx", specifier = ">=0.28.0" },
@ -908,6 +910,19 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/07/6c/aa3f2f849e01cb6a001cd8554a88d4c77c5c1a31c95bdf1cf9301e6d9ef4/defusedxml-0.7.1-py2.py3-none-any.whl", hash = "sha256:a352e7e428770286cc899e2542b6cdaedb2b4953ff269a210103ec58f6198a61", size = 25604, upload-time = "2021-03-08T10:59:24.45Z" },
]
[[package]]
name = "dingtalk-stream"
version = "0.24.3"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "aiohttp" },
{ name = "requests" },
{ name = "websockets" },
]
wheels = [
{ url = "https://files.pythonhosted.org/packages/4c/44/102dede3f371277598df6aa9725b82e3add068c729333c7a5dbc12764579/dingtalk_stream-0.24.3-py3-none-any.whl", hash = "sha256:2160403656985962878bf60cdf5adf41619f21067348e06f07a7c7eebf5943ad", size = 27813, upload-time = "2025-10-24T09:36:57.497Z" },
]
[[package]]
name = "distro"
version = "1.9.0"

7
config.example.yaml
View File

@ -964,6 +964,13 @@ run_events:
# enabled: false
# bot_id: $WECOM_BOT_ID
# bot_secret: $WECOM_BOT_SECRET
#
# dingtalk:
# enabled: false
# client_id: $DINGTALK_CLIENT_ID
# client_secret: $DINGTALK_CLIENT_SECRET
# allowed_users: [] # empty = allow all
# card_template_id: "" # Optional: AI Card template ID for streaming updates
# ============================================================================
# Guardrails Configuration