diff --git a/LICENSE b/LICENSE
new file mode 100644
index 0000000..24bdacc
--- /dev/null
+++ b/LICENSE
@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2025 nanobot contributors
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
\ No newline at end of file
diff --git a/README.md b/README.md
index 81336bc..92b3646 100644
--- a/README.md
+++ b/README.md
@@ -1,2 +1,285 @@
-# nanobot
-"nanobot: Extremely Simple and Lightweight Clawdbot Implementation"
+
+
+
nanobot: Your Lightweight Personal AI Assistant
+
+ 
+ 
+ 
+
+
+ 
+
+
+## β¨ Features
+
+
+
+ π Web Search |
+ π» Code & Files |
+ π
Scheduled Tasks |
+ π§ Memory |
+
+
+ 
|
+ 
|
+ 
|
+ 
|
+
+
+ | Search real-time info |
+ Read, write, execute |
+ Cron jobs & reminders |
+ Remember context |
+
+
+
+## π¦ Install
+
+**Install from PyPi**
+
+```bash
+pip install nanobot-ai
+```
+
+**Install from source** (recommended for development)
+
+```bash
+git clone https://github.com/HKUDS/nanobot.git
+cd nanobot
+pip install -e .
+```
+
+## π Quick Start
+
+> [!TIP]
+> Set your API key in `~/.nanobot/config.json`. Get one at [OpenRouter](https://openrouter.ai/keys).
+
+**1. Initialize**
+
+```bash
+nanobot onboard
+```
+
+**2. Configure** (`~/.nanobot/config.json`)
+
+```json
+{
+ "providers": {
+ "openrouter": {
+ "apiKey": "sk-or-v1-xxx"
+ }
+ }
+}
+```
+
+**3. Chat**
+
+```bash
+nanobot agent -m "What is 2+2?"
+```
+
+That's it! You have a working AI assistant in 2 minutes.
+
+## π¬ Chat Apps
+
+Talk to your nanobot through Telegram or WhatsApp β anytime, anywhere.
+
+| Channel | Setup |
+|---------|-------|
+| **Telegram** | Easy (just a token) |
+| **WhatsApp** | Medium (scan QR) |
+
+
+Telegram (Recommended)
+
+**1. Create a bot**
+- Open Telegram, search `@BotFather`
+- Send `/newbot`, follow prompts
+- Copy the token
+
+**2. Configure**
+
+```json
+{
+ "channels": {
+ "telegram": {
+ "enabled": true,
+ "token": "YOUR_BOT_TOKEN",
+ "allowFrom": ["YOUR_USER_ID"]
+ }
+ }
+}
+```
+
+> Get your user ID from `@userinfobot` on Telegram.
+
+**3. Run**
+
+```bash
+nanobot gateway
+```
+
+
+
+
+WhatsApp
+
+Requires **Node.js β₯18**.
+
+**1. Link device**
+
+```bash
+nanobot channels login
+# Scan QR with WhatsApp β Settings β Linked Devices
+```
+
+**2. Configure**
+
+```json
+{
+ "channels": {
+ "whatsapp": {
+ "enabled": true,
+ "allowFrom": ["+1234567890"]
+ }
+ }
+}
+```
+
+**3. Run** (two terminals)
+
+```bash
+# Terminal 1
+nanobot channels login
+
+# Terminal 2
+nanobot gateway
+```
+
+
+
+## βοΈ Configuration
+
+
+Full config example
+
+```json
+{
+ "agents": {
+ "defaults": {
+ "model": "anthropic/claude-opus-4-5"
+ }
+ },
+ "providers": {
+ "openrouter": {
+ "apiKey": "sk-or-v1-xxx"
+ }
+ },
+ "channels": {
+ "telegram": {
+ "enabled": true,
+ "token": "123456:ABC...",
+ "allowFrom": ["123456789"]
+ },
+ "whatsapp": {
+ "enabled": false
+ }
+ },
+ "tools": {
+ "web": {
+ "search": {
+ "apiKey": "BSA..."
+ }
+ }
+ }
+}
+```
+
+
+
+**API Keys:**
+- **OpenRouter**: https://openrouter.ai/keys
+- **Brave Search** (optional): https://brave.com/search/api/
+
+## CLI Reference
+
+| Command | Description |
+|---------|-------------|
+| `nanobot onboard` | Initialize config & workspace |
+| `nanobot agent -m "..."` | Chat with the agent |
+| `nanobot agent` | Interactive chat mode |
+| `nanobot gateway` | Start the gateway |
+| `nanobot status` | Show status |
+| `nanobot channels login` | Link WhatsApp (scan QR) |
+| `nanobot channels status` | Show channel status |
+
+
+Scheduled Tasks (Cron)
+
+```bash
+# Add a job
+nanobot cron add --name "daily" --message "Good morning!" --cron "0 9 * * *"
+nanobot cron add --name "hourly" --message "Check status" --every 3600
+
+# List jobs
+nanobot cron list
+
+# Remove a job
+nanobot cron remove
+```
+
+
+
+## π Project Structure
+
+```
+nanobot/
+βββ agent/ # π§ Core agent logic
+β βββ loop.py # Agent loop (LLM β tool execution)
+β βββ context.py # Prompt builder
+β βββ memory.py # Persistent memory
+β βββ skills.py # Skills loader
+β βββ tools/ # Built-in tools
+βββ skills/ # π― Bundled skills (github, weather, tmux...)
+βββ channels/ # π± Telegram, WhatsApp
+βββ bus/ # π Message routing
+βββ cron/ # β° Scheduled tasks
+βββ providers/ # π€ LLM providers (OpenRouter, etc.)
+βββ session/ # π¬ Conversation sessions
+βββ config/ # βοΈ Configuration
+βββ cli/ # π₯οΈ Commands
+```
+
+## πΊοΈ Roadmap
+
+- [ ] **Multi-modal** β See and hear (images, voice, video)
+- [ ] **Long-term memory** β Never forget important context
+- [ ] **Better reasoning** β Multi-step planning and reflection
+- [ ] **More integrations** β Discord, Slack, email, calendar
+- [ ] **Self-improvement** β Learn from feedback and mistakes
+
+**Want to help?** Pick an item and [open a PR](https://github.com/HKUDS/nanobot/pulls)!
+
+## π€ Contribute
+
+PRs welcome! The codebase is intentionally small and readable.
diff --git a/bridge/package.json b/bridge/package.json
new file mode 100644
index 0000000..e29fed8
--- /dev/null
+++ b/bridge/package.json
@@ -0,0 +1,26 @@
+{
+ "name": "nanobot-whatsapp-bridge",
+ "version": "0.1.0",
+ "description": "WhatsApp bridge for nanobot using Baileys",
+ "type": "module",
+ "main": "dist/index.js",
+ "scripts": {
+ "build": "tsc",
+ "start": "node dist/index.js",
+ "dev": "tsc && node dist/index.js"
+ },
+ "dependencies": {
+ "@whiskeysockets/baileys": "7.0.0-rc.9",
+ "ws": "^8.17.0",
+ "qrcode-terminal": "^0.12.0",
+ "pino": "^9.0.0"
+ },
+ "devDependencies": {
+ "@types/node": "^20.14.0",
+ "@types/ws": "^8.5.10",
+ "typescript": "^5.4.0"
+ },
+ "engines": {
+ "node": ">=20.0.0"
+ }
+}
diff --git a/bridge/src/index.ts b/bridge/src/index.ts
new file mode 100644
index 0000000..8db63ef
--- /dev/null
+++ b/bridge/src/index.ts
@@ -0,0 +1,50 @@
+#!/usr/bin/env node
+/**
+ * nanobot WhatsApp Bridge
+ *
+ * This bridge connects WhatsApp Web to nanobot's Python backend
+ * via WebSocket. It handles authentication, message forwarding,
+ * and reconnection logic.
+ *
+ * Usage:
+ * npm run build && npm start
+ *
+ * Or with custom settings:
+ * BRIDGE_PORT=3001 AUTH_DIR=~/.nanobot/whatsapp npm start
+ */
+
+// Polyfill crypto for Baileys in ESM
+import { webcrypto } from 'crypto';
+if (!globalThis.crypto) {
+ (globalThis as any).crypto = webcrypto;
+}
+
+import { BridgeServer } from './server.js';
+import { homedir } from 'os';
+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');
+
+console.log('π nanobot WhatsApp Bridge');
+console.log('========================\n');
+
+const server = new BridgeServer(PORT, AUTH_DIR);
+
+// Handle graceful shutdown
+process.on('SIGINT', async () => {
+ console.log('\n\nShutting down...');
+ await server.stop();
+ process.exit(0);
+});
+
+process.on('SIGTERM', async () => {
+ await server.stop();
+ process.exit(0);
+});
+
+// Start the server
+server.start().catch((error) => {
+ console.error('Failed to start bridge:', error);
+ process.exit(1);
+});
diff --git a/bridge/src/server.ts b/bridge/src/server.ts
new file mode 100644
index 0000000..c6fd599
--- /dev/null
+++ b/bridge/src/server.ts
@@ -0,0 +1,104 @@
+/**
+ * WebSocket server for Python-Node.js bridge communication.
+ */
+
+import { WebSocketServer, WebSocket } from 'ws';
+import { WhatsAppClient, InboundMessage } from './whatsapp.js';
+
+interface SendCommand {
+ type: 'send';
+ to: string;
+ text: string;
+}
+
+interface BridgeMessage {
+ type: 'message' | 'status' | 'qr' | 'error';
+ [key: string]: unknown;
+}
+
+export class BridgeServer {
+ private wss: WebSocketServer | null = null;
+ private wa: WhatsAppClient | null = null;
+ private clients: Set = new Set();
+
+ constructor(private port: number, private authDir: string) {}
+
+ async start(): Promise {
+ // Create WebSocket server
+ this.wss = new WebSocketServer({ port: this.port });
+ console.log(`π Bridge server listening on ws://localhost:${this.port}`);
+
+ // Initialize WhatsApp client
+ this.wa = new WhatsAppClient({
+ authDir: this.authDir,
+ onMessage: (msg) => this.broadcast({ type: 'message', ...msg }),
+ onQR: (qr) => this.broadcast({ type: 'qr', qr }),
+ onStatus: (status) => this.broadcast({ type: 'status', status }),
+ });
+
+ // 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);
+ });
+ });
+
+ // Connect to WhatsApp
+ await this.wa.connect();
+ }
+
+ private async handleCommand(cmd: SendCommand): Promise {
+ if (cmd.type === 'send' && this.wa) {
+ await this.wa.sendMessage(cmd.to, cmd.text);
+ }
+ }
+
+ private broadcast(msg: BridgeMessage): void {
+ const data = JSON.stringify(msg);
+ for (const client of this.clients) {
+ if (client.readyState === WebSocket.OPEN) {
+ client.send(data);
+ }
+ }
+ }
+
+ async stop(): Promise {
+ // Close all client connections
+ for (const client of this.clients) {
+ client.close();
+ }
+ this.clients.clear();
+
+ // Close WebSocket server
+ if (this.wss) {
+ this.wss.close();
+ this.wss = null;
+ }
+
+ // Disconnect WhatsApp
+ if (this.wa) {
+ await this.wa.disconnect();
+ this.wa = null;
+ }
+ }
+}
diff --git a/bridge/src/types.d.ts b/bridge/src/types.d.ts
new file mode 100644
index 0000000..3aeb18b
--- /dev/null
+++ b/bridge/src/types.d.ts
@@ -0,0 +1,3 @@
+declare module 'qrcode-terminal' {
+ export function generate(text: string, options?: { small?: boolean }): void;
+}
diff --git a/bridge/src/whatsapp.ts b/bridge/src/whatsapp.ts
new file mode 100644
index 0000000..4185632
--- /dev/null
+++ b/bridge/src/whatsapp.ts
@@ -0,0 +1,180 @@
+/**
+ * WhatsApp client wrapper using Baileys.
+ * Based on OpenClaw's working implementation.
+ */
+
+/* eslint-disable @typescript-eslint/no-explicit-any */
+import makeWASocket, {
+ DisconnectReason,
+ useMultiFileAuthState,
+ fetchLatestBaileysVersion,
+ makeCacheableSignalKeyStore,
+} from '@whiskeysockets/baileys';
+
+import { Boom } from '@hapi/boom';
+import qrcode from 'qrcode-terminal';
+import pino from 'pino';
+
+const VERSION = '0.1.0';
+
+export interface InboundMessage {
+ id: string;
+ sender: string;
+ content: string;
+ timestamp: number;
+ isGroup: boolean;
+}
+
+export interface WhatsAppClientOptions {
+ authDir: string;
+ onMessage: (msg: InboundMessage) => void;
+ onQR: (qr: string) => void;
+ onStatus: (status: string) => void;
+}
+
+export class WhatsAppClient {
+ private sock: any = null;
+ private options: WhatsAppClientOptions;
+ private reconnecting = false;
+
+ constructor(options: WhatsAppClientOptions) {
+ this.options = options;
+ }
+
+ async connect(): Promise {
+ const logger = pino({ level: 'silent' });
+ const { state, saveCreds } = await useMultiFileAuthState(this.options.authDir);
+ const { version } = await fetchLatestBaileysVersion();
+
+ console.log(`Using Baileys version: ${version.join('.')}`);
+
+ // Create socket following OpenClaw's pattern
+ this.sock = makeWASocket({
+ auth: {
+ creds: state.creds,
+ keys: makeCacheableSignalKeyStore(state.keys, logger),
+ },
+ version,
+ logger,
+ printQRInTerminal: false,
+ browser: ['nanobot', 'cli', VERSION],
+ syncFullHistory: false,
+ markOnlineOnConnect: false,
+ });
+
+ // Handle WebSocket errors
+ if (this.sock.ws && typeof this.sock.ws.on === 'function') {
+ this.sock.ws.on('error', (err: Error) => {
+ console.error('WebSocket error:', err.message);
+ });
+ }
+
+ // Handle connection updates
+ this.sock.ev.on('connection.update', async (update: any) => {
+ const { connection, lastDisconnect, qr } = update;
+
+ if (qr) {
+ // Display QR code in terminal
+ console.log('\nπ± Scan this QR code with WhatsApp (Linked Devices):\n');
+ qrcode.generate(qr, { small: true });
+ this.options.onQR(qr);
+ }
+
+ if (connection === 'close') {
+ const statusCode = (lastDisconnect?.error as Boom)?.output?.statusCode;
+ const shouldReconnect = statusCode !== DisconnectReason.loggedOut;
+
+ console.log(`Connection closed. Status: ${statusCode}, Will reconnect: ${shouldReconnect}`);
+ this.options.onStatus('disconnected');
+
+ if (shouldReconnect && !this.reconnecting) {
+ this.reconnecting = true;
+ console.log('Reconnecting in 5 seconds...');
+ setTimeout(() => {
+ this.reconnecting = false;
+ this.connect();
+ }, 5000);
+ }
+ } else if (connection === 'open') {
+ console.log('β
Connected to WhatsApp');
+ this.options.onStatus('connected');
+ }
+ });
+
+ // Save credentials on update
+ this.sock.ev.on('creds.update', saveCreds);
+
+ // Handle incoming messages
+ this.sock.ev.on('messages.upsert', async ({ messages, type }: { messages: any[]; type: string }) => {
+ if (type !== 'notify') return;
+
+ for (const msg of messages) {
+ // Skip own messages
+ if (msg.key.fromMe) continue;
+
+ // Skip status updates
+ if (msg.key.remoteJid === 'status@broadcast') continue;
+
+ const content = this.extractMessageContent(msg);
+ if (!content) continue;
+
+ const isGroup = msg.key.remoteJid?.endsWith('@g.us') || false;
+
+ this.options.onMessage({
+ id: msg.key.id || '',
+ sender: msg.key.remoteJid || '',
+ content,
+ timestamp: msg.messageTimestamp as number,
+ isGroup,
+ });
+ }
+ });
+ }
+
+ private extractMessageContent(msg: any): string | null {
+ const message = msg.message;
+ if (!message) return null;
+
+ // Text message
+ if (message.conversation) {
+ return message.conversation;
+ }
+
+ // Extended text (reply, link preview)
+ if (message.extendedTextMessage?.text) {
+ return message.extendedTextMessage.text;
+ }
+
+ // Image with caption
+ if (message.imageMessage?.caption) {
+ return `[Image] ${message.imageMessage.caption}`;
+ }
+
+ // Video with caption
+ if (message.videoMessage?.caption) {
+ return `[Video] ${message.videoMessage.caption}`;
+ }
+
+ // Document with caption
+ if (message.documentMessage?.caption) {
+ return `[Document] ${message.documentMessage.caption}`;
+ }
+
+ return null;
+ }
+
+ async sendMessage(to: string, text: string): Promise {
+ if (!this.sock) {
+ throw new Error('Not connected');
+ }
+
+ await this.sock.sendMessage(to, { text });
+ }
+
+ async disconnect(): Promise {
+ if (this.sock) {
+ this.sock.end(undefined);
+ this.sock = null;
+ }
+ }
+}
diff --git a/bridge/tsconfig.json b/bridge/tsconfig.json
new file mode 100644
index 0000000..7f472b2
--- /dev/null
+++ b/bridge/tsconfig.json
@@ -0,0 +1,16 @@
+{
+ "compilerOptions": {
+ "target": "ES2022",
+ "module": "ESNext",
+ "moduleResolution": "node",
+ "esModuleInterop": true,
+ "strict": true,
+ "skipLibCheck": true,
+ "outDir": "./dist",
+ "rootDir": "./src",
+ "declaration": true,
+ "resolveJsonModule": true
+ },
+ "include": ["src/**/*"],
+ "exclude": ["node_modules", "dist"]
+}
diff --git a/case/code.gif b/case/code.gif
new file mode 100644
index 0000000..159dad8
Binary files /dev/null and b/case/code.gif differ
diff --git a/case/memory.gif b/case/memory.gif
new file mode 100644
index 0000000..fc91f55
Binary files /dev/null and b/case/memory.gif differ
diff --git a/case/scedule.gif b/case/scedule.gif
new file mode 100644
index 0000000..a2e3073
Binary files /dev/null and b/case/scedule.gif differ
diff --git a/case/search.gif b/case/search.gif
new file mode 100644
index 0000000..fd3d067
Binary files /dev/null and b/case/search.gif differ
diff --git a/nanobot/__init__.py b/nanobot/__init__.py
new file mode 100644
index 0000000..ee0445b
--- /dev/null
+++ b/nanobot/__init__.py
@@ -0,0 +1,6 @@
+"""
+nanobot - A lightweight AI agent framework
+"""
+
+__version__ = "0.1.0"
+__logo__ = "π"
diff --git a/nanobot/__main__.py b/nanobot/__main__.py
new file mode 100644
index 0000000..c7f5620
--- /dev/null
+++ b/nanobot/__main__.py
@@ -0,0 +1,8 @@
+"""
+Entry point for running nanobot as a module: python -m nanobot
+"""
+
+from nanobot.cli.commands import app
+
+if __name__ == "__main__":
+ app()
diff --git a/nanobot/agent/__init__.py b/nanobot/agent/__init__.py
new file mode 100644
index 0000000..c3fc97b
--- /dev/null
+++ b/nanobot/agent/__init__.py
@@ -0,0 +1,8 @@
+"""Agent core module."""
+
+from nanobot.agent.loop import AgentLoop
+from nanobot.agent.context import ContextBuilder
+from nanobot.agent.memory import MemoryStore
+from nanobot.agent.skills import SkillsLoader
+
+__all__ = ["AgentLoop", "ContextBuilder", "MemoryStore", "SkillsLoader"]
diff --git a/nanobot/agent/context.py b/nanobot/agent/context.py
new file mode 100644
index 0000000..5383c35
--- /dev/null
+++ b/nanobot/agent/context.py
@@ -0,0 +1,196 @@
+"""Context builder for assembling agent prompts."""
+
+from pathlib import Path
+from typing import Any
+
+from nanobot.agent.memory import MemoryStore
+from nanobot.agent.skills import SkillsLoader
+
+
+class ContextBuilder:
+ """
+ Builds the context (system prompt + messages) for the agent.
+
+ Assembles bootstrap files, memory, skills, and conversation history
+ into a coherent prompt for the LLM.
+ """
+
+ BOOTSTRAP_FILES = ["AGENTS.md", "SOUL.md", "USER.md", "TOOLS.md", "IDENTITY.md"]
+
+ def __init__(self, workspace: Path):
+ self.workspace = workspace
+ self.memory = MemoryStore(workspace)
+ self.skills = SkillsLoader(workspace)
+
+ def build_system_prompt(self, skill_names: list[str] | None = None) -> str:
+ """
+ Build the system prompt from bootstrap files, memory, and skills.
+
+ Args:
+ skill_names: Optional list of skills to include.
+
+ Returns:
+ Complete system prompt.
+ """
+ parts = []
+
+ # Core identity
+ parts.append(self._get_identity())
+
+ # Bootstrap files
+ bootstrap = self._load_bootstrap_files()
+ if bootstrap:
+ parts.append(bootstrap)
+
+ # Memory context
+ memory = self.memory.get_memory_context()
+ if memory:
+ parts.append(f"# Memory\n\n{memory}")
+
+ # Skills - progressive loading
+ # 1. Always-loaded skills: include full content
+ always_skills = self.skills.get_always_skills()
+ if always_skills:
+ always_content = self.skills.load_skills_for_context(always_skills)
+ if always_content:
+ parts.append(f"# Active Skills\n\n{always_content}")
+
+ # 2. Available skills: only show summary (agent uses read_file to load)
+ skills_summary = self.skills.build_skills_summary()
+ if skills_summary:
+ parts.append(f"""# Skills
+
+The following skills extend your capabilities. To use a skill, read its SKILL.md file using the read_file tool.
+Skills with available="false" need dependencies installed first - you can try installing them with apt/brew.
+
+{skills_summary}""")
+
+ return "\n\n---\n\n".join(parts)
+
+ def _get_identity(self) -> str:
+ """Get the core identity section."""
+ from datetime import datetime
+ now = datetime.now().strftime("%Y-%m-%d %H:%M (%A)")
+ workspace_path = str(self.workspace.expanduser().resolve())
+
+ return f"""# nanobot π
+
+You are nanobot, a helpful AI assistant. You have access to tools that allow you to:
+- Read, write, and edit files
+- Execute shell commands
+- Search the web and fetch web pages
+- Send messages to users on chat channels
+
+## Current Time
+{now}
+
+## Workspace
+Your workspace is at: {workspace_path}
+- Memory files: {workspace_path}/memory/MEMORY.md
+- Daily notes: {workspace_path}/memory/YYYY-MM-DD.md
+- 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"""
+
+ def _load_bootstrap_files(self) -> str:
+ """Load all bootstrap files from workspace."""
+ parts = []
+
+ for filename in self.BOOTSTRAP_FILES:
+ file_path = self.workspace / filename
+ if file_path.exists():
+ content = file_path.read_text(encoding="utf-8")
+ parts.append(f"## {filename}\n\n{content}")
+
+ return "\n\n".join(parts) if parts else ""
+
+ def build_messages(
+ self,
+ history: list[dict[str, Any]],
+ current_message: str,
+ skill_names: list[str] | None = None
+ ) -> list[dict[str, Any]]:
+ """
+ Build the complete message list for an LLM call.
+
+ Args:
+ history: Previous conversation messages.
+ current_message: The new user message.
+ skill_names: Optional skills to include.
+
+ Returns:
+ List of messages including system prompt.
+ """
+ messages = []
+
+ # System prompt
+ system_prompt = self.build_system_prompt(skill_names)
+ messages.append({"role": "system", "content": system_prompt})
+
+ # History
+ messages.extend(history)
+
+ # Current message
+ messages.append({"role": "user", "content": current_message})
+
+ return messages
+
+ def add_tool_result(
+ self,
+ messages: list[dict[str, Any]],
+ tool_call_id: str,
+ tool_name: str,
+ result: str
+ ) -> list[dict[str, Any]]:
+ """
+ Add a tool result to the message list.
+
+ Args:
+ messages: Current message list.
+ tool_call_id: ID of the tool call.
+ tool_name: Name of the tool.
+ result: Tool execution result.
+
+ Returns:
+ Updated message list.
+ """
+ messages.append({
+ "role": "tool",
+ "tool_call_id": tool_call_id,
+ "name": tool_name,
+ "content": result
+ })
+ return messages
+
+ def add_assistant_message(
+ self,
+ messages: list[dict[str, Any]],
+ content: str | None,
+ tool_calls: list[dict[str, Any]] | None = None
+ ) -> list[dict[str, Any]]:
+ """
+ Add an assistant message to the message list.
+
+ Args:
+ messages: Current message list.
+ content: Message content.
+ tool_calls: Optional tool calls.
+
+ Returns:
+ Updated message list.
+ """
+ msg: dict[str, Any] = {"role": "assistant"}
+
+ if content:
+ msg["content"] = content
+
+ if tool_calls:
+ msg["tool_calls"] = tool_calls
+
+ messages.append(msg)
+ return messages
diff --git a/nanobot/agent/loop.py b/nanobot/agent/loop.py
new file mode 100644
index 0000000..4ad76a6
--- /dev/null
+++ b/nanobot/agent/loop.py
@@ -0,0 +1,213 @@
+"""Agent loop: the core processing engine."""
+
+import asyncio
+import json
+from pathlib import Path
+from typing import Any
+
+from loguru import logger
+
+from nanobot.bus.events import InboundMessage, OutboundMessage
+from nanobot.bus.queue import MessageBus
+from nanobot.providers.base import LLMProvider
+from nanobot.agent.context import ContextBuilder
+from nanobot.agent.tools.registry import ToolRegistry
+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
+from nanobot.agent.tools.message import MessageTool
+from nanobot.session.manager import SessionManager
+
+
+class AgentLoop:
+ """
+ The agent loop is the core processing engine.
+
+ It:
+ 1. Receives messages from the bus
+ 2. Builds context with history, memory, skills
+ 3. Calls the LLM
+ 4. Executes tool calls
+ 5. Sends responses back
+ """
+
+ def __init__(
+ self,
+ bus: MessageBus,
+ provider: LLMProvider,
+ workspace: Path,
+ model: str | None = None,
+ max_iterations: int = 20,
+ brave_api_key: str | None = None
+ ):
+ self.bus = bus
+ self.provider = provider
+ self.workspace = workspace
+ self.model = model or provider.get_default_model()
+ self.max_iterations = max_iterations
+ self.brave_api_key = brave_api_key
+
+ self.context = ContextBuilder(workspace)
+ self.sessions = SessionManager(workspace)
+ self.tools = ToolRegistry()
+
+ self._running = False
+ self._register_default_tools()
+
+ def _register_default_tools(self) -> None:
+ """Register the default set of tools."""
+ # File tools
+ self.tools.register(ReadFileTool())
+ self.tools.register(WriteFileTool())
+ self.tools.register(EditFileTool())
+ self.tools.register(ListDirTool())
+
+ # Shell tool
+ self.tools.register(ExecTool(working_dir=str(self.workspace)))
+
+ # Web tools
+ self.tools.register(WebSearchTool(api_key=self.brave_api_key))
+ self.tools.register(WebFetchTool())
+
+ # Message tool
+ message_tool = MessageTool(send_callback=self.bus.publish_outbound)
+ self.tools.register(message_tool)
+
+ async def run(self) -> None:
+ """Run the agent loop, processing messages from the bus."""
+ self._running = True
+ logger.info("Agent loop started")
+
+ while self._running:
+ try:
+ # Wait for next message
+ msg = await asyncio.wait_for(
+ self.bus.consume_inbound(),
+ timeout=1.0
+ )
+
+ # Process it
+ try:
+ response = await self._process_message(msg)
+ if response:
+ await self.bus.publish_outbound(response)
+ except Exception as e:
+ logger.error(f"Error processing message: {e}")
+ # Send error response
+ await self.bus.publish_outbound(OutboundMessage(
+ channel=msg.channel,
+ chat_id=msg.chat_id,
+ content=f"Sorry, I encountered an error: {str(e)}"
+ ))
+ except asyncio.TimeoutError:
+ continue
+
+ def stop(self) -> None:
+ """Stop the agent loop."""
+ self._running = False
+ logger.info("Agent loop stopping")
+
+ async def _process_message(self, msg: InboundMessage) -> OutboundMessage | None:
+ """
+ Process a single inbound message.
+
+ Args:
+ msg: The inbound message to process.
+
+ Returns:
+ The response message, or None if no response needed.
+ """
+ logger.info(f"Processing message from {msg.channel}:{msg.sender_id}")
+
+ # Get or create session
+ session = self.sessions.get_or_create(msg.session_key)
+
+ # Update message tool context
+ message_tool = self.tools.get("message")
+ if isinstance(message_tool, MessageTool):
+ message_tool.set_context(msg.channel, msg.chat_id)
+
+ # Build initial messages (use get_history for LLM-formatted messages)
+ messages = self.context.build_messages(
+ history=session.get_history(),
+ current_message=msg.content
+ )
+
+ # Agent loop
+ iteration = 0
+ final_content = None
+
+ while iteration < self.max_iterations:
+ iteration += 1
+
+ # Call LLM
+ response = await self.provider.chat(
+ messages=messages,
+ tools=self.tools.get_definitions(),
+ model=self.model
+ )
+
+ # Handle tool calls
+ if response.has_tool_calls:
+ # Add assistant message with tool calls
+ tool_call_dicts = [
+ {
+ "id": tc.id,
+ "type": "function",
+ "function": {
+ "name": tc.name,
+ "arguments": json.dumps(tc.arguments) # Must be JSON string
+ }
+ }
+ for tc in response.tool_calls
+ ]
+ messages = self.context.add_assistant_message(
+ messages, response.content, tool_call_dicts
+ )
+
+ # Execute tools
+ for tool_call in response.tool_calls:
+ logger.debug(f"Executing tool: {tool_call.name}")
+ 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
+ )
+ 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."
+
+ # Save to session
+ session.add_message("user", msg.content)
+ session.add_message("assistant", final_content)
+ self.sessions.save(session)
+
+ return OutboundMessage(
+ channel=msg.channel,
+ chat_id=msg.chat_id,
+ content=final_content
+ )
+
+ async def process_direct(self, content: str, session_key: str = "cli:direct") -> str:
+ """
+ Process a message directly (for CLI usage).
+
+ Args:
+ content: The message content.
+ session_key: Session identifier.
+
+ Returns:
+ The agent's response.
+ """
+ msg = InboundMessage(
+ channel="cli",
+ sender_id="user",
+ chat_id="direct",
+ content=content
+ )
+
+ response = await self._process_message(msg)
+ return response.content if response else ""
diff --git a/nanobot/agent/memory.py b/nanobot/agent/memory.py
new file mode 100644
index 0000000..58c079b
--- /dev/null
+++ b/nanobot/agent/memory.py
@@ -0,0 +1,110 @@
+"""Memory system for persistent agent memory."""
+
+from pathlib import Path
+from datetime import datetime
+
+from nanobot.utils.helpers import ensure_dir, today_date
+
+
+class MemoryStore:
+ """
+ Memory system for the agent.
+
+ Supports daily notes (memory/YYYY-MM-DD.md) and long-term memory (MEMORY.md).
+ Compatible with clawbot memory format.
+ """
+
+ 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")
+
+ 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 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 ""
diff --git a/nanobot/agent/skills.py b/nanobot/agent/skills.py
new file mode 100644
index 0000000..7b04924
--- /dev/null
+++ b/nanobot/agent/skills.py
@@ -0,0 +1,228 @@
+"""Skills loader for agent capabilities."""
+
+import json
+import os
+import re
+import shutil
+from pathlib import Path
+
+# Default builtin skills directory (relative to this file)
+BUILTIN_SKILLS_DIR = Path(__file__).parent.parent / "skills"
+
+
+class SkillsLoader:
+ """
+ Loader for agent skills.
+
+ Skills are markdown files (SKILL.md) that teach the agent how to use
+ specific tools or perform certain tasks.
+ """
+
+ def __init__(self, workspace: Path, builtin_skills_dir: Path | None = None):
+ self.workspace = workspace
+ self.workspace_skills = workspace / "skills"
+ self.builtin_skills = builtin_skills_dir or BUILTIN_SKILLS_DIR
+
+ def list_skills(self, filter_unavailable: bool = True) -> list[dict[str, str]]:
+ """
+ List all available skills.
+
+ Args:
+ filter_unavailable: If True, filter out skills with unmet requirements.
+
+ Returns:
+ List of skill info dicts with 'name', 'path', 'source'.
+ """
+ skills = []
+
+ # Workspace skills (highest priority)
+ if self.workspace_skills.exists():
+ for skill_dir in self.workspace_skills.iterdir():
+ if skill_dir.is_dir():
+ skill_file = skill_dir / "SKILL.md"
+ if skill_file.exists():
+ skills.append({"name": skill_dir.name, "path": str(skill_file), "source": "workspace"})
+
+ # Built-in skills
+ if self.builtin_skills and self.builtin_skills.exists():
+ for skill_dir in self.builtin_skills.iterdir():
+ if skill_dir.is_dir():
+ skill_file = skill_dir / "SKILL.md"
+ if skill_file.exists() and not any(s["name"] == skill_dir.name for s in skills):
+ skills.append({"name": skill_dir.name, "path": str(skill_file), "source": "builtin"})
+
+ # Filter by requirements
+ if filter_unavailable:
+ return [s for s in skills if self._check_requirements(self._get_ocmeta(s["name"]))]
+ return skills
+
+ def load_skill(self, name: str) -> str | None:
+ """
+ Load a skill by name.
+
+ Args:
+ name: Skill name (directory name).
+
+ Returns:
+ Skill content or None if not found.
+ """
+ # Check workspace first
+ workspace_skill = self.workspace_skills / name / "SKILL.md"
+ if workspace_skill.exists():
+ return workspace_skill.read_text(encoding="utf-8")
+
+ # Check built-in
+ if self.builtin_skills:
+ builtin_skill = self.builtin_skills / name / "SKILL.md"
+ if builtin_skill.exists():
+ return builtin_skill.read_text(encoding="utf-8")
+
+ return None
+
+ def load_skills_for_context(self, skill_names: list[str]) -> str:
+ """
+ Load specific skills for inclusion in agent context.
+
+ Args:
+ skill_names: List of skill names to load.
+
+ Returns:
+ Formatted skills content.
+ """
+ parts = []
+ for name in skill_names:
+ content = self.load_skill(name)
+ if content:
+ content = self._strip_frontmatter(content)
+ parts.append(f"### Skill: {name}\n\n{content}")
+
+ return "\n\n---\n\n".join(parts) if parts else ""
+
+ def build_skills_summary(self) -> str:
+ """
+ Build a summary of all skills (name, description, path, availability).
+
+ This is used for progressive loading - the agent can read the full
+ skill content using read_file when needed.
+
+ Returns:
+ XML-formatted skills summary.
+ """
+ all_skills = self.list_skills(filter_unavailable=False)
+ if not all_skills:
+ return ""
+
+ def escape_xml(s: str) -> str:
+ return s.replace("&", "&").replace("<", "<").replace(">", ">")
+
+ lines = [""]
+ for s in all_skills:
+ name = escape_xml(s["name"])
+ path = s["path"]
+ desc = escape_xml(self._get_skill_description(s["name"]))
+ ocmeta = self._get_ocmeta(s["name"])
+ available = self._check_requirements(ocmeta)
+
+ lines.append(f" ")
+ lines.append(f" {name}")
+ lines.append(f" {desc}")
+ lines.append(f" {path}")
+
+ # Show missing requirements for unavailable skills
+ if not available:
+ missing = self._get_missing_requirements(ocmeta)
+ if missing:
+ lines.append(f" {escape_xml(missing)}")
+
+ lines.append(f" ")
+ lines.append("")
+
+ return "\n".join(lines)
+
+ def _get_missing_requirements(self, ocmeta: dict) -> str:
+ """Get a description of missing requirements."""
+ missing = []
+ requires = ocmeta.get("requires", {})
+ for b in requires.get("bins", []):
+ if not shutil.which(b):
+ missing.append(f"CLI: {b}")
+ for env in requires.get("env", []):
+ if not os.environ.get(env):
+ missing.append(f"ENV: {env}")
+ return ", ".join(missing)
+
+ def _get_skill_description(self, name: str) -> str:
+ """Get the description of a skill from its frontmatter."""
+ meta = self.get_skill_metadata(name)
+ if meta and meta.get("description"):
+ return meta["description"]
+ return name # Fallback to skill name
+
+ def _strip_frontmatter(self, content: str) -> str:
+ """Remove YAML frontmatter from markdown content."""
+ if content.startswith("---"):
+ match = re.match(r"^---\n.*?\n---\n", content, re.DOTALL)
+ if match:
+ return content[match.end():].strip()
+ return content
+
+ def _parse_openclaw_metadata(self, raw: str) -> dict:
+ """Parse openclaw metadata JSON from frontmatter."""
+ try:
+ data = json.loads(raw)
+ return data.get("openclaw", {}) if isinstance(data, dict) else {}
+ except (json.JSONDecodeError, TypeError):
+ return {}
+
+ def _check_requirements(self, ocmeta: dict) -> bool:
+ """Check if skill requirements are met (bins, env vars)."""
+ requires = ocmeta.get("requires", {})
+ for b in requires.get("bins", []):
+ if not shutil.which(b):
+ return False
+ for env in requires.get("env", []):
+ if not os.environ.get(env):
+ return False
+ return True
+
+ def _get_ocmeta(self, name: str) -> dict:
+ """Get openclaw metadata for a skill (cached in frontmatter)."""
+ meta = self.get_skill_metadata(name) or {}
+ return self._parse_openclaw_metadata(meta.get("metadata", ""))
+
+ def get_always_skills(self) -> list[str]:
+ """Get skills marked as always=true that meet requirements."""
+ result = []
+ for s in self.list_skills(filter_unavailable=True):
+ meta = self.get_skill_metadata(s["name"]) or {}
+ ocmeta = self._parse_openclaw_metadata(meta.get("metadata", ""))
+ if ocmeta.get("always") or meta.get("always"):
+ result.append(s["name"])
+ return result
+
+ def get_skill_metadata(self, name: str) -> dict | None:
+ """
+ Get metadata from a skill's frontmatter.
+
+ Args:
+ name: Skill name.
+
+ Returns:
+ Metadata dict or None.
+ """
+ content = self.load_skill(name)
+ if not content:
+ return None
+
+ if content.startswith("---"):
+ match = re.match(r"^---\n(.*?)\n---", content, re.DOTALL)
+ if match:
+ # Simple YAML parsing
+ metadata = {}
+ for line in match.group(1).split("\n"):
+ if ":" in line:
+ key, value = line.split(":", 1)
+ metadata[key.strip()] = value.strip().strip('"\'')
+ return metadata
+
+ return None
diff --git a/nanobot/agent/tools/__init__.py b/nanobot/agent/tools/__init__.py
new file mode 100644
index 0000000..aac5d7d
--- /dev/null
+++ b/nanobot/agent/tools/__init__.py
@@ -0,0 +1,6 @@
+"""Agent tools module."""
+
+from nanobot.agent.tools.base import Tool
+from nanobot.agent.tools.registry import ToolRegistry
+
+__all__ = ["Tool", "ToolRegistry"]
diff --git a/nanobot/agent/tools/base.py b/nanobot/agent/tools/base.py
new file mode 100644
index 0000000..6fcfec6
--- /dev/null
+++ b/nanobot/agent/tools/base.py
@@ -0,0 +1,55 @@
+"""Base class for agent tools."""
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+
+class Tool(ABC):
+ """
+ Abstract base class for agent tools.
+
+ Tools are capabilities that the agent can use to interact with
+ the environment, such as reading files, executing commands, etc.
+ """
+
+ @property
+ @abstractmethod
+ def name(self) -> str:
+ """Tool name used in function calls."""
+ pass
+
+ @property
+ @abstractmethod
+ def description(self) -> str:
+ """Description of what the tool does."""
+ pass
+
+ @property
+ @abstractmethod
+ def parameters(self) -> dict[str, Any]:
+ """JSON Schema for tool parameters."""
+ pass
+
+ @abstractmethod
+ async def execute(self, **kwargs: Any) -> str:
+ """
+ Execute the tool with given parameters.
+
+ Args:
+ **kwargs: Tool-specific parameters.
+
+ Returns:
+ String result of the tool execution.
+ """
+ pass
+
+ def to_schema(self) -> dict[str, Any]:
+ """Convert tool to OpenAI function schema format."""
+ return {
+ "type": "function",
+ "function": {
+ "name": self.name,
+ "description": self.description,
+ "parameters": self.parameters,
+ }
+ }
diff --git a/nanobot/agent/tools/filesystem.py b/nanobot/agent/tools/filesystem.py
new file mode 100644
index 0000000..e141fab
--- /dev/null
+++ b/nanobot/agent/tools/filesystem.py
@@ -0,0 +1,191 @@
+"""File system tools: read, write, edit."""
+
+from pathlib import Path
+from typing import Any
+
+from nanobot.agent.tools.base import Tool
+
+
+class ReadFileTool(Tool):
+ """Tool to read file contents."""
+
+ @property
+ def name(self) -> str:
+ return "read_file"
+
+ @property
+ def description(self) -> str:
+ return "Read the contents of a file at the given path."
+
+ @property
+ def parameters(self) -> dict[str, Any]:
+ return {
+ "type": "object",
+ "properties": {
+ "path": {
+ "type": "string",
+ "description": "The file path to read"
+ }
+ },
+ "required": ["path"]
+ }
+
+ async def execute(self, path: str, **kwargs: Any) -> str:
+ try:
+ file_path = Path(path).expanduser()
+ if not file_path.exists():
+ return f"Error: File not found: {path}"
+ if not file_path.is_file():
+ return f"Error: Not a file: {path}"
+
+ content = file_path.read_text(encoding="utf-8")
+ return content
+ except PermissionError:
+ return f"Error: Permission denied: {path}"
+ except Exception as e:
+ return f"Error reading file: {str(e)}"
+
+
+class WriteFileTool(Tool):
+ """Tool to write content to a file."""
+
+ @property
+ def name(self) -> str:
+ return "write_file"
+
+ @property
+ def description(self) -> str:
+ return "Write content to a file at the given path. Creates parent directories if needed."
+
+ @property
+ def parameters(self) -> dict[str, Any]:
+ return {
+ "type": "object",
+ "properties": {
+ "path": {
+ "type": "string",
+ "description": "The file path to write to"
+ },
+ "content": {
+ "type": "string",
+ "description": "The content to write"
+ }
+ },
+ "required": ["path", "content"]
+ }
+
+ async def execute(self, path: str, content: str, **kwargs: Any) -> str:
+ try:
+ file_path = Path(path).expanduser()
+ file_path.parent.mkdir(parents=True, exist_ok=True)
+ file_path.write_text(content, encoding="utf-8")
+ return f"Successfully wrote {len(content)} bytes to {path}"
+ except PermissionError:
+ return f"Error: Permission denied: {path}"
+ except Exception as e:
+ return f"Error writing file: {str(e)}"
+
+
+class EditFileTool(Tool):
+ """Tool to edit a file by replacing text."""
+
+ @property
+ def name(self) -> str:
+ return "edit_file"
+
+ @property
+ def description(self) -> str:
+ return "Edit a file by replacing old_text with new_text. The old_text must exist exactly in the file."
+
+ @property
+ def parameters(self) -> dict[str, Any]:
+ return {
+ "type": "object",
+ "properties": {
+ "path": {
+ "type": "string",
+ "description": "The file path to edit"
+ },
+ "old_text": {
+ "type": "string",
+ "description": "The exact text to find and replace"
+ },
+ "new_text": {
+ "type": "string",
+ "description": "The text to replace with"
+ }
+ },
+ "required": ["path", "old_text", "new_text"]
+ }
+
+ async def execute(self, path: str, old_text: str, new_text: str, **kwargs: Any) -> str:
+ try:
+ file_path = Path(path).expanduser()
+ if not file_path.exists():
+ return f"Error: File not found: {path}"
+
+ content = file_path.read_text(encoding="utf-8")
+
+ if old_text not in content:
+ return f"Error: old_text not found in file. Make sure it matches exactly."
+
+ # Count occurrences
+ count = content.count(old_text)
+ if count > 1:
+ return f"Warning: old_text appears {count} times. Please provide more context to make it unique."
+
+ new_content = content.replace(old_text, new_text, 1)
+ file_path.write_text(new_content, encoding="utf-8")
+
+ return f"Successfully edited {path}"
+ except PermissionError:
+ return f"Error: Permission denied: {path}"
+ except Exception as e:
+ return f"Error editing file: {str(e)}"
+
+
+class ListDirTool(Tool):
+ """Tool to list directory contents."""
+
+ @property
+ def name(self) -> str:
+ return "list_dir"
+
+ @property
+ def description(self) -> str:
+ return "List the contents of a directory."
+
+ @property
+ def parameters(self) -> dict[str, Any]:
+ return {
+ "type": "object",
+ "properties": {
+ "path": {
+ "type": "string",
+ "description": "The directory path to list"
+ }
+ },
+ "required": ["path"]
+ }
+
+ async def execute(self, path: str, **kwargs: Any) -> str:
+ try:
+ dir_path = Path(path).expanduser()
+ if not dir_path.exists():
+ return f"Error: Directory not found: {path}"
+ if not dir_path.is_dir():
+ return f"Error: Not a directory: {path}"
+
+ items = []
+ for item in sorted(dir_path.iterdir()):
+ prefix = "π " if item.is_dir() else "π "
+ items.append(f"{prefix}{item.name}")
+
+ if not items:
+ return f"Directory {path} is empty"
+
+ return "\n".join(items)
+ except PermissionError:
+ return f"Error: Permission denied: {path}"
+ except Exception as e:
+ return f"Error listing directory: {str(e)}"
diff --git a/nanobot/agent/tools/message.py b/nanobot/agent/tools/message.py
new file mode 100644
index 0000000..347830f
--- /dev/null
+++ b/nanobot/agent/tools/message.py
@@ -0,0 +1,86 @@
+"""Message tool for sending messages to users."""
+
+from typing import Any, Callable, Awaitable
+
+from nanobot.agent.tools.base import Tool
+from nanobot.bus.events import OutboundMessage
+
+
+class MessageTool(Tool):
+ """Tool to send messages to users on chat channels."""
+
+ def __init__(
+ self,
+ send_callback: Callable[[OutboundMessage], Awaitable[None]] | None = None,
+ default_channel: str = "",
+ default_chat_id: str = ""
+ ):
+ self._send_callback = send_callback
+ self._default_channel = default_channel
+ self._default_chat_id = default_chat_id
+
+ def set_context(self, channel: str, chat_id: str) -> None:
+ """Set the current message context."""
+ self._default_channel = channel
+ self._default_chat_id = chat_id
+
+ def set_send_callback(self, callback: Callable[[OutboundMessage], Awaitable[None]]) -> None:
+ """Set the callback for sending messages."""
+ self._send_callback = callback
+
+ @property
+ def name(self) -> str:
+ return "message"
+
+ @property
+ def description(self) -> str:
+ return "Send a message to the user. Use this when you want to communicate something."
+
+ @property
+ def parameters(self) -> dict[str, Any]:
+ return {
+ "type": "object",
+ "properties": {
+ "content": {
+ "type": "string",
+ "description": "The message content to send"
+ },
+ "channel": {
+ "type": "string",
+ "description": "Optional: target channel (telegram, discord, etc.)"
+ },
+ "chat_id": {
+ "type": "string",
+ "description": "Optional: target chat/user ID"
+ }
+ },
+ "required": ["content"]
+ }
+
+ async def execute(
+ self,
+ content: str,
+ channel: str | None = None,
+ chat_id: str | None = None,
+ **kwargs: Any
+ ) -> str:
+ channel = channel or self._default_channel
+ chat_id = chat_id or self._default_chat_id
+
+ if not channel or not chat_id:
+ return "Error: No target channel/chat specified"
+
+ if not self._send_callback:
+ return "Error: Message sending not configured"
+
+ msg = OutboundMessage(
+ channel=channel,
+ chat_id=chat_id,
+ content=content
+ )
+
+ try:
+ await self._send_callback(msg)
+ return f"Message sent to {channel}:{chat_id}"
+ except Exception as e:
+ return f"Error sending message: {str(e)}"
diff --git a/nanobot/agent/tools/registry.py b/nanobot/agent/tools/registry.py
new file mode 100644
index 0000000..1e8f56d
--- /dev/null
+++ b/nanobot/agent/tools/registry.py
@@ -0,0 +1,70 @@
+"""Tool registry for dynamic tool management."""
+
+from typing import Any
+
+from nanobot.agent.tools.base import Tool
+
+
+class ToolRegistry:
+ """
+ Registry for agent tools.
+
+ Allows dynamic registration and execution of tools.
+ """
+
+ def __init__(self):
+ self._tools: dict[str, Tool] = {}
+
+ def register(self, tool: Tool) -> None:
+ """Register a tool."""
+ self._tools[tool.name] = tool
+
+ def unregister(self, name: str) -> None:
+ """Unregister a tool by name."""
+ self._tools.pop(name, None)
+
+ def get(self, name: str) -> Tool | None:
+ """Get a tool by name."""
+ return self._tools.get(name)
+
+ def has(self, name: str) -> bool:
+ """Check if a tool is registered."""
+ return name in self._tools
+
+ def get_definitions(self) -> list[dict[str, Any]]:
+ """Get all tool definitions in OpenAI format."""
+ return [tool.to_schema() for tool in self._tools.values()]
+
+ async def execute(self, name: str, params: dict[str, Any]) -> str:
+ """
+ Execute a tool by name with given parameters.
+
+ Args:
+ name: Tool name.
+ params: Tool parameters.
+
+ Returns:
+ Tool execution result as string.
+
+ Raises:
+ KeyError: If tool not found.
+ """
+ tool = self._tools.get(name)
+ if not tool:
+ return f"Error: Tool '{name}' not found"
+
+ try:
+ return await tool.execute(**params)
+ except Exception as e:
+ return f"Error executing {name}: {str(e)}"
+
+ @property
+ def tool_names(self) -> list[str]:
+ """Get list of registered tool names."""
+ return list(self._tools.keys())
+
+ def __len__(self) -> int:
+ return len(self._tools)
+
+ def __contains__(self, name: str) -> bool:
+ return name in self._tools
diff --git a/nanobot/agent/tools/shell.py b/nanobot/agent/tools/shell.py
new file mode 100644
index 0000000..bf7f064
--- /dev/null
+++ b/nanobot/agent/tools/shell.py
@@ -0,0 +1,85 @@
+"""Shell execution tool."""
+
+import asyncio
+import os
+from typing import Any
+
+from nanobot.agent.tools.base import Tool
+
+
+class ExecTool(Tool):
+ """Tool to execute shell commands."""
+
+ def __init__(self, timeout: int = 60, working_dir: str | None = None):
+ self.timeout = timeout
+ self.working_dir = working_dir
+
+ @property
+ def name(self) -> str:
+ return "exec"
+
+ @property
+ def description(self) -> str:
+ return "Execute a shell command and return its output. Use with caution."
+
+ @property
+ def parameters(self) -> dict[str, Any]:
+ return {
+ "type": "object",
+ "properties": {
+ "command": {
+ "type": "string",
+ "description": "The shell command to execute"
+ },
+ "working_dir": {
+ "type": "string",
+ "description": "Optional working directory for the command"
+ }
+ },
+ "required": ["command"]
+ }
+
+ async def execute(self, command: str, working_dir: str | None = None, **kwargs: Any) -> str:
+ cwd = working_dir or self.working_dir or os.getcwd()
+
+ try:
+ process = await asyncio.create_subprocess_shell(
+ command,
+ stdout=asyncio.subprocess.PIPE,
+ stderr=asyncio.subprocess.PIPE,
+ cwd=cwd,
+ )
+
+ try:
+ stdout, stderr = await asyncio.wait_for(
+ process.communicate(),
+ timeout=self.timeout
+ )
+ except asyncio.TimeoutError:
+ process.kill()
+ return f"Error: Command timed out after {self.timeout} seconds"
+
+ output_parts = []
+
+ if stdout:
+ output_parts.append(stdout.decode("utf-8", errors="replace"))
+
+ if stderr:
+ stderr_text = stderr.decode("utf-8", errors="replace")
+ if stderr_text.strip():
+ output_parts.append(f"STDERR:\n{stderr_text}")
+
+ if process.returncode != 0:
+ output_parts.append(f"\nExit code: {process.returncode}")
+
+ result = "\n".join(output_parts) if output_parts else "(no output)"
+
+ # Truncate very long output
+ max_len = 10000
+ if len(result) > max_len:
+ result = result[:max_len] + f"\n... (truncated, {len(result) - max_len} more chars)"
+
+ return result
+
+ except Exception as e:
+ return f"Error executing command: {str(e)}"
diff --git a/nanobot/agent/tools/web.py b/nanobot/agent/tools/web.py
new file mode 100644
index 0000000..c9d989c
--- /dev/null
+++ b/nanobot/agent/tools/web.py
@@ -0,0 +1,139 @@
+"""Web tools: web_search and web_fetch."""
+
+import html
+import json
+import os
+import re
+from typing import Any
+
+import httpx
+
+from nanobot.agent.tools.base import Tool
+
+# Shared constants
+USER_AGENT = "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_7_2) AppleWebKit/537.36"
+
+
+def _strip_tags(text: str) -> str:
+ """Remove HTML tags and decode entities."""
+ text = re.sub(r'', '', text, flags=re.I)
+ text = re.sub(r'', '', text, flags=re.I)
+ text = re.sub(r'<[^>]+>', '', text)
+ return html.unescape(text).strip()
+
+
+def _normalize(text: str) -> str:
+ """Normalize whitespace."""
+ text = re.sub(r'[ \t]+', ' ', text)
+ return re.sub(r'\n{3,}', '\n\n', text).strip()
+
+
+class WebSearchTool(Tool):
+ """Search the web using Brave Search API."""
+
+ name = "web_search"
+ description = "Search the web. Returns titles, URLs, and snippets."
+ parameters = {
+ "type": "object",
+ "properties": {
+ "query": {"type": "string", "description": "Search query"},
+ "count": {"type": "integer", "description": "Results (1-10)", "minimum": 1, "maximum": 10}
+ },
+ "required": ["query"]
+ }
+
+ def __init__(self, api_key: str | None = None, max_results: int = 5):
+ self.api_key = api_key or os.environ.get("BRAVE_API_KEY", "")
+ self.max_results = max_results
+
+ async def execute(self, query: str, count: int | None = None, **kwargs: Any) -> str:
+ if not self.api_key:
+ return "Error: BRAVE_API_KEY not configured"
+
+ try:
+ n = min(max(count or self.max_results, 1), 10)
+ async with httpx.AsyncClient() as client:
+ r = await client.get(
+ "https://api.search.brave.com/res/v1/web/search",
+ params={"q": query, "count": n},
+ headers={"Accept": "application/json", "X-Subscription-Token": self.api_key},
+ timeout=10.0
+ )
+ r.raise_for_status()
+
+ results = r.json().get("web", {}).get("results", [])
+ if not results:
+ return f"No results for: {query}"
+
+ lines = [f"Results for: {query}\n"]
+ for i, item in enumerate(results[:n], 1):
+ lines.append(f"{i}. {item.get('title', '')}\n {item.get('url', '')}")
+ if desc := item.get("description"):
+ lines.append(f" {desc}")
+ return "\n".join(lines)
+ except Exception as e:
+ return f"Error: {e}"
+
+
+class WebFetchTool(Tool):
+ """Fetch and extract content from a URL using Readability."""
+
+ name = "web_fetch"
+ description = "Fetch URL and extract readable content (HTML β markdown/text)."
+ parameters = {
+ "type": "object",
+ "properties": {
+ "url": {"type": "string", "description": "URL to fetch"},
+ "extractMode": {"type": "string", "enum": ["markdown", "text"], "default": "markdown"},
+ "maxChars": {"type": "integer", "minimum": 100}
+ },
+ "required": ["url"]
+ }
+
+ def __init__(self, max_chars: int = 50000):
+ self.max_chars = max_chars
+
+ async def execute(self, url: str, extractMode: str = "markdown", maxChars: int | None = None, **kwargs: Any) -> str:
+ from readability import Document
+
+ max_chars = maxChars or self.max_chars
+
+ try:
+ async with httpx.AsyncClient() as client:
+ r = await client.get(url, headers={"User-Agent": USER_AGENT}, follow_redirects=True, timeout=30.0)
+ r.raise_for_status()
+
+ ctype = r.headers.get("content-type", "")
+
+ # JSON
+ if "application/json" in ctype:
+ text, extractor = json.dumps(r.json(), indent=2), "json"
+ # HTML
+ elif "text/html" in ctype or r.text[:256].lower().startswith((" max_chars
+ if truncated:
+ text = text[:max_chars]
+
+ return json.dumps({"url": url, "finalUrl": str(r.url), "status": r.status_code,
+ "extractor": extractor, "truncated": truncated, "length": len(text), "text": text})
+ except Exception as e:
+ return json.dumps({"error": str(e), "url": url})
+
+ def _to_markdown(self, html: str) -> str:
+ """Convert HTML to markdown."""
+ # Convert links, headings, lists before stripping tags
+ text = re.sub(r']*href=["\']([^"\']+)["\'][^>]*>([\s\S]*?)',
+ lambda m: f'[{_strip_tags(m[2])}]({m[1]})', html, flags=re.I)
+ text = re.sub(r']*>([\s\S]*?)',
+ lambda m: f'\n{"#" * int(m[1])} {_strip_tags(m[2])}\n', text, flags=re.I)
+ text = re.sub(r']*>([\s\S]*?)', lambda m: f'\n- {_strip_tags(m[1])}', text, flags=re.I)
+ text = re.sub(r'', '\n\n', text, flags=re.I)
+ text = re.sub(r'<(br|hr)\s*/?>', '\n', text, flags=re.I)
+ return _normalize(_strip_tags(text))
diff --git a/nanobot/bus/__init__.py b/nanobot/bus/__init__.py
new file mode 100644
index 0000000..c7b282d
--- /dev/null
+++ b/nanobot/bus/__init__.py
@@ -0,0 +1,6 @@
+"""Message bus module for decoupled channel-agent communication."""
+
+from nanobot.bus.events import InboundMessage, OutboundMessage
+from nanobot.bus.queue import MessageBus
+
+__all__ = ["MessageBus", "InboundMessage", "OutboundMessage"]
diff --git a/nanobot/bus/events.py b/nanobot/bus/events.py
new file mode 100644
index 0000000..a149e20
--- /dev/null
+++ b/nanobot/bus/events.py
@@ -0,0 +1,37 @@
+"""Event types for the message bus."""
+
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Any
+
+
+@dataclass
+class InboundMessage:
+ """Message received from a chat channel."""
+
+ channel: str # telegram, discord, slack, whatsapp
+ sender_id: str # User identifier
+ chat_id: str # Chat/channel identifier
+ content: str # Message text
+ timestamp: datetime = field(default_factory=datetime.now)
+ media: list[str] = field(default_factory=list) # Media URLs
+ metadata: dict[str, Any] = field(default_factory=dict) # Channel-specific data
+
+ @property
+ def session_key(self) -> str:
+ """Unique key for session identification."""
+ return f"{self.channel}:{self.chat_id}"
+
+
+@dataclass
+class OutboundMessage:
+ """Message to send to a chat channel."""
+
+ channel: str
+ chat_id: str
+ content: str
+ reply_to: str | None = None
+ media: list[str] = field(default_factory=list)
+ metadata: dict[str, Any] = field(default_factory=dict)
+
+
diff --git a/nanobot/bus/queue.py b/nanobot/bus/queue.py
new file mode 100644
index 0000000..4123d06
--- /dev/null
+++ b/nanobot/bus/queue.py
@@ -0,0 +1,81 @@
+"""Async message queue for decoupled channel-agent communication."""
+
+import asyncio
+from typing import Callable, Awaitable
+
+from loguru import logger
+
+from nanobot.bus.events import InboundMessage, OutboundMessage
+
+
+class MessageBus:
+ """
+ Async message bus that decouples chat channels from the agent core.
+
+ Channels push messages to the inbound queue, and the agent processes
+ them and pushes responses to the outbound queue.
+ """
+
+ def __init__(self):
+ self.inbound: asyncio.Queue[InboundMessage] = asyncio.Queue()
+ self.outbound: asyncio.Queue[OutboundMessage] = asyncio.Queue()
+ self._outbound_subscribers: dict[str, list[Callable[[OutboundMessage], Awaitable[None]]]] = {}
+ self._running = False
+
+ async def publish_inbound(self, msg: InboundMessage) -> None:
+ """Publish a message from a channel to the agent."""
+ await self.inbound.put(msg)
+
+ async def consume_inbound(self) -> InboundMessage:
+ """Consume the next inbound message (blocks until available)."""
+ return await self.inbound.get()
+
+ async def publish_outbound(self, msg: OutboundMessage) -> None:
+ """Publish a response from the agent to channels."""
+ await self.outbound.put(msg)
+
+ async def consume_outbound(self) -> OutboundMessage:
+ """Consume the next outbound message (blocks until available)."""
+ return await self.outbound.get()
+
+ def subscribe_outbound(
+ self,
+ channel: str,
+ callback: Callable[[OutboundMessage], Awaitable[None]]
+ ) -> None:
+ """Subscribe to outbound messages for a specific channel."""
+ if channel not in self._outbound_subscribers:
+ self._outbound_subscribers[channel] = []
+ self._outbound_subscribers[channel].append(callback)
+
+ async def dispatch_outbound(self) -> None:
+ """
+ Dispatch outbound messages to subscribed channels.
+ Run this as a background task.
+ """
+ self._running = True
+ while self._running:
+ try:
+ msg = await asyncio.wait_for(self.outbound.get(), timeout=1.0)
+ subscribers = self._outbound_subscribers.get(msg.channel, [])
+ for callback in subscribers:
+ try:
+ await callback(msg)
+ except Exception as e:
+ logger.error(f"Error dispatching to {msg.channel}: {e}")
+ except asyncio.TimeoutError:
+ continue
+
+ def stop(self) -> None:
+ """Stop the dispatcher loop."""
+ self._running = False
+
+ @property
+ def inbound_size(self) -> int:
+ """Number of pending inbound messages."""
+ return self.inbound.qsize()
+
+ @property
+ def outbound_size(self) -> int:
+ """Number of pending outbound messages."""
+ return self.outbound.qsize()
diff --git a/nanobot/channels/__init__.py b/nanobot/channels/__init__.py
new file mode 100644
index 0000000..588169d
--- /dev/null
+++ b/nanobot/channels/__init__.py
@@ -0,0 +1,6 @@
+"""Chat channels module with plugin architecture."""
+
+from nanobot.channels.base import BaseChannel
+from nanobot.channels.manager import ChannelManager
+
+__all__ = ["BaseChannel", "ChannelManager"]
diff --git a/nanobot/channels/base.py b/nanobot/channels/base.py
new file mode 100644
index 0000000..d83367c
--- /dev/null
+++ b/nanobot/channels/base.py
@@ -0,0 +1,114 @@
+"""Base channel interface for chat platforms."""
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+from nanobot.bus.events import InboundMessage, OutboundMessage
+from nanobot.bus.queue import MessageBus
+
+
+class BaseChannel(ABC):
+ """
+ Abstract base class for chat channel implementations.
+
+ Each channel (Telegram, Discord, etc.) should implement this interface
+ to integrate with the nanobot message bus.
+ """
+
+ name: str = "base"
+
+ def __init__(self, config: Any, bus: MessageBus):
+ """
+ Initialize the channel.
+
+ Args:
+ config: Channel-specific configuration.
+ bus: The message bus for communication.
+ """
+ self.config = config
+ self.bus = bus
+ self._running = False
+
+ @abstractmethod
+ async def start(self) -> None:
+ """
+ Start the channel and begin listening for messages.
+
+ This should be a long-running async task that:
+ 1. Connects to the chat platform
+ 2. Listens for incoming messages
+ 3. Forwards messages to the bus via _handle_message()
+ """
+ pass
+
+ @abstractmethod
+ async def stop(self) -> None:
+ """Stop the channel and clean up resources."""
+ pass
+
+ @abstractmethod
+ async def send(self, msg: OutboundMessage) -> None:
+ """
+ Send a message through this channel.
+
+ Args:
+ msg: The message to send.
+ """
+ pass
+
+ def is_allowed(self, sender_id: str) -> bool:
+ """
+ Check if a sender is allowed to use this bot.
+
+ Args:
+ sender_id: The sender's identifier.
+
+ Returns:
+ True if allowed, False otherwise.
+ """
+ allow_list = getattr(self.config, "allow_from", [])
+
+ # If no allow list, allow everyone
+ if not allow_list:
+ return True
+
+ return str(sender_id) in allow_list
+
+ async def _handle_message(
+ self,
+ sender_id: str,
+ chat_id: str,
+ content: str,
+ media: list[str] | None = None,
+ metadata: dict[str, Any] | None = None
+ ) -> None:
+ """
+ Handle an incoming message from the chat platform.
+
+ This method checks permissions and forwards to the bus.
+
+ Args:
+ sender_id: The sender's identifier.
+ chat_id: The chat/channel identifier.
+ content: Message text content.
+ media: Optional list of media URLs.
+ metadata: Optional channel-specific metadata.
+ """
+ if not self.is_allowed(sender_id):
+ return
+
+ msg = InboundMessage(
+ channel=self.name,
+ sender_id=str(sender_id),
+ chat_id=str(chat_id),
+ content=content,
+ media=media or [],
+ metadata=metadata or {}
+ )
+
+ await self.bus.publish_inbound(msg)
+
+ @property
+ def is_running(self) -> bool:
+ """Check if the channel is running."""
+ return self._running
diff --git a/nanobot/channels/manager.py b/nanobot/channels/manager.py
new file mode 100644
index 0000000..04abf5f
--- /dev/null
+++ b/nanobot/channels/manager.py
@@ -0,0 +1,137 @@
+"""Channel manager for coordinating chat channels."""
+
+import asyncio
+from typing import Any
+
+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 Config
+
+
+class ChannelManager:
+ """
+ Manages chat channels and coordinates message routing.
+
+ Responsibilities:
+ - Initialize enabled channels (Telegram, WhatsApp, etc.)
+ - Start/stop channels
+ - Route outbound messages
+ """
+
+ def __init__(self, config: Config, bus: MessageBus):
+ self.config = config
+ self.bus = bus
+ self.channels: dict[str, BaseChannel] = {}
+ self._dispatch_task: asyncio.Task | None = None
+
+ self._init_channels()
+
+ def _init_channels(self) -> None:
+ """Initialize channels based on config."""
+
+ # Telegram channel
+ if self.config.channels.telegram.enabled:
+ try:
+ from nanobot.channels.telegram import TelegramChannel
+ self.channels["telegram"] = TelegramChannel(
+ self.config.channels.telegram, self.bus
+ )
+ logger.info("Telegram channel enabled")
+ except ImportError as e:
+ logger.warning(f"Telegram channel not available: {e}")
+
+ # WhatsApp channel
+ if self.config.channels.whatsapp.enabled:
+ try:
+ from nanobot.channels.whatsapp import WhatsAppChannel
+ self.channels["whatsapp"] = WhatsAppChannel(
+ self.config.channels.whatsapp, self.bus
+ )
+ logger.info("WhatsApp channel enabled")
+ except ImportError as e:
+ logger.warning(f"WhatsApp channel not available: {e}")
+
+ async def start_all(self) -> None:
+ """Start WhatsApp channel and the outbound dispatcher."""
+ if not self.channels:
+ logger.warning("No channels enabled")
+ return
+
+ # Start outbound dispatcher
+ self._dispatch_task = asyncio.create_task(self._dispatch_outbound())
+
+ # Start WhatsApp channel
+ tasks = []
+ for name, channel in self.channels.items():
+ logger.info(f"Starting {name} channel...")
+ tasks.append(asyncio.create_task(channel.start()))
+
+ # Wait for all to complete (they should run forever)
+ await asyncio.gather(*tasks, return_exceptions=True)
+
+ async def stop_all(self) -> None:
+ """Stop all channels and the dispatcher."""
+ logger.info("Stopping all channels...")
+
+ # Stop dispatcher
+ if self._dispatch_task:
+ self._dispatch_task.cancel()
+ try:
+ await self._dispatch_task
+ except asyncio.CancelledError:
+ pass
+
+ # Stop all channels
+ for name, channel in self.channels.items():
+ try:
+ await channel.stop()
+ logger.info(f"Stopped {name} channel")
+ except Exception as e:
+ logger.error(f"Error stopping {name}: {e}")
+
+ async def _dispatch_outbound(self) -> None:
+ """Dispatch outbound messages to the appropriate channel."""
+ logger.info("Outbound dispatcher started")
+
+ while True:
+ try:
+ msg = await asyncio.wait_for(
+ self.bus.consume_outbound(),
+ timeout=1.0
+ )
+
+ channel = self.channels.get(msg.channel)
+ if channel:
+ try:
+ await channel.send(msg)
+ except Exception as e:
+ logger.error(f"Error sending to {msg.channel}: {e}")
+ else:
+ logger.warning(f"Unknown channel: {msg.channel}")
+
+ except asyncio.TimeoutError:
+ continue
+ except asyncio.CancelledError:
+ break
+
+ def get_channel(self, name: str) -> BaseChannel | None:
+ """Get a channel by name."""
+ return self.channels.get(name)
+
+ def get_status(self) -> dict[str, Any]:
+ """Get status of all channels."""
+ return {
+ name: {
+ "enabled": True,
+ "running": channel.is_running
+ }
+ for name, channel in self.channels.items()
+ }
+
+ @property
+ def enabled_channels(self) -> list[str]:
+ """Get list of enabled channel names."""
+ return list(self.channels.keys())
diff --git a/nanobot/channels/telegram.py b/nanobot/channels/telegram.py
new file mode 100644
index 0000000..78a7e98
--- /dev/null
+++ b/nanobot/channels/telegram.py
@@ -0,0 +1,212 @@
+"""Telegram channel implementation using python-telegram-bot."""
+
+import asyncio
+from typing import Any
+
+from loguru import logger
+from telegram import Update
+from telegram.ext import Application, MessageHandler, filters, ContextTypes
+
+from nanobot.bus.events import OutboundMessage
+from nanobot.bus.queue import MessageBus
+from nanobot.channels.base import BaseChannel
+from nanobot.config.schema import TelegramConfig
+
+
+class TelegramChannel(BaseChannel):
+ """
+ Telegram channel using long polling.
+
+ Simple and reliable - no webhook/public IP needed.
+ """
+
+ name = "telegram"
+
+ def __init__(self, config: TelegramConfig, bus: MessageBus):
+ super().__init__(config, bus)
+ self.config: TelegramConfig = config
+ self._app: Application | None = None
+ self._chat_ids: dict[str, int] = {} # Map sender_id to chat_id for replies
+
+ async def start(self) -> None:
+ """Start the Telegram bot with long polling."""
+ if not self.config.token:
+ logger.error("Telegram bot token not configured")
+ return
+
+ self._running = True
+
+ # Build the application
+ self._app = (
+ Application.builder()
+ .token(self.config.token)
+ .build()
+ )
+
+ # Add message handler for text, photos, voice, documents
+ self._app.add_handler(
+ MessageHandler(
+ (filters.TEXT | filters.PHOTO | filters.VOICE | filters.AUDIO | filters.Document.ALL)
+ & ~filters.COMMAND,
+ self._on_message
+ )
+ )
+
+ # Add /start command handler
+ from telegram.ext import CommandHandler
+ self._app.add_handler(CommandHandler("start", self._on_start))
+
+ logger.info("Starting Telegram bot (polling mode)...")
+
+ # Initialize and start polling
+ await self._app.initialize()
+ await self._app.start()
+
+ # Get bot info
+ bot_info = await self._app.bot.get_me()
+ logger.info(f"Telegram bot @{bot_info.username} connected")
+
+ # Start polling (this runs until stopped)
+ await self._app.updater.start_polling(
+ allowed_updates=["message"],
+ drop_pending_updates=True # Ignore old messages on startup
+ )
+
+ # Keep running until stopped
+ while self._running:
+ await asyncio.sleep(1)
+
+ async def stop(self) -> None:
+ """Stop the Telegram bot."""
+ self._running = False
+
+ if self._app:
+ logger.info("Stopping Telegram bot...")
+ await self._app.updater.stop()
+ await self._app.stop()
+ await self._app.shutdown()
+ self._app = None
+
+ async def send(self, msg: OutboundMessage) -> None:
+ """Send a message through Telegram."""
+ if not self._app:
+ logger.warning("Telegram bot not running")
+ return
+
+ try:
+ # chat_id should be the Telegram chat ID (integer)
+ chat_id = int(msg.chat_id)
+ await self._app.bot.send_message(
+ chat_id=chat_id,
+ text=msg.content
+ )
+ except ValueError:
+ logger.error(f"Invalid chat_id: {msg.chat_id}")
+ except Exception as e:
+ logger.error(f"Error sending Telegram message: {e}")
+
+ async def _on_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+ """Handle /start command."""
+ if not update.message or not update.effective_user:
+ return
+
+ user = update.effective_user
+ await update.message.reply_text(
+ f"π Hi {user.first_name}! I'm nanobot.\n\n"
+ "Send me a message and I'll respond!"
+ )
+
+ async def _on_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+ """Handle incoming messages (text, photos, voice, documents)."""
+ if not update.message or not update.effective_user:
+ return
+
+ message = update.message
+ user = update.effective_user
+ chat_id = message.chat_id
+
+ # Get sender identifier (prefer username, fallback to user_id)
+ sender_id = str(user.username or user.id)
+
+ # Store chat_id for replies
+ self._chat_ids[sender_id] = chat_id
+
+ # Build content from text and/or media
+ content_parts = []
+ media_paths = []
+
+ # Text content
+ if message.text:
+ content_parts.append(message.text)
+ if message.caption:
+ content_parts.append(message.caption)
+
+ # Handle media files
+ media_file = None
+ media_type = None
+
+ if message.photo:
+ media_file = message.photo[-1] # Largest photo
+ media_type = "image"
+ elif message.voice:
+ media_file = message.voice
+ media_type = "voice"
+ elif message.audio:
+ media_file = message.audio
+ media_type = "audio"
+ elif message.document:
+ media_file = message.document
+ media_type = "file"
+
+ # Download media if present
+ if media_file and self._app:
+ try:
+ file = await self._app.bot.get_file(media_file.file_id)
+ ext = self._get_extension(media_type, getattr(media_file, 'mime_type', None))
+
+ # Save to workspace/media/
+ from pathlib import Path
+ media_dir = Path.home() / ".nanobot" / "media"
+ media_dir.mkdir(parents=True, exist_ok=True)
+
+ file_path = media_dir / f"{media_file.file_id[:16]}{ext}"
+ await file.download_to_drive(str(file_path))
+
+ media_paths.append(str(file_path))
+ content_parts.append(f"[{media_type}: {file_path}]")
+ logger.debug(f"Downloaded {media_type} to {file_path}")
+ except Exception as e:
+ logger.error(f"Failed to download media: {e}")
+ content_parts.append(f"[{media_type}: download failed]")
+
+ content = "\n".join(content_parts) if content_parts else "[empty message]"
+
+ logger.debug(f"Telegram message from {sender_id}: {content[:50]}...")
+
+ # Forward to the message bus
+ await self._handle_message(
+ sender_id=sender_id,
+ chat_id=str(chat_id),
+ content=content,
+ media=media_paths,
+ metadata={
+ "message_id": message.message_id,
+ "user_id": user.id,
+ "username": user.username,
+ "first_name": user.first_name,
+ "is_group": message.chat.type != "private"
+ }
+ )
+
+ def _get_extension(self, media_type: str, mime_type: str | None) -> str:
+ """Get file extension based on media type."""
+ if mime_type:
+ ext_map = {
+ "image/jpeg": ".jpg", "image/png": ".png", "image/gif": ".gif",
+ "audio/ogg": ".ogg", "audio/mpeg": ".mp3", "audio/mp4": ".m4a",
+ }
+ if mime_type in ext_map:
+ return ext_map[mime_type]
+
+ type_map = {"image": ".jpg", "voice": ".ogg", "audio": ".mp3", "file": ""}
+ return type_map.get(media_type, "")
diff --git a/nanobot/channels/whatsapp.py b/nanobot/channels/whatsapp.py
new file mode 100644
index 0000000..efbd3e1
--- /dev/null
+++ b/nanobot/channels/whatsapp.py
@@ -0,0 +1,136 @@
+"""WhatsApp channel implementation using Node.js bridge."""
+
+import asyncio
+import json
+from typing import Any
+
+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 WhatsAppConfig
+
+
+class WhatsAppChannel(BaseChannel):
+ """
+ WhatsApp channel that connects to a Node.js bridge.
+
+ The bridge uses @whiskeysockets/baileys to handle the WhatsApp Web protocol.
+ Communication between Python and Node.js is via WebSocket.
+ """
+
+ name = "whatsapp"
+
+ def __init__(self, config: WhatsAppConfig, bus: MessageBus):
+ super().__init__(config, bus)
+ self.config: WhatsAppConfig = config
+ self._ws = None
+ self._connected = False
+
+ async def start(self) -> None:
+ """Start the WhatsApp channel by connecting to the bridge."""
+ import websockets
+
+ bridge_url = self.config.bridge_url
+
+ logger.info(f"Connecting to WhatsApp bridge at {bridge_url}...")
+
+ self._running = True
+
+ while self._running:
+ try:
+ async with websockets.connect(bridge_url) as ws:
+ self._ws = ws
+ self._connected = True
+ logger.info("Connected to WhatsApp bridge")
+
+ # Listen for messages
+ async for message in ws:
+ try:
+ await self._handle_bridge_message(message)
+ except Exception as e:
+ logger.error(f"Error handling bridge message: {e}")
+
+ except asyncio.CancelledError:
+ break
+ except Exception as e:
+ self._connected = False
+ self._ws = None
+ logger.warning(f"WhatsApp bridge connection error: {e}")
+
+ if self._running:
+ logger.info("Reconnecting in 5 seconds...")
+ await asyncio.sleep(5)
+
+ async def stop(self) -> None:
+ """Stop the WhatsApp channel."""
+ self._running = False
+ self._connected = False
+
+ if self._ws:
+ await self._ws.close()
+ self._ws = None
+
+ async def send(self, msg: OutboundMessage) -> None:
+ """Send a message through WhatsApp."""
+ if not self._ws or not self._connected:
+ logger.warning("WhatsApp bridge not connected")
+ return
+
+ try:
+ payload = {
+ "type": "send",
+ "to": msg.chat_id,
+ "text": msg.content
+ }
+ await self._ws.send(json.dumps(payload))
+ except Exception as e:
+ logger.error(f"Error sending WhatsApp message: {e}")
+
+ async def _handle_bridge_message(self, raw: str) -> None:
+ """Handle a message from the bridge."""
+ try:
+ data = json.loads(raw)
+ except json.JSONDecodeError:
+ logger.warning(f"Invalid JSON from bridge: {raw[:100]}")
+ return
+
+ msg_type = data.get("type")
+
+ if msg_type == "message":
+ # Incoming message from WhatsApp
+ sender = data.get("sender", "")
+ content = data.get("content", "")
+
+ # sender is typically: @s.whatsapp.net
+ # Extract just the phone number as chat_id
+ chat_id = sender.split("@")[0] if "@" in sender else sender
+
+ await self._handle_message(
+ sender_id=chat_id,
+ chat_id=sender, # Use full JID for replies
+ content=content,
+ metadata={
+ "message_id": data.get("id"),
+ "timestamp": data.get("timestamp"),
+ "is_group": data.get("isGroup", False)
+ }
+ )
+
+ elif msg_type == "status":
+ # Connection status update
+ status = data.get("status")
+ logger.info(f"WhatsApp status: {status}")
+
+ if status == "connected":
+ self._connected = True
+ elif status == "disconnected":
+ self._connected = False
+
+ elif msg_type == "qr":
+ # QR code for authentication
+ logger.info("Scan QR code in the bridge terminal to connect WhatsApp")
+
+ elif msg_type == "error":
+ logger.error(f"WhatsApp bridge error: {data.get('error')}")
diff --git a/nanobot/cli/__init__.py b/nanobot/cli/__init__.py
new file mode 100644
index 0000000..b023cad
--- /dev/null
+++ b/nanobot/cli/__init__.py
@@ -0,0 +1 @@
+"""CLI module for nanobot."""
diff --git a/nanobot/cli/commands.py b/nanobot/cli/commands.py
new file mode 100644
index 0000000..05513f0
--- /dev/null
+++ b/nanobot/cli/commands.py
@@ -0,0 +1,634 @@
+"""CLI commands for nanobot."""
+
+import asyncio
+from pathlib import Path
+
+import typer
+from rich.console import Console
+from rich.table import Table
+
+from nanobot import __version__, __logo__
+
+app = typer.Typer(
+ name="nanobot",
+ help=f"{__logo__} nanobot - Personal AI Assistant",
+ no_args_is_help=True,
+)
+
+console = Console()
+
+
+def version_callback(value: bool):
+ if value:
+ console.print(f"{__logo__} nanobot v{__version__}")
+ raise typer.Exit()
+
+
+@app.callback()
+def main(
+ version: bool = typer.Option(
+ None, "--version", "-v", callback=version_callback, is_eager=True
+ ),
+):
+ """nanobot - Personal AI Assistant."""
+ pass
+
+
+# ============================================================================
+# Onboard / Setup
+# ============================================================================
+
+
+@app.command()
+def onboard():
+ """Initialize nanobot configuration and workspace."""
+ from nanobot.config.loader import get_config_path, save_config
+ from nanobot.config.schema import Config
+ from nanobot.utils.helpers import get_workspace_path
+
+ config_path = get_config_path()
+
+ if config_path.exists():
+ console.print(f"[yellow]Config already exists at {config_path}[/yellow]")
+ if not typer.confirm("Overwrite?"):
+ raise typer.Exit()
+
+ # Create default config
+ config = Config()
+ save_config(config)
+ console.print(f"[green]β[/green] Created config at {config_path}")
+
+ # Create workspace
+ workspace = get_workspace_path()
+ console.print(f"[green]β[/green] Created workspace at {workspace}")
+
+ # Create default bootstrap files
+ _create_workspace_templates(workspace)
+
+ console.print(f"\n{__logo__} nanobot is ready!")
+ console.print("\nNext steps:")
+ console.print(" 1. Add your API key to [cyan]~/.nanobot/config.json[/cyan]")
+ console.print(" Get one at: https://openrouter.ai/keys")
+ console.print(" 2. Chat: [cyan]nanobot agent -m \"Hello!\"[/cyan]")
+ console.print("\n[dim]Want Telegram/WhatsApp? See: https://github.com/HKUDS/nanobot#-chat-apps[/dim]")
+
+
+
+
+def _create_workspace_templates(workspace: Path):
+ """Create default workspace template files."""
+ templates = {
+ "AGENTS.md": """# Agent Instructions
+
+You are a helpful AI assistant. Be concise, accurate, and friendly.
+
+## Guidelines
+
+- 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
+""",
+ "SOUL.md": """# Soul
+
+I am nanobot, a lightweight AI assistant.
+
+## Personality
+
+- Helpful and friendly
+- Concise and to the point
+- Curious and eager to learn
+
+## Values
+
+- Accuracy over speed
+- User privacy and safety
+- Transparency in actions
+""",
+ "USER.md": """# User
+
+Information about the user goes here.
+
+## Preferences
+
+- Communication style: (casual/formal)
+- Timezone: (your timezone)
+- Language: (your preferred language)
+""",
+ }
+
+ for filename, content in templates.items():
+ file_path = workspace / filename
+ if not file_path.exists():
+ file_path.write_text(content)
+ console.print(f" [dim]Created {filename}[/dim]")
+
+ # Create memory directory and MEMORY.md
+ memory_dir = workspace / "memory"
+ memory_dir.mkdir(exist_ok=True)
+ memory_file = memory_dir / "MEMORY.md"
+ if not memory_file.exists():
+ memory_file.write_text("""# Long-term Memory
+
+This file stores important information that should persist across sessions.
+
+## User Information
+
+(Important facts about the user)
+
+## Preferences
+
+(User preferences learned over time)
+
+## Important Notes
+
+(Things to remember)
+""")
+ console.print(" [dim]Created memory/MEMORY.md[/dim]")
+
+
+# ============================================================================
+# Gateway / Server
+# ============================================================================
+
+
+@app.command()
+def gateway(
+ port: int = typer.Option(18789, "--port", "-p", help="Gateway port"),
+ verbose: bool = typer.Option(False, "--verbose", "-v", help="Verbose output"),
+):
+ """Start the nanobot gateway."""
+ from nanobot.config.loader import load_config, get_data_dir
+ from nanobot.bus.queue import MessageBus
+ from nanobot.providers.litellm_provider import LiteLLMProvider
+ from nanobot.agent.loop import AgentLoop
+ from nanobot.channels.manager import ChannelManager
+ from nanobot.cron.service import CronService
+ from nanobot.cron.types import CronJob
+ from nanobot.heartbeat.service import HeartbeatService
+
+ if verbose:
+ import logging
+ logging.basicConfig(level=logging.DEBUG)
+
+ console.print(f"{__logo__} Starting nanobot gateway on port {port}...")
+
+ config = load_config()
+
+ # Create components
+ bus = MessageBus()
+
+ # Create provider (supports OpenRouter, Anthropic, OpenAI)
+ api_key = config.get_api_key()
+ api_base = config.get_api_base()
+
+ if not api_key:
+ console.print("[red]Error: No API key configured.[/red]")
+ console.print("Set one in ~/.nanobot/config.json under providers.openrouter.apiKey")
+ raise typer.Exit(1)
+
+ provider = LiteLLMProvider(
+ api_key=api_key,
+ api_base=api_base,
+ default_model=config.agents.defaults.model
+ )
+
+ # Create agent
+ agent = AgentLoop(
+ bus=bus,
+ provider=provider,
+ workspace=config.workspace_path,
+ model=config.agents.defaults.model,
+ max_iterations=config.agents.defaults.max_tool_iterations,
+ brave_api_key=config.tools.web.search.api_key or None
+ )
+
+ # Create cron service
+ async def on_cron_job(job: CronJob) -> str | None:
+ """Execute a cron job through the agent."""
+ response = await agent.process_direct(
+ job.payload.message,
+ session_key=f"cron:{job.id}"
+ )
+ # Optionally deliver to channel
+ if job.payload.deliver and job.payload.to:
+ from nanobot.bus.events import OutboundMessage
+ await bus.publish_outbound(OutboundMessage(
+ channel=job.payload.channel or "whatsapp",
+ chat_id=job.payload.to,
+ content=response or ""
+ ))
+ return response
+
+ cron_store_path = get_data_dir() / "cron" / "jobs.json"
+ cron = CronService(cron_store_path, on_job=on_cron_job)
+
+ # Create heartbeat service
+ async def on_heartbeat(prompt: str) -> str:
+ """Execute heartbeat through the agent."""
+ return await agent.process_direct(prompt, session_key="heartbeat")
+
+ heartbeat = HeartbeatService(
+ workspace=config.workspace_path,
+ on_heartbeat=on_heartbeat,
+ interval_s=30 * 60, # 30 minutes
+ enabled=True
+ )
+
+ # Create channel manager
+ channels = ChannelManager(config, bus)
+
+ if channels.enabled_channels:
+ console.print(f"[green]β[/green] Channels enabled: {', '.join(channels.enabled_channels)}")
+ else:
+ console.print("[yellow]Warning: No channels enabled[/yellow]")
+
+ cron_status = cron.status()
+ if cron_status["jobs"] > 0:
+ console.print(f"[green]β[/green] Cron: {cron_status['jobs']} scheduled jobs")
+
+ console.print(f"[green]β[/green] Heartbeat: every 30m")
+
+ async def run():
+ try:
+ await cron.start()
+ await heartbeat.start()
+ await asyncio.gather(
+ agent.run(),
+ channels.start_all(),
+ )
+ except KeyboardInterrupt:
+ console.print("\nShutting down...")
+ heartbeat.stop()
+ cron.stop()
+ agent.stop()
+ await channels.stop_all()
+
+ asyncio.run(run())
+
+
+
+
+# ============================================================================
+# Agent Commands
+# ============================================================================
+
+
+@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"),
+):
+ """Interact with the agent directly."""
+ from nanobot.config.loader import load_config
+ from nanobot.bus.queue import MessageBus
+ from nanobot.providers.litellm_provider import LiteLLMProvider
+ from nanobot.agent.loop import AgentLoop
+
+ config = load_config()
+
+ api_key = config.get_api_key()
+ api_base = config.get_api_base()
+
+ if not api_key:
+ console.print("[red]Error: No API key configured.[/red]")
+ raise typer.Exit(1)
+
+ bus = MessageBus()
+ provider = LiteLLMProvider(
+ api_key=api_key,
+ api_base=api_base,
+ default_model=config.agents.defaults.model
+ )
+
+ agent_loop = AgentLoop(
+ bus=bus,
+ provider=provider,
+ workspace=config.workspace_path,
+ brave_api_key=config.tools.web.search.api_key or None
+ )
+
+ if message:
+ # Single message mode
+ async def run_once():
+ response = await agent_loop.process_direct(message, session_id)
+ console.print(f"\n{__logo__} {response}")
+
+ asyncio.run(run_once())
+ else:
+ # Interactive mode
+ console.print(f"{__logo__} Interactive mode (Ctrl+C to exit)\n")
+
+ async def run_interactive():
+ while True:
+ try:
+ user_input = console.input("[bold blue]You:[/bold blue] ")
+ if not user_input.strip():
+ continue
+
+ response = await agent_loop.process_direct(user_input, session_id)
+ console.print(f"\n{__logo__} {response}\n")
+ except KeyboardInterrupt:
+ console.print("\nGoodbye!")
+ break
+
+ asyncio.run(run_interactive())
+
+
+# ============================================================================
+# Channel Commands
+# ============================================================================
+
+
+channels_app = typer.Typer(help="Manage channels")
+app.add_typer(channels_app, name="channels")
+
+
+@channels_app.command("status")
+def channels_status():
+ """Show channel status."""
+ from nanobot.config.loader import load_config
+
+ config = load_config()
+
+ table = Table(title="Channel Status")
+ table.add_column("Channel", style="cyan")
+ table.add_column("Enabled", style="green")
+ table.add_column("Bridge URL", style="yellow")
+
+ wa = config.channels.whatsapp
+ table.add_row(
+ "WhatsApp",
+ "β" if wa.enabled else "β",
+ wa.bridge_url
+ )
+
+ console.print(table)
+
+
+def _get_bridge_dir() -> Path:
+ """Get the bridge directory, setting it up if needed."""
+ import shutil
+ import subprocess
+
+ # User's bridge location
+ user_bridge = Path.home() / ".nanobot" / "bridge"
+
+ # Check if already built
+ if (user_bridge / "dist" / "index.js").exists():
+ return user_bridge
+
+ # Check for npm
+ if not shutil.which("npm"):
+ console.print("[red]npm not found. Please install Node.js >= 18.[/red]")
+ raise typer.Exit(1)
+
+ # Find source bridge: first check package data, then source dir
+ pkg_bridge = Path(__file__).parent / "bridge" # nanobot/bridge (installed)
+ src_bridge = Path(__file__).parent.parent.parent / "bridge" # repo root/bridge (dev)
+
+ source = None
+ if (pkg_bridge / "package.json").exists():
+ source = pkg_bridge
+ elif (src_bridge / "package.json").exists():
+ source = src_bridge
+
+ if not source:
+ console.print("[red]Bridge source not found.[/red]")
+ console.print("Try reinstalling: pip install --force-reinstall nanobot")
+ raise typer.Exit(1)
+
+ console.print(f"{__logo__} Setting up bridge...")
+
+ # Copy to user directory
+ user_bridge.parent.mkdir(parents=True, exist_ok=True)
+ if user_bridge.exists():
+ shutil.rmtree(user_bridge)
+ shutil.copytree(source, user_bridge, ignore=shutil.ignore_patterns("node_modules", "dist"))
+
+ # Install and build
+ try:
+ console.print(" Installing dependencies...")
+ subprocess.run(["npm", "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)
+
+ console.print("[green]β[/green] Bridge ready\n")
+ except subprocess.CalledProcessError as e:
+ console.print(f"[red]Build failed: {e}[/red]")
+ if e.stderr:
+ console.print(f"[dim]{e.stderr.decode()[:500]}[/dim]")
+ raise typer.Exit(1)
+
+ return user_bridge
+
+
+@channels_app.command("login")
+def channels_login():
+ """Link device via QR code."""
+ import subprocess
+
+ bridge_dir = _get_bridge_dir()
+
+ console.print(f"{__logo__} Starting bridge...")
+ console.print("Scan the QR code to connect.\n")
+
+ try:
+ subprocess.run(["npm", "start"], cwd=bridge_dir, check=True)
+ 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]")
+
+
+# ============================================================================
+# Cron Commands
+# ============================================================================
+
+cron_app = typer.Typer(help="Manage scheduled tasks")
+app.add_typer(cron_app, name="cron")
+
+
+@cron_app.command("list")
+def cron_list(
+ all: bool = typer.Option(False, "--all", "-a", help="Include disabled jobs"),
+):
+ """List scheduled jobs."""
+ from nanobot.config.loader import get_data_dir
+ from nanobot.cron.service import CronService
+
+ store_path = get_data_dir() / "cron" / "jobs.json"
+ service = CronService(store_path)
+
+ jobs = service.list_jobs(include_disabled=all)
+
+ if not jobs:
+ console.print("No scheduled jobs.")
+ return
+
+ table = Table(title="Scheduled Jobs")
+ table.add_column("ID", style="cyan")
+ table.add_column("Name")
+ table.add_column("Schedule")
+ table.add_column("Status")
+ table.add_column("Next Run")
+
+ import time
+ for job in jobs:
+ # Format schedule
+ if job.schedule.kind == "every":
+ sched = f"every {(job.schedule.every_ms or 0) // 1000}s"
+ elif job.schedule.kind == "cron":
+ sched = job.schedule.expr or ""
+ else:
+ sched = "one-time"
+
+ # Format next run
+ next_run = ""
+ if job.state.next_run_at_ms:
+ next_time = time.strftime("%Y-%m-%d %H:%M", time.localtime(job.state.next_run_at_ms / 1000))
+ next_run = next_time
+
+ status = "[green]enabled[/green]" if job.enabled else "[dim]disabled[/dim]"
+
+ table.add_row(job.id, job.name, sched, status, next_run)
+
+ console.print(table)
+
+
+@cron_app.command("add")
+def cron_add(
+ name: str = typer.Option(..., "--name", "-n", help="Job name"),
+ message: str = typer.Option(..., "--message", "-m", help="Message for agent"),
+ every: int = typer.Option(None, "--every", "-e", help="Run every N seconds"),
+ cron_expr: str = typer.Option(None, "--cron", "-c", help="Cron expression (e.g. '0 9 * * *')"),
+ at: str = typer.Option(None, "--at", help="Run once at time (ISO format)"),
+ deliver: bool = typer.Option(False, "--deliver", "-d", help="Deliver response to channel"),
+ to: str = typer.Option(None, "--to", help="Recipient for delivery"),
+):
+ """Add a scheduled job."""
+ from nanobot.config.loader import get_data_dir
+ from nanobot.cron.service import CronService
+ from nanobot.cron.types import CronSchedule
+
+ # Determine schedule type
+ if every:
+ schedule = CronSchedule(kind="every", every_ms=every * 1000)
+ elif cron_expr:
+ schedule = CronSchedule(kind="cron", expr=cron_expr)
+ elif at:
+ import datetime
+ dt = datetime.datetime.fromisoformat(at)
+ schedule = CronSchedule(kind="at", at_ms=int(dt.timestamp() * 1000))
+ else:
+ console.print("[red]Error: Must specify --every, --cron, or --at[/red]")
+ raise typer.Exit(1)
+
+ store_path = get_data_dir() / "cron" / "jobs.json"
+ service = CronService(store_path)
+
+ job = service.add_job(
+ name=name,
+ schedule=schedule,
+ message=message,
+ deliver=deliver,
+ to=to,
+ )
+
+ console.print(f"[green]β[/green] Added job '{job.name}' ({job.id})")
+
+
+@cron_app.command("remove")
+def cron_remove(
+ job_id: str = typer.Argument(..., help="Job ID to remove"),
+):
+ """Remove a scheduled job."""
+ from nanobot.config.loader import get_data_dir
+ from nanobot.cron.service import CronService
+
+ store_path = get_data_dir() / "cron" / "jobs.json"
+ service = CronService(store_path)
+
+ if service.remove_job(job_id):
+ console.print(f"[green]β[/green] Removed job {job_id}")
+ else:
+ console.print(f"[red]Job {job_id} not found[/red]")
+
+
+@cron_app.command("enable")
+def cron_enable(
+ job_id: str = typer.Argument(..., help="Job ID"),
+ disable: bool = typer.Option(False, "--disable", help="Disable instead of enable"),
+):
+ """Enable or disable a job."""
+ from nanobot.config.loader import get_data_dir
+ from nanobot.cron.service import CronService
+
+ store_path = get_data_dir() / "cron" / "jobs.json"
+ service = CronService(store_path)
+
+ job = service.enable_job(job_id, enabled=not disable)
+ if job:
+ status = "disabled" if disable else "enabled"
+ console.print(f"[green]β[/green] Job '{job.name}' {status}")
+ else:
+ console.print(f"[red]Job {job_id} not found[/red]")
+
+
+@cron_app.command("run")
+def cron_run(
+ job_id: str = typer.Argument(..., help="Job ID to run"),
+ force: bool = typer.Option(False, "--force", "-f", help="Run even if disabled"),
+):
+ """Manually run a job."""
+ from nanobot.config.loader import get_data_dir
+ from nanobot.cron.service import CronService
+
+ store_path = get_data_dir() / "cron" / "jobs.json"
+ service = CronService(store_path)
+
+ async def run():
+ return await service.run_job(job_id, force=force)
+
+ if asyncio.run(run()):
+ console.print(f"[green]β[/green] Job executed")
+ else:
+ console.print(f"[red]Failed to run job {job_id}[/red]")
+
+
+# ============================================================================
+# Status Commands
+# ============================================================================
+
+
+@app.command()
+def status():
+ """Show nanobot status."""
+ from nanobot.config.loader import load_config, get_config_path
+ from nanobot.utils.helpers import get_workspace_path
+
+ config_path = get_config_path()
+ workspace = get_workspace_path()
+
+ console.print(f"{__logo__} nanobot Status\n")
+
+ console.print(f"Config: {config_path} {'[green]β[/green]' if config_path.exists() else '[red]β[/red]'}")
+ console.print(f"Workspace: {workspace} {'[green]β[/green]' if workspace.exists() else '[red]β[/red]'}")
+
+ if config_path.exists():
+ config = load_config()
+ console.print(f"Model: {config.agents.defaults.model}")
+
+ # Check API keys
+ has_openrouter = bool(config.providers.openrouter.api_key)
+ has_anthropic = bool(config.providers.anthropic.api_key)
+ has_openai = bool(config.providers.openai.api_key)
+
+ console.print(f"OpenRouter API: {'[green]β[/green]' if has_openrouter else '[dim]not set[/dim]'}")
+ console.print(f"Anthropic API: {'[green]β[/green]' if has_anthropic else '[dim]not set[/dim]'}")
+ console.print(f"OpenAI API: {'[green]β[/green]' if has_openai else '[dim]not set[/dim]'}")
+
+
+if __name__ == "__main__":
+ app()
diff --git a/nanobot/config/__init__.py b/nanobot/config/__init__.py
new file mode 100644
index 0000000..88e8e9b
--- /dev/null
+++ b/nanobot/config/__init__.py
@@ -0,0 +1,6 @@
+"""Configuration module for nanobot."""
+
+from nanobot.config.loader import load_config, get_config_path
+from nanobot.config.schema import Config
+
+__all__ = ["Config", "load_config", "get_config_path"]
diff --git a/nanobot/config/loader.py b/nanobot/config/loader.py
new file mode 100644
index 0000000..e828c20
--- /dev/null
+++ b/nanobot/config/loader.py
@@ -0,0 +1,95 @@
+"""Configuration loading utilities."""
+
+import json
+from pathlib import Path
+from typing import Any
+
+from nanobot.config.schema import Config
+
+
+def get_config_path() -> Path:
+ """Get the default configuration file path."""
+ return Path.home() / ".nanobot" / "config.json"
+
+
+def get_data_dir() -> Path:
+ """Get the nanobot data directory."""
+ from nanobot.utils.helpers import get_data_path
+ return get_data_path()
+
+
+def load_config(config_path: Path | None = None) -> Config:
+ """
+ Load configuration from file or create default.
+
+ Args:
+ config_path: Optional path to config file. Uses default if not provided.
+
+ Returns:
+ Loaded configuration object.
+ """
+ path = config_path or get_config_path()
+
+ if path.exists():
+ try:
+ with open(path) as f:
+ data = json.load(f)
+ return Config.model_validate(convert_keys(data))
+ except (json.JSONDecodeError, ValueError) as e:
+ print(f"Warning: Failed to load config from {path}: {e}")
+ print("Using default configuration.")
+
+ return Config()
+
+
+def save_config(config: Config, config_path: Path | None = None) -> None:
+ """
+ Save configuration to file.
+
+ Args:
+ config: Configuration to save.
+ config_path: Optional path to save to. Uses default if not provided.
+ """
+ path = config_path or get_config_path()
+ path.parent.mkdir(parents=True, exist_ok=True)
+
+ # Convert to clawbot-compatible format (camelCase)
+ data = config.model_dump()
+ data = convert_to_camel(data)
+
+ with open(path, "w") as f:
+ json.dump(data, f, indent=2)
+
+
+def convert_keys(data: Any) -> Any:
+ """Convert camelCase keys to snake_case for Pydantic."""
+ if isinstance(data, dict):
+ return {camel_to_snake(k): convert_keys(v) for k, v in data.items()}
+ if isinstance(data, list):
+ return [convert_keys(item) for item in data]
+ return data
+
+
+def convert_to_camel(data: Any) -> Any:
+ """Convert snake_case keys to camelCase for clawbot compatibility."""
+ if isinstance(data, dict):
+ return {snake_to_camel(k): convert_to_camel(v) for k, v in data.items()}
+ if isinstance(data, list):
+ return [convert_to_camel(item) for item in data]
+ return data
+
+
+def camel_to_snake(name: str) -> str:
+ """Convert camelCase to snake_case."""
+ result = []
+ for i, char in enumerate(name):
+ if char.isupper() and i > 0:
+ result.append("_")
+ result.append(char.lower())
+ return "".join(result)
+
+
+def snake_to_camel(name: str) -> str:
+ """Convert snake_case to camelCase."""
+ components = name.split("_")
+ return components[0] + "".join(x.title() for x in components[1:])
diff --git a/nanobot/config/schema.py b/nanobot/config/schema.py
new file mode 100644
index 0000000..865bb55
--- /dev/null
+++ b/nanobot/config/schema.py
@@ -0,0 +1,111 @@
+"""Configuration schema using Pydantic."""
+
+from pathlib import Path
+from pydantic import BaseModel, Field
+from pydantic_settings import BaseSettings
+
+
+class WhatsAppConfig(BaseModel):
+ """WhatsApp channel configuration."""
+ enabled: bool = False
+ bridge_url: str = "ws://localhost:3001"
+ allow_from: list[str] = Field(default_factory=list) # Allowed phone numbers
+
+
+class TelegramConfig(BaseModel):
+ """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
+
+
+class ChannelsConfig(BaseModel):
+ """Configuration for chat channels."""
+ whatsapp: WhatsAppConfig = Field(default_factory=WhatsAppConfig)
+ telegram: TelegramConfig = Field(default_factory=TelegramConfig)
+
+
+class AgentDefaults(BaseModel):
+ """Default agent configuration."""
+ workspace: str = "~/.nanobot/workspace"
+ model: str = "anthropic/claude-opus-4-5"
+ max_tokens: int = 8192
+ temperature: float = 0.7
+ max_tool_iterations: int = 20
+
+
+class AgentsConfig(BaseModel):
+ """Agent configuration."""
+ defaults: AgentDefaults = Field(default_factory=AgentDefaults)
+
+
+class ProviderConfig(BaseModel):
+ """LLM provider configuration."""
+ api_key: str = ""
+ api_base: str | None = None
+
+
+class ProvidersConfig(BaseModel):
+ """Configuration for LLM providers."""
+ anthropic: ProviderConfig = Field(default_factory=ProviderConfig)
+ openai: ProviderConfig = Field(default_factory=ProviderConfig)
+ openrouter: ProviderConfig = Field(default_factory=ProviderConfig)
+
+
+class GatewayConfig(BaseModel):
+ """Gateway/server configuration."""
+ host: str = "0.0.0.0"
+ port: int = 18789
+
+
+class WebSearchConfig(BaseModel):
+ """Web search tool configuration."""
+ api_key: str = "" # Brave Search API key
+ max_results: int = 5
+
+
+class WebToolsConfig(BaseModel):
+ """Web tools configuration."""
+ search: WebSearchConfig = Field(default_factory=WebSearchConfig)
+
+
+class ToolsConfig(BaseModel):
+ """Tools configuration."""
+ web: WebToolsConfig = Field(default_factory=WebToolsConfig)
+
+
+class Config(BaseSettings):
+ """
+ Root configuration for nanobot.
+
+ Compatible with clawbot configuration format for easy migration.
+ """
+ agents: AgentsConfig = Field(default_factory=AgentsConfig)
+ channels: ChannelsConfig = Field(default_factory=ChannelsConfig)
+ providers: ProvidersConfig = Field(default_factory=ProvidersConfig)
+ gateway: GatewayConfig = Field(default_factory=GatewayConfig)
+ tools: ToolsConfig = Field(default_factory=ToolsConfig)
+
+ @property
+ def workspace_path(self) -> Path:
+ """Get expanded workspace path."""
+ return Path(self.agents.defaults.workspace).expanduser()
+
+ def get_api_key(self) -> str | None:
+ """Get API key in priority order: OpenRouter > Anthropic > OpenAI."""
+ return (
+ self.providers.openrouter.api_key or
+ self.providers.anthropic.api_key or
+ self.providers.openai.api_key or
+ None
+ )
+
+ def get_api_base(self) -> str | None:
+ """Get API base URL if using OpenRouter."""
+ if self.providers.openrouter.api_key:
+ return self.providers.openrouter.api_base or "https://openrouter.ai/api/v1"
+ return None
+
+ class Config:
+ env_prefix = "NANOBOT_"
+ env_nested_delimiter = "__"
diff --git a/nanobot/cron/__init__.py b/nanobot/cron/__init__.py
new file mode 100644
index 0000000..a9d4cad
--- /dev/null
+++ b/nanobot/cron/__init__.py
@@ -0,0 +1,6 @@
+"""Cron service for scheduled agent tasks."""
+
+from nanobot.cron.service import CronService
+from nanobot.cron.types import CronJob, CronSchedule
+
+__all__ = ["CronService", "CronJob", "CronSchedule"]
diff --git a/nanobot/cron/service.py b/nanobot/cron/service.py
new file mode 100644
index 0000000..d1965a9
--- /dev/null
+++ b/nanobot/cron/service.py
@@ -0,0 +1,346 @@
+"""Cron service for scheduling agent tasks."""
+
+import asyncio
+import json
+import time
+import uuid
+from pathlib import Path
+from typing import Any, Callable, Coroutine
+
+from loguru import logger
+
+from nanobot.cron.types import CronJob, CronJobState, CronPayload, CronSchedule, CronStore
+
+
+def _now_ms() -> int:
+ return int(time.time() * 1000)
+
+
+def _compute_next_run(schedule: CronSchedule, now_ms: int) -> int | None:
+ """Compute next run time in ms."""
+ if schedule.kind == "at":
+ return schedule.at_ms if schedule.at_ms and schedule.at_ms > now_ms else None
+
+ if schedule.kind == "every":
+ if not schedule.every_ms or schedule.every_ms <= 0:
+ return None
+ # Next interval from now
+ return now_ms + schedule.every_ms
+
+ if schedule.kind == "cron" and schedule.expr:
+ try:
+ from croniter import croniter
+ cron = croniter(schedule.expr, time.time())
+ next_time = cron.get_next()
+ return int(next_time * 1000)
+ except Exception:
+ return None
+
+ return None
+
+
+class CronService:
+ """Service for managing and executing scheduled jobs."""
+
+ def __init__(
+ self,
+ store_path: Path,
+ on_job: Callable[[CronJob], Coroutine[Any, Any, str | None]] | None = None
+ ):
+ self.store_path = store_path
+ self.on_job = on_job # Callback to execute job, returns response text
+ self._store: CronStore | None = None
+ self._timer_task: asyncio.Task | None = None
+ self._running = False
+
+ def _load_store(self) -> CronStore:
+ """Load jobs from disk."""
+ if self._store:
+ return self._store
+
+ if self.store_path.exists():
+ try:
+ data = json.loads(self.store_path.read_text())
+ jobs = []
+ for j in data.get("jobs", []):
+ jobs.append(CronJob(
+ id=j["id"],
+ name=j["name"],
+ enabled=j.get("enabled", True),
+ schedule=CronSchedule(
+ kind=j["schedule"]["kind"],
+ at_ms=j["schedule"].get("atMs"),
+ every_ms=j["schedule"].get("everyMs"),
+ expr=j["schedule"].get("expr"),
+ tz=j["schedule"].get("tz"),
+ ),
+ payload=CronPayload(
+ kind=j["payload"].get("kind", "agent_turn"),
+ message=j["payload"].get("message", ""),
+ deliver=j["payload"].get("deliver", False),
+ channel=j["payload"].get("channel"),
+ to=j["payload"].get("to"),
+ ),
+ state=CronJobState(
+ next_run_at_ms=j.get("state", {}).get("nextRunAtMs"),
+ last_run_at_ms=j.get("state", {}).get("lastRunAtMs"),
+ last_status=j.get("state", {}).get("lastStatus"),
+ last_error=j.get("state", {}).get("lastError"),
+ ),
+ created_at_ms=j.get("createdAtMs", 0),
+ updated_at_ms=j.get("updatedAtMs", 0),
+ delete_after_run=j.get("deleteAfterRun", False),
+ ))
+ self._store = CronStore(jobs=jobs)
+ except Exception as e:
+ logger.warning(f"Failed to load cron store: {e}")
+ self._store = CronStore()
+ else:
+ self._store = CronStore()
+
+ return self._store
+
+ def _save_store(self) -> None:
+ """Save jobs to disk."""
+ if not self._store:
+ return
+
+ self.store_path.parent.mkdir(parents=True, exist_ok=True)
+
+ data = {
+ "version": self._store.version,
+ "jobs": [
+ {
+ "id": j.id,
+ "name": j.name,
+ "enabled": j.enabled,
+ "schedule": {
+ "kind": j.schedule.kind,
+ "atMs": j.schedule.at_ms,
+ "everyMs": j.schedule.every_ms,
+ "expr": j.schedule.expr,
+ "tz": j.schedule.tz,
+ },
+ "payload": {
+ "kind": j.payload.kind,
+ "message": j.payload.message,
+ "deliver": j.payload.deliver,
+ "channel": j.payload.channel,
+ "to": j.payload.to,
+ },
+ "state": {
+ "nextRunAtMs": j.state.next_run_at_ms,
+ "lastRunAtMs": j.state.last_run_at_ms,
+ "lastStatus": j.state.last_status,
+ "lastError": j.state.last_error,
+ },
+ "createdAtMs": j.created_at_ms,
+ "updatedAtMs": j.updated_at_ms,
+ "deleteAfterRun": j.delete_after_run,
+ }
+ for j in self._store.jobs
+ ]
+ }
+
+ self.store_path.write_text(json.dumps(data, indent=2))
+
+ async def start(self) -> None:
+ """Start the cron service."""
+ self._running = True
+ self._load_store()
+ self._recompute_next_runs()
+ self._save_store()
+ self._arm_timer()
+ logger.info(f"Cron service started with {len(self._store.jobs if self._store else [])} jobs")
+
+ def stop(self) -> None:
+ """Stop the cron service."""
+ self._running = False
+ if self._timer_task:
+ self._timer_task.cancel()
+ self._timer_task = None
+
+ def _recompute_next_runs(self) -> None:
+ """Recompute next run times for all enabled jobs."""
+ if not self._store:
+ return
+ now = _now_ms()
+ for job in self._store.jobs:
+ if job.enabled:
+ job.state.next_run_at_ms = _compute_next_run(job.schedule, now)
+
+ def _get_next_wake_ms(self) -> int | None:
+ """Get the earliest next run time across all jobs."""
+ if not self._store:
+ return None
+ times = [j.state.next_run_at_ms for j in self._store.jobs
+ if j.enabled and j.state.next_run_at_ms]
+ return min(times) if times else None
+
+ def _arm_timer(self) -> None:
+ """Schedule the next timer tick."""
+ if self._timer_task:
+ self._timer_task.cancel()
+
+ next_wake = self._get_next_wake_ms()
+ if not next_wake or not self._running:
+ return
+
+ delay_ms = max(0, next_wake - _now_ms())
+ delay_s = delay_ms / 1000
+
+ async def tick():
+ await asyncio.sleep(delay_s)
+ if self._running:
+ await self._on_timer()
+
+ self._timer_task = asyncio.create_task(tick())
+
+ async def _on_timer(self) -> None:
+ """Handle timer tick - run due jobs."""
+ if not self._store:
+ return
+
+ now = _now_ms()
+ due_jobs = [
+ j for j in self._store.jobs
+ if j.enabled and j.state.next_run_at_ms and now >= j.state.next_run_at_ms
+ ]
+
+ for job in due_jobs:
+ await self._execute_job(job)
+
+ self._save_store()
+ self._arm_timer()
+
+ async def _execute_job(self, job: CronJob) -> None:
+ """Execute a single job."""
+ start_ms = _now_ms()
+ logger.info(f"Cron: executing job '{job.name}' ({job.id})")
+
+ try:
+ response = None
+ if self.on_job:
+ response = await self.on_job(job)
+
+ job.state.last_status = "ok"
+ job.state.last_error = None
+ logger.info(f"Cron: job '{job.name}' completed")
+
+ except Exception as e:
+ job.state.last_status = "error"
+ job.state.last_error = str(e)
+ logger.error(f"Cron: job '{job.name}' failed: {e}")
+
+ job.state.last_run_at_ms = start_ms
+ job.updated_at_ms = _now_ms()
+
+ # Handle one-shot jobs
+ if job.schedule.kind == "at":
+ if job.delete_after_run:
+ self._store.jobs = [j for j in self._store.jobs if j.id != job.id]
+ else:
+ job.enabled = False
+ job.state.next_run_at_ms = None
+ else:
+ # Compute next run
+ job.state.next_run_at_ms = _compute_next_run(job.schedule, _now_ms())
+
+ # ========== Public API ==========
+
+ def list_jobs(self, include_disabled: bool = False) -> list[CronJob]:
+ """List all jobs."""
+ store = self._load_store()
+ jobs = store.jobs if include_disabled else [j for j in store.jobs if j.enabled]
+ return sorted(jobs, key=lambda j: j.state.next_run_at_ms or float('inf'))
+
+ def add_job(
+ self,
+ name: str,
+ schedule: CronSchedule,
+ message: str,
+ deliver: bool = False,
+ channel: str | None = None,
+ to: str | None = None,
+ delete_after_run: bool = False,
+ ) -> CronJob:
+ """Add a new job."""
+ store = self._load_store()
+ now = _now_ms()
+
+ job = CronJob(
+ id=str(uuid.uuid4())[:8],
+ name=name,
+ enabled=True,
+ schedule=schedule,
+ payload=CronPayload(
+ kind="agent_turn",
+ message=message,
+ deliver=deliver,
+ channel=channel,
+ to=to,
+ ),
+ state=CronJobState(next_run_at_ms=_compute_next_run(schedule, now)),
+ created_at_ms=now,
+ updated_at_ms=now,
+ delete_after_run=delete_after_run,
+ )
+
+ store.jobs.append(job)
+ self._save_store()
+ self._arm_timer()
+
+ logger.info(f"Cron: added job '{name}' ({job.id})")
+ return job
+
+ def remove_job(self, job_id: str) -> bool:
+ """Remove a job by ID."""
+ store = self._load_store()
+ before = len(store.jobs)
+ store.jobs = [j for j in store.jobs if j.id != job_id]
+ removed = len(store.jobs) < before
+
+ if removed:
+ self._save_store()
+ self._arm_timer()
+ logger.info(f"Cron: removed job {job_id}")
+
+ return removed
+
+ def enable_job(self, job_id: str, enabled: bool = True) -> CronJob | None:
+ """Enable or disable a job."""
+ store = self._load_store()
+ for job in store.jobs:
+ if job.id == job_id:
+ job.enabled = enabled
+ job.updated_at_ms = _now_ms()
+ if enabled:
+ job.state.next_run_at_ms = _compute_next_run(job.schedule, _now_ms())
+ else:
+ job.state.next_run_at_ms = None
+ self._save_store()
+ self._arm_timer()
+ return job
+ return None
+
+ async def run_job(self, job_id: str, force: bool = False) -> bool:
+ """Manually run a job."""
+ store = self._load_store()
+ for job in store.jobs:
+ if job.id == job_id:
+ if not force and not job.enabled:
+ return False
+ await self._execute_job(job)
+ self._save_store()
+ self._arm_timer()
+ return True
+ return False
+
+ def status(self) -> dict:
+ """Get service status."""
+ store = self._load_store()
+ return {
+ "enabled": self._running,
+ "jobs": len(store.jobs),
+ "next_wake_at_ms": self._get_next_wake_ms(),
+ }
diff --git a/nanobot/cron/types.py b/nanobot/cron/types.py
new file mode 100644
index 0000000..2b42060
--- /dev/null
+++ b/nanobot/cron/types.py
@@ -0,0 +1,59 @@
+"""Cron types."""
+
+from dataclasses import dataclass, field
+from typing import Literal
+
+
+@dataclass
+class CronSchedule:
+ """Schedule definition for a cron job."""
+ kind: Literal["at", "every", "cron"]
+ # For "at": timestamp in ms
+ at_ms: int | None = None
+ # For "every": interval in ms
+ every_ms: int | None = None
+ # For "cron": cron expression (e.g. "0 9 * * *")
+ expr: str | None = None
+ # Timezone for cron expressions
+ tz: str | None = None
+
+
+@dataclass
+class CronPayload:
+ """What to do when the job runs."""
+ kind: Literal["system_event", "agent_turn"] = "agent_turn"
+ message: str = ""
+ # Deliver response to channel
+ deliver: bool = False
+ channel: str | None = None # e.g. "whatsapp"
+ to: str | None = None # e.g. phone number
+
+
+@dataclass
+class CronJobState:
+ """Runtime state of a job."""
+ next_run_at_ms: int | None = None
+ last_run_at_ms: int | None = None
+ last_status: Literal["ok", "error", "skipped"] | None = None
+ last_error: str | None = None
+
+
+@dataclass
+class CronJob:
+ """A scheduled job."""
+ id: str
+ name: str
+ enabled: bool = True
+ schedule: CronSchedule = field(default_factory=lambda: CronSchedule(kind="every"))
+ payload: CronPayload = field(default_factory=CronPayload)
+ state: CronJobState = field(default_factory=CronJobState)
+ created_at_ms: int = 0
+ updated_at_ms: int = 0
+ delete_after_run: bool = False
+
+
+@dataclass
+class CronStore:
+ """Persistent store for cron jobs."""
+ version: int = 1
+ jobs: list[CronJob] = field(default_factory=list)
diff --git a/nanobot/heartbeat/__init__.py b/nanobot/heartbeat/__init__.py
new file mode 100644
index 0000000..2ecd879
--- /dev/null
+++ b/nanobot/heartbeat/__init__.py
@@ -0,0 +1,5 @@
+"""Heartbeat service for periodic agent wake-ups."""
+
+from nanobot.heartbeat.service import HeartbeatService
+
+__all__ = ["HeartbeatService"]
diff --git a/nanobot/heartbeat/service.py b/nanobot/heartbeat/service.py
new file mode 100644
index 0000000..4cb469e
--- /dev/null
+++ b/nanobot/heartbeat/service.py
@@ -0,0 +1,130 @@
+"""Heartbeat service - periodic agent wake-up to check for tasks."""
+
+import asyncio
+from pathlib import Path
+from typing import Any, Callable, Coroutine
+
+from loguru import logger
+
+# Default interval: 30 minutes
+DEFAULT_HEARTBEAT_INTERVAL_S = 30 * 60
+
+# The prompt sent to agent during heartbeat
+HEARTBEAT_PROMPT = """Read HEARTBEAT.md in your workspace (if it exists).
+Follow any instructions or tasks listed there.
+If nothing needs attention, reply with just: HEARTBEAT_OK"""
+
+# Token that indicates "nothing to do"
+HEARTBEAT_OK_TOKEN = "HEARTBEAT_OK"
+
+
+def _is_heartbeat_empty(content: str | None) -> bool:
+ """Check if HEARTBEAT.md has no actionable content."""
+ if not content:
+ return True
+
+ # Lines to skip: empty, headers, HTML comments, empty checkboxes
+ skip_patterns = {"- [ ]", "* [ ]", "- [x]", "* [x]"}
+
+ for line in content.split("\n"):
+ line = line.strip()
+ if not line or line.startswith("#") or line.startswith("
+
+
+## Completed
+
+
+
diff --git a/workspace/SOUL.md b/workspace/SOUL.md
new file mode 100644
index 0000000..59403e7
--- /dev/null
+++ b/workspace/SOUL.md
@@ -0,0 +1,21 @@
+# Soul
+
+I am nanobot π, a personal AI assistant.
+
+## Personality
+
+- Helpful and friendly
+- Concise and to the point
+- Curious and eager to learn
+
+## Values
+
+- Accuracy over speed
+- User privacy and safety
+- Transparency in actions
+
+## Communication Style
+
+- Be clear and direct
+- Explain reasoning when helpful
+- Ask clarifying questions when needed
diff --git a/workspace/TOOLS.md b/workspace/TOOLS.md
new file mode 100644
index 0000000..9915561
--- /dev/null
+++ b/workspace/TOOLS.md
@@ -0,0 +1,138 @@
+# Available Tools
+
+This document describes the tools available to nanobot.
+
+## File Operations
+
+### read_file
+Read the contents of a file.
+```
+read_file(path: str) -> str
+```
+
+### write_file
+Write content to a file (creates parent directories if needed).
+```
+write_file(path: str, content: str) -> str
+```
+
+### edit_file
+Edit a file by replacing specific text.
+```
+edit_file(path: str, old_text: str, new_text: str) -> str
+```
+
+### list_dir
+List contents of a directory.
+```
+list_dir(path: str) -> str
+```
+
+## Shell Execution
+
+### exec
+Execute a shell command and return output.
+```
+exec(command: str, working_dir: str = None) -> str
+```
+
+**Safety Notes:**
+- Commands have a 60-second timeout
+- Output is truncated at 10,000 characters
+- Use with caution for destructive operations
+
+## Web Access
+
+### web_search
+Search the web using DuckDuckGo.
+```
+web_search(query: str) -> str
+```
+
+Returns top 5 search results with titles, URLs, and snippets.
+
+### web_fetch
+Fetch and extract main content from a URL.
+```
+web_fetch(url: str) -> str
+```
+
+**Notes:**
+- Content is extracted using trafilatura
+- Output is truncated at 8,000 characters
+
+## Communication
+
+### message
+Send a message to the user (used internally).
+```
+message(content: str, channel: str = None, chat_id: str = None) -> str
+```
+
+## Scheduled Reminders (Cron)
+
+Use the `exec` tool to create scheduled reminders with `nanobot cron add`:
+
+### Set a recurring reminder
+```bash
+# Every day at 9am
+nanobot cron add --name "morning" --message "Good morning! βοΈ" --cron "0 9 * * *"
+
+# Every 2 hours
+nanobot cron add --name "water" --message "Drink water! π§" --every 7200
+```
+
+### Set a one-time reminder
+```bash
+# At a specific time (ISO format)
+nanobot cron add --name "meeting" --message "Meeting starts now!" --at "2025-01-31T15:00:00"
+```
+
+### Manage reminders
+```bash
+nanobot cron list # List all jobs
+nanobot cron remove # Remove a job
+```
+
+## Heartbeat Task Management
+
+The `HEARTBEAT.md` file in the workspace is checked every 30 minutes.
+Use file operations to manage periodic tasks:
+
+### Add a heartbeat task
+```python
+# Append a new task
+edit_file(
+ path="HEARTBEAT.md",
+ old_text="## Example Tasks",
+ new_text="- [ ] New periodic task here\n\n## Example Tasks"
+)
+```
+
+### Remove a heartbeat task
+```python
+# Remove a specific task
+edit_file(
+ path="HEARTBEAT.md",
+ old_text="- [ ] Task to remove\n",
+ new_text=""
+)
+```
+
+### Rewrite all tasks
+```python
+# Replace the entire file
+write_file(
+ path="HEARTBEAT.md",
+ content="# Heartbeat Tasks\n\n- [ ] Task 1\n- [ ] Task 2\n"
+)
+```
+
+---
+
+## Adding Custom Tools
+
+To add custom tools:
+1. Create a class that extends `Tool` in `nanobot/agent/tools/`
+2. Implement `name`, `description`, `parameters`, and `execute`
+3. Register it in `AgentLoop._register_default_tools()`
diff --git a/workspace/USER.md b/workspace/USER.md
new file mode 100644
index 0000000..671ec49
--- /dev/null
+++ b/workspace/USER.md
@@ -0,0 +1,49 @@
+# User Profile
+
+Information about the user to help personalize interactions.
+
+## Basic Information
+
+- **Name**: (your name)
+- **Timezone**: (your timezone, e.g., UTC+8)
+- **Language**: (preferred language)
+
+## Preferences
+
+### Communication Style
+
+- [ ] Casual
+- [ ] Professional
+- [ ] Technical
+
+### Response Length
+
+- [ ] Brief and concise
+- [ ] Detailed explanations
+- [ ] Adaptive based on question
+
+### Technical Level
+
+- [ ] Beginner
+- [ ] Intermediate
+- [ ] Expert
+
+## Work Context
+
+- **Primary Role**: (your role, e.g., developer, researcher)
+- **Main Projects**: (what you're working on)
+- **Tools You Use**: (IDEs, languages, frameworks)
+
+## Topics of Interest
+
+-
+-
+-
+
+## Special Instructions
+
+(Any specific instructions for how the assistant should behave)
+
+---
+
+*Edit this file to customize nanobot's behavior for your needs.*
diff --git a/workspace/memory/MEMORY.md b/workspace/memory/MEMORY.md
new file mode 100644
index 0000000..fd2ca96
--- /dev/null
+++ b/workspace/memory/MEMORY.md
@@ -0,0 +1,23 @@
+# Long-term Memory
+
+This file stores important information that should persist across sessions.
+
+## User Information
+
+(Important facts about the user)
+
+## Preferences
+
+(User preferences learned over time)
+
+## Project Context
+
+(Information about ongoing projects)
+
+## Important Notes
+
+(Things to remember)
+
+---
+
+*This file is automatically updated by nanobot when important information should be remembered.*