Building OpenClaw From First Principles

A deep dive into personal AI assistant architecture

This tutorial teaches you how persistent, tool-using AI assistants actually work by building one from scratch. We'll start with a basic chatbot and incrementally add features until we understand every major component of OpenClaw's architecture.

What you'll learn:

How to give AI assistants persistent memory across conversations
How tool calling works and why it's powerful
How to build a gateway that connects multiple messaging platforms
How to manage state, sessions, and context windows
How to implement security boundaries for dangerous operations
How to architect multi-agent systems

Prerequisites: Basic Python or TypeScript knowledge. Familiarity with LLM APIs helpful but not required.

The Problem Space
Layer 0: The Simplest Chatbot
Layer 1: Persistent Sessions
Layer 2: Adding Personality (SOUL)
Layer 3: Tool Calling
Layer 4: Security & Permissions
Layer 5: The Gateway Pattern
Layer 6: Context Window Management
Layer 7: Cross-Session Memory
Layer 8: Concurrency & Command Queue
Layer 9: Scheduled Tasks (Cron)
Layer 10: Multi-Agent Routing
Production Patterns
Complete Implementation

The Problem Space

When you use ChatGPT or Claude in a browser, you face fundamental limitations:

You go to it, it doesn't come to you - You must open a browser tab
It can't do anything - It can only respond with text
It lives in isolation - Separate from where you actually communicate
It forgets everything - Each session is independent
One size fits all - Same personality for every use case

What if instead:

The AI lived in your messaging apps (WhatsApp, Telegram, Discord, Slack)?
It could execute commands, browse the web, manage files?
It remembered who you are across all conversations?
It could act on schedules and wake up automatically?
You could have different AI personalities for different contexts?

This is what OpenClaw does. Let's build it from scratch.

Layer 0: The Simplest Chatbot

Let's start at absolute zero: an AI that responds to messages.

# bot_v0.py - The simplest possible AI bot
import os
import anthropic
from telegram import Update
from telegram.ext import Application, MessageHandler, filters

client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

async def handle_message(update: Update, context):
    user_message = update.message.text
    
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": user_message}]
    )
    
    await update.message.reply_text(response.content[0].text)

# Start the bot
app = Application.builder().token(os.getenv("TELEGRAM_BOT_TOKEN")).build()
app.add_handler(MessageHandler(filters.TEXT, handle_message))
app.run_polling()

Run it:

export ANTHROPIC_API_KEY="your-key"
export TELEGRAM_BOT_TOKEN="your-token"
python bot_v0.py

Send a message on Telegram → bot responds. Simple!

The problem: This is stateless. Every message is a fresh conversation. Ask "what did I say earlier?" and the bot has no idea.

Layer 1: Persistent Sessions

To give the bot memory, we need to track conversation history per user.

The JSONL Format

OpenClaw uses JSONL (JSON Lines) for session storage. Each line is one message. Why?

Append-only - Just write to the end of the file
Crash-safe - If the process crashes mid-write, you lose at most one line
Human-readable - Open it in a text editor and see the conversation
Easy parsing - Read line by line, no need to load entire file

Example session file:

{"role":"user","content":"My name is Mehdi","timestamp":"2024-01-15T10:00:00Z"}
{"role":"assistant","content":"Nice to meet you, Mehdi!","timestamp":"2024-01-15T10:00:05Z"}
{"role":"user","content":"What's my name?","timestamp":"2024-01-15T11:30:00Z"}
{"role":"assistant","content":"Your name is Mehdi!","timestamp":"2024-01-15T11:30:03Z"}

Implementation

# bot_v1.py - Bot with persistent sessions
import json
import os
import anthropic
from telegram import Update
from telegram.ext import Application, MessageHandler, filters

client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

# Session storage
SESSIONS_DIR = "./sessions"
os.makedirs(SESSIONS_DIR, exist_ok=True)

def get_session_path(user_id):
    """Get the path to a user's session file."""
    return os.path.join(SESSIONS_DIR, f"{user_id}.jsonl")

def load_session(user_id):
    """Load conversation history from disk."""
    path = get_session_path(user_id)
    messages = []
    
    if os.path.exists(path):
        with open(path, "r") as f:
            for line in f:
                if line.strip():
                    messages.append(json.loads(line))
    
    return messages

def append_to_session(user_id, message):
    """Append a single message to the session file (crash-safe)."""
    path = get_session_path(user_id)
    with open(path, "a") as f:
        f.write(json.dumps(message) + "\n")

async def handle_message(update: Update, context):
    user_id = str(update.effective_user.id)
    user_message = update.message.text
    
    # Load existing conversation
    messages = load_session(user_id)
    
    # Add new user message
    user_msg = {"role": "user", "content": user_message}
    messages.append(user_msg)
    append_to_session(user_id, user_msg)
    
    # Call the AI with full history
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        messages=messages
    )
    
    # Save assistant response
    assistant_msg = {
        "role": "assistant", 
        "content": response.content[0].text
    }
    messages.append(assistant_msg)
    append_to_session(user_id, assistant_msg)
    
    await update.message.reply_text(response.content[0].text)

app = Application.builder().token(os.getenv("TELEGRAM_BOT_TOKEN")).build()
app.add_handler(MessageHandler(filters.TEXT, handle_message))
app.run_polling()

Try it:

You: My name is Mehdi
Bot: Nice to meet you, Mehdi!

[hours later, or even after restarting the bot...]

You: What's my name?
Bot: Your name is Mehdi!

Key Insight: Storage Location

In OpenClaw, session files live at:

~/.openclaw/agents/<agentId>/sessions/<sessionId>.jsonl

Each session maps to exactly one file. Restart the process, and everything is still there.

Session ID format:

"telegram:user123"                    # DM on Telegram
"telegram:group456:user123"           # Group chat on Telegram  
"whatsapp:+1234567890"                # WhatsApp DM
"discord:guild789:channel999:user111" # Discord

The hierarchical structure lets you:

Find all sessions for a user
Find all sessions in a group
Route messages correctly

Layer 2: Adding Personality (SOUL)

Our bot works, but it's generic. It has no consistent personality or boundaries.

OpenClaw solves this with SOUL.md - a markdown file that defines the agent's identity.

What is SOUL.md?

It's the system prompt that gets injected on every API call. But instead of hardcoding it, you write it in a file that can evolve.

Example SOUL.md:

# Who You Are

**Name:** Jarvis
**Role:** Personal AI assistant

## Personality

- Be genuinely helpful, not performatively helpful
- Skip the "Great question!" - just help
- Have opinions. You're allowed to disagree
- Be concise when needed, thorough when it matters

## Boundaries

- Private things stay private
- When in doubt, ask before acting externally
- You're not the user's voice - be careful about sending messages on their behalf

## Memory

You have access to long-term memory via tools.
Remember important details from conversations.
Write them down if they matter.

## Workspace

Your workspace is at ~/.openclaw/workspace
You can read and write files there to persist information.

Implementation

# bot_v2.py - Bot with personality

SOUL = """
# Who You Are

**Name:** Jarvis  
**Role:** Personal AI assistant

## Personality
- Be genuinely helpful, not performatively helpful
- Skip the "Great question!" - just help
- Have opinions. You're allowed to disagree
- Be concise when needed, thorough when it matters

## Boundaries
- Private things stay private
- When in doubt, ask before acting externally
- You're not the user's voice - be careful about sending messages on their behalf

## Memory
Remember important details from conversations.
Write them down if they matter.
"""

async def handle_message(update: Update, context):
    user_id = str(update.effective_user.id)
    messages = load_session(user_id)
    
    user_msg = {"role": "user", "content": update.message.text}
    messages.append(user_msg)
    append_to_session(user_id, user_msg)
    
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        system=SOUL,  # <-- personality injected here
        messages=messages
    )
    
    assistant_msg = {"role": "assistant", "content": response.content[0].text}
    messages.append(assistant_msg)
    append_to_session(user_id, assistant_msg)
    
    await update.message.reply_text(response.content[0].text)

In OpenClaw

The SOUL.md lives in the workspace:

~/.openclaw/workspace/SOUL.md

It gets loaded at session start and injected as the system prompt.

Key insight: The more specific your SOUL, the more consistent the agent's behavior.

Vague: "Be helpful"
Better: "Be the assistant you'd actually want to talk to. Concise when needed, thorough when it matters. Not a corporate drone. Not a sycophant. Just... good."

Layer 3: Tool Calling

A bot that can only talk is limited. What if it could do things?

The Tool Calling Mechanism

The Anthropic API (and OpenAI's) supports function calling:

You define tools with JSON schemas
The LLM decides when to use them
You execute the tool
You send the result back
The LLM continues with the result

User: "What files are in my home directory?"
  ↓
Claude: [uses bash tool] {"command": "ls ~"}
  ↓
Tool executes: "Documents/\nDownloads/\n..."
  ↓
Claude: "You have several folders: Documents/, Downloads/, ..."

Tool Definition

TOOLS = [
    {
        "name": "bash",
        "description": "Execute a bash command and return stdout/stderr",
        "input_schema": {
            "type": "object",
            "properties": {
                "command": {
                    "type": "string",
                    "description": "The bash command to execute"
                }
            },
            "required": ["command"]
        }
    },
    {
        "name": "read_file",
        "description": "Read a file from the filesystem",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string",
                    "description": "Path to the file"
                }
            },
            "required": ["path"]
        }
    },
    {
        "name": "write_file",
        "description": "Write content to a file",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "content": {"type": "string"}
            },
            "required": ["path", "content"]
        }
    }
]

Tool Execution

import subprocess

def execute_tool(name, tool_input):
    """Execute a tool and return the result."""
    
    if name == "bash":
        result = subprocess.run(
            tool_input["command"],
            shell=True,
            capture_output=True,
            text=True,
            timeout=30
        )
        return result.stdout + result.stderr
    
    elif name == "read_file":
        with open(tool_input["path"], "r") as f:
            return f.read()
    
    elif name == "write_file":
        os.makedirs(os.path.dirname(tool_input["path"]) or ".", exist_ok=True)
        with open(tool_input["path"], "w") as f:
            f.write(tool_input["content"])
        return f"Wrote to {tool_input['path']}"
    
    return f"Unknown tool: {name}"

The Agent Loop

def serialize_content(content):
    """Convert API response content blocks to JSON-serializable format."""
    serialized = []
    for block in content:
        if hasattr(block, "text"):
            serialized.append({"type": "text", "text": block.text})
        elif block.type == "tool_use":
            serialized.append({
                "type": "tool_use",
                "id": block.id,
                "name": block.name,
                "input": block.input
            })
    return serialized

def run_agent_turn(messages, system_prompt):
    """
    Run one full agent turn.
    May involve multiple tool calls in a loop until the agent is done.
    """
    while True:
        # Call Claude with tools
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            system=system_prompt,
            tools=TOOLS,
            messages=messages
        )
        
        content = serialize_content(response.content)
        
        # If the AI is done (no tool use), return the text
        if response.stop_reason == "end_turn":
            text = ""
            for block in response.content:
                if hasattr(block, "text"):
                    text += block.text
            messages.append({"role": "assistant", "content": content})
            return text, messages
        
        # Process tool calls
        if response.stop_reason == "tool_use":
            # Add assistant message with tool use
            messages.append({"role": "assistant", "content": content})
            
            # Execute each tool
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    print(f"  🔧 Tool: {block.name}({json.dumps(block.input)})")
                    result = execute_tool(block.name, block.input)
                    print(f"     → {str(result)[:100]}")
                    
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": str(result)
                    })
            
            # Add tool results as user message
            messages.append({"role": "user", "content": tool_results})
            
            # Loop continues - call Claude again with the results

Updated Message Handler

async def handle_message(update: Update, context):
    user_id = str(update.effective_user.id)
    messages = load_session(user_id)
    
    # Add user message
    messages.append({"role": "user", "content": update.message.text})
    
    # Run agent turn (handles tool calling loop)
    response_text, messages = run_agent_turn(messages, SOUL)
    
    # Save the full conversation (including tool calls)
    save_session(user_id, messages)
    
    await update.message.reply_text(response_text)

Try It

You: Create a file called hello.py that prints "Hello World", then run it

Bot: [uses write_file to create hello.py]
     [uses bash to execute it]
     Done! Created hello.py and ran it. Output: "Hello World"

Key insight: The AI decided which tools to use, in what order, and synthesized the results into a natural response.

OpenClaw's Tool Catalog

src/agents/
  ├── bash-tools.exec.ts        # Shell command execution
  └── bash-tools.process.ts     # Long-running process management

src/browser/
  ├── browser-tool.ts           # Web automation
  └── cdp-client.ts             # Chrome DevTools Protocol

src/canvas-host/
  └── canvas-tool.ts            # Visual workspace control

src/memory/
  └── memory-tools.ts           # Cross-session memory

Every tool follows the same pattern: schema + description + execution function.

Layer 4: Security & Permissions

We're executing bash commands from Telegram messages. That's dangerous.

What if someone gets access to your Telegram and sends: rm -rf /?

We need permission controls.

Approach: Tiered Safety

Safe commands - Auto-approved (ls, cat, pwd, date)
Previously approved - Remembered from past approvals
Dangerous patterns - Require explicit user approval

Implementation

import re

# Safe commands (read-only, harmless)
SAFE_COMMANDS = {
    "ls", "cat", "head", "tail", "wc", "date", 
    "whoami", "echo", "pwd", "which"
}

# Dangerous patterns
DANGEROUS_PATTERNS = [
    r"\brm\b",           # file deletion
    r"\bsudo\b",         # privilege escalation
    r"\bchmod\b",        # permission changes
    r"\bcurl.*\|.*sh",   # pipe to shell
    r"\bdd\b",           # disk operations
]

# Persistent approval storage
APPROVALS_FILE = "./exec-approvals.json"

def load_approvals():
    """Load the approval allowlist."""
    if os.path.exists(APPROVALS_FILE):
        with open(APPROVALS_FILE) as f:
            return json.load(f)
    return {"allowed": [], "denied": []}

def save_approval(command, approved):
    """Remember an approval decision."""
    approvals = load_approvals()
    key = "allowed" if approved else "denied"
    if command not in approvals[key]:
        approvals[key].append(command)
    with open(APPROVALS_FILE, "w") as f:
        json.dump(approvals, f, indent=2)

def check_command_safety(command):
    """
    Returns:
        "safe" - auto-approved
        "approved" - previously approved
        "needs_approval" - requires user confirmation
    """
    # Extract base command
    base_cmd = command.strip().split()[0] if command.strip() else ""
    
    # Check if it's in safe list
    if base_cmd in SAFE_COMMANDS:
        return "safe"
    
    # Check if previously approved
    approvals = load_approvals()
    if command in approvals["allowed"]:
        return "approved"
    
    # Check for dangerous patterns
    for pattern in DANGEROUS_PATTERNS:
        if re.search(pattern, command):
            return "needs_approval"
    
    return "needs_approval"

Updated Tool Execution

def execute_tool(name, tool_input):
    if name == "bash":
        cmd = tool_input["command"]
        safety = check_command_safety(cmd)
        
        if safety == "needs_approval":
            # In a real implementation, you'd prompt the user via Telegram
            # For now, we'll just deny
            print(f"  ⚠️  Blocked: {cmd} (needs approval)")
            return "Permission denied. Command requires approval."
        
        # Execute if safe or approved
        result = subprocess.run(
            cmd, shell=True, 
            capture_output=True, 
            text=True, 
            timeout=30
        )
        return result.stdout + result.stderr
    
    # ... other tools ...

OpenClaw's Permission Model

OpenClaw extends this with three tiers:

"ask" - Prompt user for approval (blocks execution)
"record" - Log but allow (audit trail)
"ignore" - Auto-allow (for trusted environments)

It also supports glob patterns:

{
  "allowed": [
    "git *",           // Approve all git commands
    "npm install *",   // Approve all npm installs
    "ls *"             // Approve all ls commands
  ]
}

Layer 5: The Gateway Pattern

So far we have a Telegram bot. But what if you also want the AI on:

Discord
WhatsApp
Slack
HTTP API
WebSocket

You could write separate bots for each. But then:

Separate sessions (AI on Telegram doesn't know what you discussed on Discord)
Separate memory
Duplicate code
Configuration hell

The solution: A Gateway - one central process managing all channels.

Key Insight

Look at our run_agent_turn function. It doesn't know anything about Telegram:

def run_agent_turn(messages, system_prompt):
    # Takes messages, returns text
    # No Telegram, no Discord, no platform-specific code

The agent logic is already decoupled from the channel!

Proof: Add a Second Channel

Let's add an HTTP API alongside Telegram, both using the same agent and sessions:

from flask import Flask, request, jsonify
import threading

flask_app = Flask(__name__)

@flask_app.route("/chat", methods=["POST"])
def chat():
    """HTTP endpoint for chatting with the agent."""
    data = request.json
    user_id = data["user_id"]
    
    # Same session loading as Telegram
    messages = load_session(user_id)
    messages.append({"role": "user", "content": data["message"]})
    
    # Same agent turn logic
    response_text, messages = run_agent_turn(messages, SOUL)
    
    # Same session saving
    save_session(user_id, messages)
    
    return jsonify({"response": response_text})

# Run HTTP server in background thread
threading.Thread(
    target=lambda: flask_app.run(port=5000), 
    daemon=True
).start()

# Telegram bot runs as before
app = Application.builder().token(os.getenv("TELEGRAM_BOT_TOKEN")).build()
app.add_handler(MessageHandler(filters.TEXT, handle_message))
app.run_polling()

Try it:

# Via Telegram
You: My name is Mehdi
Bot: Nice to meet you, Mehdi!

# Via HTTP (use your Telegram user ID to hit the same session)
curl -X POST http://127.0.0.1:5000/chat \
  -H "Content-Type: application/json" \
  -d '{"user_id": "YOUR_TELEGRAM_USER_ID", "message": "What is my name?"}'

# Response:
{"response": "Your name is Mehdi!"}

Same agent, same sessions, same memory. Two different interfaces.

OpenClaw's Gateway

The production gateway is a WebSocket server:

┌──────────────────────────────────┐
│      Gateway (WS Server)         │
│      ws://localhost:18789        │
└────────────┬─────────────────────┘
             │
    ┌────────┼───────┬──────┬──────┐
    │        │       │      │      │
Telegram  Discord  Slack  HTTP  WebChat

Why WebSocket?

Bidirectional communication
Real-time updates
Streaming responses
Connection state management

Each channel is a WebSocket client that:

Connects to the gateway
Sends normalized messages
Receives responses
Handles platform-specific formatting

Gateway Protocol Example

// Channel → Gateway
{
  type: "inbound_message",
  data: {
    channelId: "telegram",
    userId: "user123",
    content: "Hello!",
    timestamp: "2024-01-15T10:00:00Z"
  }
}

// Gateway → Channel
{
  type: "outbound_message",
  data: {
    channelId: "telegram",
    target: "user123",
    content: "Hi! How can I help?"
  }
}

All platform differences are abstracted away. The Gateway only sees:

Messages coming in
Messages going out
Session management
Tool execution

Layer 6: Context Window Management

As conversations grow, we hit a problem: token limits.

Claude Sonnet has a 200K token context window. After weeks of chatting, your session might have:

1000+ messages
500K+ tokens
Exceeds the limit!

The Solution: Compaction

When the session gets too long:

Keep recent messages (last 50)
Summarize old messages
Replace old section with summary

def estimate_tokens(messages):
    """Rough token estimate: ~4 chars per token."""
    total_chars = sum(len(json.dumps(m)) for m in messages)
    return total_chars // 4

def compact_session(user_id, messages):
    """Summarize old messages when context gets too long."""
    
    # Check if compaction needed (80% of 200K = 160K tokens)
    if estimate_tokens(messages) < 160_000:
        return messages  # No compaction needed
    
    # Split: old half vs recent half
    split_point = len(messages) // 2
    old_messages = messages[:split_point]
    recent_messages = messages[split_point:]
    
    print(f"  📦 Compacting session (too many tokens)...")
    
    # Ask Claude to summarize the old messages
    summary_response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2000,
        messages=[{
            "role": "user",
            "content": (
                "Summarize this conversation concisely. Preserve:\n"
                "- Key facts about the user (name, preferences, background)\n"
                "- Important decisions made\n"
                "- Open tasks or TODOs\n"
                "- Context needed for future conversations\n\n"
                f"{json.dumps(old_messages, indent=2)}"
            )
        }]
    )
    
    # Create compacted session
    summary_message = {
        "role": "user",
        "content": f"[Previous conversation summary]\n{summary_response.content[0].text}"
    }
    
    compacted = [summary_message] + recent_messages
    
    # Save compacted version
    save_session(user_id, compacted)
    
    return compacted

Add Compaction Check

async def handle_message(update: Update, context):
    user_id = str(update.effective_user.id)
    
    # Load session
    messages = load_session(user_id)
    
    # Check if compaction needed
    messages = compact_session(user_id, messages)
    
    # Add new message
    messages.append({"role": "user", "content": update.message.text})
    
    # Run agent turn
    response_text, messages = run_agent_turn(messages, SOUL)
    
    # Save
    save_session(user_id, messages)
    
    await update.message.reply_text(response_text)

Testing Compaction

To test without chatting for hours, temporarily lower the threshold:

if estimate_tokens(messages) < 1000:  # lowered for testing

Have a 10-15 message conversation, then watch the old messages get summarized.

OpenClaw's Approach

OpenClaw's compaction is more sophisticated:

Chunk-based - Splits into multiple chunks, summarizes each
Safety margin - Uses 80% of limit to account for estimation errors
Configurable - Can configure compaction strategy per agent
Preserves tool calls - Keeps recent tool use blocks for context

Layer 7: Cross-Session Memory

Session history gives you conversation memory. But what happens when:

You start a new session
You reset the current session
You switch to a different channel

Everything is gone! We need persistent knowledge that survives session resets.

The Approach: Memory as Files

Give the agent tools to:

Save facts to long-term storage
Search facts when needed

Memory Tools

MEMORY_DIR = "./memory"

# Add to TOOLS list:
{
    "name": "save_memory",
    "description": "Save important information to long-term memory. "
                   "Use for user preferences, key facts, and anything "
                   "worth remembering across sessions.",
    "input_schema": {
        "type": "object",
        "properties": {
            "key": {
                "type": "string",
                "description": "Short label (e.g. 'user-preferences', 'project-notes')"
            },
            "content": {
                "type": "string",
                "description": "The information to remember"
            }
        },
        "required": ["key", "content"]
    }
},
{
    "name": "memory_search",
    "description": "Search long-term memory for relevant information. "
                   "Use at the start of conversations to recall context.",
    "input_schema": {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "What to search for"
            }
        },
        "required": ["query"]
    }
}

Memory Execution

def execute_tool(name, tool_input):
    # ... existing tools ...
    
    elif name == "save_memory":
        os.makedirs(MEMORY_DIR, exist_ok=True)
        filepath = os.path.join(MEMORY_DIR, f"{tool_input['key']}.md")
        
        with open(filepath, "w") as f:
            f.write(tool_input["content"])
        
        return f"Saved to memory: {tool_input['key']}"
    
    elif name == "memory_search":
        query = tool_input["query"].lower()
        results = []
        
        if os.path.exists(MEMORY_DIR):
            for filename in os.listdir(MEMORY_DIR):
                if filename.endswith(".md"):
                    filepath = os.path.join(MEMORY_DIR, filename)
                    with open(filepath, "r") as f:
                        content = f.read()
                    
                    # Simple keyword search
                    if any(word in content.lower() for word in query.split()):
                        results.append(f"--- {filename} ---\n{content}")
        
        return "\n\n".join(results) if results else "No matching memories found."

Update SOUL to Mention Memory

SOUL = """
# Who You Are
... existing personality ...

## Memory

You have a long-term memory system:
- Use `save_memory` to store important information 
  (user preferences, key facts, project details)
- Use `memory_search` at the start of conversations 
  to recall context from previous sessions
- Memory files are stored as markdown in ./memory/
"""

Try It

You: Remember that my favorite restaurant is Sushi Ran 
     and I prefer weekend reservations

Bot: [uses save_memory]
     Got it - saved your restaurant preferences.

[Reset session or restart bot]

You: Where should we go for dinner?

Bot: [uses memory_search("restaurant dinner favorite")]
     How about Sushi Ran? I know it's your favorite. 
     Want to book a weekend spot?

Key insight: Memory persists because it's in files, not in the session.

OpenClaw's Production Memory

OpenClaw uses a hybrid approach:

Vector search - Semantic similarity using embeddings
Keyword search - Exact matching with FTS5
Knowledge graph - Entities and relationships

~/.openclaw/memory/
  ├── graph.json        # Entities + relations
  └── vectors.lance/    # LanceDB vector database

Example knowledge graph:

{
  "entities": [
    {
      "id": "user_mehdi",
      "type": "person",
      "observations": [
        "Staff Software Engineer at Meta",
        "Works on Wearables AI team",
        "Located in Redwood City, CA"
      ]
    },
    {
      "id": "project_habit_detection",
      "type": "project",
      "observations": [
        "Habit detection pipeline for AR glasses",
        "Has January deadline"
      ]
    }
  ],
  "relations": [
    {
      "from": "user_mehdi",
      "to": "project_habit_detection",
      "type": "working_on"
    }
  ]
}

This enables queries like:

"What am I working on?" → finds the relation → returns project details
"Tell me about the AR project" → semantic search → finds habit detection

Layer 8: Concurrency & Command Queue

Subtle but critical problem: What happens when two messages arrive simultaneously?

Example:

Send "check my calendar" on Telegram
Send "what's the weather" via HTTP
Both try to load the same session
Both try to append to it
Data corruption!

The Fix: Per-Session Locks

Only one message processes at a time per session. Different sessions can still run in parallel.

from collections import defaultdict
import threading

# Create a lock per session
session_locks = defaultdict(threading.Lock)

Wrap Message Handling

async def handle_message(update: Update, context):
    user_id = str(update.effective_user.id)
    
    # Acquire lock for this session
    with session_locks[user_id]:
        messages = load_session(user_id)
        messages = compact_session(user_id, messages)
        messages.append({"role": "user", "content": update.message.text})
        
        response_text, messages = run_agent_turn(messages, SOUL)
        
        save_session(user_id, messages)
    
    # Lock released here
    await update.message.reply_text(response_text)

Do the Same for HTTP

@flask_app.route("/chat", methods=["POST"])
def chat():
    data = request.json
    user_id = data["user_id"]
    
    with session_locks[user_id]:
        messages = load_session(user_id)
        messages = compact_session(user_id, messages)
        messages.append({"role": "user", "content": data["message"]})
        
        response_text, messages = run_agent_turn(messages, SOUL)
        
        save_session(user_id, messages)
    
    return jsonify({"response": response_text})

That's it! Five lines of setup. Now:

Messages for the same user queue up
Messages for different users run in parallel
No race conditions

OpenClaw's Queue System

OpenClaw extends this with lane-based queues:

Lane 1: Real-time messages (highest priority)
Lane 2: Scheduled tasks (cron jobs)
Lane 3: Background jobs (sub-agents)

This ensures heartbeats never block user messages.

Layer 9: Scheduled Tasks (Cron)

So far, the agent only responds when you talk to it. But what if you want it to:

Check email every morning
Summarize calendar before meetings
Monitor website changes
Send daily briefings

You need scheduled execution.

Using the schedule Library

import schedule
import time

def setup_heartbeats():
    """Configure recurring agent tasks."""
    
    def morning_briefing():
        print("\n⏰ Heartbeat: morning briefing")
        
        # Use isolated session key (don't pollute main chat)
        session_key = "cron:morning-briefing"
        
        with session_locks[session_key]:
            messages = load_session(session_key)
            messages.append({
                "role": "user",
                "content": "Good morning! Check today's date and give me a motivational quote."
            })
            
            response_text, messages = run_agent_turn(messages, SOUL)
            save_session(session_key, messages)
        
        print(f"🤖 {response_text}\n")
        # In production, send this to Telegram/Discord
    
    # Schedule daily at 7:30 AM
    schedule.every().day.at("07:30").do(morning_briefing)
    
    # Run scheduler in background thread
    def scheduler_loop():
        while True:
            schedule.run_pending()
            time.sleep(60)
    
    threading.Thread(target=scheduler_loop, daemon=True).start()

# Call during startup
setup_heartbeats()

Key insight: Each heartbeat uses its own session key (cron:morning-briefing).

This keeps scheduled tasks from cluttering your main conversation history.

Try It

For testing, schedule every minute:

schedule.every(1).minutes.do(morning_briefing)

You'll see the heartbeat fire in your terminal.

OpenClaw's Cron

OpenClaw supports full cron expressions:

heartbeats:
  - name: morning-check
    schedule: "30 7 * * *"  # 7:30 AM daily
    agent: main-agent
    prompt: "Check calendar and summarize today's meetings"
  
  - name: afternoon-reminder
    schedule: "0 14 * * 1-5"  # 2 PM weekdays
    agent: work-agent
    prompt: "Review open tasks and send status update"

Layer 10: Multi-Agent Routing

One agent is useful. But as you add more tasks, you'll find that a single personality and toolset can't handle everything well.

Example scenarios:

Personal agent - Full tools, casual tone
Work agent - Limited tools, professional tone
Research agent - No bash, focus on web search
Public bot - No tools, basic queries only

The Solution: Multiple Agents with Routing

Each agent has:

Its own SOUL
Its own session namespace
Its own tool access

You route messages based on:

Channel
User
Message content (prefix commands)

Implementation

AGENTS = {
    "main": {
        "name": "Jarvis",
        "soul": SOUL,  # our existing SOUL
        "session_prefix": "agent:main",
        "tools": TOOLS,  # all tools
    },
    "researcher": {
        "name": "Scout",
        "soul": """
You are Scout, a research specialist.
Your job: find information and cite sources.
Every claim needs evidence.
Use tools to gather data. Be thorough but concise.
Save important findings to memory for other agents to reference.
        """,
        "session_prefix": "agent:researcher",
        "tools": [t for t in TOOLS if t["name"] in ["web_search", "save_memory", "memory_search"]],
    },
}

def resolve_agent(message_text):
    """Route messages to the right agent based on prefix commands."""
    if message_text.startswith("/research "):
        return "researcher", message_text[len("/research "):]
    
    return "main", message_text

Updated Message Handler

async def handle_message(update: Update, context):
    user_id = str(update.effective_user.id)
    
    # Route to appropriate agent
    agent_id, message_text = resolve_agent(update.message.text)
    agent = AGENTS[agent_id]
    
    # Use agent-specific session key
    session_key = f"{agent['session_prefix']}:{user_id}"
    
    with session_locks[session_key]:
        messages = load_session(session_key)
        messages = compact_session(session_key, messages)
        messages.append({"role": "user", "content": message_text})
        
        response_text, messages = run_agent_turn(
            messages, 
            agent["soul"]
        )
        
        save_session(session_key, messages)
    
    # Indicate which agent responded
    await update.message.reply_text(f"[{agent['name']}] {response_text}")

Try It

You: What's the weather like?
[Jarvis] It's a nice day! I'd check a weather service for exact details.

You: /research What are the best practices for Python async programming?
[Scout] Here's what I found...
       [uses web_search, save_memory to gather and store findings]
       The key practices are: 1) Use asyncio.gather for concurrent tasks...

You: What did Scout find about Python async?
[Jarvis] [uses memory_search]
        Scout's research found that the key async best practices are...

Key insight: Each agent has its own conversation history, but they share the same memory directory. Scout saves research findings; Jarvis can search for them later.

OpenClaw's Routing

OpenClaw supports sophisticated routing rules:

routing:
  defaultAgent: personal-agent
  
  rules:
    # Priority 100 - specific user
    - channelId: telegram
      userId: "mehdi_telegram"
      agentId: personal-agent
    
    # Priority 90 - work Slack
    - channelId: slack
      groupId: "work_team_channel"
      agentId: work-agent
    
    # Priority 80 - public Discord
    - channelId: discord
      groupId: "public_server"
      agentId: public-bot
    
    # Priority 50 - admin commands
    - pattern: "^!admin"
      agentId: admin-agent

Rules are evaluated by priority. First match wins.

Production Patterns

Now that we understand the fundamentals, let's look at production patterns in OpenClaw.

1. WebSocket Protocol

Instead of HTTP polling, OpenClaw uses WebSocket for:

Bidirectional communication
Real-time streaming
Connection state

Message types:

// Channel → Gateway
type InboundMessage = {
  type: "inbound_message",
  data: {
    channelId: string,
    userId: string,
    content: string,
    media?: Attachment[]
  }
}

// Gateway → Channel
type OutboundMessage = {
  type: "outbound_message",
  data: {
    channelId: string,
    target: string,
    content: string
  }
}

// Tool execution
type ToolExecution = {
  type: "tool_execute",
  data: {
    toolName: string,
    input: unknown
  }
}

2. Streaming Responses

For real-time responses, OpenClaw streams chunks as they arrive:

// Agent → Gateway (streaming)
{
  type: "agent_response_chunk",
  data: {
    sessionId: string,
    chunk: string,
    sequenceId: number
  }
}

// Some channels support edit-in-place
// Telegram: Edit the same message as new chunks arrive
// Discord: Same
// WhatsApp: Send final message only

3. Browser Control with Semantic Snapshots

Instead of sending 5MB screenshots, OpenClaw uses accessibility trees:

// Semantic snapshot (text representation)
{
  role: "heading",
  name: "Welcome to GitHub",
  level: 1
},
{
  role: "button",
  name: "Sign In",
  ref: "ref_1"  // <- Agent can click this
},
{
  role: "textbox",
  name: "Email",
  ref: "ref_2"
}

Agent says: "click ref_1" → clicks exact button. No ambiguity.

This is ~100x smaller than screenshots in tokens.

4. Session Scoping

OpenClaw supports configurable session scoping:

main - All DMs share one session (simple)
per-peer - Each person gets one session across channels
per-channel-peer - Each person per channel gets own session

Example:

sessionScope: per-peer
identityLinks:
  - telegram:user123
    discord:user456
    # Same person, sessions merge

5. Skills System

Users can install skills (plugins) that add tools:

skills/
  ├── github/
  │   ├── SKILL.md           # Claude reads this
  │   ├── skill.json         # Metadata
  │   └── tools/
  │       ├── create-issue.ts
  │       └── search-repos.ts
  │
  └── notion/
      ├── SKILL.md
      ├── skill.json
      └── tools/
          └── search-pages.ts

Install: openclaw skill install github

The SKILL.md gets appended to the system prompt so Claude knows how to use it.

6. Sub-Agent Spawning

Agents can spawn child agents for focused tasks:

# Main agent can call:
sessions_spawn(
    agent="researcher",
    prompt="Research Python async best practices",
    timeout=300
)

# Child runs in isolated session
# Returns result to parent

This enables delegation patterns.

7. Rate Limiting

interface RateLimits {
  maxMessagesPerHour: number;
  maxMessagesPerUser: number;
  maxTokensPerDay: number;
}

// Per agent configuration
agents: {
  "public-bot": {
    rateLimit: {
      maxMessagesPerHour: 20,
      maxMessagesPerUser: 5
    }
  }
}

Complete Implementation

Here's a complete, runnable implementation combining everything we've built:

#!/usr/bin/env python3
"""
mini-openclaw.py - A minimal OpenClaw implementation
Demonstrates all core concepts from first principles

Run: pip install anthropic python-telegram-bot flask schedule
     python mini-openclaw.py
"""

import anthropic
import subprocess
import json
import os
import re
import threading
import time
import schedule
from collections import defaultdict
from datetime import datetime
from telegram import Update
from telegram.ext import Application, MessageHandler, filters
from flask import Flask, request, jsonify

# ═══════════════════════════════════════════════════════════════
# Configuration
# ═══════════════════════════════════════════════════════════════

client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

WORKSPACE = os.path.expanduser("~/.mini-openclaw")
SESSIONS_DIR = os.path.join(WORKSPACE, "sessions")
MEMORY_DIR = os.path.join(WORKSPACE, "memory")
APPROVALS_FILE = os.path.join(WORKSPACE, "exec-approvals.json")

# ═══════════════════════════════════════════════════════════════
# Agent Definitions
# ═══════════════════════════════════════════════════════════════

AGENTS = {
    "main": {
        "name": "Jarvis",
        "model": "claude-sonnet-4-20250514",
        "soul": f"""
# Who You Are

**Name:** Jarvis
**Role:** Personal AI assistant

## Personality
- Be genuinely helpful, not performatively helpful
- Skip the "Great question!" - just help
- Have opinions. You're allowed to disagree
- Be concise when needed, thorough when it matters

## Boundaries
- Private things stay private
- When in doubt, ask before acting externally
- You're not the user's voice - be careful in group chats

## Memory
Your workspace is {WORKSPACE}
Use save_memory to store important information across sessions.
Use memory_search at the start of conversations to recall context.

## Tools
You have access to bash, file operations, and memory tools.
Use them proactively to be helpful.
        """,
        "session_prefix": "agent:main",
    },
    "researcher": {
        "name": "Scout",
        "model": "claude-sonnet-4-20250514",
        "soul": """
# Who You Are

**Name:** Scout
**Role:** Research specialist

## Mission
Find information and cite sources.
Every claim needs evidence.
Be thorough but concise.

## Tools
Use tools to gather data.
Save important findings with save_memory for other agents to reference.
        """,
        "session_prefix": "agent:researcher",
    },
}

# ═══════════════════════════════════════════════════════════════
# Tools
# ═══════════════════════════════════════════════════════════════

TOOLS = [
    {
        "name": "bash",
        "description": "Execute a bash command and return stdout/stderr",
        "input_schema": {
            "type": "object",
            "properties": {
                "command": {
                    "type": "string",
                    "description": "The bash command to execute"
                }
            },
            "required": ["command"]
        }
    },
    {
        "name": "read_file",
        "description": "Read a file from the filesystem",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string",
                    "description": "Path to the file"
                }
            },
            "required": ["path"]
        }
    },
    {
        "name": "write_file",
        "description": "Write content to a file (creates directories if needed)",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "Path to the file"},
                "content": {"type": "string", "description": "Content to write"}
            },
            "required": ["path", "content"]
        }
    },
    {
        "name": "save_memory",
        "description": "Save important information to long-term memory",
        "input_schema": {
            "type": "object",
            "properties": {
                "key": {
                    "type": "string",
                    "description": "Short label (e.g. 'user-preferences')"
                },
                "content": {
                    "type": "string",
                    "description": "The information to remember"
                }
            },
            "required": ["key", "content"]
        }
    },
    {
        "name": "memory_search",
        "description": "Search long-term memory for relevant information",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "What to search for"
                }
            },
            "required": ["query"]
        }
    },
]

# ═══════════════════════════════════════════════════════════════
# Permission Controls
# ═══════════════════════════════════════════════════════════════

SAFE_COMMANDS = {
    "ls", "cat", "head", "tail", "wc", "date", "whoami",
    "echo", "pwd", "which", "git", "python", "node", "npm"
}

DANGEROUS_PATTERNS = [
    r"\brm\b", r"\bsudo\b", r"\bchmod\b",
    r"\bcurl.*\|.*sh", r"\bdd\b"
]

def load_approvals():
    if os.path.exists(APPROVALS_FILE):
        with open(APPROVALS_FILE) as f:
            return json.load(f)
    return {"allowed": [], "denied": []}

def save_approval(command, approved):
    approvals = load_approvals()
    key = "allowed" if approved else "denied"
    if command not in approvals[key]:
        approvals[key].append(command)
    with open(APPROVALS_FILE, "w") as f:
        json.dump(approvals, f, indent=2)

def check_command_safety(command):
    base_cmd = command.strip().split()[0] if command.strip() else ""
    
    if base_cmd in SAFE_COMMANDS:
        return "safe"
    
    approvals = load_approvals()
    if command in approvals["allowed"]:
        return "approved"
    
    for pattern in DANGEROUS_PATTERNS:
        if re.search(pattern, command):
            return "needs_approval"
    
    return "needs_approval"

# ═══════════════════════════════════════════════════════════════
# Tool Execution
# ═══════════════════════════════════════════════════════════════

def execute_tool(name, tool_input):
    """Execute a tool and return the result."""
    
    if name == "bash":
        cmd = tool_input["command"]
        safety = check_command_safety(cmd)
        
        if safety == "needs_approval":
            print(f"\n  ⚠️  Command: {cmd}")
            confirm = input("  Allow? (y/n): ").strip().lower()
            if confirm != "y":
                save_approval(cmd, False)
                return "Permission denied by user."
            save_approval(cmd, True)
        
        try:
            result = subprocess.run(
                cmd, shell=True,
                capture_output=True,
                text=True,
                timeout=30
            )
            output = result.stdout + result.stderr
            return output if output else "(no output)"
        except subprocess.TimeoutExpired:
            return "Command timed out after 30 seconds"
        except Exception as e:
            return f"Error: {e}"
    
    elif name == "read_file":
        try:
            with open(tool_input["path"], "r") as f:
                return f.read()[:10000]  # Limit to 10KB
        except Exception as e:
            return f"Error: {e}"
    
    elif name == "write_file":
        try:
            dirpath = os.path.dirname(tool_input["path"])
            if dirpath:
                os.makedirs(dirpath, exist_ok=True)
            with open(tool_input["path"], "w") as f:
                f.write(tool_input["content"])
            return f"Wrote to {tool_input['path']}"
        except Exception as e:
            return f"Error: {e}"
    
    elif name == "save_memory":
        os.makedirs(MEMORY_DIR, exist_ok=True)
        filepath = os.path.join(MEMORY_DIR, f"{tool_input['key']}.md")
        with open(filepath, "w") as f:
            f.write(tool_input["content"])
        return f"Saved to memory: {tool_input['key']}"
    
    elif name == "memory_search":
        query = tool_input["query"].lower()
        results = []
        
        if os.path.exists(MEMORY_DIR):
            for filename in os.listdir(MEMORY_DIR):
                if filename.endswith(".md"):
                    filepath = os.path.join(MEMORY_DIR, filename)
                    with open(filepath, "r") as f:
                        content = f.read()
                    
                    if any(word in content.lower() for word in query.split()):
                        results.append(f"--- {filename} ---\n{content}")
        
        return "\n\n".join(results) if results else "No matching memories found."
    
    return f"Unknown tool: {name}"

# ═══════════════════════════════════════════════════════════════
# Session Management
# ═══════════════════════════════════════════════════════════════

def get_session_path(session_key):
    os.makedirs(SESSIONS_DIR, exist_ok=True)
    safe_key = session_key.replace(":", "_").replace("/", "_")
    return os.path.join(SESSIONS_DIR, f"{safe_key}.jsonl")

def load_session(session_key):
    """Load conversation history from JSONL file."""
    path = get_session_path(session_key)
    messages = []
    
    if os.path.exists(path):
        with open(path, "r") as f:
            for line in f:
                if line.strip():
                    try:
                        messages.append(json.loads(line))
                    except json.JSONDecodeError:
                        continue
    
    return messages

def append_message(session_key, message):
    """Append a single message (crash-safe)."""
    with open(get_session_path(session_key), "a") as f:
        f.write(json.dumps(message) + "\n")

def save_session(session_key, messages):
    """Overwrite session with full message list."""
    with open(get_session_path(session_key), "w") as f:
        for msg in messages:
            f.write(json.dumps(msg) + "\n")

# ═══════════════════════════════════════════════════════════════
# Context Window Management
# ═══════════════════════════════════════════════════════════════

def estimate_tokens(messages):
    """Rough token estimate: ~4 chars per token."""
    total_chars = sum(len(json.dumps(m)) for m in messages)
    return total_chars // 4

def compact_session(session_key, messages):
    """Summarize old messages when context gets too long."""
    
    if estimate_tokens(messages) < 160_000:  # 80% of 200K
        return messages
    
    split_point = len(messages) // 2
    old_messages = messages[:split_point]
    recent_messages = messages[split_point:]
    
    print(f"\n  📦 Compacting session (too many tokens)...")
    
    summary_response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2000,
        messages=[{
            "role": "user",
            "content": (
                "Summarize this conversation concisely. Preserve:\n"
                "- Key facts about the user\n"
                "- Important decisions made\n"
                "- Open tasks or TODOs\n\n"
                f"{json.dumps(old_messages, indent=2)}"
            )
        }]
    )
    
    summary_message = {
        "role": "user",
        "content": f"[Previous conversation summary]\n{summary_response.content[0].text}"
    }
    
    compacted = [summary_message] + recent_messages
    save_session(session_key, compacted)
    
    return compacted

# ═══════════════════════════════════════════════════════════════
# Command Queue (Concurrency Control)
# ═══════════════════════════════════════════════════════════════

session_locks = defaultdict(threading.Lock)

# ═══════════════════════════════════════════════════════════════
# Agent Loop
# ═══════════════════════════════════════════════════════════════

def serialize_content(content):
    """Convert API response content blocks to JSON-serializable format."""
    serialized = []
    for block in content:
        if hasattr(block, "text"):
            serialized.append({"type": "text", "text": block.text})
        elif block.type == "tool_use":
            serialized.append({
                "type": "tool_use",
                "id": block.id,
                "name": block.name,
                "input": block.input
            })
    return serialized

def run_agent_turn(session_key, user_text, agent_config):
    """
    Run a full agent turn: load session, call LLM in a loop, save.
    Handles tool calling automatically.
    """
    with session_locks[session_key]:
        # Load and compact session
        messages = load_session(session_key)
        messages = compact_session(session_key, messages)
        
        # Add user message
        user_msg = {"role": "user", "content": user_text}
        messages.append(user_msg)
        append_message(session_key, user_msg)
        
        # Agent loop (max 20 turns to prevent infinite loops)
        for turn in range(20):
            response = client.messages.create(
                model=agent_config["model"],
                max_tokens=4096,
                system=agent_config["soul"],
                tools=TOOLS,
                messages=messages
            )
            
            content = serialize_content(response.content)
            assistant_msg = {"role": "assistant", "content": content}
            messages.append(assistant_msg)
            append_message(session_key, assistant_msg)
            
            # If done, return text
            if response.stop_reason == "end_turn":
                return "".join(
                    b.text for b in response.content 
                    if hasattr(b, "text")
                )
            
            # Process tool calls
            if response.stop_reason == "tool_use":
                tool_results = []
                
                for block in response.content:
                    if block.type == "tool_use":
                        print(f"  🔧 {block.name}: {json.dumps(block.input)[:100]}")
                        result = execute_tool(block.name, block.input)
                        display = str(result)[:150]
                        print(f"     → {display}")
                        
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": str(result)
                        })
                
                # Add tool results
                results_msg = {"role": "user", "content": tool_results}
                messages.append(results_msg)
                append_message(session_key, results_msg)
                
                # Loop continues - call Claude again
        
        return "(max turns reached)"

# ═══════════════════════════════════════════════════════════════
# Multi-Agent Routing
# ═══════════════════════════════════════════════════════════════

def resolve_agent(message_text):
    """Route messages to the right agent based on prefix commands."""
    if message_text.startswith("/research "):
        return "researcher", message_text[len("/research "):]
    
    return "main", message_text

# ═══════════════════════════════════════════════════════════════
# Scheduled Tasks (Cron)
# ═══════════════════════════════════════════════════════════════

def setup_heartbeats():
    """Configure recurring agent tasks."""
    
    def morning_check():
        print("\n⏰ Heartbeat: morning check")
        result = run_agent_turn(
            "cron:morning-check",
            "Good morning! Check today's date and give me a motivational quote.",
            AGENTS["main"]
        )
        print(f"🤖 {result}\n")
    
    # Schedule daily at 7:30 AM
    schedule.every().day.at("07:30").do(morning_check)
    
    # Run scheduler loop in background
    def scheduler_loop():
        while True:
            schedule.run_pending()
            time.sleep(60)
    
    threading.Thread(target=scheduler_loop, daemon=True).start()

# ═══════════════════════════════════════════════════════════════
# Telegram Channel
# ═══════════════════════════════════════════════════════════════

async def handle_telegram_message(update: Update, context):
    """Handle incoming Telegram messages."""
    user_id = str(update.effective_user.id)
    
    # Route to appropriate agent
    agent_id, message_text = resolve_agent(update.message.text)
    agent = AGENTS[agent_id]
    
    # Use agent-specific session key
    session_key = f"{agent['session_prefix']}:telegram:{user_id}"
    
    # Run agent turn
    response_text = run_agent_turn(session_key, message_text, agent)
    
    # Send response
    await update.message.reply_text(f"[{agent['name']}] {response_text}")

# ═══════════════════════════════════════════════════════════════
# HTTP Channel (Gateway Pattern Demo)
# ═══════════════════════════════════════════════════════════════

flask_app = Flask(__name__)

@flask_app.route("/chat", methods=["POST"])
def chat():
    """HTTP endpoint for chatting with the agent."""
    data = request.json
    user_id = data["user_id"]
    message_text = data["message"]
    
    # Route to appropriate agent
    agent_id, message_text = resolve_agent(message_text)
    agent = AGENTS[agent_id]
    
    # Use agent-specific session key
    session_key = f"{agent['session_prefix']}:http:{user_id}"
    
    # Run agent turn
    response_text = run_agent_turn(session_key, message_text, agent)
    
    return jsonify({
        "agent": agent["name"],
        "response": response_text
    })

# ═══════════════════════════════════════════════════════════════
# Main Entry Point
# ═══════════════════════════════════════════════════════════════

def main():
    """Start all channels and services."""
    
    # Create workspace directories
    for directory in [WORKSPACE, SESSIONS_DIR, MEMORY_DIR]:
        os.makedirs(directory, exist_ok=True)
    
    # Start heartbeats
    setup_heartbeats()
    
    # Start HTTP server in background
    threading.Thread(
        target=lambda: flask_app.run(port=5000, debug=False),
        daemon=True
    ).start()
    
    print("Mini OpenClaw")
    print(f"  Agents: {', '.join(a['name'] for a in AGENTS.values())}")
    print(f"  Workspace: {WORKSPACE}")
    print(f"  HTTP API: http://localhost:5000/chat")
    print(f"  Telegram: Starting...\n")
    
    # Start Telegram bot
    telegram_token = os.getenv("TELEGRAM_BOT_TOKEN")
    if telegram_token:
        app = Application.builder().token(telegram_token).build()
        app.add_handler(MessageHandler(filters.TEXT, handle_telegram_message))
        app.run_polling()
    else:
        print("  Warning: TELEGRAM_BOT_TOKEN not set. HTTP-only mode.")
        print("  Try: curl -X POST http://localhost:5000/chat \\")
        print("         -H 'Content-Type: application/json' \\")
        print("         -d '{\"user_id\": \"test\", \"message\": \"Hello!\"}'")
        
        # Keep running for HTTP
        while True:
            time.sleep(1)

if __name__ == "__main__":
    main()

Running the Complete Implementation

# Install dependencies
pip install anthropic python-telegram-bot flask schedule

# Set API keys
export ANTHROPIC_API_KEY="your-key"
export TELEGRAM_BOT_TOKEN="your-token"  # Optional

# Run
python mini-openclaw.py

Try It Out

Via Telegram:

You: Remember that I prefer sushi for dinner

[Jarvis] [uses save_memory]
Got it - saved your preference.

You: /research What are the best sushi restaurants in SF?

[Scout] [uses web_search, save_memory]
Here's what I found...

You: What did Scout find?

[Jarvis] [uses memory_search]
Scout's research found...

Via HTTP:

curl -X POST http://localhost:5000/chat \
  -H "Content-Type: application/json" \
  -d '{"user_id": "test", "message": "My name is Mehdi"}'

# Later...
curl -X POST http://localhost:5000/chat \
  -H "Content-Type: application/json" \
  -d '{"user_id": "test", "message": "What is my name?"}'

# Response:
{"agent": "Jarvis", "response": "Your name is Mehdi!"}

What We've Built

Starting from a simple chatbot, we built every major component of OpenClaw:

Persistent Sessions - JSONL files for crash-safe conversation memory
SOUL.md - Personality files that transform generic AI into specific agents
Tool Calling - Structured tool definitions + agent loop for execution
Security - Permission controls with allowlists and approval flow
Gateway Pattern - One agent, multiple interfaces (Telegram, HTTP, etc.)
Context Compaction - Automatic summarization when conversations grow too long
Cross-Session Memory - File-based storage that survives session resets
Command Queue - Per-session locking to prevent race conditions
Scheduled Tasks - Cron-like heartbeats for recurring agent runs
Multi-Agent Routing - Multiple agent configurations with message routing

Each emerged from a practical problem:

"The AI can't remember" → Sessions
"It's too generic" → SOUL.md
"It can only talk" → Tools
"It's dangerous" → Permissions
"I want it everywhere" → Gateway
"Conversations too long" → Compaction
"It forgets between sessions" → Memory
"Race conditions" → Queue
"I want automation" → Cron
"One agent can't do everything" → Multi-agent

Key Architectural Insights

1. Separation of Concerns

State (Gateway)  ≠  Compute (Agent)  ≠  Interface (Channels)
     ↓                    ↓                     ↓
  Sessions          Tool Execution        Telegram/HTTP
  Memory            LLM Calls             Discord/Slack
  Config            Agent Loop            WebSocket

This separation enables:

Multiple channels sharing one agent
Agent restarts without losing state
Easy testing (mock channels, real agent)

2. Append-Only Data Structures

JSONL files:  Crash-safe, human-readable, streamable
Approvals:    Additive security (never remove, only add)
Memory files: Each file is independent, no corruption risk

3. Pull-Based Retrieval

Don't push all context to the agent. Let the agent pull what it needs:

# Bad: Push everything
system_prompt = load_all_memory()  # 100KB of text

# Good: Pull on demand
tools = [memory_search_tool]  # Agent decides what to search

4. Idempotent Operations

Tools should be safe to retry:

# Idempotent
write_file(path, content)  # Same result if called twice

# Not idempotent
append_to_file(path, content)  # Different result each time

5. Explicit State Transitions

Make state changes visible:

# Explicit
messages = compact_session(session_key, messages)
save_session(session_key, messages)

# Implicit (harder to debug)
auto_compact_and_save(session_key)  # What happened?

Comparison: Our Implementation vs OpenClaw

Feature	Our Implementation	OpenClaw Production
Storage	JSONL files	JSONL + SQLite
Memory	Keyword search	Vector search + FTS5
Channels	Telegram + HTTP	10+ platforms + WebSocket
Compaction	Simple half-split	Chunk-based with safety margin
Routing	Prefix commands	Priority-based rule engine
Tools	5 basic tools	30+ tools + plugin system
Concurrency	Per-session locks	Lane-based queues
Deployment	Single process	Multi-process + systemd

Our implementation has all the core concepts. OpenClaw adds production hardening.

Going Further

If You Want to Build Your Own

Start with one channel - Get Telegram or Discord working first
Add tools incrementally - Start with read/write files, then bash
Add memory when you need it - Once sessions reset, add persistence
Add channels when you outgrow one - Gateway pattern emerges naturally
Add agents when tasks specialize - Don't start with 10 agents

If You Want to Use OpenClaw

Install it:

npm install -g openclaw@latest
openclaw onboard --install-daemon

It handles all the edge cases we glossed over:

OAuth for channels
Browser automation with CDP
Sub-agent spawning
Vector memory search
Production logging
Error recovery
Rate limiting
Multi-user deployment

Conclusion

OpenClaw isn't magic. It's a thoughtful application of simple patterns:

JSONL for sessions - Append-only, crash-safe
System prompts for personality - SOUL.md
Function calling for capabilities - Tool definitions + execution
Files for memory - Simple, durable, inspectable
WebSocket for coordination - Real-time, bidirectional
Locks for concurrency - Per-session serialization

Each piece solves a specific problem. Together, they create a system that feels alive - an AI that:

Lives where you communicate
Remembers who you are
Can actually do things
Runs 24/7 on your own infrastructure

You could've invented OpenClaw. Now you know how.

Author: Tutorial created from first principles analysis of OpenClaw
License: MIT (OpenClaw is MIT licensed)
Source: https://github.com/openclaw/openclaw

mehdimashayekhi/OpenClaw-Fundamentals.md

Building OpenClaw From First Principles

Table of Contents

The Problem Space

Layer 0: The Simplest Chatbot

Layer 1: Persistent Sessions

The JSONL Format

Implementation

Key Insight: Storage Location

Layer 2: Adding Personality (SOUL)

What is SOUL.md?

Implementation

In OpenClaw

Layer 3: Tool Calling

The Tool Calling Mechanism

Tool Definition

Tool Execution

The Agent Loop

Updated Message Handler

Try It

OpenClaw's Tool Catalog

Layer 4: Security & Permissions

Approach: Tiered Safety

Implementation

Updated Tool Execution

OpenClaw's Permission Model

Layer 5: The Gateway Pattern

Key Insight

Proof: Add a Second Channel

OpenClaw's Gateway

Gateway Protocol Example

Layer 6: Context Window Management

The Solution: Compaction

Add Compaction Check

Testing Compaction

OpenClaw's Approach

Layer 7: Cross-Session Memory

The Approach: Memory as Files

Memory Tools

Memory Execution

Update SOUL to Mention Memory

Try It

OpenClaw's Production Memory

Layer 8: Concurrency & Command Queue

The Fix: Per-Session Locks

Wrap Message Handling

Do the Same for HTTP

OpenClaw's Queue System

Layer 9: Scheduled Tasks (Cron)

Using the schedule Library

Try It

OpenClaw's Cron

Layer 10: Multi-Agent Routing

The Solution: Multiple Agents with Routing

Implementation

Updated Message Handler

Try It

OpenClaw's Routing

Production Patterns

1. WebSocket Protocol

2. Streaming Responses

3. Browser Control with Semantic Snapshots

4. Session Scoping

5. Skills System

6. Sub-Agent Spawning

7. Rate Limiting

Complete Implementation

Running the Complete Implementation

Try It Out

What We've Built

Key Architectural Insights

1. Separation of Concerns

2. Append-Only Data Structures

3. Pull-Based Retrieval

4. Idempotent Operations

5. Explicit State Transitions

Comparison: Our Implementation vs OpenClaw

Going Further

If You Want to Build Your Own

If You Want to Use OpenClaw