Hua/nanobot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

merge: resolve PR #1966 conflicts with main

Browse Source
This commit is contained in:
Xubin Ren
2026-03-14 15:32:19 +00:00
parent a8fbea6a95 f1a82c0165
commit 44af7eca3f
38 changed files with 2163 additions and 320 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -5,7 +5,6 @@
*.pyc
dist/
build/
docs/
*.egg-info/
*.egg
*.pycs

32
README.md
View File

@@ -216,7 +216,9 @@ That's it! You have a working AI assistant in 2 minutes.
## 💬 Chat Apps
Connect nanobot to your favorite chat platform.
Connect nanobot to your favorite chat platform. Want to build your own? See the [Channel Plugin Guide](.docs/CHANNEL_PLUGIN_GUIDE.md).
> Channel plugin support is available in the `main` branch; not yet published to PyPI.
| Channel | What you need |
|---------|---------------|
@@ -544,6 +546,7 @@ Uses **botpy SDK** with WebSocket — no public IP required. Currently supports
**3. Configure**
> - `allowFrom`: Add your openid (find it in nanobot logs when you message the bot). Use `["*"]` for public access.
> - `msgFormat`: Optional. Use `"plain"` (default) for maximum compatibility with legacy QQ clients, or `"markdown"` for richer formatting on newer clients.
> - For production: submit a review in the bot console and publish. See [QQ Bot Docs](https://bot.q.qq.com/wiki/) for the full publishing flow.
```json
@@ -553,7 +556,8 @@ Uses **botpy SDK** with WebSocket — no public IP required. Currently supports
"enabled": true,
"appId": "YOUR_APP_ID",
"secret": "YOUR_APP_SECRET",
"allowFrom": ["YOUR_OPENID"]
"allowFrom": ["YOUR_OPENID"],
"msgFormat": "plain"
}
}
}
@@ -1108,6 +1112,28 @@ Use `toolTimeout` to override the default 30s per-call timeout for slow servers:
}
```
Use `enabledTools` to register only a subset of tools from an MCP server:
```json
{
"tools": {
"mcpServers": {
"filesystem": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/dir"],
"enabledTools": ["read_file", "mcp_filesystem_write_file"]
}
}
}
}
```
`enabledTools` accepts either the raw MCP tool name (for example `read_file`) or the wrapped nanobot tool name (for example `mcp_filesystem_write_file`).
- Omit `enabledTools`, or set it to `["*"]`, to register all tools.
- Set `enabledTools` to `[]` to register no tools from that server.
- Set `enabledTools` to a non-empty list of names to register only that subset.
MCP tools are automatically discovered and registered on startup. The LLM can use them alongside built-in tools — no extra configuration needed.
@@ -1370,7 +1396,7 @@ nanobot/
│ ├── subagent.py # Background task execution
│ └── tools/ # Built-in tools (incl. spawn)
├── skills/ # 🎯 Bundled skills (github, weather, tmux...)
├── channels/ # 📱 Chat channel integrations
├── channels/ # 📱 Chat channel integrations (supports plugins)
├── bus/ # 🚌 Message routing
├── cron/ # ⏰ Scheduled tasks
├── heartbeat/ # 💓 Proactive wake-up

254
docs/CHANNEL_PLUGIN_GUIDE.md Normal file
View File

@@ -0,0 +1,254 @@
# Channel Plugin Guide
Build a custom nanobot channel in three steps: subclass, package, install.
## How It Works
nanobot discovers channel plugins via Python [entry points](https://packaging.python.org/en/latest/specifications/entry-points/). When `nanobot gateway` starts, it scans:
1. Built-in channels in `nanobot/channels/`
2. External packages registered under the `nanobot.channels` entry point group
If a matching config section has `"enabled": true`, the channel is instantiated and started.
## Quick Start
We'll build a minimal webhook channel that receives messages via HTTP POST and sends replies back.
### Project Structure
```
nanobot-channel-webhook/
├── nanobot_channel_webhook/
│ ├── __init__.py # re-export WebhookChannel
│ └── channel.py # channel implementation
└── pyproject.toml
```
### 1. Create Your Channel
```python
# nanobot_channel_webhook/__init__.py
from nanobot_channel_webhook.channel import WebhookChannel
__all__ = ["WebhookChannel"]
```
```python
# nanobot_channel_webhook/channel.py
import asyncio
from typing import Any
from aiohttp import web
from loguru import logger
from nanobot.channels.base import BaseChannel
from nanobot.bus.events import OutboundMessage
class WebhookChannel(BaseChannel):
name = "webhook"
display_name = "Webhook"
@classmethod
def default_config(cls) -> dict[str, Any]:
return {"enabled": False, "port": 9000, "allowFrom": []}
async def start(self) -> None:
"""Start an HTTP server that listens for incoming messages.
IMPORTANT: start() must block forever (or until stop() is called).
If it returns, the channel is considered dead.
"""
self._running = True
port = self.config.get("port", 9000)
app = web.Application()
app.router.add_post("/message", self._on_request)
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, "0.0.0.0", port)
await site.start()
logger.info("Webhook listening on :{}", port)
# Block until stopped
while self._running:
await asyncio.sleep(1)
await runner.cleanup()
async def stop(self) -> None:
self._running = False
async def send(self, msg: OutboundMessage) -> None:
"""Deliver an outbound message.
msg.content — markdown text (convert to platform format as needed)
msg.media — list of local file paths to attach
msg.chat_id — the recipient (same chat_id you passed to _handle_message)
msg.metadata — may contain "_progress": True for streaming chunks
"""
logger.info("[webhook] -> {}: {}", msg.chat_id, msg.content[:80])
# In a real plugin: POST to a callback URL, send via SDK, etc.
async def _on_request(self, request: web.Request) -> web.Response:
"""Handle an incoming HTTP POST."""
body = await request.json()
sender = body.get("sender", "unknown")
chat_id = body.get("chat_id", sender)
text = body.get("text", "")
media = body.get("media", []) # list of URLs
# This is the key call: validates allowFrom, then puts the
# message onto the bus for the agent to process.
await self._handle_message(
sender_id=sender,
chat_id=chat_id,
content=text,
media=media,
)
return web.json_response({"ok": True})
```
### 2. Register the Entry Point
```toml
# pyproject.toml
[project]
name = "nanobot-channel-webhook"
version = "0.1.0"
dependencies = ["nanobot", "aiohttp"]
[project.entry-points."nanobot.channels"]
webhook = "nanobot_channel_webhook:WebhookChannel"
[build-system]
requires = ["setuptools"]
build-backend = "setuptools.backends._legacy:_Backend"
```
The key (`webhook`) becomes the config section name. The value points to your `BaseChannel` subclass.
### 3. Install & Configure
```bash
pip install -e .
nanobot plugins list # verify "Webhook" shows as "plugin"
nanobot onboard # auto-adds default config for detected plugins
```
Edit `~/.nanobot/config.json`:
```json
{
"channels": {
"webhook": {
"enabled": true,
"port": 9000,
"allowFrom": ["*"]
}
}
}
```
### 4. Run & Test
```bash
nanobot gateway
```
In another terminal:
```bash
curl -X POST http://localhost:9000/message \
-H "Content-Type: application/json" \
-d '{"sender": "user1", "chat_id": "user1", "text": "Hello!"}'
```
The agent receives the message and processes it. Replies arrive in your `send()` method.
## BaseChannel API
### Required (abstract)
| Method | Description |
|--------|-------------|
| `async start()` | **Must block forever.** Connect to platform, listen for messages, call `_handle_message()` on each. If this returns, the channel is dead. |
| `async stop()` | Set `self._running = False` and clean up. Called when gateway shuts down. |
| `async send(msg: OutboundMessage)` | Deliver an outbound message to the platform. |
### Provided by Base
| Method / Property | Description |
|-------------------|-------------|
| `_handle_message(sender_id, chat_id, content, media?, metadata?, session_key?)` | **Call this when you receive a message.** Checks `is_allowed()`, then publishes to the bus. |
| `is_allowed(sender_id)` | Checks against `config["allowFrom"]`; `"*"` allows all, `[]` denies all. |
| `default_config()` (classmethod) | Returns default config dict for `nanobot onboard`. Override to declare your fields. |
| `transcribe_audio(file_path)` | Transcribes audio via Groq Whisper (if configured). |
| `is_running` | Returns `self._running`. |
### Message Types
```python
@dataclass
class OutboundMessage:
channel: str # your channel name
chat_id: str # recipient (same value you passed to _handle_message)
content: str # markdown text — convert to platform format as needed
media: list[str] # local file paths to attach (images, audio, docs)
metadata: dict # may contain: "_progress" (bool) for streaming chunks,
# "message_id" for reply threading
```
## Config
Your channel receives config as a plain `dict`. Access fields with `.get()`:
```python
async def start(self) -> None:
port = self.config.get("port", 9000)
token = self.config.get("token", "")
```
`allowFrom` is handled automatically by `_handle_message()` — you don't need to check it yourself.
Override `default_config()` so `nanobot onboard` auto-populates `config.json`:
```python
@classmethod
def default_config(cls) -> dict[str, Any]:
return {"enabled": False, "port": 9000, "allowFrom": []}
```
If not overridden, the base class returns `{"enabled": false}`.
## Naming Convention
| What | Format | Example |
|------|--------|---------|
| PyPI package | `nanobot-channel-{name}` | `nanobot-channel-webhook` |
| Entry point key | `{name}` | `webhook` |
| Config section | `channels.{name}` | `channels.webhook` |
| Python package | `nanobot_channel_{name}` | `nanobot_channel_webhook` |
## Local Development
```bash
git clone https://github.com/you/nanobot-channel-webhook
cd nanobot-channel-webhook
pip install -e .
nanobot plugins list # should show "Webhook" as "plugin"
nanobot gateway # test end-to-end
```
## Verify
```bash
$ nanobot plugins list
Name Source Enabled
telegram builtin yes
discord builtin no
webhook plugin yes
```

3
nanobot/agent/loop.py
View File

@@ -266,6 +266,9 @@ class AgentLoop:
msg = await asyncio.wait_for(self.bus.consume_inbound(), timeout=1.0)
except asyncio.TimeoutError:
continue
except Exception as e:
logger.warning("Error consuming inbound message: {}, continuing...", e)
continue
cmd = msg.content.strip().lower()
if cmd == "/stop":

38
nanobot/agent/tools/mcp.py
View File

@@ -138,11 +138,47 @@ async def connect_mcp_servers(
await session.initialize()
tools = await session.list_tools()
enabled_tools = set(cfg.enabled_tools)
allow_all_tools = "*" in enabled_tools
registered_count = 0
matched_enabled_tools: set[str] = set()
available_raw_names = [tool_def.name for tool_def in tools.tools]
available_wrapped_names = [f"mcp_{name}_{tool_def.name}" for tool_def in tools.tools]
for tool_def in tools.tools:
wrapped_name = f"mcp_{name}_{tool_def.name}"
if (
not allow_all_tools
and tool_def.name not in enabled_tools
and wrapped_name not in enabled_tools
):
logger.debug(
"MCP: skipping tool '{}' from server '{}' (not in enabledTools)",
wrapped_name,
name,
)
continue
wrapper = MCPToolWrapper(session, name, tool_def, tool_timeout=cfg.tool_timeout)
registry.register(wrapper)
logger.debug("MCP: registered tool '{}' from server '{}'", wrapper.name, name)
registered_count += 1
if enabled_tools:
if tool_def.name in enabled_tools:
matched_enabled_tools.add(tool_def.name)
if wrapped_name in enabled_tools:
matched_enabled_tools.add(wrapped_name)
logger.info("MCP server '{}': connected, {} tools registered", name, len(tools.tools))
if enabled_tools and not allow_all_tools:
unmatched_enabled_tools = sorted(enabled_tools - matched_enabled_tools)
if unmatched_enabled_tools:
logger.warning(
"MCP server '{}': enabledTools entries not found: {}. Available raw names: {}. "
"Available wrapped names: {}",
name,
", ".join(unmatched_enabled_tools),
", ".join(available_raw_names) or "(none)",
", ".join(available_wrapped_names) or "(none)",
)
logger.info("MCP server '{}': connected, {} tools registered", name, registered_count)
except Exception as e:
logger.error("MCP server '{}': failed to connect: {}", name, e)

5
nanobot/channels/base.py
View File

@@ -128,6 +128,11 @@ class BaseChannel(ABC):
await self.bus.publish_inbound(msg)
@classmethod
def default_config(cls) -> dict[str, Any]:
"""Return default config for onboard. Override in plugins to auto-populate config.json."""
return {"enabled": False}
@property
def is_running(self) -> bool:
"""Check if the channel is running."""

20
nanobot/channels/dingtalk.py
View File

@@ -11,11 +11,12 @@ from urllib.parse import unquote, urlparse
import httpx
from loguru import logger
from pydantic import Field
from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.base import BaseChannel
from nanobot.config.schema import DingTalkConfig
from nanobot.config.schema import Base
try:
from dingtalk_stream import (
@@ -102,6 +103,15 @@ class NanobotDingTalkHandler(CallbackHandler):
return AckMessage.STATUS_OK, "Error"
class DingTalkConfig(Base):
"""DingTalk channel configuration using Stream mode."""
enabled: bool = False
client_id: str = ""
client_secret: str = ""
allow_from: list[str] = Field(default_factory=list)
class DingTalkChannel(BaseChannel):
"""
DingTalk channel using Stream Mode.
@@ -119,7 +129,13 @@ class DingTalkChannel(BaseChannel):
_AUDIO_EXTS = {".amr", ".mp3", ".wav", ".ogg", ".m4a", ".aac"}
_VIDEO_EXTS = {".mp4", ".mov", ".avi", ".mkv", ".webm"}
def __init__(self, config: DingTalkConfig, bus: MessageBus):
@classmethod
def default_config(cls) -> dict[str, Any]:
return DingTalkConfig().model_dump(by_alias=True)
def __init__(self, config: Any, bus: MessageBus):
if isinstance(config, dict):
config = DingTalkConfig.model_validate(config)
super().__init__(config, bus)
self.config: DingTalkConfig = config
self._client: Any = None

24
nanobot/channels/discord.py
View File

@@ -3,9 +3,10 @@
import asyncio
import json
from pathlib import Path
from typing import Any
from typing import Any, Literal
import httpx
from pydantic import Field
import websockets
from loguru import logger
@@ -13,7 +14,7 @@ from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.base import BaseChannel
from nanobot.config.paths import get_media_dir
from nanobot.config.schema import DiscordConfig
from nanobot.config.schema import Base
from nanobot.utils.helpers import split_message
DISCORD_API_BASE = "https://discord.com/api/v10"
@@ -21,13 +22,30 @@ MAX_ATTACHMENT_BYTES = 20 * 1024 * 1024 # 20MB
MAX_MESSAGE_LEN = 2000 # Discord message character limit
class DiscordConfig(Base):
"""Discord channel configuration."""
enabled: bool = False
token: str = ""
allow_from: list[str] = Field(default_factory=list)
gateway_url: str = "wss://gateway.discord.gg/?v=10&encoding=json"
intents: int = 37377
group_policy: Literal["mention", "open"] = "mention"
class DiscordChannel(BaseChannel):
"""Discord channel using Gateway websocket."""
name = "discord"
display_name = "Discord"
def __init__(self, config: DiscordConfig, bus: MessageBus):
@classmethod
def default_config(cls) -> dict[str, Any]:
return DiscordConfig().model_dump(by_alias=True)
def __init__(self, config: Any, bus: MessageBus):
if isinstance(config, dict):
config = DiscordConfig.model_validate(config)
super().__init__(config, bus)
self.config: DiscordConfig = config
self._ws: websockets.WebSocketClientProtocol | None = None

40
nanobot/channels/email.py
View File

@@ -15,11 +15,41 @@ from email.utils import parseaddr
from typing import Any
from loguru import logger
from pydantic import Field
from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.base import BaseChannel
from nanobot.config.schema import EmailConfig
from nanobot.config.schema import Base
class EmailConfig(Base):
"""Email channel configuration (IMAP inbound + SMTP outbound)."""
enabled: bool = False
consent_granted: bool = False
imap_host: str = ""
imap_port: int = 993
imap_username: str = ""
imap_password: str = ""
imap_mailbox: str = "INBOX"
imap_use_ssl: bool = True
smtp_host: str = ""
smtp_port: int = 587
smtp_username: str = ""
smtp_password: str = ""
smtp_use_tls: bool = True
smtp_use_ssl: bool = False
from_address: str = ""
auto_reply_enabled: bool = True
poll_interval_seconds: int = 30
mark_seen: bool = True
max_body_chars: int = 12000
subject_prefix: str = "Re: "
allow_from: list[str] = Field(default_factory=list)
class EmailChannel(BaseChannel):
@@ -51,7 +81,13 @@ class EmailChannel(BaseChannel):
"Dec",
)
def __init__(self, config: EmailConfig, bus: MessageBus):
@classmethod
def default_config(cls) -> dict[str, Any]:
return EmailConfig().model_dump(by_alias=True)
def __init__(self, config: Any, bus: MessageBus):
if isinstance(config, dict):
config = EmailConfig.model_validate(config)
super().__init__(config, bus)
self.config: EmailConfig = config
self._last_subject_by_chat: dict[str, str] = {}

165
nanobot/channels/feishu.py
View File

@@ -7,7 +7,7 @@ import re
import threading
from collections import OrderedDict
from pathlib import Path
from typing import Any
from typing import Any, Literal
from loguru import logger
@@ -15,7 +15,8 @@ from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.base import BaseChannel
from nanobot.config.paths import get_media_dir
from nanobot.config.schema import FeishuConfig
from nanobot.config.schema import Base
from pydantic import Field
import importlib.util
@@ -231,6 +232,20 @@ def _extract_post_text(content_json: dict) -> str:
return text
class FeishuConfig(Base):
"""Feishu/Lark channel configuration using WebSocket long connection."""
enabled: bool = False
app_id: str = ""
app_secret: str = ""
encrypt_key: str = ""
verification_token: str = ""
allow_from: list[str] = Field(default_factory=list)
react_emoji: str = "THUMBSUP"
group_policy: Literal["open", "mention"] = "mention"
reply_to_message: bool = False # If True, bot replies quote the user's original message
class FeishuChannel(BaseChannel):
"""
Feishu/Lark channel using WebSocket long connection.
@@ -246,7 +261,13 @@ class FeishuChannel(BaseChannel):
name = "feishu"
display_name = "Feishu"
def __init__(self, config: FeishuConfig, bus: MessageBus):
@classmethod
def default_config(cls) -> dict[str, Any]:
return FeishuConfig().model_dump(by_alias=True)
def __init__(self, config: Any, bus: MessageBus):
if isinstance(config, dict):
config = FeishuConfig.model_validate(config)
super().__init__(config, bus)
self.config: FeishuConfig = config
self._client: Any = None
@@ -786,6 +807,77 @@ class FeishuChannel(BaseChannel):
return None, f"[{msg_type}: download failed]"
_REPLY_CONTEXT_MAX_LEN = 200
def _get_message_content_sync(self, message_id: str) -> str | None:
"""Fetch the text content of a Feishu message by ID (synchronous).
Returns a "[Reply to: ...]" context string, or None on failure.
"""
from lark_oapi.api.im.v1 import GetMessageRequest
try:
request = GetMessageRequest.builder().message_id(message_id).build()
response = self._client.im.v1.message.get(request)
if not response.success():
logger.debug(
"Feishu: could not fetch parent message {}: code={}, msg={}",
message_id, response.code, response.msg,
)
return None
items = getattr(response.data, "items", None)
if not items:
return None
msg_obj = items[0]
raw_content = getattr(msg_obj, "body", None)
raw_content = getattr(raw_content, "content", None) if raw_content else None
if not raw_content:
return None
try:
content_json = json.loads(raw_content)
except (json.JSONDecodeError, TypeError):
return None
msg_type = getattr(msg_obj, "msg_type", "")
if msg_type == "text":
text = content_json.get("text", "").strip()
elif msg_type == "post":
text, _ = _extract_post_content(content_json)
text = text.strip()
else:
text = ""
if not text:
return None
if len(text) > self._REPLY_CONTEXT_MAX_LEN:
text = text[: self._REPLY_CONTEXT_MAX_LEN] + "..."
return f"[Reply to: {text}]"
except Exception as e:
logger.debug("Feishu: error fetching parent message {}: {}", message_id, e)
return None
def _reply_message_sync(self, parent_message_id: str, msg_type: str, content: str) -> bool:
"""Reply to an existing Feishu message using the Reply API (synchronous)."""
from lark_oapi.api.im.v1 import ReplyMessageRequest, ReplyMessageRequestBody
try:
request = ReplyMessageRequest.builder() \
.message_id(parent_message_id) \
.request_body(
ReplyMessageRequestBody.builder()
.msg_type(msg_type)
.content(content)
.build()
).build()
response = self._client.im.v1.message.reply(request)
if not response.success():
logger.error(
"Failed to reply to Feishu message {}: code={}, msg={}, log_id={}",
parent_message_id, response.code, response.msg, response.get_log_id()
)
return False
logger.debug("Feishu reply sent to message {}", parent_message_id)
return True
except Exception as e:
logger.error("Error replying to Feishu message {}: {}", parent_message_id, e)
return False
def _send_message_sync(self, receive_id_type: str, receive_id: str, msg_type: str, content: str) -> bool:
"""Send a single message (text/image/file/interactive) synchronously."""
from lark_oapi.api.im.v1 import CreateMessageRequest, CreateMessageRequestBody
@@ -822,12 +914,38 @@ class FeishuChannel(BaseChannel):
receive_id_type = "chat_id" if msg.chat_id.startswith("oc_") else "open_id"
loop = asyncio.get_running_loop()
# Handle tool hint messages as code blocks in interactive cards
# Handle tool hint messages as code blocks in interactive cards.
# These are progress-only messages and should bypass normal reply routing.
if msg.metadata.get("_tool_hint"):
if msg.content and msg.content.strip():
await self._send_tool_hint_card(receive_id_type, msg.chat_id, msg.content.strip())
await self._send_tool_hint_card(
receive_id_type, msg.chat_id, msg.content.strip()
)
return
# Determine whether the first message should quote the user's message.
# Only the very first send (media or text) in this call uses reply; subsequent
# chunks/media fall back to plain create to avoid redundant quote bubbles.
reply_message_id: str | None = None
if (
self.config.reply_to_message
and not msg.metadata.get("_progress", False)
):
reply_message_id = msg.metadata.get("message_id") or None
first_send = True # tracks whether the reply has already been used
def _do_send(m_type: str, content: str) -> None:
"""Send via reply (first message) or create (subsequent)."""
nonlocal first_send
if reply_message_id and first_send:
first_send = False
ok = self._reply_message_sync(reply_message_id, m_type, content)
if ok:
return
# Fall back to regular send if reply fails
self._send_message_sync(receive_id_type, msg.chat_id, m_type, content)
for file_path in msg.media:
if not os.path.isfile(file_path):
logger.warning("Media file not found: {}", file_path)
@@ -837,8 +955,8 @@ class FeishuChannel(BaseChannel):
key = await loop.run_in_executor(None, self._upload_image_sync, file_path)
if key:
await loop.run_in_executor(
None, self._send_message_sync,
receive_id_type, msg.chat_id, "image", json.dumps({"image_key": key}, ensure_ascii=False),
None, _do_send,
"image", json.dumps({"image_key": key}, ensure_ascii=False),
)
else:
key = await loop.run_in_executor(None, self._upload_file_sync, file_path)
@@ -850,8 +968,8 @@ class FeishuChannel(BaseChannel):
else:
media_type = "file"
await loop.run_in_executor(
None, self._send_message_sync,
receive_id_type, msg.chat_id, media_type, json.dumps({"file_key": key}, ensure_ascii=False),
None, _do_send,
media_type, json.dumps({"file_key": key}, ensure_ascii=False),
)
if msg.content and msg.content.strip():
@@ -860,18 +978,12 @@ class FeishuChannel(BaseChannel):
if fmt == "text":
# Short plain text – send as simple text message
text_body = json.dumps({"text": msg.content.strip()}, ensure_ascii=False)
await loop.run_in_executor(
None, self._send_message_sync,
receive_id_type, msg.chat_id, "text", text_body,
)
await loop.run_in_executor(None, _do_send, "text", text_body)
elif fmt == "post":
# Medium content with links – send as rich-text post
post_body = self._markdown_to_post(msg.content)
await loop.run_in_executor(
None, self._send_message_sync,
receive_id_type, msg.chat_id, "post", post_body,
)
await loop.run_in_executor(None, _do_send, "post", post_body)
else:
# Complex / long content – send as interactive card
@@ -879,8 +991,8 @@ class FeishuChannel(BaseChannel):
for chunk in self._split_elements_by_table_limit(elements):
card = {"config": {"wide_screen_mode": True}, "elements": chunk}
await loop.run_in_executor(
None, self._send_message_sync,
receive_id_type, msg.chat_id, "interactive", json.dumps(card, ensure_ascii=False),
None, _do_send,
"interactive", json.dumps(card, ensure_ascii=False),
)
except Exception as e:
@@ -975,6 +1087,19 @@ class FeishuChannel(BaseChannel):
else:
content_parts.append(MSG_TYPE_MAP.get(msg_type, f"[{msg_type}]"))
# Extract reply context (parent/root message IDs)
parent_id = getattr(message, "parent_id", None) or None
root_id = getattr(message, "root_id", None) or None
# Prepend quoted message text when the user replied to another message
if parent_id and self._client:
loop = asyncio.get_running_loop()
reply_ctx = await loop.run_in_executor(
None, self._get_message_content_sync, parent_id
)
if reply_ctx:
content_parts.insert(0, reply_ctx)
content = "\n".join(content_parts) if content_parts else ""
if not content and not media_paths:
@@ -991,6 +1116,8 @@ class FeishuChannel(BaseChannel):
"message_id": message_id,
"chat_type": chat_type,
"msg_type": msg_type,
"parent_id": parent_id,
"root_id": root_id,
}
)

24
nanobot/channels/manager.py
View File

@@ -31,23 +31,29 @@ class ChannelManager:
self._init_channels()
def _init_channels(self) -> None:
"""Initialize channels discovered via pkgutil scan."""
from nanobot.channels.registry import discover_channel_names, load_channel_class
"""Initialize channels discovered via pkgutil scan + entry_points plugins."""
from nanobot.channels.registry import discover_all
groq_key = self.config.providers.groq.api_key
for modname in discover_channel_names():
section = getattr(self.config.channels, modname, None)
if not section or not getattr(section, "enabled", False):
for name, cls in discover_all().items():
section = getattr(self.config.channels, name, None)
if section is None:
continue
enabled = (
section.get("enabled", False)
if isinstance(section, dict)
else getattr(section, "enabled", False)
)
if not enabled:
continue
try:
cls = load_channel_class(modname)
channel = cls(section, self.bus)
channel.transcription_api_key = groq_key
self.channels[modname] = channel
self.channels[name] = channel
logger.info("{} channel enabled", cls.display_name)
except ImportError as e:
logger.warning("{} channel not available: {}", modname, e)
except Exception as e:
logger.warning("{} channel not available: {}", name, e)
self._validate_allow_from()

27
nanobot/channels/matrix.py
View File

@@ -4,9 +4,10 @@ import asyncio
import logging
import mimetypes
from pathlib import Path
from typing import Any, TypeAlias
from typing import Any, Literal, TypeAlias
from loguru import logger
from pydantic import Field
try:
import nh3
@@ -40,6 +41,7 @@ from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.base import BaseChannel
from nanobot.config.paths import get_data_dir, get_media_dir
from nanobot.config.schema import Base
from nanobot.utils.helpers import safe_filename
TYPING_NOTICE_TIMEOUT_MS = 30_000
@@ -143,12 +145,33 @@ def _configure_nio_logging_bridge() -> None:
nio_logger.propagate = False
class MatrixConfig(Base):
"""Matrix (Element) channel configuration."""
enabled: bool = False
homeserver: str = "https://matrix.org"
access_token: str = ""
user_id: str = ""
device_id: str = ""
e2ee_enabled: bool = True
sync_stop_grace_seconds: int = 2
max_media_bytes: int = 20 * 1024 * 1024
allow_from: list[str] = Field(default_factory=list)
group_policy: Literal["open", "mention", "allowlist"] = "open"
group_allow_from: list[str] = Field(default_factory=list)
allow_room_mentions: bool = False
class MatrixChannel(BaseChannel):
"""Matrix (Element) channel using long-polling sync."""
name = "matrix"
display_name = "Matrix"
@classmethod
def default_config(cls) -> dict[str, Any]:
return MatrixConfig().model_dump(by_alias=True)
def __init__(
self,
config: Any,
@@ -157,6 +180,8 @@ class MatrixChannel(BaseChannel):
restrict_to_workspace: bool = False,
workspace: str | Path | None = None,
):
if isinstance(config, dict):
config = MatrixConfig.model_validate(config)
super().__init__(config, bus)
self.client: AsyncClient | None = None
self._sync_task: asyncio.Task | None = None

54
nanobot/channels/mochat.py
View File

@@ -16,7 +16,8 @@ from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.base import BaseChannel
from nanobot.config.paths import get_runtime_subdir
from nanobot.config.schema import MochatConfig
from nanobot.config.schema import Base
from pydantic import Field
try:
import socketio
@@ -208,6 +209,49 @@ def parse_timestamp(value: Any) -> int | None:
return None
# ---------------------------------------------------------------------------
# Config classes
# ---------------------------------------------------------------------------
class MochatMentionConfig(Base):
"""Mochat mention behavior configuration."""
require_in_groups: bool = False
class MochatGroupRule(Base):
"""Mochat per-group mention requirement."""
require_mention: bool = False
class MochatConfig(Base):
"""Mochat channel configuration."""
enabled: bool = False
base_url: str = "https://mochat.io"
socket_url: str = ""
socket_path: str = "/socket.io"
socket_disable_msgpack: bool = False
socket_reconnect_delay_ms: int = 1000
socket_max_reconnect_delay_ms: int = 10000
socket_connect_timeout_ms: int = 10000
refresh_interval_ms: int = 30000
watch_timeout_ms: int = 25000
watch_limit: int = 100
retry_delay_ms: int = 500
max_retry_attempts: int = 0
claw_token: str = ""
agent_user_id: str = ""
sessions: list[str] = Field(default_factory=list)
panels: list[str] = Field(default_factory=list)
allow_from: list[str] = Field(default_factory=list)
mention: MochatMentionConfig = Field(default_factory=MochatMentionConfig)
groups: dict[str, MochatGroupRule] = Field(default_factory=dict)
reply_delay_mode: str = "non-mention"
reply_delay_ms: int = 120000
# ---------------------------------------------------------------------------
# Channel
# ---------------------------------------------------------------------------
@@ -218,7 +262,13 @@ class MochatChannel(BaseChannel):
name = "mochat"
display_name = "Mochat"
def __init__(self, config: MochatConfig, bus: MessageBus):
@classmethod
def default_config(cls) -> dict[str, Any]:
return MochatConfig().model_dump(by_alias=True)
def __init__(self, config: Any, bus: MessageBus):
if isinstance(config, dict):
config = MochatConfig.model_validate(config)
super().__init__(config, bus)
self.config: MochatConfig = config
self._http: httpx.AsyncClient | None = None

48
nanobot/channels/qq.py
View File

@@ -2,14 +2,15 @@
import asyncio
from collections import deque
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Any, Literal
from loguru import logger
from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.base import BaseChannel
from nanobot.config.schema import QQConfig
from nanobot.config.schema import Base
from pydantic import Field
try:
import botpy
@@ -50,13 +51,29 @@ def _make_bot_class(channel: "QQChannel") -> "type[botpy.Client]":
return _Bot
class QQConfig(Base):
"""QQ channel configuration using botpy SDK."""
enabled: bool = False
app_id: str = ""
secret: str = ""
allow_from: list[str] = Field(default_factory=list)
msg_format: Literal["plain", "markdown"] = "plain"
class QQChannel(BaseChannel):
"""QQ channel using botpy SDK with WebSocket connection."""
name = "qq"
display_name = "QQ"
def __init__(self, config: QQConfig, bus: MessageBus):
@classmethod
def default_config(cls) -> dict[str, Any]:
return QQConfig().model_dump(by_alias=True)
def __init__(self, config: Any, bus: MessageBus):
if isinstance(config, dict):
config = QQConfig.model_validate(config)
super().__init__(config, bus)
self.config: QQConfig = config
self._client: "botpy.Client | None" = None
@@ -110,22 +127,27 @@ class QQChannel(BaseChannel):
try:
msg_id = msg.metadata.get("message_id")
self._msg_seq += 1
msg_type = self._chat_type_cache.get(msg.chat_id, "c2c")
if msg_type == "group":
use_markdown = self.config.msg_format == "markdown"
payload: dict[str, Any] = {
"msg_type": 2 if use_markdown else 0,
"msg_id": msg_id,
"msg_seq": self._msg_seq,
}
if use_markdown:
payload["markdown"] = {"content": msg.content}
else:
payload["content"] = msg.content
chat_type = self._chat_type_cache.get(msg.chat_id, "c2c")
if chat_type == "group":
await self._client.api.post_group_message(
group_openid=msg.chat_id,
msg_type=0,
content=msg.content,
msg_id=msg_id,
msg_seq=self._msg_seq,
**payload,
)
else:
await self._client.api.post_c2c_message(
openid=msg.chat_id,
msg_type=0,
content=msg.content,
msg_id=msg_id,
msg_seq=self._msg_seq,
**payload,
)
except Exception as e:
logger.error("Error sending QQ message: {}", e)

40
nanobot/channels/registry.py
View File

@@ -1,4 +1,4 @@
"""Auto-discovery for channel modules — no hardcoded registry."""
"""Auto-discovery for built-in channel modules and external plugins."""
from __future__ import annotations
@@ -6,6 +6,8 @@ import importlib
import pkgutil
from typing import TYPE_CHECKING
from loguru import logger
if TYPE_CHECKING:
from nanobot.channels.base import BaseChannel
@@ -13,7 +15,7 @@ _INTERNAL = frozenset({"base", "manager", "registry"})
def discover_channel_names() -> list[str]:
"""Return all channel module names by scanning the package (zero imports)."""
"""Return all built-in channel module names by scanning the package (zero imports)."""
import nanobot.channels as pkg
return [
@@ -33,3 +35,37 @@ def load_channel_class(module_name: str) -> type[BaseChannel]:
if isinstance(obj, type) and issubclass(obj, _Base) and obj is not _Base:
return obj
raise ImportError(f"No BaseChannel subclass in nanobot.channels.{module_name}")
def discover_plugins() -> dict[str, type[BaseChannel]]:
"""Discover external channel plugins registered via entry_points."""
from importlib.metadata import entry_points
plugins: dict[str, type[BaseChannel]] = {}
for ep in entry_points(group="nanobot.channels"):
try:
cls = ep.load()
plugins[ep.name] = cls
except Exception as e:
logger.warning("Failed to load channel plugin '{}': {}", ep.name, e)
return plugins
def discover_all() -> dict[str, type[BaseChannel]]:
"""Return all channels: built-in (pkgutil) merged with external (entry_points).
Built-in channels take priority — an external plugin cannot shadow a built-in name.
"""
builtin: dict[str, type[BaseChannel]] = {}
for modname in discover_channel_names():
try:
builtin[modname] = load_channel_class(modname)
except ImportError as e:
logger.debug("Skipping built-in channel '{}': {}", modname, e)
external = discover_plugins()
shadowed = set(external) & set(builtin)
if shadowed:
logger.warning("Plugin(s) shadowed by built-in channels (ignored): {}", shadowed)
return {**external, **builtin}

37
nanobot/channels/slack.py
View File

@@ -13,8 +13,35 @@ from slackify_markdown import slackify_markdown
from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from pydantic import Field
from nanobot.channels.base import BaseChannel
from nanobot.config.schema import SlackConfig
from nanobot.config.schema import Base
class SlackDMConfig(Base):
"""Slack DM policy configuration."""
enabled: bool = True
policy: str = "open"
allow_from: list[str] = Field(default_factory=list)
class SlackConfig(Base):
"""Slack channel configuration."""
enabled: bool = False
mode: str = "socket"
webhook_path: str = "/slack/events"
bot_token: str = ""
app_token: str = ""
user_token_read_only: bool = True
reply_in_thread: bool = True
react_emoji: str = "eyes"
allow_from: list[str] = Field(default_factory=list)
group_policy: str = "mention"
group_allow_from: list[str] = Field(default_factory=list)
dm: SlackDMConfig = Field(default_factory=SlackDMConfig)
class SlackChannel(BaseChannel):
@@ -23,7 +50,13 @@ class SlackChannel(BaseChannel):
name = "slack"
display_name = "Slack"
def __init__(self, config: SlackConfig, bus: MessageBus):
@classmethod
def default_config(cls) -> dict[str, Any]:
return SlackConfig().model_dump(by_alias=True)
def __init__(self, config: Any, bus: MessageBus):
if isinstance(config, dict):
config = SlackConfig.model_validate(config)
super().__init__(config, bus)
self.config: SlackConfig = config
self._web_client: AsyncWebClient | None = None

27
nanobot/channels/telegram.py
View File

@@ -6,8 +6,10 @@ import asyncio
import re
import time
import unicodedata
from typing import Any, Literal
from loguru import logger
from pydantic import Field
from telegram import BotCommand, ReplyParameters, Update
from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters
from telegram.request import HTTPXRequest
@@ -16,7 +18,7 @@ from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.base import BaseChannel
from nanobot.config.paths import get_media_dir
from nanobot.config.schema import TelegramConfig
from nanobot.config.schema import Base
from nanobot.utils.helpers import split_message
TELEGRAM_MAX_MESSAGE_LEN = 4000 # Telegram message character limit
@@ -148,6 +150,17 @@ def _markdown_to_telegram_html(text: str) -> str:
return text
class TelegramConfig(Base):
"""Telegram channel configuration."""
enabled: bool = False
token: str = ""
allow_from: list[str] = Field(default_factory=list)
proxy: str | None = None
reply_to_message: bool = False
group_policy: Literal["open", "mention"] = "mention"
class TelegramChannel(BaseChannel):
"""
Telegram channel using long polling.
@@ -167,7 +180,13 @@ class TelegramChannel(BaseChannel):
BotCommand("restart", "Restart the bot"),
]
def __init__(self, config: TelegramConfig, bus: MessageBus):
@classmethod
def default_config(cls) -> dict[str, Any]:
return TelegramConfig().model_dump(by_alias=True)
def __init__(self, config: Any, bus: MessageBus):
if isinstance(config, dict):
config = TelegramConfig.model_validate(config)
super().__init__(config, bus)
self.config: TelegramConfig = config
self._app: Application | None = None
@@ -434,6 +453,7 @@ class TelegramChannel(BaseChannel):
"🐈 nanobot commands:\n"
"/new — Start a new conversation\n"
"/stop — Stop the current task\n"
"/restart — Restart the bot\n"
"/help — Show available commands"
)
@@ -514,7 +534,8 @@ class TelegramChannel(BaseChannel):
getattr(media_file, "file_name", None),
)
media_dir = get_media_dir("telegram")
file_path = media_dir / f"{media_file.file_id[:16]}{ext}"
unique_id = getattr(media_file, "file_unique_id", media_file.file_id)
file_path = media_dir / f"{unique_id}{ext}"
await file.download_to_drive(str(file_path))
path_str = str(file_path)
if media_type in ("voice", "audio"):

21
nanobot/channels/wecom.py
View File

@@ -12,10 +12,21 @@ from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.base import BaseChannel
from nanobot.config.paths import get_media_dir
from nanobot.config.schema import WecomConfig
from nanobot.config.schema import Base
from pydantic import Field
WECOM_AVAILABLE = importlib.util.find_spec("wecom_aibot_sdk") is not None
class WecomConfig(Base):
"""WeCom (Enterprise WeChat) AI Bot channel configuration."""
enabled: bool = False
bot_id: str = ""
secret: str = ""
allow_from: list[str] = Field(default_factory=list)
welcome_message: str = ""
# Message type display mapping
MSG_TYPE_MAP = {
"image": "[image]",
@@ -38,7 +49,13 @@ class WecomChannel(BaseChannel):
name = "wecom"
display_name = "WeCom"
def __init__(self, config: WecomConfig, bus: MessageBus):
@classmethod
def default_config(cls) -> dict[str, Any]:
return WecomConfig().model_dump(by_alias=True)
def __init__(self, config: Any, bus: MessageBus):
if isinstance(config, dict):
config = WecomConfig.model_validate(config)
super().__init__(config, bus)
self.config: WecomConfig = config
self._client: Any = None

23
nanobot/channels/whatsapp.py
View File

@@ -4,13 +4,25 @@ import asyncio
import json
import mimetypes
from collections import OrderedDict
from typing import Any
from loguru import logger
from pydantic import Field
from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.base import BaseChannel
from nanobot.config.schema import WhatsAppConfig
from nanobot.config.schema import Base
class WhatsAppConfig(Base):
"""WhatsApp channel configuration."""
enabled: bool = False
bridge_url: str = "ws://localhost:3001"
bridge_token: str = ""
allow_from: list[str] = Field(default_factory=list)
class WhatsAppChannel(BaseChannel):
@@ -24,9 +36,14 @@ class WhatsAppChannel(BaseChannel):
name = "whatsapp"
display_name = "WhatsApp"
def __init__(self, config: WhatsAppConfig, bus: MessageBus):
@classmethod
def default_config(cls) -> dict[str, Any]:
return WhatsAppConfig().model_dump(by_alias=True)
def __init__(self, config: Any, bus: MessageBus):
if isinstance(config, dict):
config = WhatsAppConfig.model_validate(config)
super().__init__(config, bus)
self.config: WhatsAppConfig = config
self._ws = None
self._connected = False
self._processed_message_ids: OrderedDict[str, None] = OrderedDict()

136
nanobot/cli/commands.py
View File

@@ -6,6 +6,7 @@ import select
import signal
import sys
from pathlib import Path
from typing import Any
# Force UTF-8 encoding for Windows console
if sys.platform == "win32":
@@ -240,6 +241,8 @@ def onboard():
console.print("[dim]Config template now uses `maxTokens` + `contextWindowTokens`; `memoryWindow` is no longer a runtime setting.[/dim]")
_onboard_plugins(config_path)
# Create workspace
workspace = get_workspace_path()
@@ -257,7 +260,42 @@ def onboard():
console.print("\n[dim]Want Telegram/WhatsApp? See: https://github.com/HKUDS/nanobot#-chat-apps[/dim]")
def _merge_missing_defaults(existing: Any, defaults: Any) -> Any:
"""Recursively fill in missing values from defaults without overwriting user config."""
if not isinstance(existing, dict) or not isinstance(defaults, dict):
return existing
merged = dict(existing)
for key, value in defaults.items():
if key not in merged:
merged[key] = value
else:
merged[key] = _merge_missing_defaults(merged[key], value)
return merged
def _onboard_plugins(config_path: Path) -> None:
"""Inject default config for all discovered channels (built-in + plugins)."""
import json
from nanobot.channels.registry import discover_all
all_channels = discover_all()
if not all_channels:
return
with open(config_path, encoding="utf-8") as f:
data = json.load(f)
channels = data.setdefault("channels", {})
for name, cls in all_channels.items():
if name not in channels:
channels[name] = cls.default_config()
else:
channels[name] = _merge_missing_defaults(channels[name], cls.default_config())
with open(config_path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
def _make_provider(config: Config):
@@ -410,13 +448,14 @@ def gateway(
"""Execute a cron job through the agent."""
from nanobot.agent.tools.cron import CronTool
from nanobot.agent.tools.message import MessageTool
from nanobot.utils.evaluator import evaluate_response
reminder_note = (
"[Scheduled Task] Timer finished.\n\n"
f"Task '{job.name}' has been triggered.\n"
f"Scheduled instruction: {job.payload.message}"
)
# Prevent the agent from scheduling new cron jobs during execution
cron_tool = agent.tools.get("cron")
cron_token = None
if isinstance(cron_tool, CronTool):
@@ -437,11 +476,15 @@ def gateway(
return response
if job.payload.deliver and job.payload.to and response:
should_notify = await evaluate_response(
response, job.payload.message, provider, agent.model,
)
if should_notify:
from nanobot.bus.events import OutboundMessage
await bus.publish_outbound(OutboundMessage(
channel=job.payload.channel or "cli",
chat_id=job.payload.to,
content=response
content=response,
))
return response
cron.on_job = on_cron_job
@@ -521,6 +564,10 @@ def gateway(
)
except KeyboardInterrupt:
console.print("\nShutting down...")
except Exception:
import traceback
console.print("\n[red]Error: Gateway crashed unexpectedly[/red]")
console.print(traceback.format_exc())
finally:
await agent.close_mcp()
heartbeat.stop()
@@ -731,7 +778,7 @@ app.add_typer(channels_app, name="channels")
@channels_app.command("status")
def channels_status():
"""Show channel status."""
from nanobot.channels.registry import discover_channel_names, load_channel_class
from nanobot.channels.registry import discover_all
from nanobot.config.loader import load_config
config = load_config()
@@ -740,16 +787,16 @@ def channels_status():
table.add_column("Channel", style="cyan")
table.add_column("Enabled", style="green")
for modname in sorted(discover_channel_names()):
section = getattr(config.channels, modname, None)
enabled = section and getattr(section, "enabled", False)
try:
cls = load_channel_class(modname)
display = cls.display_name
except ImportError:
display = modname.title()
for name, cls in sorted(discover_all().items()):
section = getattr(config.channels, name, None)
if section is None:
enabled = False
elif isinstance(section, dict):
enabled = section.get("enabled", False)
else:
enabled = getattr(section, "enabled", False)
table.add_row(
display,
cls.display_name,
"[green]\u2713[/green]" if enabled else "[dim]\u2717[/dim]",
)
@@ -771,7 +818,8 @@ def _get_bridge_dir() -> Path:
return user_bridge
# Check for npm
if not shutil.which("npm"):
npm_path = shutil.which("npm")
if not npm_path:
console.print("[red]npm not found. Please install Node.js >= 18.[/red]")
raise typer.Exit(1)
@@ -801,10 +849,10 @@ def _get_bridge_dir() -> Path:
# Install and build
try:
console.print(" Installing dependencies...")
subprocess.run(["npm", "install"], cwd=user_bridge, check=True, capture_output=True)
subprocess.run([npm_path, "install"], cwd=user_bridge, check=True, capture_output=True)
console.print(" Building...")
subprocess.run(["npm", "run", "build"], cwd=user_bridge, check=True, capture_output=True)
subprocess.run([npm_path, "run", "build"], cwd=user_bridge, check=True, capture_output=True)
console.print("[green]✓[/green] Bridge ready\n")
except subprocess.CalledProcessError as e:
@@ -819,6 +867,7 @@ def _get_bridge_dir() -> Path:
@channels_app.command("login")
def channels_login():
"""Link device via QR code."""
import shutil
import subprocess
from nanobot.config.loader import load_config
@@ -831,16 +880,63 @@ def channels_login():
console.print("Scan the QR code to connect.\n")
env = {**os.environ}
if config.channels.whatsapp.bridge_token:
env["BRIDGE_TOKEN"] = config.channels.whatsapp.bridge_token
wa_cfg = getattr(config.channels, "whatsapp", None) or {}
bridge_token = wa_cfg.get("bridgeToken", "") if isinstance(wa_cfg, dict) else getattr(wa_cfg, "bridge_token", "")
if bridge_token:
env["BRIDGE_TOKEN"] = bridge_token
env["AUTH_DIR"] = str(get_runtime_subdir("whatsapp-auth"))
npm_path = shutil.which("npm")
if not npm_path:
console.print("[red]npm not found. Please install Node.js.[/red]")
raise typer.Exit(1)
try:
subprocess.run(["npm", "start"], cwd=bridge_dir, check=True, env=env)
subprocess.run([npm_path, "start"], cwd=bridge_dir, check=True, env=env)
except subprocess.CalledProcessError as e:
console.print(f"[red]Bridge failed: {e}[/red]")
except FileNotFoundError:
console.print("[red]npm not found. Please install Node.js.[/red]")
# ============================================================================
# Plugin Commands
# ============================================================================
plugins_app = typer.Typer(help="Manage channel plugins")
app.add_typer(plugins_app, name="plugins")
@plugins_app.command("list")
def plugins_list():
"""List all discovered channels (built-in and plugins)."""
from nanobot.channels.registry import discover_all, discover_channel_names
from nanobot.config.loader import load_config
config = load_config()
builtin_names = set(discover_channel_names())
all_channels = discover_all()
table = Table(title="Channel Plugins")
table.add_column("Name", style="cyan")
table.add_column("Source", style="magenta")
table.add_column("Enabled", style="green")
for name in sorted(all_channels):
cls = all_channels[name]
source = "builtin" if name in builtin_names else "plugin"
section = getattr(config.channels, name, None)
if section is None:
enabled = False
elif isinstance(section, dict):
enabled = section.get("enabled", False)
else:
enabled = getattr(section, "enabled", False)
table.add_row(
cls.display_name,
source,
"[green]yes[/green]" if enabled else "[dim]no[/dim]",
)
console.print(table)
# ============================================================================

218
nanobot/config/schema.py
View File

@@ -14,219 +14,17 @@ class Base(BaseModel):
model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)
class WhatsAppConfig(Base):
"""WhatsApp channel configuration."""
enabled: bool = False
bridge_url: str = "ws://localhost:3001"
bridge_token: str = "" # Shared token for bridge auth (optional, recommended)
allow_from: list[str] = Field(default_factory=list) # Allowed phone numbers
class TelegramConfig(Base):
"""Telegram channel configuration."""
enabled: bool = False
token: str = "" # Bot token from @BotFather
allow_from: list[str] = Field(default_factory=list) # Allowed user IDs or usernames
proxy: str | None = (
None # HTTP/SOCKS5 proxy URL, e.g. "http://127.0.0.1:7890" or "socks5://127.0.0.1:1080"
)
reply_to_message: bool = False # If true, bot replies quote the original message
group_policy: Literal["open", "mention"] = "mention" # "mention" responds when @mentioned or replied to, "open" responds to all
class FeishuConfig(Base):
"""Feishu/Lark channel configuration using WebSocket long connection."""
enabled: bool = False
app_id: str = "" # App ID from Feishu Open Platform
app_secret: str = "" # App Secret from Feishu Open Platform
encrypt_key: str = "" # Encrypt Key for event subscription (optional)
verification_token: str = "" # Verification Token for event subscription (optional)
allow_from: list[str] = Field(default_factory=list) # Allowed user open_ids
react_emoji: str = (
"THUMBSUP" # Emoji type for message reactions (e.g. THUMBSUP, OK, DONE, SMILE)
)
group_policy: Literal["open", "mention"] = "mention" # "mention" responds when @mentioned, "open" responds to all
class DingTalkConfig(Base):
"""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(Base):
"""Discord channel configuration."""
enabled: bool = False
token: str = "" # Bot token from Discord Developer Portal
allow_from: list[str] = Field(default_factory=list) # Allowed user IDs
gateway_url: str = "wss://gateway.discord.gg/?v=10&encoding=json"
intents: int = 37377 # GUILDS + GUILD_MESSAGES + DIRECT_MESSAGES + MESSAGE_CONTENT
group_policy: Literal["mention", "open"] = "mention"
class MatrixConfig(Base):
"""Matrix (Element) channel configuration."""
enabled: bool = False
homeserver: str = "https://matrix.org"
access_token: str = ""
user_id: str = "" # @bot:matrix.org
device_id: str = ""
e2ee_enabled: bool = True # Enable Matrix E2EE support (encryption + encrypted room handling).
sync_stop_grace_seconds: int = (
2 # Max seconds to wait for sync_forever to stop gracefully before cancellation fallback.
)
max_media_bytes: int = (
20 * 1024 * 1024
) # Max attachment size accepted for Matrix media handling (inbound + outbound).
allow_from: list[str] = Field(default_factory=list)
group_policy: Literal["open", "mention", "allowlist"] = "open"
group_allow_from: list[str] = Field(default_factory=list)
allow_room_mentions: bool = False
class EmailConfig(Base):
"""Email channel configuration (IMAP inbound + SMTP outbound)."""
enabled: bool = False
consent_granted: bool = False # Explicit owner permission to access mailbox data
# IMAP (receive)
imap_host: str = ""
imap_port: int = 993
imap_username: str = ""
imap_password: str = ""
imap_mailbox: str = "INBOX"
imap_use_ssl: bool = True
# SMTP (send)
smtp_host: str = ""
smtp_port: int = 587
smtp_username: str = ""
smtp_password: str = ""
smtp_use_tls: bool = True
smtp_use_ssl: bool = False
from_address: str = ""
# Behavior
auto_reply_enabled: bool = (
True # If false, inbound email is read but no automatic reply is sent
)
poll_interval_seconds: int = 30
mark_seen: bool = True
max_body_chars: int = 12000
subject_prefix: str = "Re: "
allow_from: list[str] = Field(default_factory=list) # Allowed sender email addresses
class MochatMentionConfig(Base):
"""Mochat mention behavior configuration."""
require_in_groups: bool = False
class MochatGroupRule(Base):
"""Mochat per-group mention requirement."""
require_mention: bool = False
class MochatConfig(Base):
"""Mochat channel configuration."""
enabled: bool = False
base_url: str = "https://mochat.io"
socket_url: str = ""
socket_path: str = "/socket.io"
socket_disable_msgpack: bool = False
socket_reconnect_delay_ms: int = 1000
socket_max_reconnect_delay_ms: int = 10000
socket_connect_timeout_ms: int = 10000
refresh_interval_ms: int = 30000
watch_timeout_ms: int = 25000
watch_limit: int = 100
retry_delay_ms: int = 500
max_retry_attempts: int = 0 # 0 means unlimited retries
claw_token: str = ""
agent_user_id: str = ""
sessions: list[str] = Field(default_factory=list)
panels: list[str] = Field(default_factory=list)
allow_from: list[str] = Field(default_factory=list)
mention: MochatMentionConfig = Field(default_factory=MochatMentionConfig)
groups: dict[str, MochatGroupRule] = Field(default_factory=dict)
reply_delay_mode: str = "non-mention" # off | non-mention
reply_delay_ms: int = 120000
class SlackDMConfig(Base):
"""Slack DM policy configuration."""
enabled: bool = True
policy: str = "open" # "open" or "allowlist"
allow_from: list[str] = Field(default_factory=list) # Allowed Slack user IDs
class SlackConfig(Base):
"""Slack channel configuration."""
enabled: bool = False
mode: str = "socket" # "socket" supported
webhook_path: str = "/slack/events"
bot_token: str = "" # xoxb-...
app_token: str = "" # xapp-...
user_token_read_only: bool = True
reply_in_thread: bool = True
react_emoji: str = "eyes"
allow_from: list[str] = Field(default_factory=list) # Allowed Slack user IDs (sender-level)
group_policy: str = "mention" # "mention", "open", "allowlist"
group_allow_from: list[str] = Field(default_factory=list) # Allowed channel IDs if allowlist
dm: SlackDMConfig = Field(default_factory=SlackDMConfig)
class QQConfig(Base):
"""QQ channel configuration using botpy SDK."""
enabled: bool = False
app_id: str = "" # 机器人 ID (AppID) from q.qq.com
secret: str = "" # 机器人密钥 (AppSecret) from q.qq.com
allow_from: list[str] = Field(
default_factory=list
) # Allowed user openids (empty = public access)
class WecomConfig(Base):
"""WeCom (Enterprise WeChat) AI Bot channel configuration."""
enabled: bool = False
bot_id: str = "" # Bot ID from WeCom AI Bot platform
secret: str = "" # Bot Secret from WeCom AI Bot platform
allow_from: list[str] = Field(default_factory=list) # Allowed user IDs
welcome_message: str = "" # Welcome message for enter_chat event
class ChannelsConfig(Base):
"""Configuration for chat channels."""
"""Configuration for chat channels.
Built-in and plugin channel configs are stored as extra fields (dicts).
Each channel parses its own config in __init__.
"""
model_config = ConfigDict(extra="allow")
send_progress: bool = True # stream agent's text progress to the channel
send_tool_hints: bool = False # stream tool-call hints (e.g. read_file("…"))
whatsapp: WhatsAppConfig = Field(default_factory=WhatsAppConfig)
telegram: TelegramConfig = Field(default_factory=TelegramConfig)
discord: DiscordConfig = Field(default_factory=DiscordConfig)
feishu: FeishuConfig = Field(default_factory=FeishuConfig)
mochat: MochatConfig = Field(default_factory=MochatConfig)
dingtalk: DingTalkConfig = Field(default_factory=DingTalkConfig)
email: EmailConfig = Field(default_factory=EmailConfig)
slack: SlackConfig = Field(default_factory=SlackConfig)
qq: QQConfig = Field(default_factory=QQConfig)
matrix: MatrixConfig = Field(default_factory=MatrixConfig)
wecom: WecomConfig = Field(default_factory=WecomConfig)
class AgentDefaults(Base):
@@ -342,7 +140,7 @@ class MCPServerConfig(Base):
url: str = "" # HTTP/SSE: endpoint URL
headers: dict[str, str] = Field(default_factory=dict) # HTTP/SSE: custom headers
tool_timeout: int = 30 # seconds before a tool call is cancelled
enabled_tools: list[str] = Field(default_factory=lambda: ["*"]) # Only register these tools; accepts raw MCP names or wrapped mcp_<server>_<tool> names; ["*"] = all tools; [] = no tools
class ToolsConfig(Base):
"""Tools configuration."""

11
nanobot/heartbeat/service.py
View File

@@ -139,6 +139,8 @@ class HeartbeatService:
async def _tick(self) -> None:
"""Execute a single heartbeat tick."""
from nanobot.utils.evaluator import evaluate_response
content = self._read_heartbeat_file()
if not content:
logger.debug("Heartbeat: HEARTBEAT.md missing or empty")
@@ -156,9 +158,16 @@ class HeartbeatService:
logger.info("Heartbeat: tasks found, executing...")
if self.on_execute:
response = await self.on_execute(tasks)
if response and self.on_notify:
if response:
should_notify = await evaluate_response(
response, tasks, self.provider, self.model,
)
if should_notify and self.on_notify:
logger.info("Heartbeat: completed, delivering response")
await self.on_notify(response)
else:
logger.info("Heartbeat: silenced by post-run evaluation")
except Exception:
logger.exception("Heartbeat execution failed")

92
nanobot/utils/evaluator.py Normal file
View File

@@ -0,0 +1,92 @@
"""Post-run evaluation for background tasks (heartbeat & cron).
After the agent executes a background task, this module makes a lightweight
LLM call to decide whether the result warrants notifying the user.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from loguru import logger
if TYPE_CHECKING:
from nanobot.providers.base import LLMProvider
_EVALUATE_TOOL = [
{
"type": "function",
"function": {
"name": "evaluate_notification",
"description": "Decide whether the user should be notified about this background task result.",
"parameters": {
"type": "object",
"properties": {
"should_notify": {
"type": "boolean",
"description": "true = result contains actionable/important info the user should see; false = routine or empty, safe to suppress",
},
"reason": {
"type": "string",
"description": "One-sentence reason for the decision",
},
},
"required": ["should_notify"],
},
},
}
]
_SYSTEM_PROMPT = (
"You are a notification gate for a background agent. "
"You will be given the original task and the agent's response. "
"Call the evaluate_notification tool to decide whether the user "
"should be notified.\n\n"
"Notify when the response contains actionable information, errors, "
"completed deliverables, or anything the user explicitly asked to "
"be reminded about.\n\n"
"Suppress when the response is a routine status check with nothing "
"new, a confirmation that everything is normal, or essentially empty."
)
async def evaluate_response(
response: str,
task_context: str,
provider: LLMProvider,
model: str,
) -> bool:
"""Decide whether a background-task result should be delivered to the user.
Uses a lightweight tool-call LLM request (same pattern as heartbeat
``_decide()``). Falls back to ``True`` (notify) on any failure so
that important messages are never silently dropped.
"""
try:
llm_response = await provider.chat_with_retry(
messages=[
{"role": "system", "content": _SYSTEM_PROMPT},
{"role": "user", "content": (
f"## Original task\n{task_context}\n\n"
f"## Agent response\n{response}"
)},
],
tools=_EVALUATE_TOOL,
model=model,
max_tokens=256,
temperature=0.0,
)
if not llm_response.has_tool_calls:
logger.warning("evaluate_response: no tool call returned, defaulting to notify")
return True
args = llm_response.tool_calls[0].arguments
should_notify = args.get("should_notify", True)
reason = args.get("reason", "")
logger.info("evaluate_response: should_notify={}, reason={}", should_notify, reason)
return bool(should_notify)
except Exception:
logger.exception("evaluate_response failed, defaulting to notify")
return True

2
pyproject.toml
View File

@@ -50,7 +50,7 @@ dependencies = [
[project.optional-dependencies]
wecom = [
"wecom-aibot-sdk-python>=0.1.2",
"wecom-aibot-sdk-python>=0.1.5",
]
matrix = [
"matrix-nio[e2e]>=0.25.2",

228
tests/test_channel_plugins.py Normal file
View File

@@ -0,0 +1,228 @@
"""Tests for channel plugin discovery, merging, and config compatibility."""
from __future__ import annotations
from types import SimpleNamespace
from unittest.mock import patch
import pytest
from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.base import BaseChannel
from nanobot.channels.manager import ChannelManager
from nanobot.config.schema import ChannelsConfig
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
class _FakePlugin(BaseChannel):
name = "fakeplugin"
display_name = "Fake Plugin"
async def start(self) -> None:
pass
async def stop(self) -> None:
pass
async def send(self, msg: OutboundMessage) -> None:
pass
class _FakeTelegram(BaseChannel):
"""Plugin that tries to shadow built-in telegram."""
name = "telegram"
display_name = "Fake Telegram"
async def start(self) -> None:
pass
async def stop(self) -> None:
pass
async def send(self, msg: OutboundMessage) -> None:
pass
def _make_entry_point(name: str, cls: type):
"""Create a mock entry point that returns *cls* on load()."""
ep = SimpleNamespace(name=name, load=lambda _cls=cls: _cls)
return ep
# ---------------------------------------------------------------------------
# ChannelsConfig extra="allow"
# ---------------------------------------------------------------------------
def test_channels_config_accepts_unknown_keys():
cfg = ChannelsConfig.model_validate({
"myplugin": {"enabled": True, "token": "abc"},
})
extra = cfg.model_extra
assert extra is not None
assert extra["myplugin"]["enabled"] is True
assert extra["myplugin"]["token"] == "abc"
def test_channels_config_getattr_returns_extra():
cfg = ChannelsConfig.model_validate({"myplugin": {"enabled": True}})
section = getattr(cfg, "myplugin", None)
assert isinstance(section, dict)
assert section["enabled"] is True
def test_channels_config_builtin_fields_removed():
"""After decoupling, ChannelsConfig has no explicit channel fields."""
cfg = ChannelsConfig()
assert not hasattr(cfg, "telegram")
assert cfg.send_progress is True
assert cfg.send_tool_hints is False
# ---------------------------------------------------------------------------
# discover_plugins
# ---------------------------------------------------------------------------
_EP_TARGET = "importlib.metadata.entry_points"
def test_discover_plugins_loads_entry_points():
from nanobot.channels.registry import discover_plugins
ep = _make_entry_point("line", _FakePlugin)
with patch(_EP_TARGET, return_value=[ep]):
result = discover_plugins()
assert "line" in result
assert result["line"] is _FakePlugin
def test_discover_plugins_handles_load_error():
from nanobot.channels.registry import discover_plugins
def _boom():
raise RuntimeError("broken")
ep = SimpleNamespace(name="broken", load=_boom)
with patch(_EP_TARGET, return_value=[ep]):
result = discover_plugins()
assert "broken" not in result
# ---------------------------------------------------------------------------
# discover_all — merge & priority
# ---------------------------------------------------------------------------
def test_discover_all_includes_builtins():
from nanobot.channels.registry import discover_all, discover_channel_names
with patch(_EP_TARGET, return_value=[]):
result = discover_all()
# discover_all() only returns channels that are actually available (dependencies installed)
# discover_channel_names() returns all built-in channel names
# So we check that all actually loaded channels are in the result
for name in result:
assert name in discover_channel_names()
def test_discover_all_includes_external_plugin():
from nanobot.channels.registry import discover_all
ep = _make_entry_point("line", _FakePlugin)
with patch(_EP_TARGET, return_value=[ep]):
result = discover_all()
assert "line" in result
assert result["line"] is _FakePlugin
def test_discover_all_builtin_shadows_plugin():
from nanobot.channels.registry import discover_all
ep = _make_entry_point("telegram", _FakeTelegram)
with patch(_EP_TARGET, return_value=[ep]):
result = discover_all()
assert "telegram" in result
assert result["telegram"] is not _FakeTelegram
# ---------------------------------------------------------------------------
# Manager _init_channels with dict config (plugin scenario)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_manager_loads_plugin_from_dict_config():
"""ChannelManager should instantiate a plugin channel from a raw dict config."""
from nanobot.channels.manager import ChannelManager
fake_config = SimpleNamespace(
channels=ChannelsConfig.model_validate({
"fakeplugin": {"enabled": True, "allowFrom": ["*"]},
}),
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
)
with patch(
"nanobot.channels.registry.discover_all",
return_value={"fakeplugin": _FakePlugin},
):
mgr = ChannelManager.__new__(ChannelManager)
mgr.config = fake_config
mgr.bus = MessageBus()
mgr.channels = {}
mgr._dispatch_task = None
mgr._init_channels()
assert "fakeplugin" in mgr.channels
assert isinstance(mgr.channels["fakeplugin"], _FakePlugin)
@pytest.mark.asyncio
async def test_manager_skips_disabled_plugin():
fake_config = SimpleNamespace(
channels=ChannelsConfig.model_validate({
"fakeplugin": {"enabled": False},
}),
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
)
with patch(
"nanobot.channels.registry.discover_all",
return_value={"fakeplugin": _FakePlugin},
):
mgr = ChannelManager.__new__(ChannelManager)
mgr.config = fake_config
mgr.bus = MessageBus()
mgr.channels = {}
mgr._dispatch_task = None
mgr._init_channels()
assert "fakeplugin" not in mgr.channels
# ---------------------------------------------------------------------------
# Built-in channel default_config() and dict->Pydantic conversion
# ---------------------------------------------------------------------------
def test_builtin_channel_default_config():
"""Built-in channels expose default_config() returning a dict with 'enabled': False."""
from nanobot.channels.telegram import TelegramChannel
cfg = TelegramChannel.default_config()
assert isinstance(cfg, dict)
assert cfg["enabled"] is False
assert "token" in cfg
def test_builtin_channel_init_from_dict():
"""Built-in channels accept a raw dict and convert to Pydantic internally."""
from nanobot.channels.telegram import TelegramChannel
bus = MessageBus()
ch = TelegramChannel({"enabled": False, "token": "test-tok", "allowFrom": ["*"]}, bus)
assert ch.config.token == "test-tok"
assert ch.config.allow_from == ["*"]

44
tests/test_config_migration.py
View File

@@ -1,4 +1,5 @@
import json
from types import SimpleNamespace
from typer.testing import CliRunner
@@ -86,3 +87,46 @@ def test_onboard_refresh_rewrites_legacy_config_template(tmp_path, monkeypatch)
assert defaults["maxTokens"] == 3333
assert defaults["contextWindowTokens"] == 65_536
assert "memoryWindow" not in defaults
def test_onboard_refresh_backfills_missing_channel_fields(tmp_path, monkeypatch) -> None:
config_path = tmp_path / "config.json"
workspace = tmp_path / "workspace"
config_path.write_text(
json.dumps(
{
"channels": {
"qq": {
"enabled": False,
"appId": "",
"secret": "",
"allowFrom": [],
}
}
}
),
encoding="utf-8",
)
monkeypatch.setattr("nanobot.config.loader.get_config_path", lambda: config_path)
monkeypatch.setattr("nanobot.cli.commands.get_workspace_path", lambda: workspace)
monkeypatch.setattr(
"nanobot.channels.registry.discover_all",
lambda: {
"qq": SimpleNamespace(
default_config=lambda: {
"enabled": False,
"appId": "",
"secret": "",
"allowFrom": [],
"msgFormat": "plain",
}
)
},
)
result = runner.invoke(app, ["onboard"], input="n\n")
assert result.exit_code == 0
saved = json.loads(config_path.read_text(encoding="utf-8"))
assert saved["channels"]["qq"]["msgFormat"] == "plain"

2
tests/test_dingtalk_channel.py
View File

@@ -6,7 +6,7 @@ import pytest
from nanobot.bus.queue import MessageBus
import nanobot.channels.dingtalk as dingtalk_module
from nanobot.channels.dingtalk import DingTalkChannel, NanobotDingTalkHandler
from nanobot.config.schema import DingTalkConfig
from nanobot.channels.dingtalk import DingTalkConfig
class _FakeResponse:

2
tests/test_email_channel.py
View File

@@ -6,7 +6,7 @@ import pytest
from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.email import EmailChannel
from nanobot.config.schema import EmailConfig
from nanobot.channels.email import EmailConfig
def _make_config() -> EmailConfig:

63
tests/test_evaluator.py Normal file
View File

@@ -0,0 +1,63 @@
import pytest
from nanobot.utils.evaluator import evaluate_response
from nanobot.providers.base import LLMProvider, LLMResponse, ToolCallRequest
class DummyProvider(LLMProvider):
def __init__(self, responses: list[LLMResponse]):
super().__init__()
self._responses = list(responses)
async def chat(self, *args, **kwargs) -> LLMResponse:
if self._responses:
return self._responses.pop(0)
return LLMResponse(content="", tool_calls=[])
def get_default_model(self) -> str:
return "test-model"
def _eval_tool_call(should_notify: bool, reason: str = "") -> LLMResponse:
return LLMResponse(
content="",
tool_calls=[
ToolCallRequest(
id="eval_1",
name="evaluate_notification",
arguments={"should_notify": should_notify, "reason": reason},
)
],
)
@pytest.mark.asyncio
async def test_should_notify_true() -> None:
provider = DummyProvider([_eval_tool_call(True, "user asked to be reminded")])
result = await evaluate_response("Task completed with results", "check emails", provider, "m")
assert result is True
@pytest.mark.asyncio
async def test_should_notify_false() -> None:
provider = DummyProvider([_eval_tool_call(False, "routine check, nothing new")])
result = await evaluate_response("All clear, no updates", "check status", provider, "m")
assert result is False
@pytest.mark.asyncio
async def test_fallback_on_error() -> None:
class FailingProvider(DummyProvider):
async def chat(self, *args, **kwargs) -> LLMResponse:
raise RuntimeError("provider down")
provider = FailingProvider([])
result = await evaluate_response("some response", "some task", provider, "m")
assert result is True
@pytest.mark.asyncio
async def test_no_tool_call_fallback() -> None:
provider = DummyProvider([LLMResponse(content="I think you should notify", tool_calls=[])])
result = await evaluate_response("some response", "some task", provider, "m")
assert result is True

392
tests/test_feishu_reply.py Normal file
View File

@@ -0,0 +1,392 @@
"""Tests for Feishu message reply (quote) feature."""
import asyncio
import json
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import pytest
from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.feishu import FeishuChannel, FeishuConfig
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_feishu_channel(reply_to_message: bool = False) -> FeishuChannel:
config = FeishuConfig(
enabled=True,
app_id="cli_test",
app_secret="secret",
allow_from=["*"],
reply_to_message=reply_to_message,
)
channel = FeishuChannel(config, MessageBus())
channel._client = MagicMock()
# _loop is only used by the WebSocket thread bridge; not needed for unit tests
channel._loop = None
return channel
def _make_feishu_event(
*,
message_id: str = "om_001",
chat_id: str = "oc_abc",
chat_type: str = "p2p",
msg_type: str = "text",
content: str = '{"text": "hello"}',
sender_open_id: str = "ou_alice",
parent_id: str | None = None,
root_id: str | None = None,
):
message = SimpleNamespace(
message_id=message_id,
chat_id=chat_id,
chat_type=chat_type,
message_type=msg_type,
content=content,
parent_id=parent_id,
root_id=root_id,
mentions=[],
)
sender = SimpleNamespace(
sender_type="user",
sender_id=SimpleNamespace(open_id=sender_open_id),
)
return SimpleNamespace(event=SimpleNamespace(message=message, sender=sender))
def _make_get_message_response(text: str, msg_type: str = "text", success: bool = True):
"""Build a fake im.v1.message.get response object."""
body = SimpleNamespace(content=json.dumps({"text": text}))
item = SimpleNamespace(msg_type=msg_type, body=body)
data = SimpleNamespace(items=[item])
resp = MagicMock()
resp.success.return_value = success
resp.data = data
resp.code = 0
resp.msg = "ok"
return resp
# ---------------------------------------------------------------------------
# Config tests
# ---------------------------------------------------------------------------
def test_feishu_config_reply_to_message_defaults_false() -> None:
assert FeishuConfig().reply_to_message is False
def test_feishu_config_reply_to_message_can_be_enabled() -> None:
config = FeishuConfig(reply_to_message=True)
assert config.reply_to_message is True
# ---------------------------------------------------------------------------
# _get_message_content_sync tests
# ---------------------------------------------------------------------------
def test_get_message_content_sync_returns_reply_prefix() -> None:
channel = _make_feishu_channel()
channel._client.im.v1.message.get.return_value = _make_get_message_response("what time is it?")
result = channel._get_message_content_sync("om_parent")
assert result == "[Reply to: what time is it?]"
def test_get_message_content_sync_truncates_long_text() -> None:
channel = _make_feishu_channel()
long_text = "x" * (FeishuChannel._REPLY_CONTEXT_MAX_LEN + 50)
channel._client.im.v1.message.get.return_value = _make_get_message_response(long_text)
result = channel._get_message_content_sync("om_parent")
assert result is not None
assert result.endswith("...]")
inner = result[len("[Reply to: ") : -1]
assert len(inner) == FeishuChannel._REPLY_CONTEXT_MAX_LEN + len("...")
def test_get_message_content_sync_returns_none_on_api_failure() -> None:
channel = _make_feishu_channel()
resp = MagicMock()
resp.success.return_value = False
resp.code = 230002
resp.msg = "bot not in group"
channel._client.im.v1.message.get.return_value = resp
result = channel._get_message_content_sync("om_parent")
assert result is None
def test_get_message_content_sync_returns_none_for_non_text_type() -> None:
channel = _make_feishu_channel()
body = SimpleNamespace(content=json.dumps({"image_key": "img_1"}))
item = SimpleNamespace(msg_type="image", body=body)
data = SimpleNamespace(items=[item])
resp = MagicMock()
resp.success.return_value = True
resp.data = data
channel._client.im.v1.message.get.return_value = resp
result = channel._get_message_content_sync("om_parent")
assert result is None
def test_get_message_content_sync_returns_none_when_empty_text() -> None:
channel = _make_feishu_channel()
channel._client.im.v1.message.get.return_value = _make_get_message_response(" ")
result = channel._get_message_content_sync("om_parent")
assert result is None
# ---------------------------------------------------------------------------
# _reply_message_sync tests
# ---------------------------------------------------------------------------
def test_reply_message_sync_returns_true_on_success() -> None:
channel = _make_feishu_channel()
resp = MagicMock()
resp.success.return_value = True
channel._client.im.v1.message.reply.return_value = resp
ok = channel._reply_message_sync("om_parent", "text", '{"text":"hi"}')
assert ok is True
channel._client.im.v1.message.reply.assert_called_once()
def test_reply_message_sync_returns_false_on_api_error() -> None:
channel = _make_feishu_channel()
resp = MagicMock()
resp.success.return_value = False
resp.code = 400
resp.msg = "bad request"
resp.get_log_id.return_value = "log_x"
channel._client.im.v1.message.reply.return_value = resp
ok = channel._reply_message_sync("om_parent", "text", '{"text":"hi"}')
assert ok is False
def test_reply_message_sync_returns_false_on_exception() -> None:
channel = _make_feishu_channel()
channel._client.im.v1.message.reply.side_effect = RuntimeError("network error")
ok = channel._reply_message_sync("om_parent", "text", '{"text":"hi"}')
assert ok is False
# ---------------------------------------------------------------------------
# send() — reply routing tests
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_send_uses_reply_api_when_configured() -> None:
channel = _make_feishu_channel(reply_to_message=True)
reply_resp = MagicMock()
reply_resp.success.return_value = True
channel._client.im.v1.message.reply.return_value = reply_resp
await channel.send(OutboundMessage(
channel="feishu",
chat_id="oc_abc",
content="hello",
metadata={"message_id": "om_001"},
))
channel._client.im.v1.message.reply.assert_called_once()
channel._client.im.v1.message.create.assert_not_called()
@pytest.mark.asyncio
async def test_send_uses_create_api_when_reply_disabled() -> None:
channel = _make_feishu_channel(reply_to_message=False)
create_resp = MagicMock()
create_resp.success.return_value = True
channel._client.im.v1.message.create.return_value = create_resp
await channel.send(OutboundMessage(
channel="feishu",
chat_id="oc_abc",
content="hello",
metadata={"message_id": "om_001"},
))
channel._client.im.v1.message.create.assert_called_once()
channel._client.im.v1.message.reply.assert_not_called()
@pytest.mark.asyncio
async def test_send_uses_create_api_when_no_message_id() -> None:
channel = _make_feishu_channel(reply_to_message=True)
create_resp = MagicMock()
create_resp.success.return_value = True
channel._client.im.v1.message.create.return_value = create_resp
await channel.send(OutboundMessage(
channel="feishu",
chat_id="oc_abc",
content="hello",
metadata={},
))
channel._client.im.v1.message.create.assert_called_once()
channel._client.im.v1.message.reply.assert_not_called()
@pytest.mark.asyncio
async def test_send_skips_reply_for_progress_messages() -> None:
channel = _make_feishu_channel(reply_to_message=True)
create_resp = MagicMock()
create_resp.success.return_value = True
channel._client.im.v1.message.create.return_value = create_resp
await channel.send(OutboundMessage(
channel="feishu",
chat_id="oc_abc",
content="thinking...",
metadata={"message_id": "om_001", "_progress": True},
))
channel._client.im.v1.message.create.assert_called_once()
channel._client.im.v1.message.reply.assert_not_called()
@pytest.mark.asyncio
async def test_send_fallback_to_create_when_reply_fails() -> None:
channel = _make_feishu_channel(reply_to_message=True)
reply_resp = MagicMock()
reply_resp.success.return_value = False
reply_resp.code = 400
reply_resp.msg = "error"
reply_resp.get_log_id.return_value = "log_x"
channel._client.im.v1.message.reply.return_value = reply_resp
create_resp = MagicMock()
create_resp.success.return_value = True
channel._client.im.v1.message.create.return_value = create_resp
await channel.send(OutboundMessage(
channel="feishu",
chat_id="oc_abc",
content="hello",
metadata={"message_id": "om_001"},
))
# reply attempted first, then falls back to create
channel._client.im.v1.message.reply.assert_called_once()
channel._client.im.v1.message.create.assert_called_once()
# ---------------------------------------------------------------------------
# _on_message — parent_id / root_id metadata tests
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_on_message_captures_parent_and_root_id_in_metadata() -> None:
channel = _make_feishu_channel()
channel._processed_message_ids.clear()
channel._client.im.v1.message.react.return_value = MagicMock(success=lambda: True)
captured = []
async def _capture(**kwargs):
captured.append(kwargs)
channel._handle_message = _capture
with patch.object(channel, "_add_reaction", return_value=None):
await channel._on_message(
_make_feishu_event(
parent_id="om_parent",
root_id="om_root",
)
)
assert len(captured) == 1
meta = captured[0]["metadata"]
assert meta["parent_id"] == "om_parent"
assert meta["root_id"] == "om_root"
assert meta["message_id"] == "om_001"
@pytest.mark.asyncio
async def test_on_message_parent_and_root_id_none_when_absent() -> None:
channel = _make_feishu_channel()
channel._processed_message_ids.clear()
captured = []
async def _capture(**kwargs):
captured.append(kwargs)
channel._handle_message = _capture
with patch.object(channel, "_add_reaction", return_value=None):
await channel._on_message(_make_feishu_event())
assert len(captured) == 1
meta = captured[0]["metadata"]
assert meta["parent_id"] is None
assert meta["root_id"] is None
@pytest.mark.asyncio
async def test_on_message_prepends_reply_context_when_parent_id_present() -> None:
channel = _make_feishu_channel()
channel._processed_message_ids.clear()
channel._client.im.v1.message.get.return_value = _make_get_message_response("original question")
captured = []
async def _capture(**kwargs):
captured.append(kwargs)
channel._handle_message = _capture
with patch.object(channel, "_add_reaction", return_value=None):
await channel._on_message(
_make_feishu_event(
content='{"text": "my answer"}',
parent_id="om_parent",
)
)
assert len(captured) == 1
content = captured[0]["content"]
assert content.startswith("[Reply to: original question]")
assert "my answer" in content
@pytest.mark.asyncio
async def test_on_message_no_extra_api_call_when_no_parent_id() -> None:
channel = _make_feishu_channel()
channel._processed_message_ids.clear()
captured = []
async def _capture(**kwargs):
captured.append(kwargs)
channel._handle_message = _capture
with patch.object(channel, "_add_reaction", return_value=None):
await channel._on_message(_make_feishu_event())
channel._client.im.v1.message.get.assert_not_called()
assert len(captured) == 1

6
tests/test_filesystem_tools.py
View File

@@ -222,8 +222,10 @@ class TestListDirTool:
@pytest.mark.asyncio
async def test_recursive(self, tool, populated_dir):
result = await tool.execute(path=str(populated_dir), recursive=True)
assert "src/main.py" in result
assert "src/utils.py" in result
# Normalize path separators for cross-platform compatibility
normalized = result.replace("\\", "/")
assert "src/main.py" in normalized
assert "src/utils.py" in normalized
assert "README.md" in result
# Ignored dirs should not appear
assert ".git" not in result

92
tests/test_heartbeat_service.py
View File

@@ -123,6 +123,98 @@ async def test_trigger_now_returns_none_when_decision_is_skip(tmp_path) -> None:
assert await service.trigger_now() is None
@pytest.mark.asyncio
async def test_tick_notifies_when_evaluator_says_yes(tmp_path, monkeypatch) -> None:
"""Phase 1 run -> Phase 2 execute -> Phase 3 evaluate=notify -> on_notify called."""
(tmp_path / "HEARTBEAT.md").write_text("- [ ] check deployments", encoding="utf-8")
provider = DummyProvider([
LLMResponse(
content="",
tool_calls=[
ToolCallRequest(
id="hb_1",
name="heartbeat",
arguments={"action": "run", "tasks": "check deployments"},
)
],
),
])
executed: list[str] = []
notified: list[str] = []
async def _on_execute(tasks: str) -> str:
executed.append(tasks)
return "deployment failed on staging"
async def _on_notify(response: str) -> None:
notified.append(response)
service = HeartbeatService(
workspace=tmp_path,
provider=provider,
model="openai/gpt-4o-mini",
on_execute=_on_execute,
on_notify=_on_notify,
)
async def _eval_notify(*a, **kw):
return True
monkeypatch.setattr("nanobot.utils.evaluator.evaluate_response", _eval_notify)
await service._tick()
assert executed == ["check deployments"]
assert notified == ["deployment failed on staging"]
@pytest.mark.asyncio
async def test_tick_suppresses_when_evaluator_says_no(tmp_path, monkeypatch) -> None:
"""Phase 1 run -> Phase 2 execute -> Phase 3 evaluate=silent -> on_notify NOT called."""
(tmp_path / "HEARTBEAT.md").write_text("- [ ] check status", encoding="utf-8")
provider = DummyProvider([
LLMResponse(
content="",
tool_calls=[
ToolCallRequest(
id="hb_1",
name="heartbeat",
arguments={"action": "run", "tasks": "check status"},
)
],
),
])
executed: list[str] = []
notified: list[str] = []
async def _on_execute(tasks: str) -> str:
executed.append(tasks)
return "everything is fine, no issues"
async def _on_notify(response: str) -> None:
notified.append(response)
service = HeartbeatService(
workspace=tmp_path,
provider=provider,
model="openai/gpt-4o-mini",
on_execute=_on_execute,
on_notify=_on_notify,
)
async def _eval_silent(*a, **kw):
return False
monkeypatch.setattr("nanobot.utils.evaluator.evaluate_response", _eval_silent)
await service._tick()
assert executed == ["check status"]
assert notified == []
@pytest.mark.asyncio
async def test_decide_retries_transient_error_then_succeeds(tmp_path, monkeypatch) -> None:
provider = DummyProvider([

2
tests/test_matrix_channel.py
View File

@@ -12,7 +12,7 @@ from nanobot.channels.matrix import (
TYPING_NOTICE_TIMEOUT_MS,
MatrixChannel,
)
from nanobot.config.schema import MatrixConfig
from nanobot.channels.matrix import MatrixConfig
_ROOM_SEND_UNSET = object()

187
tests/test_mcp_tool.py
View File

@@ -1,12 +1,15 @@
from __future__ import annotations
import asyncio
from contextlib import AsyncExitStack, asynccontextmanager
import sys
from types import ModuleType, SimpleNamespace
import pytest
from nanobot.agent.tools.mcp import MCPToolWrapper
from nanobot.agent.tools.mcp import MCPToolWrapper, connect_mcp_servers
from nanobot.agent.tools.registry import ToolRegistry
from nanobot.config.schema import MCPServerConfig
class _FakeTextContent:
@@ -14,12 +17,63 @@ class _FakeTextContent:
self.text = text
@pytest.fixture
def fake_mcp_runtime() -> dict[str, object | None]:
return {"session": None}
@pytest.fixture(autouse=True)
def _fake_mcp_module(monkeypatch: pytest.MonkeyPatch) -> None:
def _fake_mcp_module(
monkeypatch: pytest.MonkeyPatch, fake_mcp_runtime: dict[str, object | None]
) -> None:
mod = ModuleType("mcp")
mod.types = SimpleNamespace(TextContent=_FakeTextContent)
class _FakeStdioServerParameters:
def __init__(self, command: str, args: list[str], env: dict | None = None) -> None:
self.command = command
self.args = args
self.env = env
class _FakeClientSession:
def __init__(self, _read: object, _write: object) -> None:
self._session = fake_mcp_runtime["session"]
async def __aenter__(self) -> object:
return self._session
async def __aexit__(self, exc_type, exc, tb) -> bool:
return False
@asynccontextmanager
async def _fake_stdio_client(_params: object):
yield object(), object()
@asynccontextmanager
async def _fake_sse_client(_url: str, httpx_client_factory=None):
yield object(), object()
@asynccontextmanager
async def _fake_streamable_http_client(_url: str, http_client=None):
yield object(), object(), object()
mod.ClientSession = _FakeClientSession
mod.StdioServerParameters = _FakeStdioServerParameters
monkeypatch.setitem(sys.modules, "mcp", mod)
client_mod = ModuleType("mcp.client")
stdio_mod = ModuleType("mcp.client.stdio")
stdio_mod.stdio_client = _fake_stdio_client
sse_mod = ModuleType("mcp.client.sse")
sse_mod.sse_client = _fake_sse_client
streamable_http_mod = ModuleType("mcp.client.streamable_http")
streamable_http_mod.streamable_http_client = _fake_streamable_http_client
monkeypatch.setitem(sys.modules, "mcp.client", client_mod)
monkeypatch.setitem(sys.modules, "mcp.client.stdio", stdio_mod)
monkeypatch.setitem(sys.modules, "mcp.client.sse", sse_mod)
monkeypatch.setitem(sys.modules, "mcp.client.streamable_http", streamable_http_mod)
def _make_wrapper(session: object, *, timeout: float = 0.1) -> MCPToolWrapper:
tool_def = SimpleNamespace(
@@ -97,3 +151,132 @@ async def test_execute_handles_generic_exception() -> None:
result = await wrapper.execute()
assert result == "(MCP tool call failed: RuntimeError)"
def _make_tool_def(name: str) -> SimpleNamespace:
return SimpleNamespace(
name=name,
description=f"{name} tool",
inputSchema={"type": "object", "properties": {}},
)
def _make_fake_session(tool_names: list[str]) -> SimpleNamespace:
async def initialize() -> None:
return None
async def list_tools() -> SimpleNamespace:
return SimpleNamespace(tools=[_make_tool_def(name) for name in tool_names])
return SimpleNamespace(initialize=initialize, list_tools=list_tools)
@pytest.mark.asyncio
async def test_connect_mcp_servers_enabled_tools_supports_raw_names(
fake_mcp_runtime: dict[str, object | None],
) -> None:
fake_mcp_runtime["session"] = _make_fake_session(["demo", "other"])
registry = ToolRegistry()
stack = AsyncExitStack()
await stack.__aenter__()
try:
await connect_mcp_servers(
{"test": MCPServerConfig(command="fake", enabled_tools=["demo"])},
registry,
stack,
)
finally:
await stack.aclose()
assert registry.tool_names == ["mcp_test_demo"]
@pytest.mark.asyncio
async def test_connect_mcp_servers_enabled_tools_defaults_to_all(
fake_mcp_runtime: dict[str, object | None],
) -> None:
fake_mcp_runtime["session"] = _make_fake_session(["demo", "other"])
registry = ToolRegistry()
stack = AsyncExitStack()
await stack.__aenter__()
try:
await connect_mcp_servers(
{"test": MCPServerConfig(command="fake")},
registry,
stack,
)
finally:
await stack.aclose()
assert registry.tool_names == ["mcp_test_demo", "mcp_test_other"]
@pytest.mark.asyncio
async def test_connect_mcp_servers_enabled_tools_supports_wrapped_names(
fake_mcp_runtime: dict[str, object | None],
) -> None:
fake_mcp_runtime["session"] = _make_fake_session(["demo", "other"])
registry = ToolRegistry()
stack = AsyncExitStack()
await stack.__aenter__()
try:
await connect_mcp_servers(
{"test": MCPServerConfig(command="fake", enabled_tools=["mcp_test_demo"])},
registry,
stack,
)
finally:
await stack.aclose()
assert registry.tool_names == ["mcp_test_demo"]
@pytest.mark.asyncio
async def test_connect_mcp_servers_enabled_tools_empty_list_registers_none(
fake_mcp_runtime: dict[str, object | None],
) -> None:
fake_mcp_runtime["session"] = _make_fake_session(["demo", "other"])
registry = ToolRegistry()
stack = AsyncExitStack()
await stack.__aenter__()
try:
await connect_mcp_servers(
{"test": MCPServerConfig(command="fake", enabled_tools=[])},
registry,
stack,
)
finally:
await stack.aclose()
assert registry.tool_names == []
@pytest.mark.asyncio
async def test_connect_mcp_servers_enabled_tools_warns_on_unknown_entries(
fake_mcp_runtime: dict[str, object | None], monkeypatch: pytest.MonkeyPatch
) -> None:
fake_mcp_runtime["session"] = _make_fake_session(["demo"])
registry = ToolRegistry()
warnings: list[str] = []
def _warning(message: str, *args: object) -> None:
warnings.append(message.format(*args))
monkeypatch.setattr("nanobot.agent.tools.mcp.logger.warning", _warning)
stack = AsyncExitStack()
await stack.__aenter__()
try:
await connect_mcp_servers(
{"test": MCPServerConfig(command="fake", enabled_tools=["unknown"])},
registry,
stack,
)
finally:
await stack.aclose()
assert registry.tool_names == []
assert warnings
assert "enabledTools entries not found: unknown" in warnings[-1]
assert "Available raw names: demo" in warnings[-1]
assert "Available wrapped names: mcp_test_demo" in warnings[-1]

31
tests/test_qq_channel.py
View File

@@ -5,7 +5,7 @@ import pytest
from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.qq import QQChannel
from nanobot.config.schema import QQConfig
from nanobot.channels.qq import QQConfig
class _FakeApi:
@@ -94,3 +94,32 @@ async def test_send_c2c_message_uses_plain_text_c2c_api_with_msg_seq() -> None:
"msg_seq": 2,
}
assert not channel._client.api.group_calls
@pytest.mark.asyncio
async def test_send_group_message_uses_markdown_when_configured() -> None:
channel = QQChannel(
QQConfig(app_id="app", secret="secret", allow_from=["*"], msg_format="markdown"),
MessageBus(),
)
channel._client = _FakeClient()
channel._chat_type_cache["group123"] = "group"
await channel.send(
OutboundMessage(
channel="qq",
chat_id="group123",
content="**hello**",
metadata={"message_id": "msg1"},
)
)
assert len(channel._client.api.group_calls) == 1
call = channel._client.api.group_calls[0]
assert call == {
"group_openid": "group123",
"msg_type": 2,
"markdown": {"content": "**hello**"},
"msg_id": "msg1",
"msg_seq": 2,
}

2
tests/test_slack_channel.py
View File

@@ -5,7 +5,7 @@ import pytest
from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.slack import SlackChannel
from nanobot.config.schema import SlackConfig
from nanobot.channels.slack import SlackConfig
class _FakeAsyncWebClient:

68
tests/test_telegram_channel.py
View File

@@ -8,7 +8,7 @@ import pytest
from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.telegram import TELEGRAM_REPLY_CONTEXT_MAX_LEN, TelegramChannel
from nanobot.config.schema import TelegramConfig
from nanobot.channels.telegram import TelegramConfig
class _FakeHTTPXRequest:
@@ -446,6 +446,56 @@ async def test_download_message_media_returns_path_when_download_succeeds(
assert "[image:" in parts[0]
@pytest.mark.asyncio
async def test_download_message_media_uses_file_unique_id_when_available(
monkeypatch, tmp_path
) -> None:
media_dir = tmp_path / "media" / "telegram"
media_dir.mkdir(parents=True)
monkeypatch.setattr(
"nanobot.channels.telegram.get_media_dir",
lambda channel=None: media_dir if channel else tmp_path / "media",
)
downloaded: dict[str, str] = {}
async def _download_to_drive(path: str) -> None:
downloaded["path"] = path
channel = TelegramChannel(
TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]),
MessageBus(),
)
app = _FakeApp(lambda: None)
app.bot.get_file = AsyncMock(
return_value=SimpleNamespace(download_to_drive=_download_to_drive)
)
channel._app = app
msg = SimpleNamespace(
photo=[
SimpleNamespace(
file_id="file-id-that-should-not-be-used",
file_unique_id="stable-unique-id",
mime_type="image/jpeg",
file_name=None,
)
],
voice=None,
audio=None,
document=None,
video=None,
video_note=None,
animation=None,
)
paths, parts = await channel._download_message_media(msg)
assert downloaded["path"].endswith("stable-unique-id.jpg")
assert paths == [str(media_dir / "stable-unique-id.jpg")]
assert parts == [f"[image: {media_dir / 'stable-unique-id.jpg'}]"]
@pytest.mark.asyncio
async def test_on_message_attaches_reply_to_media_when_available(monkeypatch, tmp_path) -> None:
"""When user replies to a message with media, that media is downloaded and attached to the turn."""
@@ -597,3 +647,19 @@ async def test_forward_command_does_not_inject_reply_context() -> None:
assert len(handled) == 1
assert handled[0]["content"] == "/new"
@pytest.mark.asyncio
async def test_on_help_includes_restart_command() -> None:
channel = TelegramChannel(
TelegramConfig(enabled=True, token="123:abc", allow_from=["*"], group_policy="open"),
MessageBus(),
)
update = _make_telegram_update(text="/help", chat_type="private")
update.message.reply_text = AsyncMock()
await channel._on_help(update, None)
update.message.reply_text.assert_awaited_once()
help_text = update.message.reply_text.await_args.args[0]
assert "/restart" in help_text

8
tests/test_tool_validation.py
View File

@@ -379,9 +379,11 @@ async def test_exec_always_returns_exit_code() -> None:
async def test_exec_head_tail_truncation() -> None:
"""Long output should preserve both head and tail."""
tool = ExecTool()
# Generate output that exceeds _MAX_OUTPUT
big = "A" * 6000 + "\n" + "B" * 6000
result = await tool.execute(command=f"echo '{big}'")
# Generate output that exceeds _MAX_OUTPUT (10_000 chars)
# Use python to generate output to avoid command line length limits
result = await tool.execute(
command="python -c \"print('A' * 6000 + '\\n' + 'B' * 6000)\""
)
assert "chars truncated" in result
# Head portion should start with As
assert result.startswith("A")