Merge PR #219: add DingTalk channel support

README.md

@@ -16,7 +16,7 @@
 
 ⚡️ Delivers core agent functionality in just **~4,000** lines of code — **99% smaller** than Clawdbot's 430k+ lines.
 
-📏 Real-time line count: **3,423 lines** (run `bash core_agent_lines.sh` to verify anytime)
+📏 Real-time line count: **3,429 lines** (run `bash core_agent_lines.sh` to verify anytime)
 
 ## 📢 News
 
@@ -293,10 +293,6 @@ nanobot gateway
 
 Uses **WebSocket** long connection — no public IP required.
 
-```bash
-pip install nanobot-ai[feishu]
-```
-
 **1. Create a Feishu bot**
 - Visit [Feishu Open Platform](https://open.feishu.cn/app)
 - Create a new app → Enable **Bot** capability
@@ -337,6 +333,45 @@ nanobot gateway
 
 </details>
 
+<details>
+<summary><b>DingTalk (钉钉)</b></summary>
+
+Uses **Stream Mode** — no public IP required.
+
+**1. Create a DingTalk bot**
+- Visit [DingTalk Open Platform](https://open-dev.dingtalk.com/)
+- Create a new app -> Add **Robot** capability
+- **Configuration**:
+  - Toggle **Stream Mode** ON
+- **Permissions**: Add necessary permissions for sending messages
+- Get **AppKey** (Client ID) and **AppSecret** (Client Secret) from "Credentials"
+- Publish the app
+
+**2. Configure**
+
+```json
+{
+  "channels": {
+    "dingtalk": {
+      "enabled": true,
+      "clientId": "YOUR_APP_KEY",
+      "clientSecret": "YOUR_APP_SECRET",
+      "allowFrom": []
+    }
+  }
+}
+```
+
+> `allowFrom`: Leave empty to allow all users, or add `["staffId"]` to restrict access.
+
+**3. Run**
+
+```bash
+nanobot gateway
+```
+
+</details>
+
 ## ⚙️ Configuration
 
 Config file: `~/.nanobot/config.json`

nanobot/channels/dingtalk.py

@@ -0,0 +1,238 @@
+"""DingTalk/DingDing channel implementation using Stream Mode."""
+
+import asyncio
+import json
+import time
+from typing import Any
+
+from loguru import logger
+import httpx
+
+from nanobot.bus.events import OutboundMessage
+from nanobot.bus.queue import MessageBus
+from nanobot.channels.base import BaseChannel
+from nanobot.config.schema import DingTalkConfig
+
+try:
+    from dingtalk_stream import (
+        DingTalkStreamClient,
+        Credential,
+        CallbackHandler,
+        CallbackMessage,
+        AckMessage,
+    )
+    from dingtalk_stream.chatbot import ChatbotMessage
+
+    DINGTALK_AVAILABLE = True
+except ImportError:
+    DINGTALK_AVAILABLE = False
+    # Fallback so class definitions don't crash at module level
+    CallbackHandler = object  # type: ignore[assignment,misc]
+    CallbackMessage = None  # type: ignore[assignment,misc]
+    AckMessage = None  # type: ignore[assignment,misc]
+    ChatbotMessage = None  # type: ignore[assignment,misc]
+
+
+class NanobotDingTalkHandler(CallbackHandler):
+    """
+    Standard DingTalk Stream SDK Callback Handler.
+    Parses incoming messages and forwards them to the Nanobot channel.
+    """
+
+    def __init__(self, channel: "DingTalkChannel"):
+        super().__init__()
+        self.channel = channel
+
+    async def process(self, message: CallbackMessage):
+        """Process incoming stream message."""
+        try:
+            # Parse using SDK's ChatbotMessage for robust handling
+            chatbot_msg = ChatbotMessage.from_dict(message.data)
+
+            # Extract text content; fall back to raw dict if SDK object is empty
+            content = ""
+            if chatbot_msg.text:
+                content = chatbot_msg.text.content.strip()
+            if not content:
+                content = message.data.get("text", {}).get("content", "").strip()
+
+            if not content:
+                logger.warning(
+                    f"Received empty or unsupported message type: {chatbot_msg.message_type}"
+                )
+                return AckMessage.STATUS_OK, "OK"
+
+            sender_id = chatbot_msg.sender_staff_id or chatbot_msg.sender_id
+            sender_name = chatbot_msg.sender_nick or "Unknown"
+
+            logger.info(f"Received DingTalk message from {sender_name} ({sender_id}): {content}")
+
+            # Forward to Nanobot via _on_message (non-blocking).
+            # Store reference to prevent GC before task completes.
+            task = asyncio.create_task(
+                self.channel._on_message(content, sender_id, sender_name)
+            )
+            self.channel._background_tasks.add(task)
+            task.add_done_callback(self.channel._background_tasks.discard)
+
+            return AckMessage.STATUS_OK, "OK"
+
+        except Exception as e:
+            logger.error(f"Error processing DingTalk message: {e}")
+            # Return OK to avoid retry loop from DingTalk server
+            return AckMessage.STATUS_OK, "Error"
+
+
+class DingTalkChannel(BaseChannel):
+    """
+    DingTalk channel using Stream Mode.
+
+    Uses WebSocket to receive events via `dingtalk-stream` SDK.
+    Uses direct HTTP API to send messages (SDK is mainly for receiving).
+
+    Note: Currently only supports private (1:1) chat. Group messages are
+    received but replies are sent back as private messages to the sender.
+    """
+
+    name = "dingtalk"
+
+    def __init__(self, config: DingTalkConfig, bus: MessageBus):
+        super().__init__(config, bus)
+        self.config: DingTalkConfig = config
+        self._client: Any = None
+        self._http: httpx.AsyncClient | None = None
+
+        # Access Token management for sending messages
+        self._access_token: str | None = None
+        self._token_expiry: float = 0
+
+        # Hold references to background tasks to prevent GC
+        self._background_tasks: set[asyncio.Task] = set()
+
+    async def start(self) -> None:
+        """Start the DingTalk bot with Stream Mode."""
+        try:
+            if not DINGTALK_AVAILABLE:
+                logger.error(
+                    "DingTalk Stream SDK not installed. Run: pip install dingtalk-stream"
+                )
+                return
+
+            if not self.config.client_id or not self.config.client_secret:
+                logger.error("DingTalk client_id and client_secret not configured")
+                return
+
+            self._running = True
+            self._http = httpx.AsyncClient()
+
+            logger.info(
+                f"Initializing DingTalk Stream Client with Client ID: {self.config.client_id}..."
+            )
+            credential = Credential(self.config.client_id, self.config.client_secret)
+            self._client = DingTalkStreamClient(credential)
+
+            # Register standard handler
+            handler = NanobotDingTalkHandler(self)
+            self._client.register_callback_handler(ChatbotMessage.TOPIC, handler)
+
+            logger.info("DingTalk bot started with Stream Mode")
+
+            # client.start() is an async infinite loop handling the websocket connection
+            await self._client.start()
+
+        except Exception as e:
+            logger.exception(f"Failed to start DingTalk channel: {e}")
+
+    async def stop(self) -> None:
+        """Stop the DingTalk bot."""
+        self._running = False
+        # Close the shared HTTP client
+        if self._http:
+            await self._http.aclose()
+            self._http = None
+        # Cancel outstanding background tasks
+        for task in self._background_tasks:
+            task.cancel()
+        self._background_tasks.clear()
+
+    async def _get_access_token(self) -> str | None:
+        """Get or refresh Access Token."""
+        if self._access_token and time.time() < self._token_expiry:
+            return self._access_token
+
+        url = "https://api.dingtalk.com/v1.0/oauth2/accessToken"
+        data = {
+            "appKey": self.config.client_id,
+            "appSecret": self.config.client_secret,
+        }
+
+        if not self._http:
+            logger.warning("DingTalk HTTP client not initialized, cannot refresh token")
+            return None
+
+        try:
+            resp = await self._http.post(url, json=data)
+            resp.raise_for_status()
+            res_data = resp.json()
+            self._access_token = res_data.get("accessToken")
+            # Expire 60s early to be safe
+            self._token_expiry = time.time() + int(res_data.get("expireIn", 7200)) - 60
+            return self._access_token
+        except Exception as e:
+            logger.error(f"Failed to get DingTalk access token: {e}")
+            return None
+
+    async def send(self, msg: OutboundMessage) -> None:
+        """Send a message through DingTalk."""
+        token = await self._get_access_token()
+        if not token:
+            return
+
+        # oToMessages/batchSend: sends to individual users (private chat)
+        # https://open.dingtalk.com/document/orgapp/robot-batch-send-messages
+        url = "https://api.dingtalk.com/v1.0/robot/oToMessages/batchSend"
+
+        headers = {"x-acs-dingtalk-access-token": token}
+
+        data = {
+            "robotCode": self.config.client_id,
+            "userIds": [msg.chat_id],  # chat_id is the user's staffId
+            "msgKey": "sampleMarkdown",
+            "msgParam": json.dumps({
+                "text": msg.content,
+                "title": "Nanobot Reply",
+            }),
+        }
+
+        if not self._http:
+            logger.warning("DingTalk HTTP client not initialized, cannot send")
+            return
+
+        try:
+            resp = await self._http.post(url, json=data, headers=headers)
+            if resp.status_code != 200:
+                logger.error(f"DingTalk send failed: {resp.text}")
+            else:
+                logger.debug(f"DingTalk message sent to {msg.chat_id}")
+        except Exception as e:
+            logger.error(f"Error sending DingTalk message: {e}")
+
+    async def _on_message(self, content: str, sender_id: str, sender_name: str) -> None:
+        """Handle incoming message (called by NanobotDingTalkHandler).
+
+        Delegates to BaseChannel._handle_message() which enforces allow_from
+        permission checks before publishing to the bus.
+        """
+        try:
+            logger.info(f"DingTalk inbound: {content} from {sender_name}")
+            await self._handle_message(
+                sender_id=sender_id,
+                chat_id=sender_id,  # For private chat, chat_id == sender_id
+                content=str(content),
+                metadata={
+                    "sender_name": sender_name,
+                    "platform": "dingtalk",
+                },
+            )
+        except Exception as e:
+            logger.error(f"Error publishing DingTalk message: {e}")

nanobot/channels/manager.py

@@ -84,6 +84,17 @@ class ChannelManager:
                 logger.info("Feishu channel enabled")
             except ImportError as e:
                 logger.warning(f"Feishu channel not available: {e}")
+
+        # DingTalk channel
+        if self.config.channels.dingtalk.enabled:
+            try:
+                from nanobot.channels.dingtalk import DingTalkChannel
+                self.channels["dingtalk"] = DingTalkChannel(
+                    self.config.channels.dingtalk, self.bus
+                )
+                logger.info("DingTalk channel enabled")
+            except ImportError as e:
+                logger.warning(f"DingTalk channel not available: {e}")
     
     async def _start_channel(self, name: str, channel: BaseChannel) -> None:
         """Start a channel and log any exceptions."""

nanobot/config/schema.py

@@ -30,6 +30,14 @@ class FeishuConfig(BaseModel):
     allow_from: list[str] = Field(default_factory=list)  # Allowed user open_ids
 
 
+class DingTalkConfig(BaseModel):
+    """DingTalk channel configuration using Stream mode."""
+    enabled: bool = False
+    client_id: str = ""  # AppKey
+    client_secret: str = ""  # AppSecret
+    allow_from: list[str] = Field(default_factory=list)  # Allowed staff_ids
+
+
 class DiscordConfig(BaseModel):
     """Discord channel configuration."""
     enabled: bool = False
@@ -45,6 +53,7 @@ class ChannelsConfig(BaseModel):
     telegram: TelegramConfig = Field(default_factory=TelegramConfig)
     discord: DiscordConfig = Field(default_factory=DiscordConfig)
     feishu: FeishuConfig = Field(default_factory=FeishuConfig)
+    dingtalk: DingTalkConfig = Field(default_factory=DingTalkConfig)
 
 
 class AgentDefaults(BaseModel):

pyproject.toml

@@ -28,6 +28,7 @@ dependencies = [
     "readability-lxml>=0.8.0",
     "rich>=13.0.0",
     "croniter>=2.0.0",
+    "dingtalk-stream>=0.4.0",
     "python-telegram-bot[socks]>=21.0",
     "lark-oapi>=1.0.0",
     "socksio>=1.0.0",