Architecture

Overview

nanobot is designed to be ultra-lightweight (~10,000 total lines, ~4,000 core agent lines) while delivering full AI agent capabilities.

Core Philosophy

Design Principles

Simplicity: Every line of code should be necessary and understandable
Modularity: Components are loosely coupled and independently replaceable
Extensibility: Easy to add new providers, channels, and tools
Research-Ready: Clean, readable code for experimentation and learning

Architecture Diagram

┌─────────────────────────────────────────────────────────────┐
│                         Channels                            │
│  Telegram │ Discord │ WhatsApp │ Feishu │ Email │ ...      │
└─────────────┬───────────────────────────────────────────────┘
              │
              ▼
┌─────────────────────────────────────────────────────────────┐
│                       Message Bus                            │
│              (InboundMessage ↔ OutboundMessage)             │
└─────────────┬───────────────────────────────────────────────┘
              │
              ▼
┌─────────────────────────────────────────────────────────────┐
│                       Agent Loop                             │
│  ┌──────────────────────────────────────────────────────┐  │
│  │  1. Receive message                                   │  │
│  │  2. Build context (history + memory + skills)        │  │
│  │  3. Call LLM                                          │  │
│  │  4. Execute tools                                     │  │
│  │  5. Send response                                     │  │
│  └──────────────────────────────────────────────────────┘  │
└─────────────┬───────────────────────────────────────────────┘
              │
    ┌─────────┼─────────┬─────────────┐
    ▼         ▼         ▼             ▼
┌────────┐ ┌──────┐ ┌────────┐  ┌─────────┐
│Provider│ │Tools │ │Session │  │ Memory  │
│ (LLM)  │ │      │ │Manager │  │  Store  │
└────────┘ └──────┘ └────────┘  └─────────┘

Component Architecture

1. Channels (`nanobot/channels/`)

Purpose: Connect nanobot to chat platforms. Pattern: All channels implement BaseChannel:

class BaseChannel(ABC):
    async def start(self) -> None:
        """Connect and start listening."""
    
    async def send(self, msg: OutboundMessage) -> None:
        """Send a message to the platform."""
    
    async def stop(self) -> None:
        """Clean up and disconnect."""

Available Channels:

telegram.py - Telegram bot (polling)
discord.py - Discord bot (WebSocket)
whatsapp.py - WhatsApp (via bridge)
feishu.py - Feishu/Lark (WebSocket)
email.py - Email (IMAP/SMTP)
slack.py - Slack (Socket Mode)
qq.py - QQ (WebSocket)
dingtalk.py - DingTalk (Stream Mode)
matrix.py - Matrix/Element (E2EE support)
mochat.py - Mochat/Claw IM (Socket.IO)

Key Features:

Access control via allowFrom lists
Media handling (images, voice, files)
Typing indicators and reactions
Thread/group support

2. Message Bus (`nanobot/bus/`)

Purpose: Decouple channels from agent logic. Key Classes:

class InboundMessage:
    channel: str       # "telegram", "discord", etc.
    sender_id: str     # User identifier
    chat_id: str       # Chat/thread identifier
    content: str       # Message text
    media: list[str]   # Media file paths
    metadata: dict     # Channel-specific data

class OutboundMessage:
    channel: str
    chat_id: str
    content: str
    media: list[str]
    metadata: dict

class MessageBus:
    async def publish_inbound(msg: InboundMessage)
    async def publish_outbound(msg: OutboundMessage)

Flow:

Channel receives message from platform
Channel publishes InboundMessage to bus
AgentLoop processes message
AgentLoop publishes OutboundMessage to bus
Channel sends message to platform

3. Agent Loop (`nanobot/agent/loop.py`)

Purpose: Core processing engine. Main Loop:

while iterations < max_iterations:
    # 1. Build context
    context = context_builder.build(
        history=session.history,
        memory=memory_store.get(),
        skills=skills_loader.load()
    )
    
    # 2. Call LLM
    response = await provider.chat(
        messages=context,
        tools=tool_registry.get_schemas(),
        model=model
    )
    
    # 3. Execute tools
    if response.has_tool_calls:
        results = await execute_tools(response.tool_calls)
        session.add_tool_results(results)
        continue  # Loop back to LLM
    
    # 4. Send response
    await bus.publish_outbound(OutboundMessage(...))
    break

Key Features:

Session management (per user/chat)
Memory integration (MEMORY.md)
Skill loading (from ~/.nanobot/skills/)
Tool execution with timeouts
Subagent spawning for background tasks
MCP server integration

4. Providers (`nanobot/providers/`)

Purpose: Abstract LLM API differences. Pattern: All providers implement LLMProvider:

class LLMProvider(ABC):
    async def chat(
        messages: list[dict],
        tools: list[dict] | None,
        model: str,
        max_tokens: int,
        temperature: float
    ) -> LLMResponse:
        """Send a chat completion request."""

Implementations:

litellm_provider.py - LiteLLM wrapper (most providers)
custom_provider.py - Direct OpenAI-compatible API
openai_codex_provider.py - OAuth-based providers

Registry Pattern (registry.py):

ProviderSpec(
    name="deepseek",
    keywords=("deepseek",),
    env_key="DEEPSEEK_API_KEY",
    litellm_prefix="deepseek",
    ...
)

Adding a provider requires:

Add ProviderSpec to PROVIDERS tuple
Add config field to ProvidersConfig

No if-elif chains needed!

5. Tools (`nanobot/agent/tools/`)

Purpose: Give the agent capabilities. Built-in Tools:

shell.py - Execute shell commands
filesystem.py - Read/write/edit/list files
web.py - Web search and fetch
spawn.py - Create background subagents
cron.py - Schedule tasks
message.py - Send messages to channels
mcp.py - Model Context Protocol servers

Pattern: All tools implement BaseTool:

class BaseTool(ABC):
    @abstractmethod
    async def execute(self, **kwargs) -> str:
        """Execute the tool and return result."""
    
    @abstractmethod
    def get_schema(self) -> dict:
        """Return JSON schema for the tool."""

Tool Registry:

registry = ToolRegistry()
registry.register(ExecTool())
registry.register(ReadFileTool())
# ...

schemas = registry.get_schemas()  # For LLM
result = await registry.execute("exec", {"command": "ls"})  # Execute

6. Session Management (`nanobot/session/`)

Purpose: Track conversation history per user/chat.

class Session:
    history: list[dict]  # OpenAI message format
    
    def add_user_message(content: str, media: list)
    def add_assistant_message(content: str)
    def add_tool_call(name: str, args: dict)
    def add_tool_result(call_id: str, result: str)
    def save()  # Persist to ~/.nanobot/sessions/{session_id}.json

Features:

Per-user sessions (isolated history)
Thread-scoped sessions (Discord/Slack threads)
Automatic persistence
Session reset via /new command

7. Memory (`nanobot/agent/memory.py`)

Purpose: Long-term persistent memory. Implementation:

Stored in ~/.nanobot/workspace/MEMORY.md
Agent can read/update via prompts
Markdown format for human readability
Injected into every context

Usage Pattern:

## Important Facts
- User's name: Alice
- Preferred language: Python
- Timezone: UTC-8

## Ongoing Tasks
- Building a web scraper
- Learning about asyncio

8. Context Builder (`nanobot/agent/context.py`)

Purpose: Assemble prompts for the LLM. Context Structure:

[
    {"role": "system", "content": system_prompt},
    {"role": "system", "content": memory_content},
    {"role": "system", "content": skills_content},
    ...session.history,  # Previous messages
    {"role": "user", "content": current_message}
]

Features:

Token counting and truncation
Skill injection
Memory injection
Tool schema formatting

9. Cron Service (`nanobot/cron/`)

Purpose: Schedule periodic tasks. Features:

Natural language scheduling (“every day at 9am”)
Persistent job storage (~/.nanobot/workspace/cron/jobs.json)
Agent can create/list/delete jobs via CronTool

Example:

{
  "jobs": [
    {
      "id": "abc123",
      "description": "Check weather",
      "schedule": "0 9 * * *",
      "task": "Fetch weather and send summary"
    }
  ]
}

10. Heartbeat (`nanobot/heartbeat/`)

Purpose: Proactive task execution. How It Works:

Gateway wakes up every 30 minutes
Reads ~/.nanobot/workspace/HEARTBEAT.md
If file has tasks, spawns agent to execute them
Results sent to most recent chat channel

Example HEARTBEAT.md:

## Periodic Tasks
- [ ] Check for urgent emails
- [ ] Review GitHub notifications

Data Flow Example

User sends “What’s the weather?” via Telegram:

1. TelegramChannel receives update
   ├─ Downloads any media
   ├─ Checks allowFrom access control
   └─ Publishes InboundMessage to bus

2. MessageBus routes to AgentLoop
   └─ AgentLoop.process_message()

3. AgentLoop builds context
   ├─ Load session history
   ├─ Load MEMORY.md
   ├─ Load skills/*.md
   └─ Add current message

4. AgentLoop calls LLM
   ├─ LiteLLMProvider.chat()
   └─ Returns tool_call: web_search("weather")

5. AgentLoop executes tool
   ├─ ToolRegistry.execute("web_search", ...)
   └─ Returns: "Sunny, 72°F"

6. AgentLoop calls LLM again
   └─ Returns: "It's sunny and 72°F today!"

7. AgentLoop publishes OutboundMessage

8. MessageBus routes to TelegramChannel

9. TelegramChannel sends message
   └─ bot.send_message(chat_id, "It's sunny...")

Extension Points

Adding a New Channel

Create nanobot/channels/mychannel.py
Implement BaseChannel
Add config to config/schema.py
Register in channels/manager.py

See Adding Channels guide.

Adding a New Provider

Add ProviderSpec to providers/registry.py
Add config field to config/schema.py

See Adding Providers guide.

Adding a New Tool

Create nanobot/agent/tools/mytool.py
Implement BaseTool
Register in AgentLoop.__init__

class MyTool(BaseTool):
    async def execute(self, param: str) -> str:
        return f"Result: {param}"
    
    def get_schema(self) -> dict:
        return {
            "type": "function",
            "function": {
                "name": "my_tool",
                "description": "Does something useful",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "param": {"type": "string"}
                    },
                    "required": ["param"]
                }
            }
        }

Performance Characteristics

Memory Usage

Base: ~50MB (Python runtime + libraries)
Per session: ~1KB (message history)
Per channel: ~5-10MB (WebSocket connections)

Startup Time

CLI mode: Less than 1 second
Gateway mode: 2-5 seconds (channel connections)

Response Time

Network latency: 10-50ms (channel → bus → agent)
LLM latency: 0.5-3 seconds (provider-dependent)
Tool execution: Varies by tool (shell: less than 1s, web: 1-5s)

Security Architecture

Access Control

Channel-level: allowFrom lists
Tool-level: restrictToWorkspace sandbox
File operations: Path validation and workspace restriction

Workspace Isolation

When restrictToWorkspace: true:

All file operations restricted to ~/.nanobot/workspace/
Shell commands allowed but file access controlled
Path traversal attempts blocked

API Key Management

Stored in ~/.nanobot/config.json (600 permissions)
Never logged or exposed to users
Environment variables set per-request, not globally

Logging and Debugging

Log Levels

logger.debug("Detailed info")      # Development
logger.info("Normal operation")   # Production
logger.warning("Unexpected")      # Issues
logger.error("Failure")           # Errors

Enable Debug Logging

# In Python code
from loguru import logger
logger.remove()
logger.add(sys.stderr, level="DEBUG")

# Or use agent with --logs flag
nanobot agent --logs -m "Test"

Testing Strategy

Currently manual testing focused:

CLI testing: nanobot agent -m "..."
Gateway testing: Real channel integration
Provider testing: Multiple LLM providers
Tool testing: Execute each tool manually

Future: Add unit tests for core components.

Performance Optimization

Current Optimizations

Prompt caching: Anthropic prompt caching support
Session persistence: Avoid re-reading from disk
Tool timeouts: Prevent hanging on slow operations
Connection pooling: Reuse HTTP connections

Future Optimizations

Streaming responses: Real-time output
Parallel tool execution: Run independent tools concurrently
Message batching: Reduce bus overhead

Code Metrics

# Count core agent lines
bash core_agent_lines.sh
# ~4,000 lines

# Total project lines
wc -l nanobot/**/*.py
# ~10,000 lines

99% smaller than alternative agent frameworks!

Contributing

Resources

Overview

Core Philosophy

Design Principles

Architecture Diagram

Component Architecture

1. Channels (`nanobot/channels/`)

2. Message Bus (`nanobot/bus/`)

3. Agent Loop (`nanobot/agent/loop.py`)

4. Providers (`nanobot/providers/`)

5. Tools (`nanobot/agent/tools/`)

6. Session Management (`nanobot/session/`)

7. Memory (`nanobot/agent/memory.py`)

8. Context Builder (`nanobot/agent/context.py`)

9. Cron Service (`nanobot/cron/`)

10. Heartbeat (`nanobot/heartbeat/`)

Data Flow Example

Extension Points

Adding a New Channel

Adding a New Provider

Adding a New Tool

Performance Characteristics

Memory Usage

Startup Time

Response Time

Security Architecture

Access Control

Workspace Isolation

API Key Management

Logging and Debugging

Log Levels

Enable Debug Logging

Testing Strategy

Performance Optimization

Current Optimizations

Future Optimizations

Code Metrics

Build docs developers (and LLMs) love

Contributing

Resources

​Overview

​Core Philosophy

​Design Principles

​Architecture Diagram

​Component Architecture

​1. Channels (nanobot/channels/)

​2. Message Bus (nanobot/bus/)

​3. Agent Loop (nanobot/agent/loop.py)

​4. Providers (nanobot/providers/)

​5. Tools (nanobot/agent/tools/)

​6. Session Management (nanobot/session/)

​7. Memory (nanobot/agent/memory.py)

​8. Context Builder (nanobot/agent/context.py)

​9. Cron Service (nanobot/cron/)

​10. Heartbeat (nanobot/heartbeat/)

​Data Flow Example

​Extension Points

​Adding a New Channel

​Adding a New Provider

​Adding a New Tool

​Performance Characteristics

​Memory Usage

​Startup Time

​Response Time

​Security Architecture

​Access Control

​Workspace Isolation

​API Key Management

​Logging and Debugging

​Log Levels

​Enable Debug Logging

​Testing Strategy

​Performance Optimization

​Current Optimizations

​Future Optimizations

​Code Metrics

Build docs developers (and LLMs) love

Overview

Core Philosophy

Design Principles

Architecture Diagram

Component Architecture

1. Channels (`nanobot/channels/`)

2. Message Bus (`nanobot/bus/`)

3. Agent Loop (`nanobot/agent/loop.py`)

4. Providers (`nanobot/providers/`)

5. Tools (`nanobot/agent/tools/`)

6. Session Management (`nanobot/session/`)

7. Memory (`nanobot/agent/memory.py`)

8. Context Builder (`nanobot/agent/context.py`)

9. Cron Service (`nanobot/cron/`)

10. Heartbeat (`nanobot/heartbeat/`)

Data Flow Example

Extension Points

Adding a New Channel

Adding a New Provider

Adding a New Tool

Performance Characteristics

Memory Usage

Startup Time

Response Time

Security Architecture

Access Control

Workspace Isolation

API Key Management

Logging and Debugging

Log Levels

Enable Debug Logging

Testing Strategy

Performance Optimization

Current Optimizations

Future Optimizations

Code Metrics