Compare commits

..
Author SHA1 Message Date
Re-bin 202f0a3144 bump: 0.1.3.post7 2026-02-13 06:17:22 +00:00
Xubin RenandGitHub 92191ad2a9 Merge pull request #587 from HKUDS/fix/whatsapp-bridge-security
fix(security): bind WhatsApp bridge to localhost + optional token auth
2026-02-13 13:41:27 +08:00
Re-bin fd7e477b18 fix(security): bind WhatsApp bridge to localhost + optional token auth 2026-02-13 05:37:56 +00:00
Xubin RenandGitHub 5c398c5faf Merge pull request #567 from 3927o/feature/better-fallback-message
Add max iterations info to fallback message
2026-02-13 12:55:14 +08:00
Re-bin 32c9431191 fix: align CLI session_id default to "cli:direct" for backward compatibility 2026-02-13 04:13:16 +00:00
Re-bin 64feec6656 Merge PR #569: feat: add /new command with memory consolidation 2026-02-13 03:31:26 +00:00
Re-bin 903caaa642 feat: unified slash commands (/new, /help) across all channels 2026-02-13 03:30:21 +00:00
我惹你的温andGitHub 0fc4f109bf Merge branch 'HKUDS:main' into feat/add_new_command 2026-02-13 01:35:07 +08:00
worenidewen 24a90af6d3 feat: add /new command 2026-02-13 01:24:48 +08:00
3927o dbbbecb25c feat: improve fallback message when max iterations reached 2026-02-12 23:57:34 +08:00
Re-bin 890d7cf853 docs: news about the redesigned memory system 2026-02-12 15:28:07 +00:00
Xubin RenandGitHub dd4c06bea5 Merge pull request #565 for redesign memory system
feat: redesign memory system — two-layer architecture with grep-based retrieval
2026-02-12 23:17:26 +08:00
Re-bin 94c21fc235 feat: redesign memory system — two-layer architecture with grep-based retrieval 2026-02-12 15:02:52 +00:00
Xubin RenandGitHub a05e58cf79 Merge pull request #543 to add edit_file tool and time context to sub agent
fix(subagent): add edit_file tool and time context to sub agent
2026-02-12 15:53:10 +08:00
Re-bin de3324807f fix(subagent): add edit_file tool and time context to sub agent 2026-02-12 07:49:36 +00:00
Re-bin cc427261d9 Merge PR #533: feat(cron): add 'at' parameter for one-time scheduled tasks 2026-02-12 06:50:53 +00:00
Re-bin 7087947e0e feat(cron): add one-time 'at' schedule to skill docs and show timezone in system prompt 2026-02-12 06:50:44 +00:00
Re-bin 73935da95f Merge branch 'main' into pr-533 2026-02-12 06:36:49 +00:00
Xubin RenandGitHub da93729d41 Merge pull request #538 from Re-bin/main
feat: add interleaved chain-of-thought to agent loop
2026-02-12 14:28:58 +08:00
Re-bin d335494212 feat: add interleaved chain-of-thought to agent loop 2026-02-12 06:25:25 +00:00
3927o a66fa650a1 feat(cron): add 'at' parameter for one-time scheduled tasks 2026-02-12 11:06:57 +08:00
Re-bin b429bf9381 fix: improve long-running stability for various channels 2026-02-12 01:20:57 +00:00
Re-bin dd63337a83 Merge PR #516: fix Pydantic V2 deprecation warning 2026-02-11 14:55:17 +00:00
Re-bin cdc37e2f5e Merge branch 'main' into pr-516 2026-02-11 14:54:37 +00:00
Re-bin 554ba81473 docs: update agent community tips 2026-02-11 14:39:20 +00:00
Sergio Sánchez Vallés cbab72ab72 fix: pydantic deprecation configdict 2026-02-11 13:01:29 +01:00
Re-bin c8831a1e1e Merge PR #488: refactor CLI input with prompt_toolkit 2026-02-11 09:38:11 +00:00
Re-bin 9d304d8a41 refactor: remove Panel border from CLI output for cleaner copy-paste 2026-02-11 09:37:49 +00:00
张涔熙 33930d1265 feat(cli): revert panel removal (keep frame), preserve input rewrite 2026-02-11 11:44:37 +08:00
张涔熙 3561b6a63d feat(cli): rewrite input layer with prompt_toolkit and polish UI
- Replaces fragile input() hacks with robust prompt_toolkit.PromptSession
- Native support for multiline paste, history, and clean display
- Restores animated spinner in _thinking_ctx (now safe)
- Replaces boxed Panel with clean header for easier copying
- Adds prompt-toolkit dependency
- Adds new unit tests for input layer
2026-02-11 11:44:37 +08:00
Re-bin ea1d2d763a Merge PR #307: feat: add MiniMax support 2026-02-10 16:39:10 +00:00
Re-bin 19b19d0d4a docs: update minimax tips 2026-02-10 16:35:50 +00:00
Re-bin 39dd7feb28 resolve conflicts with main and adapt MiniMax 2026-02-10 16:27:10 +00:00
chaohuang-aiandGitHub f8de53c7c1 Update README.md 2026-02-10 20:46:13 +08:00
chaohuang-aiandGitHub eca16947be Update README.md 2026-02-10 19:51:46 +08:00
chaohuang-aiandGitHub ca7d6bf1ab Update README.md 2026-02-10 19:51:12 +08:00
chaohuang-aiandGitHub 9ee65cd681 Update README.md 2026-02-10 19:50:47 +08:00
chaohuang-aiandGitHub 08b9270e0a Update README.md 2026-02-10 19:50:09 +08:00
Re-bin c98ca70d30 docs: update provider tips 2026-02-10 08:38:36 +00:00
Re-bin ef1b062be5 fix: create skills dir on onboard 2026-02-10 07:42:39 +00:00
Re-bin 8626caff74 fix: prevent safety guard from blocking relative paths in exec tool 2026-02-10 07:39:15 +00:00
Xubin RenandGitHub caf7a1a532 Merge pull request #439 from Mrart/dinktalk
fixed dingtalk exception.
2026-02-10 15:23:36 +08:00
Re-bin cd4eeb1d20 docs: update mochat guidelines 2026-02-10 07:22:03 +00:00
Re-bin ccf3896a5b Merge pull request #389 to support MoChat 2026-02-10 07:06:34 +00:00
Re-bin ba2bdb080d refactor: streamline mochat channel 2026-02-10 07:06:04 +00:00
Re-bin d1f0615282 resolve conflicts with main; remove test_mochat_channel.py 2026-02-10 06:52:52 +00:00
ouyangwulin f634658707 fixed dingtalk exception. 2026-02-10 11:10:00 +08:00
Re-bin a779f8c453 docs: update release news 2026-02-10 03:08:17 +00:00
Re-bin 76e51ca8de docs: release v0.1.3.post6 2026-02-10 03:07:27 +00:00
tjb-tech 866942eedd fix: update agentUserId in README and change base_url to HTTPS in configuration 2026-02-09 09:12:53 +00:00
tjb-tech ef7972b6d3 Merge origin/main into feat/mochat-channel 2026-02-09 09:01:25 +00:00
tjb-tech 3779225917 refactor(channels): rename moltchat integration to mochat 2026-02-09 08:50:17 +00:00
tjb-tech 20b8a2fc58 feat(channels): add Moltchat websocket channel with polling fallback 2026-02-09 08:46:47 +00:00
Vincent Wu 3c8eadffed feat: add MiniMax provider support via LiteLLM 2026-02-08 03:55:24 +08:00
27 changed files with 1494 additions and 314 deletions
+80 -7
View File
@@ -12,14 +12,16 @@
</p>
</div>
🐈 **nanobot** is an **ultra-lightweight** personal AI assistant inspired by [Clawdbot](https://github.com/openclaw/openclaw)
🐈 **nanobot** is an **ultra-lightweight** personal AI assistant inspired by [OpenClaw](https://github.com/openclaw/openclaw)
⚡️ Delivers core agent functionality in just **~4,000** lines of code — **99% smaller** than Clawdbot's 430k+ lines.
📏 Real-time line count: **3,510 lines** (run `bash core_agent_lines.sh` to verify anytime)
📏 Real-time line count: **3,582 lines** (run `bash core_agent_lines.sh` to verify anytime)
## 📢 News
- **2026-02-12** 🧠 Redesigned memory system — Less code, more reliable. Join the [discussion](https://github.com/HKUDS/nanobot/discussions/566) about it!
- **2026-02-10** 🎉 Released v0.1.3.post6 with improvements! Check the updates [notes](https://github.com/HKUDS/nanobot/releases/tag/v0.1.3.post6) and our [roadmap](https://github.com/HKUDS/nanobot/discussions/431).
- **2026-02-09** 💬 Added Slack, Email, and QQ support — nanobot now supports multiple chat platforms!
- **2026-02-08** 🔧 Refactored Providers—adding a new LLM provider now takes just 2 simple steps! Check [here](#providers).
- **2026-02-07** 🚀 Released v0.1.3.post5 with Qwen support & several key improvements! Check [here](https://github.com/HKUDS/nanobot/releases/tag/v0.1.3.post5) for details.
@@ -94,7 +96,7 @@ pip install nanobot-ai
> [!TIP]
> Set your API key in `~/.nanobot/config.json`.
> Get API keys: [OpenRouter](https://openrouter.ai/keys) (Global) · [DashScope](https://dashscope.console.aliyun.com) (Qwen) · [Brave Search](https://brave.com/search/api/) (optional, for web search)
> Get API keys: [OpenRouter](https://openrouter.ai/keys) (Global) · [Brave Search](https://brave.com/search/api/) (optional, for web search)
**1. Initialize**
@@ -167,7 +169,7 @@ nanobot agent -m "Hello from my local LLM!"
## 💬 Chat Apps
Talk to your nanobot through Telegram, Discord, WhatsApp, Feishu, DingTalk, Slack, Email, or QQ — anytime, anywhere.
Talk to your nanobot through Telegram, Discord, WhatsApp, Feishu, Mochat, DingTalk, Slack, Email, or QQ — anytime, anywhere.
| Channel | Setup |
|---------|-------|
@@ -175,6 +177,7 @@ Talk to your nanobot through Telegram, Discord, WhatsApp, Feishu, DingTalk, Slac
| **Discord** | Easy (bot token + intents) |
| **WhatsApp** | Medium (scan QR) |
| **Feishu** | Medium (app credentials) |
| **Mochat** | Medium (claw token + websocket) |
| **DingTalk** | Medium (app credentials) |
| **Slack** | Medium (bot + app tokens) |
| **Email** | Medium (IMAP/SMTP credentials) |
@@ -214,6 +217,63 @@ nanobot gateway
</details>
<details>
<summary><b>Mochat (Claw IM)</b></summary>
Uses **Socket.IO WebSocket** by default, with HTTP polling fallback.
**1. Ask nanobot to set up Mochat for you**
Simply send this message to nanobot (replace `xxx@xxx` with your real email):
```
Read https://raw.githubusercontent.com/HKUDS/MoChat/refs/heads/main/skills/nanobot/skill.md and register on MoChat. My Email account is xxx@xxx Bind me as your owner and DM me on MoChat.
```
nanobot will automatically register, configure `~/.nanobot/config.json`, and connect to Mochat.
**2. Restart gateway**
```bash
nanobot gateway
```
That's it — nanobot handles the rest!
<br>
<details>
<summary>Manual configuration (advanced)</summary>
If you prefer to configure manually, add the following to `~/.nanobot/config.json`:
> Keep `claw_token` private. It should only be sent in `X-Claw-Token` header to your Mochat API endpoint.
```json
{
"channels": {
"mochat": {
"enabled": true,
"base_url": "https://mochat.io",
"socket_url": "https://mochat.io",
"socket_path": "/socket.io",
"claw_token": "claw_xxx",
"agent_user_id": "6982abcdef",
"sessions": ["*"],
"panels": ["*"],
"reply_delay_mode": "non-mention",
"reply_delay_ms": 120000
}
}
}
```
</details>
</details>
<details>
<summary><b>Discord</b></summary>
@@ -514,6 +574,17 @@ nanobot gateway
</details>
## 🌐 Agent Social Network
🐈 nanobot is capable of linking to the agent social network (agent community). **Just send one message and your nanobot joins automatically!**
| Platform | How to Join (send this message to your bot) |
|----------|-------------|
| [**Moltbook**](https://www.moltbook.com/) | `Read https://moltbook.com/skill.md and follow the instructions to join Moltbook` |
| [**ClawdChat**](https://clawdchat.ai/) | `Read https://clawdchat.ai/skill.md and follow the instructions to join ClawdChat` |
Simply send the command above to your nanobot (via CLI or any chat channel), and it will handle the rest.
## ⚙️ Configuration
Config file: `~/.nanobot/config.json`
@@ -523,6 +594,7 @@ Config file: `~/.nanobot/config.json`
> [!TIP]
> - **Groq** provides free voice transcription via Whisper. If configured, Telegram voice messages will be automatically transcribed.
> - **Zhipu Coding Plan**: If you're on Zhipu's coding plan, set `"apiBase": "https://open.bigmodel.cn/api/coding/paas/v4"` in your zhipu provider config.
> - **MiniMax (Mainland China)**: If your API key is from MiniMax's mainland China platform (minimaxi.com), set `"apiBase": "https://api.minimaxi.com/v1"` in your minimax provider config.
| Provider | Purpose | Get API Key |
|----------|---------|-------------|
@@ -532,6 +604,7 @@ Config file: `~/.nanobot/config.json`
| `deepseek` | LLM (DeepSeek direct) | [platform.deepseek.com](https://platform.deepseek.com) |
| `groq` | LLM + **Voice transcription** (Whisper) | [console.groq.com](https://console.groq.com) |
| `gemini` | LLM (Gemini direct) | [aistudio.google.com](https://aistudio.google.com) |
| `minimax` | LLM (MiniMax direct) | [platform.minimax.io](https://platform.minimax.io) |
| `aihubmix` | LLM (API gateway, access to all models) | [aihubmix.com](https://aihubmix.com) |
| `dashscope` | LLM (Qwen) | [dashscope.console.aliyun.com](https://dashscope.console.aliyun.com) |
| `moonshot` | LLM (Moonshot/Kimi) | [platform.moonshot.cn](https://platform.moonshot.cn) |
@@ -643,7 +716,7 @@ docker run -v ~/.nanobot:/root/.nanobot --rm nanobot onboard
# Edit config on host to add API keys
vim ~/.nanobot/config.json
# Run gateway (connects to Telegram/WhatsApp)
# Run gateway (connects to enabled channels, e.g. Telegram/Discord/Mochat)
docker run -v ~/.nanobot:/root/.nanobot -p 18790:18790 nanobot gateway
# Or run a single command
@@ -663,7 +736,7 @@ nanobot/
│ ├── subagent.py # Background task execution
│ └── tools/ # Built-in tools (incl. spawn)
├── skills/ # 🎯 Bundled skills (github, weather, tmux...)
├── channels/ # 📱 WhatsApp integration
├── channels/ # 📱 Chat channel integrations
├── bus/ # 🚌 Message routing
├── cron/ # ⏰ Scheduled tasks
├── heartbeat/ # 💓 Proactive wake-up
@@ -689,7 +762,7 @@ PRs welcome! The codebase is intentionally small and readable. 🤗
### Contributors
<a href="https://github.com/HKUDS/nanobot/graphs/contributors">
<img src="https://contrib.rocks/image?repo=HKUDS/nanobot&max=100&columns=12" />
<img src="https://contrib.rocks/image?repo=HKUDS/nanobot&max=100&columns=12&updated=20260210" alt="Contributors" />
</a>
+3 -3
View File
@@ -95,8 +95,8 @@ File operations have path traversal protection, but:
- Consider using a firewall to restrict outbound connections if needed
**WhatsApp Bridge:**
- The bridge runs on `localhost:3001` by default
- If exposing to network, use proper authentication and TLS
- The bridge binds to `127.0.0.1:3001` (localhost only, not accessible from external network)
- Set `bridgeToken` in config to enable shared-secret authentication between Python and Node.js
- Keep authentication data in `~/.nanobot/whatsapp-auth` secure (mode 0700)
### 6. Dependency Security
@@ -224,7 +224,7 @@ If you suspect a security breach:
✅ **Secure Communication**
- HTTPS for all external API calls
- TLS for Telegram API
- WebSocket security for WhatsApp bridge
- WhatsApp bridge: localhost-only binding + optional token auth
## Known Limitations
+2 -1
View File
@@ -25,11 +25,12 @@ import { join } from 'path';
const PORT = parseInt(process.env.BRIDGE_PORT || '3001', 10);
const AUTH_DIR = process.env.AUTH_DIR || join(homedir(), '.nanobot', 'whatsapp-auth');
const TOKEN = process.env.BRIDGE_TOKEN || undefined;
console.log('🐈 nanobot WhatsApp Bridge');
console.log('========================\n');
const server = new BridgeServer(PORT, AUTH_DIR);
const server = new BridgeServer(PORT, AUTH_DIR, TOKEN);
// Handle graceful shutdown
process.on('SIGINT', async () => {
+52 -27
View File
@@ -1,5 +1,6 @@
/**
* WebSocket server for Python-Node.js bridge communication.
* Security: binds to 127.0.0.1 only; optional BRIDGE_TOKEN auth.
*/
import { WebSocketServer, WebSocket } from 'ws';
@@ -21,12 +22,13 @@ export class BridgeServer {
private wa: WhatsAppClient | null = null;
private clients: Set<WebSocket> = new Set();
constructor(private port: number, private authDir: string) {}
constructor(private port: number, private authDir: string, private token?: string) {}
async start(): Promise<void> {
// Create WebSocket server
this.wss = new WebSocketServer({ port: this.port });
console.log(`🌉 Bridge server listening on ws://localhost:${this.port}`);
// Bind to localhost only — never expose to external network
this.wss = new WebSocketServer({ host: '127.0.0.1', port: this.port });
console.log(`🌉 Bridge server listening on ws://127.0.0.1:${this.port}`);
if (this.token) console.log('🔒 Token authentication enabled');
// Initialize WhatsApp client
this.wa = new WhatsAppClient({
@@ -38,35 +40,58 @@ export class BridgeServer {
// Handle WebSocket connections
this.wss.on('connection', (ws) => {
console.log('🔗 Python client connected');
this.clients.add(ws);
ws.on('message', async (data) => {
try {
const cmd = JSON.parse(data.toString()) as SendCommand;
await this.handleCommand(cmd);
ws.send(JSON.stringify({ type: 'sent', to: cmd.to }));
} catch (error) {
console.error('Error handling command:', error);
ws.send(JSON.stringify({ type: 'error', error: String(error) }));
}
});
ws.on('close', () => {
console.log('🔌 Python client disconnected');
this.clients.delete(ws);
});
ws.on('error', (error) => {
console.error('WebSocket error:', error);
this.clients.delete(ws);
});
if (this.token) {
// Require auth handshake as first message
const timeout = setTimeout(() => ws.close(4001, 'Auth timeout'), 5000);
ws.once('message', (data) => {
clearTimeout(timeout);
try {
const msg = JSON.parse(data.toString());
if (msg.type === 'auth' && msg.token === this.token) {
console.log('🔗 Python client authenticated');
this.setupClient(ws);
} else {
ws.close(4003, 'Invalid token');
}
} catch {
ws.close(4003, 'Invalid auth message');
}
});
} else {
console.log('🔗 Python client connected');
this.setupClient(ws);
}
});
// Connect to WhatsApp
await this.wa.connect();
}
private setupClient(ws: WebSocket): void {
this.clients.add(ws);
ws.on('message', async (data) => {
try {
const cmd = JSON.parse(data.toString()) as SendCommand;
await this.handleCommand(cmd);
ws.send(JSON.stringify({ type: 'sent', to: cmd.to }));
} catch (error) {
console.error('Error handling command:', error);
ws.send(JSON.stringify({ type: 'error', error: String(error) }));
}
});
ws.on('close', () => {
console.log('🔌 Python client disconnected');
this.clients.delete(ws);
});
ws.on('error', (error) => {
console.error('WebSocket error:', error);
this.clients.delete(ws);
});
}
private async handleCommand(cmd: SendCommand): Promise<void> {
if (cmd.type === 'send' && this.wa) {
await this.wa.sendMessage(cmd.to, cmd.text);
+8 -5
View File
@@ -73,7 +73,9 @@ Skills with available="false" need dependencies installed first - you can try in
def _get_identity(self) -> str:
"""Get the core identity section."""
from datetime import datetime
import time as _time
now = datetime.now().strftime("%Y-%m-%d %H:%M (%A)")
tz = _time.strftime("%Z") or "UTC"
workspace_path = str(self.workspace.expanduser().resolve())
system = platform.system()
runtime = f"{'macOS' if system == 'Darwin' else system} {platform.machine()}, Python {platform.python_version()}"
@@ -88,23 +90,24 @@ You are nanobot, a helpful AI assistant. You have access to tools that allow you
- Spawn subagents for complex background tasks
## Current Time
{now}
{now} ({tz})
## Runtime
{runtime}
## Workspace
Your workspace is at: {workspace_path}
- Memory files: {workspace_path}/memory/MEMORY.md
- Daily notes: {workspace_path}/memory/YYYY-MM-DD.md
- Long-term memory: {workspace_path}/memory/MEMORY.md
- History log: {workspace_path}/memory/HISTORY.md (grep-searchable)
- Custom skills: {workspace_path}/skills/{{skill-name}}/SKILL.md
IMPORTANT: When responding to direct questions or conversations, reply directly with your text response.
Only use the 'message' tool when you need to send a message to a specific chat channel (like WhatsApp).
For normal conversation, just respond with text - do not call the message tool.
Always be helpful, accurate, and concise. When using tools, explain what you're doing.
When remembering something, write to {workspace_path}/memory/MEMORY.md"""
Always be helpful, accurate, and concise. When using tools, think step by step: what you know, what you need, and why you chose this tool.
When remembering something important, write to {workspace_path}/memory/MEMORY.md
To recall past events, grep {workspace_path}/memory/HISTORY.md"""
def _load_bootstrap_files(self) -> str:
"""Load all bootstrap files from workspace."""
+104 -9
View File
@@ -18,6 +18,7 @@ from nanobot.agent.tools.web import WebSearchTool, WebFetchTool
from nanobot.agent.tools.message import MessageTool
from nanobot.agent.tools.spawn import SpawnTool
from nanobot.agent.tools.cron import CronTool
from nanobot.agent.memory import MemoryStore
from nanobot.agent.subagent import SubagentManager
from nanobot.session.manager import SessionManager
@@ -41,6 +42,7 @@ class AgentLoop:
workspace: Path,
model: str | None = None,
max_iterations: int = 20,
memory_window: int = 50,
brave_api_key: str | None = None,
exec_config: "ExecToolConfig | None" = None,
cron_service: "CronService | None" = None,
@@ -54,6 +56,7 @@ class AgentLoop:
self.workspace = workspace
self.model = model or provider.get_default_model()
self.max_iterations = max_iterations
self.memory_window = memory_window
self.brave_api_key = brave_api_key
self.exec_config = exec_config or ExecToolConfig()
self.cron_service = cron_service
@@ -141,12 +144,13 @@ class AgentLoop:
self._running = False
logger.info("Agent loop stopping")
async def _process_message(self, msg: InboundMessage) -> OutboundMessage | None:
async def _process_message(self, msg: InboundMessage, session_key: str | None = None) -> OutboundMessage | None:
"""
Process a single inbound message.
Args:
msg: The inbound message to process.
session_key: Override session key (used by process_direct).
Returns:
The response message, or None if no response needed.
@@ -160,7 +164,24 @@ class AgentLoop:
logger.info(f"Processing message from {msg.channel}:{msg.sender_id}: {preview}")
# Get or create session
session = self.sessions.get_or_create(msg.session_key)
key = session_key or msg.session_key
session = self.sessions.get_or_create(key)
# Handle slash commands
cmd = msg.content.strip().lower()
if cmd == "/new":
await self._consolidate_memory(session, archive_all=True)
session.clear()
self.sessions.save(session)
return OutboundMessage(channel=msg.channel, chat_id=msg.chat_id,
content="🐈 New session started. Memory consolidated.")
if cmd == "/help":
return OutboundMessage(channel=msg.channel, chat_id=msg.chat_id,
content="🐈 nanobot commands:\n/new — Start a new conversation\n/help — Show available commands")
# Consolidate memory before processing if session is too large
if len(session.messages) > self.memory_window:
await self._consolidate_memory(session)
# Update tool contexts
message_tool = self.tools.get("message")
@@ -187,6 +208,7 @@ class AgentLoop:
# Agent loop
iteration = 0
final_content = None
tools_used: list[str] = []
while iteration < self.max_iterations:
iteration += 1
@@ -219,27 +241,34 @@ class AgentLoop:
# Execute tools
for tool_call in response.tool_calls:
tools_used.append(tool_call.name)
args_str = json.dumps(tool_call.arguments, ensure_ascii=False)
logger.info(f"Tool call: {tool_call.name}({args_str[:200]})")
result = await self.tools.execute(tool_call.name, tool_call.arguments)
messages = self.context.add_tool_result(
messages, tool_call.id, tool_call.name, result
)
# Interleaved CoT: reflect before next action
messages.append({"role": "user", "content": "Reflect on the results and decide next steps."})
else:
# No tool calls, we're done
final_content = response.content
break
if final_content is None:
final_content = "I've completed processing but have no response to give."
if iteration >= self.max_iterations:
final_content = f"Reached {self.max_iterations} iterations without completion."
else:
final_content = "I've completed processing but have no response to give."
# Log response preview
preview = final_content[:120] + "..." if len(final_content) > 120 else final_content
logger.info(f"Response to {msg.channel}:{msg.sender_id}: {preview}")
# Save to session
# Save to session (include tool names so consolidation sees what happened)
session.add_message("user", msg.content)
session.add_message("assistant", final_content)
session.add_message("assistant", final_content,
tools_used=tools_used if tools_used else None)
self.sessions.save(session)
return OutboundMessage(
@@ -330,6 +359,8 @@ class AgentLoop:
messages = self.context.add_tool_result(
messages, tool_call.id, tool_call.name, result
)
# Interleaved CoT: reflect before next action
messages.append({"role": "user", "content": "Reflect on the results and decide next steps."})
else:
final_content = response.content
break
@@ -348,6 +379,70 @@ class AgentLoop:
content=final_content
)
async def _consolidate_memory(self, session, archive_all: bool = False) -> None:
"""Consolidate old messages into MEMORY.md + HISTORY.md, then trim session."""
if not session.messages:
return
memory = MemoryStore(self.workspace)
if archive_all:
old_messages = session.messages
keep_count = 0
else:
keep_count = min(10, max(2, self.memory_window // 2))
old_messages = session.messages[:-keep_count]
if not old_messages:
return
logger.info(f"Memory consolidation started: {len(session.messages)} messages, archiving {len(old_messages)}, keeping {keep_count}")
# Format messages for LLM (include tool names when available)
lines = []
for m in old_messages:
if not m.get("content"):
continue
tools = f" [tools: {', '.join(m['tools_used'])}]" if m.get("tools_used") else ""
lines.append(f"[{m.get('timestamp', '?')[:16]}] {m['role'].upper()}{tools}: {m['content']}")
conversation = "\n".join(lines)
current_memory = memory.read_long_term()
prompt = f"""You are a memory consolidation agent. Process this conversation and return a JSON object with exactly two keys:
1. "history_entry": A paragraph (2-5 sentences) summarizing the key events/decisions/topics. Start with a timestamp like [YYYY-MM-DD HH:MM]. Include enough detail to be useful when found by grep search later.
2. "memory_update": The updated long-term memory content. Add any new facts: user location, preferences, personal info, habits, project context, technical decisions, tools/services used. If nothing new, return the existing content unchanged.
## Current Long-term Memory
{current_memory or "(empty)"}
## Conversation to Process
{conversation}
Respond with ONLY valid JSON, no markdown fences."""
try:
response = await self.provider.chat(
messages=[
{"role": "system", "content": "You are a memory consolidation agent. Respond only with valid JSON."},
{"role": "user", "content": prompt},
],
model=self.model,
)
text = (response.content or "").strip()
if text.startswith("```"):
text = text.split("\n", 1)[-1].rsplit("```", 1)[0].strip()
result = json.loads(text)
if entry := result.get("history_entry"):
memory.append_history(entry)
if update := result.get("memory_update"):
if update != current_memory:
memory.write_long_term(update)
session.messages = session.messages[-keep_count:] if keep_count else []
self.sessions.save(session)
logger.info(f"Memory consolidation done, session trimmed to {len(session.messages)} messages")
except Exception as e:
logger.error(f"Memory consolidation failed: {e}")
async def process_direct(
self,
content: str,
@@ -360,9 +455,9 @@ class AgentLoop:
Args:
content: The message content.
session_key: Session identifier.
channel: Source channel (for context).
chat_id: Source chat ID (for context).
session_key: Session identifier (overrides channel:chat_id for session lookup).
channel: Source channel (for tool context routing).
chat_id: Source chat ID (for tool context routing).
Returns:
The agent's response.
@@ -374,5 +469,5 @@ class AgentLoop:
content=content
)
response = await self._process_message(msg)
response = await self._process_message(msg, session_key=session_key)
return response.content if response else ""
+12 -91
View File
@@ -1,109 +1,30 @@
"""Memory system for persistent agent memory."""
from pathlib import Path
from datetime import datetime
from nanobot.utils.helpers import ensure_dir, today_date
from nanobot.utils.helpers import ensure_dir
class MemoryStore:
"""
Memory system for the agent.
Supports daily notes (memory/YYYY-MM-DD.md) and long-term memory (MEMORY.md).
"""
"""Two-layer memory: MEMORY.md (long-term facts) + HISTORY.md (grep-searchable log)."""
def __init__(self, workspace: Path):
self.workspace = workspace
self.memory_dir = ensure_dir(workspace / "memory")
self.memory_file = self.memory_dir / "MEMORY.md"
def get_today_file(self) -> Path:
"""Get path to today's memory file."""
return self.memory_dir / f"{today_date()}.md"
def read_today(self) -> str:
"""Read today's memory notes."""
today_file = self.get_today_file()
if today_file.exists():
return today_file.read_text(encoding="utf-8")
return ""
def append_today(self, content: str) -> None:
"""Append content to today's memory notes."""
today_file = self.get_today_file()
if today_file.exists():
existing = today_file.read_text(encoding="utf-8")
content = existing + "\n" + content
else:
# Add header for new day
header = f"# {today_date()}\n\n"
content = header + content
today_file.write_text(content, encoding="utf-8")
self.history_file = self.memory_dir / "HISTORY.md"
def read_long_term(self) -> str:
"""Read long-term memory (MEMORY.md)."""
if self.memory_file.exists():
return self.memory_file.read_text(encoding="utf-8")
return ""
def write_long_term(self, content: str) -> None:
"""Write to long-term memory (MEMORY.md)."""
self.memory_file.write_text(content, encoding="utf-8")
def get_recent_memories(self, days: int = 7) -> str:
"""
Get memories from the last N days.
Args:
days: Number of days to look back.
Returns:
Combined memory content.
"""
from datetime import timedelta
memories = []
today = datetime.now().date()
for i in range(days):
date = today - timedelta(days=i)
date_str = date.strftime("%Y-%m-%d")
file_path = self.memory_dir / f"{date_str}.md"
if file_path.exists():
content = file_path.read_text(encoding="utf-8")
memories.append(content)
return "\n\n---\n\n".join(memories)
def list_memory_files(self) -> list[Path]:
"""List all memory files sorted by date (newest first)."""
if not self.memory_dir.exists():
return []
files = list(self.memory_dir.glob("????-??-??.md"))
return sorted(files, reverse=True)
def append_history(self, entry: str) -> None:
with open(self.history_file, "a", encoding="utf-8") as f:
f.write(entry.rstrip() + "\n\n")
def get_memory_context(self) -> str:
"""
Get memory context for the agent.
Returns:
Formatted memory context including long-term and recent memories.
"""
parts = []
# Long-term memory
long_term = self.read_long_term()
if long_term:
parts.append("## Long-term Memory\n" + long_term)
# Today's notes
today = self.read_today()
if today:
parts.append("## Today's Notes\n" + today)
return "\n\n".join(parts) if parts else ""
return f"## Long-term Memory\n{long_term}" if long_term else ""
+11 -4
View File
@@ -12,7 +12,7 @@ from nanobot.bus.events import InboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.providers.base import LLMProvider
from nanobot.agent.tools.registry import ToolRegistry
from nanobot.agent.tools.filesystem import ReadFileTool, WriteFileTool, ListDirTool
from nanobot.agent.tools.filesystem import ReadFileTool, WriteFileTool, EditFileTool, ListDirTool
from nanobot.agent.tools.shell import ExecTool
from nanobot.agent.tools.web import WebSearchTool, WebFetchTool
@@ -101,6 +101,7 @@ class SubagentManager:
allowed_dir = self.workspace if self.restrict_to_workspace else None
tools.register(ReadFileTool(allowed_dir=allowed_dir))
tools.register(WriteFileTool(allowed_dir=allowed_dir))
tools.register(EditFileTool(allowed_dir=allowed_dir))
tools.register(ListDirTool(allowed_dir=allowed_dir))
tools.register(ExecTool(
working_dir=str(self.workspace),
@@ -210,12 +211,17 @@ Summarize this naturally for the user. Keep it brief (1-2 sentences). Do not men
def _build_subagent_prompt(self, task: str) -> str:
"""Build a focused system prompt for the subagent."""
from datetime import datetime
import time as _time
now = datetime.now().strftime("%Y-%m-%d %H:%M (%A)")
tz = _time.strftime("%Z") or "UTC"
return f"""# Subagent
You are a subagent spawned by the main agent to complete a specific task.
## Current Time
{now} ({tz})
## Your Task
{task}
You are a subagent spawned by the main agent to complete a specific task.
## Rules
1. Stay focused - complete only the assigned task, nothing else
@@ -236,6 +242,7 @@ You are a subagent spawned by the main agent to complete a specific task.
## Workspace
Your workspace is at: {self.workspace}
Skills are available at: {self.workspace}/skills/ (read SKILL.md files as needed)
When you have completed the task, provide a clear summary of your findings or actions."""
+16 -3
View File
@@ -50,6 +50,10 @@ class CronTool(Tool):
"type": "string",
"description": "Cron expression like '0 9 * * *' (for scheduled tasks)"
},
"at": {
"type": "string",
"description": "ISO datetime for one-time execution (e.g. '2026-02-12T10:30:00')"
},
"job_id": {
"type": "string",
"description": "Job ID (for remove)"
@@ -64,30 +68,38 @@ class CronTool(Tool):
message: str = "",
every_seconds: int | None = None,
cron_expr: str | None = None,
at: str | None = None,
job_id: str | None = None,
**kwargs: Any
) -> str:
if action == "add":
return self._add_job(message, every_seconds, cron_expr)
return self._add_job(message, every_seconds, cron_expr, at)
elif action == "list":
return self._list_jobs()
elif action == "remove":
return self._remove_job(job_id)
return f"Unknown action: {action}"
def _add_job(self, message: str, every_seconds: int | None, cron_expr: str | None) -> str:
def _add_job(self, message: str, every_seconds: int | None, cron_expr: str | None, at: str | None) -> str:
if not message:
return "Error: message is required for add"
if not self._channel or not self._chat_id:
return "Error: no session context (channel/chat_id)"
# Build schedule
delete_after = False
if every_seconds:
schedule = CronSchedule(kind="every", every_ms=every_seconds * 1000)
elif cron_expr:
schedule = CronSchedule(kind="cron", expr=cron_expr)
elif at:
from datetime import datetime
dt = datetime.fromisoformat(at)
at_ms = int(dt.timestamp() * 1000)
schedule = CronSchedule(kind="at", at_ms=at_ms)
delete_after = True
else:
return "Error: either every_seconds or cron_expr is required"
return "Error: either every_seconds, cron_expr, or at is required"
job = self._cron.add_job(
name=message[:30],
@@ -96,6 +108,7 @@ class CronTool(Tool):
deliver=True,
channel=self._channel,
to=self._chat_id,
delete_after_run=delete_after,
)
return f"Created job '{job.name}' (id: {job.id})"
+6 -3
View File
@@ -128,14 +128,17 @@ class ExecTool(Tool):
cwd_path = Path(cwd).resolve()
win_paths = re.findall(r"[A-Za-z]:\\[^\\\"']+", cmd)
posix_paths = re.findall(r"/[^\s\"']+", cmd)
# Only match absolute paths — avoid false positives on relative
# paths like ".venv/bin/python" where "/bin/python" would be
# incorrectly extracted by the old pattern.
posix_paths = re.findall(r"(?:^|[\s|>])(/[^\s\"'>]+)", cmd)
for raw in win_paths + posix_paths:
try:
p = Path(raw).resolve()
p = Path(raw.strip()).resolve()
except Exception:
continue
if cwd_path not in p.parents and p != cwd_path:
if p.is_absolute() and cwd_path not in p.parents and p != cwd_path:
return "Error: Command blocked by safety guard (path outside working dir)"
return None
+9 -2
View File
@@ -137,8 +137,15 @@ class DingTalkChannel(BaseChannel):
logger.info("DingTalk bot started with Stream Mode")
# client.start() is an async infinite loop handling the websocket connection
await self._client.start()
# Reconnect loop: restart stream if SDK exits or crashes
while self._running:
try:
await self._client.start()
except Exception as e:
logger.warning(f"DingTalk stream error: {e}")
if self._running:
logger.info("Reconnecting DingTalk stream in 5 seconds...")
await asyncio.sleep(5)
except Exception as e:
logger.exception(f"Failed to start DingTalk channel: {e}")
+8 -5
View File
@@ -98,12 +98,15 @@ class FeishuChannel(BaseChannel):
log_level=lark.LogLevel.INFO
)
# Start WebSocket client in a separate thread
# Start WebSocket client in a separate thread with reconnect loop
def run_ws():
try:
self._ws_client.start()
except Exception as e:
logger.error(f"Feishu WebSocket error: {e}")
while self._running:
try:
self._ws_client.start()
except Exception as e:
logger.warning(f"Feishu WebSocket error: {e}")
if self._running:
import time; time.sleep(5)
self._ws_thread = threading.Thread(target=run_ws, daemon=True)
self._ws_thread.start()
+14 -7
View File
@@ -3,7 +3,7 @@
from __future__ import annotations
import asyncio
from typing import Any, TYPE_CHECKING
from typing import Any
from loguru import logger
@@ -12,9 +12,6 @@ from nanobot.bus.queue import MessageBus
from nanobot.channels.base import BaseChannel
from nanobot.config.schema import Config
if TYPE_CHECKING:
from nanobot.session.manager import SessionManager
class ChannelManager:
"""
@@ -26,10 +23,9 @@ class ChannelManager:
- Route outbound messages
"""
def __init__(self, config: Config, bus: MessageBus, session_manager: "SessionManager | None" = None):
def __init__(self, config: Config, bus: MessageBus):
self.config = config
self.bus = bus
self.session_manager = session_manager
self.channels: dict[str, BaseChannel] = {}
self._dispatch_task: asyncio.Task | None = None
@@ -46,7 +42,6 @@ class ChannelManager:
self.config.channels.telegram,
self.bus,
groq_api_key=self.config.providers.groq.api_key,
session_manager=self.session_manager,
)
logger.info("Telegram channel enabled")
except ImportError as e:
@@ -85,6 +80,18 @@ class ChannelManager:
except ImportError as e:
logger.warning(f"Feishu channel not available: {e}")
# Mochat channel
if self.config.channels.mochat.enabled:
try:
from nanobot.channels.mochat import MochatChannel
self.channels["mochat"] = MochatChannel(
self.config.channels.mochat, self.bus
)
logger.info("Mochat channel enabled")
except ImportError as e:
logger.warning(f"Mochat channel not available: {e}")
# DingTalk channel
if self.config.channels.dingtalk.enabled:
try:
+895
View File
@@ -0,0 +1,895 @@
"""Mochat channel implementation using Socket.IO with HTTP polling fallback."""
from __future__ import annotations
import asyncio
import json
from collections import deque
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any
import httpx
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 MochatConfig
from nanobot.utils.helpers import get_data_path
try:
import socketio
SOCKETIO_AVAILABLE = True
except ImportError:
socketio = None
SOCKETIO_AVAILABLE = False
try:
import msgpack # noqa: F401
MSGPACK_AVAILABLE = True
except ImportError:
MSGPACK_AVAILABLE = False
MAX_SEEN_MESSAGE_IDS = 2000
CURSOR_SAVE_DEBOUNCE_S = 0.5
# ---------------------------------------------------------------------------
# Data classes
# ---------------------------------------------------------------------------
@dataclass
class MochatBufferedEntry:
"""Buffered inbound entry for delayed dispatch."""
raw_body: str
author: str
sender_name: str = ""
sender_username: str = ""
timestamp: int | None = None
message_id: str = ""
group_id: str = ""
@dataclass
class DelayState:
"""Per-target delayed message state."""
entries: list[MochatBufferedEntry] = field(default_factory=list)
lock: asyncio.Lock = field(default_factory=asyncio.Lock)
timer: asyncio.Task | None = None
@dataclass
class MochatTarget:
"""Outbound target resolution result."""
id: str
is_panel: bool
# ---------------------------------------------------------------------------
# Pure helpers
# ---------------------------------------------------------------------------
def _safe_dict(value: Any) -> dict:
"""Return *value* if it's a dict, else empty dict."""
return value if isinstance(value, dict) else {}
def _str_field(src: dict, *keys: str) -> str:
"""Return the first non-empty str value found for *keys*, stripped."""
for k in keys:
v = src.get(k)
if isinstance(v, str) and v.strip():
return v.strip()
return ""
def _make_synthetic_event(
message_id: str, author: str, content: Any,
meta: Any, group_id: str, converse_id: str,
timestamp: Any = None, *, author_info: Any = None,
) -> dict[str, Any]:
"""Build a synthetic ``message.add`` event dict."""
payload: dict[str, Any] = {
"messageId": message_id, "author": author,
"content": content, "meta": _safe_dict(meta),
"groupId": group_id, "converseId": converse_id,
}
if author_info is not None:
payload["authorInfo"] = _safe_dict(author_info)
return {
"type": "message.add",
"timestamp": timestamp or datetime.utcnow().isoformat(),
"payload": payload,
}
def normalize_mochat_content(content: Any) -> str:
"""Normalize content payload to text."""
if isinstance(content, str):
return content.strip()
if content is None:
return ""
try:
return json.dumps(content, ensure_ascii=False)
except TypeError:
return str(content)
def resolve_mochat_target(raw: str) -> MochatTarget:
"""Resolve id and target kind from user-provided target string."""
trimmed = (raw or "").strip()
if not trimmed:
return MochatTarget(id="", is_panel=False)
lowered = trimmed.lower()
cleaned, forced_panel = trimmed, False
for prefix in ("mochat:", "group:", "channel:", "panel:"):
if lowered.startswith(prefix):
cleaned = trimmed[len(prefix):].strip()
forced_panel = prefix in {"group:", "channel:", "panel:"}
break
if not cleaned:
return MochatTarget(id="", is_panel=False)
return MochatTarget(id=cleaned, is_panel=forced_panel or not cleaned.startswith("session_"))
def extract_mention_ids(value: Any) -> list[str]:
"""Extract mention ids from heterogeneous mention payload."""
if not isinstance(value, list):
return []
ids: list[str] = []
for item in value:
if isinstance(item, str):
if item.strip():
ids.append(item.strip())
elif isinstance(item, dict):
for key in ("id", "userId", "_id"):
candidate = item.get(key)
if isinstance(candidate, str) and candidate.strip():
ids.append(candidate.strip())
break
return ids
def resolve_was_mentioned(payload: dict[str, Any], agent_user_id: str) -> bool:
"""Resolve mention state from payload metadata and text fallback."""
meta = payload.get("meta")
if isinstance(meta, dict):
if meta.get("mentioned") is True or meta.get("wasMentioned") is True:
return True
for f in ("mentions", "mentionIds", "mentionedUserIds", "mentionedUsers"):
if agent_user_id and agent_user_id in extract_mention_ids(meta.get(f)):
return True
if not agent_user_id:
return False
content = payload.get("content")
if not isinstance(content, str) or not content:
return False
return f"<@{agent_user_id}>" in content or f"@{agent_user_id}" in content
def resolve_require_mention(config: MochatConfig, session_id: str, group_id: str) -> bool:
"""Resolve mention requirement for group/panel conversations."""
groups = config.groups or {}
for key in (group_id, session_id, "*"):
if key and key in groups:
return bool(groups[key].require_mention)
return bool(config.mention.require_in_groups)
def build_buffered_body(entries: list[MochatBufferedEntry], is_group: bool) -> str:
"""Build text body from one or more buffered entries."""
if not entries:
return ""
if len(entries) == 1:
return entries[0].raw_body
lines: list[str] = []
for entry in entries:
if not entry.raw_body:
continue
if is_group:
label = entry.sender_name.strip() or entry.sender_username.strip() or entry.author
if label:
lines.append(f"{label}: {entry.raw_body}")
continue
lines.append(entry.raw_body)
return "\n".join(lines).strip()
def parse_timestamp(value: Any) -> int | None:
"""Parse event timestamp to epoch milliseconds."""
if not isinstance(value, str) or not value.strip():
return None
try:
return int(datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp() * 1000)
except ValueError:
return None
# ---------------------------------------------------------------------------
# Channel
# ---------------------------------------------------------------------------
class MochatChannel(BaseChannel):
"""Mochat channel using socket.io with fallback polling workers."""
name = "mochat"
def __init__(self, config: MochatConfig, bus: MessageBus):
super().__init__(config, bus)
self.config: MochatConfig = config
self._http: httpx.AsyncClient | None = None
self._socket: Any = None
self._ws_connected = self._ws_ready = False
self._state_dir = get_data_path() / "mochat"
self._cursor_path = self._state_dir / "session_cursors.json"
self._session_cursor: dict[str, int] = {}
self._cursor_save_task: asyncio.Task | None = None
self._session_set: set[str] = set()
self._panel_set: set[str] = set()
self._auto_discover_sessions = self._auto_discover_panels = False
self._cold_sessions: set[str] = set()
self._session_by_converse: dict[str, str] = {}
self._seen_set: dict[str, set[str]] = {}
self._seen_queue: dict[str, deque[str]] = {}
self._delay_states: dict[str, DelayState] = {}
self._fallback_mode = False
self._session_fallback_tasks: dict[str, asyncio.Task] = {}
self._panel_fallback_tasks: dict[str, asyncio.Task] = {}
self._refresh_task: asyncio.Task | None = None
self._target_locks: dict[str, asyncio.Lock] = {}
# ---- lifecycle ---------------------------------------------------------
async def start(self) -> None:
"""Start Mochat channel workers and websocket connection."""
if not self.config.claw_token:
logger.error("Mochat claw_token not configured")
return
self._running = True
self._http = httpx.AsyncClient(timeout=30.0)
self._state_dir.mkdir(parents=True, exist_ok=True)
await self._load_session_cursors()
self._seed_targets_from_config()
await self._refresh_targets(subscribe_new=False)
if not await self._start_socket_client():
await self._ensure_fallback_workers()
self._refresh_task = asyncio.create_task(self._refresh_loop())
while self._running:
await asyncio.sleep(1)
async def stop(self) -> None:
"""Stop all workers and clean up resources."""
self._running = False
if self._refresh_task:
self._refresh_task.cancel()
self._refresh_task = None
await self._stop_fallback_workers()
await self._cancel_delay_timers()
if self._socket:
try:
await self._socket.disconnect()
except Exception:
pass
self._socket = None
if self._cursor_save_task:
self._cursor_save_task.cancel()
self._cursor_save_task = None
await self._save_session_cursors()
if self._http:
await self._http.aclose()
self._http = None
self._ws_connected = self._ws_ready = False
async def send(self, msg: OutboundMessage) -> None:
"""Send outbound message to session or panel."""
if not self.config.claw_token:
logger.warning("Mochat claw_token missing, skip send")
return
parts = ([msg.content.strip()] if msg.content and msg.content.strip() else [])
if msg.media:
parts.extend(m for m in msg.media if isinstance(m, str) and m.strip())
content = "\n".join(parts).strip()
if not content:
return
target = resolve_mochat_target(msg.chat_id)
if not target.id:
logger.warning("Mochat outbound target is empty")
return
is_panel = (target.is_panel or target.id in self._panel_set) and not target.id.startswith("session_")
try:
if is_panel:
await self._api_send("/api/claw/groups/panels/send", "panelId", target.id,
content, msg.reply_to, self._read_group_id(msg.metadata))
else:
await self._api_send("/api/claw/sessions/send", "sessionId", target.id,
content, msg.reply_to)
except Exception as e:
logger.error(f"Failed to send Mochat message: {e}")
# ---- config / init helpers ---------------------------------------------
def _seed_targets_from_config(self) -> None:
sessions, self._auto_discover_sessions = self._normalize_id_list(self.config.sessions)
panels, self._auto_discover_panels = self._normalize_id_list(self.config.panels)
self._session_set.update(sessions)
self._panel_set.update(panels)
for sid in sessions:
if sid not in self._session_cursor:
self._cold_sessions.add(sid)
@staticmethod
def _normalize_id_list(values: list[str]) -> tuple[list[str], bool]:
cleaned = [str(v).strip() for v in values if str(v).strip()]
return sorted({v for v in cleaned if v != "*"}), "*" in cleaned
# ---- websocket ---------------------------------------------------------
async def _start_socket_client(self) -> bool:
if not SOCKETIO_AVAILABLE:
logger.warning("python-socketio not installed, Mochat using polling fallback")
return False
serializer = "default"
if not self.config.socket_disable_msgpack:
if MSGPACK_AVAILABLE:
serializer = "msgpack"
else:
logger.warning("msgpack not installed but socket_disable_msgpack=false; using JSON")
client = socketio.AsyncClient(
reconnection=True,
reconnection_attempts=self.config.max_retry_attempts or None,
reconnection_delay=max(0.1, self.config.socket_reconnect_delay_ms / 1000.0),
reconnection_delay_max=max(0.1, self.config.socket_max_reconnect_delay_ms / 1000.0),
logger=False, engineio_logger=False, serializer=serializer,
)
@client.event
async def connect() -> None:
self._ws_connected, self._ws_ready = True, False
logger.info("Mochat websocket connected")
subscribed = await self._subscribe_all()
self._ws_ready = subscribed
await (self._stop_fallback_workers() if subscribed else self._ensure_fallback_workers())
@client.event
async def disconnect() -> None:
if not self._running:
return
self._ws_connected = self._ws_ready = False
logger.warning("Mochat websocket disconnected")
await self._ensure_fallback_workers()
@client.event
async def connect_error(data: Any) -> None:
logger.error(f"Mochat websocket connect error: {data}")
@client.on("claw.session.events")
async def on_session_events(payload: dict[str, Any]) -> None:
await self._handle_watch_payload(payload, "session")
@client.on("claw.panel.events")
async def on_panel_events(payload: dict[str, Any]) -> None:
await self._handle_watch_payload(payload, "panel")
for ev in ("notify:chat.inbox.append", "notify:chat.message.add",
"notify:chat.message.update", "notify:chat.message.recall",
"notify:chat.message.delete"):
client.on(ev, self._build_notify_handler(ev))
socket_url = (self.config.socket_url or self.config.base_url).strip().rstrip("/")
socket_path = (self.config.socket_path or "/socket.io").strip().lstrip("/")
try:
self._socket = client
await client.connect(
socket_url, transports=["websocket"], socketio_path=socket_path,
auth={"token": self.config.claw_token},
wait_timeout=max(1.0, self.config.socket_connect_timeout_ms / 1000.0),
)
return True
except Exception as e:
logger.error(f"Failed to connect Mochat websocket: {e}")
try:
await client.disconnect()
except Exception:
pass
self._socket = None
return False
def _build_notify_handler(self, event_name: str):
async def handler(payload: Any) -> None:
if event_name == "notify:chat.inbox.append":
await self._handle_notify_inbox_append(payload)
elif event_name.startswith("notify:chat.message."):
await self._handle_notify_chat_message(payload)
return handler
# ---- subscribe ---------------------------------------------------------
async def _subscribe_all(self) -> bool:
ok = await self._subscribe_sessions(sorted(self._session_set))
ok = await self._subscribe_panels(sorted(self._panel_set)) and ok
if self._auto_discover_sessions or self._auto_discover_panels:
await self._refresh_targets(subscribe_new=True)
return ok
async def _subscribe_sessions(self, session_ids: list[str]) -> bool:
if not session_ids:
return True
for sid in session_ids:
if sid not in self._session_cursor:
self._cold_sessions.add(sid)
ack = await self._socket_call("com.claw.im.subscribeSessions", {
"sessionIds": session_ids, "cursors": self._session_cursor,
"limit": self.config.watch_limit,
})
if not ack.get("result"):
logger.error(f"Mochat subscribeSessions failed: {ack.get('message', 'unknown error')}")
return False
data = ack.get("data")
items: list[dict[str, Any]] = []
if isinstance(data, list):
items = [i for i in data if isinstance(i, dict)]
elif isinstance(data, dict):
sessions = data.get("sessions")
if isinstance(sessions, list):
items = [i for i in sessions if isinstance(i, dict)]
elif "sessionId" in data:
items = [data]
for p in items:
await self._handle_watch_payload(p, "session")
return True
async def _subscribe_panels(self, panel_ids: list[str]) -> bool:
if not self._auto_discover_panels and not panel_ids:
return True
ack = await self._socket_call("com.claw.im.subscribePanels", {"panelIds": panel_ids})
if not ack.get("result"):
logger.error(f"Mochat subscribePanels failed: {ack.get('message', 'unknown error')}")
return False
return True
async def _socket_call(self, event_name: str, payload: dict[str, Any]) -> dict[str, Any]:
if not self._socket:
return {"result": False, "message": "socket not connected"}
try:
raw = await self._socket.call(event_name, payload, timeout=10)
except Exception as e:
return {"result": False, "message": str(e)}
return raw if isinstance(raw, dict) else {"result": True, "data": raw}
# ---- refresh / discovery -----------------------------------------------
async def _refresh_loop(self) -> None:
interval_s = max(1.0, self.config.refresh_interval_ms / 1000.0)
while self._running:
await asyncio.sleep(interval_s)
try:
await self._refresh_targets(subscribe_new=self._ws_ready)
except Exception as e:
logger.warning(f"Mochat refresh failed: {e}")
if self._fallback_mode:
await self._ensure_fallback_workers()
async def _refresh_targets(self, subscribe_new: bool) -> None:
if self._auto_discover_sessions:
await self._refresh_sessions_directory(subscribe_new)
if self._auto_discover_panels:
await self._refresh_panels(subscribe_new)
async def _refresh_sessions_directory(self, subscribe_new: bool) -> None:
try:
response = await self._post_json("/api/claw/sessions/list", {})
except Exception as e:
logger.warning(f"Mochat listSessions failed: {e}")
return
sessions = response.get("sessions")
if not isinstance(sessions, list):
return
new_ids: list[str] = []
for s in sessions:
if not isinstance(s, dict):
continue
sid = _str_field(s, "sessionId")
if not sid:
continue
if sid not in self._session_set:
self._session_set.add(sid)
new_ids.append(sid)
if sid not in self._session_cursor:
self._cold_sessions.add(sid)
cid = _str_field(s, "converseId")
if cid:
self._session_by_converse[cid] = sid
if not new_ids:
return
if self._ws_ready and subscribe_new:
await self._subscribe_sessions(new_ids)
if self._fallback_mode:
await self._ensure_fallback_workers()
async def _refresh_panels(self, subscribe_new: bool) -> None:
try:
response = await self._post_json("/api/claw/groups/get", {})
except Exception as e:
logger.warning(f"Mochat getWorkspaceGroup failed: {e}")
return
raw_panels = response.get("panels")
if not isinstance(raw_panels, list):
return
new_ids: list[str] = []
for p in raw_panels:
if not isinstance(p, dict):
continue
pt = p.get("type")
if isinstance(pt, int) and pt != 0:
continue
pid = _str_field(p, "id", "_id")
if pid and pid not in self._panel_set:
self._panel_set.add(pid)
new_ids.append(pid)
if not new_ids:
return
if self._ws_ready and subscribe_new:
await self._subscribe_panels(new_ids)
if self._fallback_mode:
await self._ensure_fallback_workers()
# ---- fallback workers --------------------------------------------------
async def _ensure_fallback_workers(self) -> None:
if not self._running:
return
self._fallback_mode = True
for sid in sorted(self._session_set):
t = self._session_fallback_tasks.get(sid)
if not t or t.done():
self._session_fallback_tasks[sid] = asyncio.create_task(self._session_watch_worker(sid))
for pid in sorted(self._panel_set):
t = self._panel_fallback_tasks.get(pid)
if not t or t.done():
self._panel_fallback_tasks[pid] = asyncio.create_task(self._panel_poll_worker(pid))
async def _stop_fallback_workers(self) -> None:
self._fallback_mode = False
tasks = [*self._session_fallback_tasks.values(), *self._panel_fallback_tasks.values()]
for t in tasks:
t.cancel()
if tasks:
await asyncio.gather(*tasks, return_exceptions=True)
self._session_fallback_tasks.clear()
self._panel_fallback_tasks.clear()
async def _session_watch_worker(self, session_id: str) -> None:
while self._running and self._fallback_mode:
try:
payload = await self._post_json("/api/claw/sessions/watch", {
"sessionId": session_id, "cursor": self._session_cursor.get(session_id, 0),
"timeoutMs": self.config.watch_timeout_ms, "limit": self.config.watch_limit,
})
await self._handle_watch_payload(payload, "session")
except asyncio.CancelledError:
break
except Exception as e:
logger.warning(f"Mochat watch fallback error ({session_id}): {e}")
await asyncio.sleep(max(0.1, self.config.retry_delay_ms / 1000.0))
async def _panel_poll_worker(self, panel_id: str) -> None:
sleep_s = max(1.0, self.config.refresh_interval_ms / 1000.0)
while self._running and self._fallback_mode:
try:
resp = await self._post_json("/api/claw/groups/panels/messages", {
"panelId": panel_id, "limit": min(100, max(1, self.config.watch_limit)),
})
msgs = resp.get("messages")
if isinstance(msgs, list):
for m in reversed(msgs):
if not isinstance(m, dict):
continue
evt = _make_synthetic_event(
message_id=str(m.get("messageId") or ""),
author=str(m.get("author") or ""),
content=m.get("content"),
meta=m.get("meta"), group_id=str(resp.get("groupId") or ""),
converse_id=panel_id, timestamp=m.get("createdAt"),
author_info=m.get("authorInfo"),
)
await self._process_inbound_event(panel_id, evt, "panel")
except asyncio.CancelledError:
break
except Exception as e:
logger.warning(f"Mochat panel polling error ({panel_id}): {e}")
await asyncio.sleep(sleep_s)
# ---- inbound event processing ------------------------------------------
async def _handle_watch_payload(self, payload: dict[str, Any], target_kind: str) -> None:
if not isinstance(payload, dict):
return
target_id = _str_field(payload, "sessionId")
if not target_id:
return
lock = self._target_locks.setdefault(f"{target_kind}:{target_id}", asyncio.Lock())
async with lock:
prev = self._session_cursor.get(target_id, 0) if target_kind == "session" else 0
pc = payload.get("cursor")
if target_kind == "session" and isinstance(pc, int) and pc >= 0:
self._mark_session_cursor(target_id, pc)
raw_events = payload.get("events")
if not isinstance(raw_events, list):
return
if target_kind == "session" and target_id in self._cold_sessions:
self._cold_sessions.discard(target_id)
return
for event in raw_events:
if not isinstance(event, dict):
continue
seq = event.get("seq")
if target_kind == "session" and isinstance(seq, int) and seq > self._session_cursor.get(target_id, prev):
self._mark_session_cursor(target_id, seq)
if event.get("type") == "message.add":
await self._process_inbound_event(target_id, event, target_kind)
async def _process_inbound_event(self, target_id: str, event: dict[str, Any], target_kind: str) -> None:
payload = event.get("payload")
if not isinstance(payload, dict):
return
author = _str_field(payload, "author")
if not author or (self.config.agent_user_id and author == self.config.agent_user_id):
return
if not self.is_allowed(author):
return
message_id = _str_field(payload, "messageId")
seen_key = f"{target_kind}:{target_id}"
if message_id and self._remember_message_id(seen_key, message_id):
return
raw_body = normalize_mochat_content(payload.get("content")) or "[empty message]"
ai = _safe_dict(payload.get("authorInfo"))
sender_name = _str_field(ai, "nickname", "email")
sender_username = _str_field(ai, "agentId")
group_id = _str_field(payload, "groupId")
is_group = bool(group_id)
was_mentioned = resolve_was_mentioned(payload, self.config.agent_user_id)
require_mention = target_kind == "panel" and is_group and resolve_require_mention(self.config, target_id, group_id)
use_delay = target_kind == "panel" and self.config.reply_delay_mode == "non-mention"
if require_mention and not was_mentioned and not use_delay:
return
entry = MochatBufferedEntry(
raw_body=raw_body, author=author, sender_name=sender_name,
sender_username=sender_username, timestamp=parse_timestamp(event.get("timestamp")),
message_id=message_id, group_id=group_id,
)
if use_delay:
delay_key = seen_key
if was_mentioned:
await self._flush_delayed_entries(delay_key, target_id, target_kind, "mention", entry)
else:
await self._enqueue_delayed_entry(delay_key, target_id, target_kind, entry)
return
await self._dispatch_entries(target_id, target_kind, [entry], was_mentioned)
# ---- dedup / buffering -------------------------------------------------
def _remember_message_id(self, key: str, message_id: str) -> bool:
seen_set = self._seen_set.setdefault(key, set())
seen_queue = self._seen_queue.setdefault(key, deque())
if message_id in seen_set:
return True
seen_set.add(message_id)
seen_queue.append(message_id)
while len(seen_queue) > MAX_SEEN_MESSAGE_IDS:
seen_set.discard(seen_queue.popleft())
return False
async def _enqueue_delayed_entry(self, key: str, target_id: str, target_kind: str, entry: MochatBufferedEntry) -> None:
state = self._delay_states.setdefault(key, DelayState())
async with state.lock:
state.entries.append(entry)
if state.timer:
state.timer.cancel()
state.timer = asyncio.create_task(self._delay_flush_after(key, target_id, target_kind))
async def _delay_flush_after(self, key: str, target_id: str, target_kind: str) -> None:
await asyncio.sleep(max(0, self.config.reply_delay_ms) / 1000.0)
await self._flush_delayed_entries(key, target_id, target_kind, "timer", None)
async def _flush_delayed_entries(self, key: str, target_id: str, target_kind: str, reason: str, entry: MochatBufferedEntry | None) -> None:
state = self._delay_states.setdefault(key, DelayState())
async with state.lock:
if entry:
state.entries.append(entry)
current = asyncio.current_task()
if state.timer and state.timer is not current:
state.timer.cancel()
state.timer = None
entries = state.entries[:]
state.entries.clear()
if entries:
await self._dispatch_entries(target_id, target_kind, entries, reason == "mention")
async def _dispatch_entries(self, target_id: str, target_kind: str, entries: list[MochatBufferedEntry], was_mentioned: bool) -> None:
if not entries:
return
last = entries[-1]
is_group = bool(last.group_id)
body = build_buffered_body(entries, is_group) or "[empty message]"
await self._handle_message(
sender_id=last.author, chat_id=target_id, content=body,
metadata={
"message_id": last.message_id, "timestamp": last.timestamp,
"is_group": is_group, "group_id": last.group_id,
"sender_name": last.sender_name, "sender_username": last.sender_username,
"target_kind": target_kind, "was_mentioned": was_mentioned,
"buffered_count": len(entries),
},
)
async def _cancel_delay_timers(self) -> None:
for state in self._delay_states.values():
if state.timer:
state.timer.cancel()
self._delay_states.clear()
# ---- notify handlers ---------------------------------------------------
async def _handle_notify_chat_message(self, payload: Any) -> None:
if not isinstance(payload, dict):
return
group_id = _str_field(payload, "groupId")
panel_id = _str_field(payload, "converseId", "panelId")
if not group_id or not panel_id:
return
if self._panel_set and panel_id not in self._panel_set:
return
evt = _make_synthetic_event(
message_id=str(payload.get("_id") or payload.get("messageId") or ""),
author=str(payload.get("author") or ""),
content=payload.get("content"), meta=payload.get("meta"),
group_id=group_id, converse_id=panel_id,
timestamp=payload.get("createdAt"), author_info=payload.get("authorInfo"),
)
await self._process_inbound_event(panel_id, evt, "panel")
async def _handle_notify_inbox_append(self, payload: Any) -> None:
if not isinstance(payload, dict) or payload.get("type") != "message":
return
detail = payload.get("payload")
if not isinstance(detail, dict):
return
if _str_field(detail, "groupId"):
return
converse_id = _str_field(detail, "converseId")
if not converse_id:
return
session_id = self._session_by_converse.get(converse_id)
if not session_id:
await self._refresh_sessions_directory(self._ws_ready)
session_id = self._session_by_converse.get(converse_id)
if not session_id:
return
evt = _make_synthetic_event(
message_id=str(detail.get("messageId") or payload.get("_id") or ""),
author=str(detail.get("messageAuthor") or ""),
content=str(detail.get("messagePlainContent") or detail.get("messageSnippet") or ""),
meta={"source": "notify:chat.inbox.append", "converseId": converse_id},
group_id="", converse_id=converse_id, timestamp=payload.get("createdAt"),
)
await self._process_inbound_event(session_id, evt, "session")
# ---- cursor persistence ------------------------------------------------
def _mark_session_cursor(self, session_id: str, cursor: int) -> None:
if cursor < 0 or cursor < self._session_cursor.get(session_id, 0):
return
self._session_cursor[session_id] = cursor
if not self._cursor_save_task or self._cursor_save_task.done():
self._cursor_save_task = asyncio.create_task(self._save_cursor_debounced())
async def _save_cursor_debounced(self) -> None:
await asyncio.sleep(CURSOR_SAVE_DEBOUNCE_S)
await self._save_session_cursors()
async def _load_session_cursors(self) -> None:
if not self._cursor_path.exists():
return
try:
data = json.loads(self._cursor_path.read_text("utf-8"))
except Exception as e:
logger.warning(f"Failed to read Mochat cursor file: {e}")
return
cursors = data.get("cursors") if isinstance(data, dict) else None
if isinstance(cursors, dict):
for sid, cur in cursors.items():
if isinstance(sid, str) and isinstance(cur, int) and cur >= 0:
self._session_cursor[sid] = cur
async def _save_session_cursors(self) -> None:
try:
self._state_dir.mkdir(parents=True, exist_ok=True)
self._cursor_path.write_text(json.dumps({
"schemaVersion": 1, "updatedAt": datetime.utcnow().isoformat(),
"cursors": self._session_cursor,
}, ensure_ascii=False, indent=2) + "\n", "utf-8")
except Exception as e:
logger.warning(f"Failed to save Mochat cursor file: {e}")
# ---- HTTP helpers ------------------------------------------------------
async def _post_json(self, path: str, payload: dict[str, Any]) -> dict[str, Any]:
if not self._http:
raise RuntimeError("Mochat HTTP client not initialized")
url = f"{self.config.base_url.strip().rstrip('/')}{path}"
response = await self._http.post(url, headers={
"Content-Type": "application/json", "X-Claw-Token": self.config.claw_token,
}, json=payload)
if not response.is_success:
raise RuntimeError(f"Mochat HTTP {response.status_code}: {response.text[:200]}")
try:
parsed = response.json()
except Exception:
parsed = response.text
if isinstance(parsed, dict) and isinstance(parsed.get("code"), int):
if parsed["code"] != 200:
msg = str(parsed.get("message") or parsed.get("name") or "request failed")
raise RuntimeError(f"Mochat API error: {msg} (code={parsed['code']})")
data = parsed.get("data")
return data if isinstance(data, dict) else {}
return parsed if isinstance(parsed, dict) else {}
async def _api_send(self, path: str, id_key: str, id_val: str,
content: str, reply_to: str | None, group_id: str | None = None) -> dict[str, Any]:
"""Unified send helper for session and panel messages."""
body: dict[str, Any] = {id_key: id_val, "content": content}
if reply_to:
body["replyTo"] = reply_to
if group_id:
body["groupId"] = group_id
return await self._post_json(path, body)
@staticmethod
def _read_group_id(metadata: dict[str, Any]) -> str | None:
if not isinstance(metadata, dict):
return None
value = metadata.get("group_id") or metadata.get("groupId")
return value.strip() if isinstance(value, str) and value.strip() else None
+9 -6
View File
@@ -75,12 +75,15 @@ class QQChannel(BaseChannel):
logger.info("QQ bot started (C2C private message)")
async def _run_bot(self) -> None:
"""Run the bot connection."""
try:
await self._client.start(appid=self.config.app_id, secret=self.config.secret)
except Exception as e:
logger.error(f"QQ auth failed, check AppID/Secret at q.qq.com: {e}")
self._running = False
"""Run the bot connection with auto-reconnect."""
while self._running:
try:
await self._client.start(appid=self.config.app_id, secret=self.config.secret)
except Exception as e:
logger.warning(f"QQ bot error: {e}")
if self._running:
logger.info("Reconnecting QQ bot in 5 seconds...")
await asyncio.sleep(5)
async def stop(self) -> None:
"""Stop the QQ bot."""
+18 -43
View File
@@ -4,20 +4,16 @@ from __future__ import annotations
import asyncio
import re
from typing import TYPE_CHECKING
from loguru import logger
from telegram import BotCommand, Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes
from telegram.request import HTTPXRequest
from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.base import BaseChannel
from nanobot.config.schema import TelegramConfig
if TYPE_CHECKING:
from nanobot.session.manager import SessionManager
def _markdown_to_telegram_html(text: str) -> str:
"""
@@ -94,7 +90,7 @@ class TelegramChannel(BaseChannel):
# Commands registered with Telegram's command menu
BOT_COMMANDS = [
BotCommand("start", "Start the bot"),
BotCommand("reset", "Reset conversation history"),
BotCommand("new", "Start a new conversation"),
BotCommand("help", "Show available commands"),
]
@@ -103,12 +99,10 @@ class TelegramChannel(BaseChannel):
config: TelegramConfig,
bus: MessageBus,
groq_api_key: str = "",
session_manager: SessionManager | None = None,
):
super().__init__(config, bus)
self.config: TelegramConfig = config
self.groq_api_key = groq_api_key
self.session_manager = session_manager
self._app: Application | None = None
self._chat_ids: dict[str, int] = {} # Map sender_id to chat_id for replies
self._typing_tasks: dict[str, asyncio.Task] = {} # chat_id -> typing loop task
@@ -121,16 +115,18 @@ class TelegramChannel(BaseChannel):
self._running = True
# Build the application
builder = Application.builder().token(self.config.token)
# Build the application with larger connection pool to avoid pool-timeout on long runs
req = HTTPXRequest(connection_pool_size=16, pool_timeout=5.0, connect_timeout=30.0, read_timeout=30.0)
builder = Application.builder().token(self.config.token).request(req).get_updates_request(req)
if self.config.proxy:
builder = builder.proxy(self.config.proxy).get_updates_proxy(self.config.proxy)
self._app = builder.build()
self._app.add_error_handler(self._on_error)
# Add command handlers
self._app.add_handler(CommandHandler("start", self._on_start))
self._app.add_handler(CommandHandler("reset", self._on_reset))
self._app.add_handler(CommandHandler("help", self._on_help))
self._app.add_handler(CommandHandler("new", self._forward_command))
self._app.add_handler(CommandHandler("help", self._forward_command))
# Add message handler for text, photos, voice, documents
self._app.add_handler(
@@ -226,40 +222,15 @@ class TelegramChannel(BaseChannel):
"Type /help to see available commands."
)
async def _on_reset(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Handle /reset command — clear conversation history."""
async def _forward_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Forward slash commands to the bus for unified handling in AgentLoop."""
if not update.message or not update.effective_user:
return
chat_id = str(update.message.chat_id)
session_key = f"{self.name}:{chat_id}"
if self.session_manager is None:
logger.warning("/reset called but session_manager is not available")
await update.message.reply_text("⚠️ Session management is not available.")
return
session = self.session_manager.get_or_create(session_key)
msg_count = len(session.messages)
session.clear()
self.session_manager.save(session)
logger.info(f"Session reset for {session_key} (cleared {msg_count} messages)")
await update.message.reply_text("🔄 Conversation history cleared. Let's start fresh!")
async def _on_help(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Handle /help command — show available commands."""
if not update.message:
return
help_text = (
"🐈 <b>nanobot commands</b>\n\n"
"/start — Start the bot\n"
"/reset — Reset conversation history\n"
"/help — Show this help message\n\n"
"Just send me a text message to chat!"
await self._handle_message(
sender_id=str(update.effective_user.id),
chat_id=str(update.message.chat_id),
content=update.message.text,
)
await update.message.reply_text(help_text, parse_mode="HTML")
async def _on_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Handle incoming messages (text, photos, voice, documents)."""
@@ -386,6 +357,10 @@ class TelegramChannel(BaseChannel):
except Exception as e:
logger.debug(f"Typing indicator stopped for {chat_id}: {e}")
async def _on_error(self, update: object, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Log polling / handler errors instead of silently swallowing them."""
logger.error(f"Telegram error: {context.error}")
def _get_extension(self, media_type: str, mime_type: str | None) -> str:
"""Get file extension based on media type."""
if mime_type:
+3
View File
@@ -42,6 +42,9 @@ class WhatsAppChannel(BaseChannel):
try:
async with websockets.connect(bridge_url) as ws:
self._ws = ws
# Send auth token if configured
if self.config.bridge_token:
await ws.send(json.dumps({"type": "auth", "token": self.config.bridge_token}))
self._connected = True
logger.info("Connected to WhatsApp bridge")
+75 -78
View File
@@ -1,7 +1,6 @@
"""CLI commands for nanobot."""
import asyncio
import atexit
import os
import signal
from pathlib import Path
@@ -11,10 +10,14 @@ import sys
import typer
from rich.console import Console
from rich.markdown import Markdown
from rich.panel import Panel
from rich.table import Table
from rich.text import Text
from prompt_toolkit import PromptSession
from prompt_toolkit.formatted_text import HTML
from prompt_toolkit.history import FileHistory
from prompt_toolkit.patch_stdout import patch_stdout
from nanobot import __version__, __logo__
app = typer.Typer(
@@ -27,13 +30,10 @@ console = Console()
EXIT_COMMANDS = {"exit", "quit", "/exit", "/quit", ":q"}
# ---------------------------------------------------------------------------
# Lightweight CLI input: readline for arrow keys / history, termios for flush
# CLI input: prompt_toolkit for editing, paste, history, and display
# ---------------------------------------------------------------------------
_READLINE = None
_HISTORY_FILE: Path | None = None
_HISTORY_HOOK_REGISTERED = False
_USING_LIBEDIT = False
_PROMPT_SESSION: PromptSession | None = None
_SAVED_TERM_ATTRS = None # original termios settings, restored on exit
@@ -64,15 +64,6 @@ def _flush_pending_tty_input() -> None:
return
def _save_history() -> None:
if _READLINE is None or _HISTORY_FILE is None:
return
try:
_READLINE.write_history_file(str(_HISTORY_FILE))
except Exception:
return
def _restore_terminal() -> None:
"""Restore terminal to its original state (echo, line buffering, etc.)."""
if _SAVED_TERM_ATTRS is None:
@@ -84,11 +75,11 @@ def _restore_terminal() -> None:
pass
def _enable_line_editing() -> None:
"""Enable readline for arrow keys, line editing, and persistent history."""
global _READLINE, _HISTORY_FILE, _HISTORY_HOOK_REGISTERED, _USING_LIBEDIT, _SAVED_TERM_ATTRS
def _init_prompt_session() -> None:
"""Create the prompt_toolkit session with persistent file history."""
global _PROMPT_SESSION, _SAVED_TERM_ATTRS
# Save terminal state before readline touches it
# Save terminal state so we can restore it on exit
try:
import termios
_SAVED_TERM_ATTRS = termios.tcgetattr(sys.stdin.fileno())
@@ -97,43 +88,12 @@ def _enable_line_editing() -> None:
history_file = Path.home() / ".nanobot" / "history" / "cli_history"
history_file.parent.mkdir(parents=True, exist_ok=True)
_HISTORY_FILE = history_file
try:
import readline
except ImportError:
return
_READLINE = readline
_USING_LIBEDIT = "libedit" in (readline.__doc__ or "").lower()
try:
if _USING_LIBEDIT:
readline.parse_and_bind("bind ^I rl_complete")
else:
readline.parse_and_bind("tab: complete")
readline.parse_and_bind("set editing-mode emacs")
except Exception:
pass
try:
readline.read_history_file(str(history_file))
except Exception:
pass
if not _HISTORY_HOOK_REGISTERED:
atexit.register(_save_history)
_HISTORY_HOOK_REGISTERED = True
def _prompt_text() -> str:
"""Build a readline-friendly colored prompt."""
if _READLINE is None:
return "You: "
# libedit on macOS does not honor GNU readline non-printing markers.
if _USING_LIBEDIT:
return "\033[1;34mYou:\033[0m "
return "\001\033[1;34m\002You:\001\033[0m\002 "
_PROMPT_SESSION = PromptSession(
history=FileHistory(str(history_file)),
enable_open_in_editor=False,
multiline=False, # Enter submits (single line mode)
)
def _print_agent_response(response: str, render_markdown: bool) -> None:
@@ -141,15 +101,8 @@ def _print_agent_response(response: str, render_markdown: bool) -> None:
content = response or ""
body = Markdown(content) if render_markdown else Text(content)
console.print()
console.print(
Panel(
body,
title=f"{__logo__} nanobot",
title_align="left",
border_style="cyan",
padding=(0, 1),
)
)
console.print(f"[cyan]{__logo__} nanobot[/cyan]")
console.print(body)
console.print()
@@ -159,13 +112,25 @@ def _is_exit_command(command: str) -> bool:
async def _read_interactive_input_async() -> str:
"""Read user input with arrow keys and history (runs input() in a thread)."""
"""Read user input using prompt_toolkit (handles paste, history, display).
prompt_toolkit natively handles:
- Multiline paste (bracketed paste mode)
- History navigation (up/down arrows)
- Clean display (no ghost characters or artifacts)
"""
if _PROMPT_SESSION is None:
raise RuntimeError("Call _init_prompt_session() first")
try:
return await asyncio.to_thread(input, _prompt_text())
with patch_stdout():
return await _PROMPT_SESSION.prompt_async(
HTML("<b fg='ansiblue'>You:</b> "),
)
except EOFError as exc:
raise KeyboardInterrupt from exc
def version_callback(value: bool):
if value:
console.print(f"{__logo__} nanobot v{__version__}")
@@ -235,7 +200,7 @@ You are a helpful AI assistant. Be concise, accurate, and friendly.
- Always explain what you're doing before taking actions
- Ask for clarification when the request is ambiguous
- Use tools to help accomplish tasks
- Remember important information in your memory files
- Remember important information in memory/MEMORY.md; past events are logged in memory/HISTORY.md
""",
"SOUL.md": """# Soul
@@ -293,6 +258,15 @@ This file stores important information that should persist across sessions.
(Things to remember)
""")
console.print(" [dim]Created memory/MEMORY.md[/dim]")
history_file = memory_dir / "HISTORY.md"
if not history_file.exists():
history_file.write_text("")
console.print(" [dim]Created memory/HISTORY.md[/dim]")
# Create skills directory for custom user skills
skills_dir = workspace / "skills"
skills_dir.mkdir(exist_ok=True)
def _make_provider(config):
@@ -355,6 +329,7 @@ def gateway(
workspace=config.workspace_path,
model=config.agents.defaults.model,
max_iterations=config.agents.defaults.max_tool_iterations,
memory_window=config.agents.defaults.memory_window,
brave_api_key=config.tools.web.search.api_key or None,
exec_config=config.tools.exec,
cron_service=cron,
@@ -394,7 +369,7 @@ def gateway(
)
# Create channel manager
channels = ChannelManager(config, bus, session_manager=session_manager)
channels = ChannelManager(config, bus)
if channels.enabled_channels:
console.print(f"[green]✓[/green] Channels enabled: {', '.join(channels.enabled_channels)}")
@@ -435,7 +410,7 @@ def gateway(
@app.command()
def agent(
message: str = typer.Option(None, "--message", "-m", help="Message to send to the agent"),
session_id: str = typer.Option("cli:default", "--session", "-s", help="Session ID"),
session_id: str = typer.Option("cli:direct", "--session", "-s", help="Session ID"),
markdown: bool = typer.Option(True, "--markdown/--no-markdown", help="Render assistant output as Markdown"),
logs: bool = typer.Option(False, "--logs/--no-logs", help="Show nanobot runtime logs during chat"),
):
@@ -459,6 +434,9 @@ def agent(
bus=bus,
provider=provider,
workspace=config.workspace_path,
model=config.agents.defaults.model,
max_iterations=config.agents.defaults.max_tool_iterations,
memory_window=config.agents.defaults.memory_window,
brave_api_key=config.tools.web.search.api_key or None,
exec_config=config.tools.exec,
restrict_to_workspace=config.tools.restrict_to_workspace,
@@ -469,6 +447,7 @@ def agent(
if logs:
from contextlib import nullcontext
return nullcontext()
# Animated spinner is safe to use with prompt_toolkit input handling
return console.status("[dim]nanobot is thinking...[/dim]", spinner="dots")
if message:
@@ -481,13 +460,10 @@ def agent(
asyncio.run(run_once())
else:
# Interactive mode
_enable_line_editing()
_init_prompt_session()
console.print(f"{__logo__} Interactive mode (type [bold]exit[/bold] or [bold]Ctrl+C[/bold] to quit)\n")
# input() runs in a worker thread that can't be cancelled.
# Without this handler, asyncio.run() would hang waiting for it.
def _exit_on_sigint(signum, frame):
_save_history()
_restore_terminal()
console.print("\nGoodbye!")
os._exit(0)
@@ -504,7 +480,6 @@ def agent(
continue
if _is_exit_command(command):
_save_history()
_restore_terminal()
console.print("\nGoodbye!")
break
@@ -513,12 +488,10 @@ def agent(
response = await agent_loop.process_direct(user_input, session_id)
_print_agent_response(response, render_markdown=markdown)
except KeyboardInterrupt:
_save_history()
_restore_terminal()
console.print("\nGoodbye!")
break
except EOFError:
_save_history()
_restore_terminal()
console.print("\nGoodbye!")
break
@@ -561,6 +534,24 @@ def channels_status():
"" if dc.enabled else "",
dc.gateway_url
)
# Feishu
fs = config.channels.feishu
fs_config = f"app_id: {fs.app_id[:10]}..." if fs.app_id else "[dim]not configured[/dim]"
table.add_row(
"Feishu",
"" if fs.enabled else "",
fs_config
)
# Mochat
mc = config.channels.mochat
mc_base = mc.base_url or "[dim]not configured[/dim]"
table.add_row(
"Mochat",
"" if mc.enabled else "",
mc_base
)
# Telegram
tg = config.channels.telegram
@@ -645,14 +636,20 @@ def _get_bridge_dir() -> Path:
def channels_login():
"""Link device via QR code."""
import subprocess
from nanobot.config.loader import load_config
config = load_config()
bridge_dir = _get_bridge_dir()
console.print(f"{__logo__} Starting bridge...")
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
try:
subprocess.run(["npm", "start"], cwd=bridge_dir, check=True)
subprocess.run(["npm", "start"], cwd=bridge_dir, check=True, env=env)
except subprocess.CalledProcessError as e:
console.print(f"[red]Bridge failed: {e}[/red]")
except FileNotFoundError:
+45 -4
View File
@@ -1,7 +1,7 @@
"""Configuration schema using Pydantic."""
from pathlib import Path
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, ConfigDict
from pydantic_settings import BaseSettings
@@ -9,6 +9,7 @@ class WhatsAppConfig(BaseModel):
"""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
@@ -77,6 +78,42 @@ class EmailConfig(BaseModel):
allow_from: list[str] = Field(default_factory=list) # Allowed sender email addresses
class MochatMentionConfig(BaseModel):
"""Mochat mention behavior configuration."""
require_in_groups: bool = False
class MochatGroupRule(BaseModel):
"""Mochat per-group mention requirement."""
require_mention: bool = False
class MochatConfig(BaseModel):
"""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(BaseModel):
"""Slack DM policy configuration."""
enabled: bool = True
@@ -111,6 +148,7 @@ class ChannelsConfig(BaseModel):
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)
@@ -124,6 +162,7 @@ class AgentDefaults(BaseModel):
max_tokens: int = 8192
temperature: float = 0.7
max_tool_iterations: int = 20
memory_window: int = 50
class AgentsConfig(BaseModel):
@@ -150,6 +189,7 @@ class ProvidersConfig(BaseModel):
vllm: ProviderConfig = Field(default_factory=ProviderConfig)
gemini: ProviderConfig = Field(default_factory=ProviderConfig)
moonshot: ProviderConfig = Field(default_factory=ProviderConfig)
minimax: ProviderConfig = Field(default_factory=ProviderConfig)
aihubmix: ProviderConfig = Field(default_factory=ProviderConfig) # AiHubMix API gateway
@@ -243,6 +283,7 @@ class Config(BaseSettings):
return spec.default_api_base
return None
class Config:
env_prefix = "NANOBOT_"
env_nested_delimiter = "__"
model_config = ConfigDict(
env_prefix="NANOBOT_",
env_nested_delimiter="__"
)
+1 -1
View File
@@ -15,7 +15,7 @@ class LiteLLMProvider(LLMProvider):
"""
LLM provider using LiteLLM for multi-provider support.
Supports OpenRouter, Anthropic, OpenAI, Gemini, and many other providers through
Supports OpenRouter, Anthropic, OpenAI, Gemini, MiniMax, and many other providers through
a unified interface. Provider-specific logic is driven by the registry
(see providers/registry.py) — no if-elif chains needed here.
"""
+19
View File
@@ -241,6 +241,25 @@ PROVIDERS: tuple[ProviderSpec, ...] = (
),
),
# MiniMax: needs "minimax/" prefix for LiteLLM routing.
# Uses OpenAI-compatible API at api.minimax.io/v1.
ProviderSpec(
name="minimax",
keywords=("minimax",),
env_key="MINIMAX_API_KEY",
display_name="MiniMax",
litellm_prefix="minimax", # MiniMax-M2.1 → minimax/MiniMax-M2.1
skip_prefixes=("minimax/", "openrouter/"),
env_extras=(),
is_gateway=False,
is_local=False,
detect_by_key_prefix="",
detect_by_base_keyword="",
default_api_base="https://api.minimax.io/v1",
strip_model_prefix=False,
model_overrides=(),
),
# === Local deployment (matched by config key, NOT by api_base) =========
# vLLM / any OpenAI-compatible local server.
+8 -1
View File
@@ -7,10 +7,11 @@ description: Schedule reminders and recurring tasks.
Use the `cron` tool to schedule reminders or recurring tasks.
## Two Modes
## Three Modes
1. **Reminder** - message is sent directly to user
2. **Task** - message is a task description, agent executes and sends result
3. **One-time** - runs once at a specific time, then auto-deletes
## Examples
@@ -24,6 +25,11 @@ Dynamic task (agent executes each time):
cron(action="add", message="Check HKUDS/nanobot GitHub stars and report", every_seconds=600)
```
One-time scheduled task (compute ISO datetime from current time):
```
cron(action="add", message="Remind me about the meeting", at="<ISO datetime>")
```
List/remove:
```
cron(action="list")
@@ -38,3 +44,4 @@ cron(action="remove", job_id="abc123")
| every hour | every_seconds: 3600 |
| every day at 8am | cron_expr: "0 8 * * *" |
| weekdays at 5pm | cron_expr: "0 17 * * 1-5" |
| at a specific time | at: ISO datetime string (compute from current time) |
+31
View File
@@ -0,0 +1,31 @@
---
name: memory
description: Two-layer memory system with grep-based recall.
always: true
---
# Memory
## Structure
- `memory/MEMORY.md` — Long-term facts (preferences, project context, relationships). Always loaded into your context.
- `memory/HISTORY.md` — Append-only event log. NOT loaded into context. Search it with grep.
## Search Past Events
```bash
grep -i "keyword" memory/HISTORY.md
```
Use the `exec` tool to run grep. Combine patterns: `grep -iE "meeting|deadline" memory/HISTORY.md`
## When to Update MEMORY.md
Write important facts immediately using `edit_file` or `write_file`:
- User preferences ("I prefer dark mode")
- Project context ("The API uses OAuth2")
- Relationships ("Alice is the project lead")
## Auto-consolidation
Old conversations are automatically summarized and appended to HISTORY.md when the session grows large. Long-term facts are extracted to MEMORY.md. You don't need to manage this.
-11
View File
@@ -37,23 +37,12 @@ def get_sessions_path() -> Path:
return ensure_dir(get_data_path() / "sessions")
def get_memory_path(workspace: Path | None = None) -> Path:
"""Get the memory directory within the workspace."""
ws = workspace or get_workspace_path()
return ensure_dir(ws / "memory")
def get_skills_path(workspace: Path | None = None) -> Path:
"""Get the skills directory within the workspace."""
ws = workspace or get_workspace_path()
return ensure_dir(ws / "skills")
def today_date() -> str:
"""Get today's date in YYYY-MM-DD format."""
return datetime.now().strftime("%Y-%m-%d")
def timestamp() -> str:
"""Get current timestamp in ISO format."""
return datetime.now().isoformat()
+5 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "nanobot-ai"
version = "0.1.3.post6"
version = "0.1.3.post7"
description = "A lightweight personal AI assistant framework"
requires-python = ">=3.11"
license = {text = "MIT"}
@@ -32,8 +32,12 @@ dependencies = [
"python-telegram-bot[socks]>=21.0",
"lark-oapi>=1.0.0",
"socksio>=1.0.0",
"python-socketio>=5.11.0",
"msgpack>=1.0.8",
"slack-sdk>=3.26.0",
"qq-botpy>=1.0.0",
"python-socks[asyncio]>=2.4.0",
"prompt-toolkit>=3.0.0",
]
[project.optional-dependencies]
+58
View File
@@ -0,0 +1,58 @@
import asyncio
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from prompt_toolkit.formatted_text import HTML
from nanobot.cli import commands
@pytest.fixture
def mock_prompt_session():
"""Mock the global prompt session."""
mock_session = MagicMock()
mock_session.prompt_async = AsyncMock()
with patch("nanobot.cli.commands._PROMPT_SESSION", mock_session):
yield mock_session
@pytest.mark.asyncio
async def test_read_interactive_input_async_returns_input(mock_prompt_session):
"""Test that _read_interactive_input_async returns the user input from prompt_session."""
mock_prompt_session.prompt_async.return_value = "hello world"
result = await commands._read_interactive_input_async()
assert result == "hello world"
mock_prompt_session.prompt_async.assert_called_once()
args, _ = mock_prompt_session.prompt_async.call_args
assert isinstance(args[0], HTML) # Verify HTML prompt is used
@pytest.mark.asyncio
async def test_read_interactive_input_async_handles_eof(mock_prompt_session):
"""Test that EOFError converts to KeyboardInterrupt."""
mock_prompt_session.prompt_async.side_effect = EOFError()
with pytest.raises(KeyboardInterrupt):
await commands._read_interactive_input_async()
def test_init_prompt_session_creates_session():
"""Test that _init_prompt_session initializes the global session."""
# Ensure global is None before test
commands._PROMPT_SESSION = None
with patch("nanobot.cli.commands.PromptSession") as MockSession, \
patch("nanobot.cli.commands.FileHistory") as MockHistory, \
patch("pathlib.Path.home") as mock_home:
mock_home.return_value = MagicMock()
commands._init_prompt_session()
assert commands._PROMPT_SESSION is not None
MockSession.assert_called_once()
_, kwargs = MockSession.call_args
assert kwargs["multiline"] is False
assert kwargs["enable_open_in_editor"] is False
+2 -2
View File
@@ -20,8 +20,8 @@ You have access to:
## Memory
- Use `memory/` directory for daily notes
- Use `MEMORY.md` for long-term information
- `memory/MEMORY.md` — long-term facts (preferences, context, relationships)
- `memory/HISTORY.md` — append-only event log, search with grep to recall past events
## Scheduled Reminders