OpenClaw: The Open-Source AI Agent Revolutionizing Task Automation

# Gateway handles multi-channel message routing
class Gateway:
    def __init__(self):
        self.channels = {
            'telegram': TelegramChannel(),
            'discord': DiscordChannel(),
            'slack': SlackChannel(),
            'webhook': WebhookChannel()
        }
    
    async def route_message(self, message: Message) -> Response:
        # Transform message to standard format
        standardized = self.channels[message.source].receive(message)
        
        # Pass to agent for processing
        result = await self.agent.process(standardized)
        
        # Route response back to source
        return await self.channels[message.source].send(result)

class Agent:
    def __init__(self, model_provider: str = "openai"):
        self.model = self._init_model(model_provider)
        self.memory = Memory()
        self.skills = SkillsRegistry()
    
    async def process(self, request: Request) -> Response:
        # Load context from memory
        context = await self.memory.load_context(request.user_id)
        
        # Analyze request and plan actions
        plan = await self.model.plan(request, context)
        
        # Execute each step in the plan
        results = []
        for step in plan.steps:
            skill = self.skills.get(step.skill_name)
            result = await skill.execute(step.params)
            results.append(result)
            
            # Update memory with progress
            await self.memory.save_step(request.user_id, step, result)
        
        return Response(results=results, summary=await self.model.summarize(results))

# Example skill definition
skills:
  - name: "browser_control"
    description: "Control browser for web automation"
    actions:
      - navigate
      - click
      - type
      - screenshot
      - extract_data
      
  - name: "file_operations"
    description: "Read, write, and manage files"
    actions:
      - read
      - write
      - append
      - delete
      - list
      - search
      
  - name: "shell_execution"
    description: "Execute shell commands"
    actions:
      - run
      - run_background
      - kill
      
  - name: "email_manager"
    description: "Manage email operations"
    actions:
      - send
      - read
      - categorize
      - archive

class Memory:
    def __init__(self, storage_path: str = "./memory"):
        self.vector_store = ChromaDB(storage_path)
        self.key_value_store = SQLite(storage_path)
    
    async def save_context(self, user_id: str, data: dict):
        # Save structured data
        self.key_value_store.set(f"user:{user_id}", data)
        
        # Embed and store for semantic search
        embedding = await self.embed(data)
        await self.vector_store.upsert(user_id, embedding)
    
    async def recall(self, user_id: str, query: str) -> list:
        # Semantic search through memory
        results = await self.vector_store.search(query, top_k=5)
        return results

# System requirements
- Node.js 20+
- Python 3.10+
- uv package manager
- For browser control: Chrome/Edge/Firefox

# 1. Clone the repository
git clone https://github.com/openshell-ai/openclaw.git
cd openclaw

# 2. Install dependencies
uv sync

# 3. Configure environment
cp .env.example .env

# 4. Edit .env with your settings

# .env configuration
# LLM Provider
LLM_PROVIDER=anthropic
ANTHROPIC_API_KEY=sk-ant-...

# Or OpenAI
# LLM_PROVIDER=openai
# OPENAI_API_KEY=sk-...

# Storage (local)
MEMORY_TYPE=local
STORAGE_PATH=./data

# Channels (enable what you need)
ENABLE_TELEGRAM=true
TELEGRAM_BOT_TOKEN=...

ENABLE_DISCORD=true
DISCORD_BOT_TOKEN=...

# Browser automation
BROWSER_TYPE=chromium
HEADLESS=false

# Security
API_SECRET=your-secret-key
ALLOWED_USERS=user1,user2

# Start the agent
uv run openclaw

# Or with specific config
uv run openclaw --config production.yaml

# Check status
curl http://localhost:8080/health

# docker-compose.yaml
version: '3.8'
services:
  openclaw:
    image: openclaw/openclaw:latest
    volumes:
      - ./data:/app/data
      - /var/run/docker.sock:/var/run/docker.sock
    environment:
      - LLM_PROVIDER=anthropic
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
    ports:
      - "8080:8080"

docker-compose up -d

# Example: Automate web form filling
async def fill_form(context: Context):
    await browser.navigate("https://example.com/form")
    
    # Wait for page load
    await browser.wait_for_selector("#name")
    
    # Fill fields
    await browser.type("#name", context.user_name)
    await browser.type("#email", context.user_email)
    await browser.type("#message", context.message)
    
    # Submit
    await browser.click("#submit")
    
    # Extract result
    result = await browser.extract(".success-message")
    return result

# Example: Generate weekly report
async def generate_weekly_report(context: Context):
    # Read data files
    sales = await files.read_json("data/sales.json")
    users = await files.read_csv("data/users.csv")
    
    # Process data
    total = sum(s['amount'] for s in sales)
    new_users = len(users[users['created'] > context.last_week])
    
    # Generate markdown report
    report = f"""
# Weekly Report - {context.week}

## Summary
- Total Sales: ${total:,.2f}
- New Users: {new_users}

## Details
{build_table(sales)}
"""
    
    # Write report
    await files.write(f"reports/week-{context.week}.md", report)
    
    # Also export to PDF
    await files.convert_markdown_to_pdf(report, "reports/week-{context.week}.pdf")
    
    return "Report generated successfully"

# Example: Deploy application
async def deploy_app(context: Context):
    # Run build
    result = await shell.run("npm run build")
    if result.exit_code != 0:
        raise Error(f"Build failed: {result.stderr}")
    
    # Run tests
    result = await shell.run("npm test")
    if result.exit_code != 0:
        raise Error(f"Tests failed: {result.stderr}")
    
    # Deploy to production
    result = await shell.run("./deploy.sh production")
    
    return f"Deployed successfully: {result.stdout}"

# scheduled_tasks.yaml
schedules:
  - name: "Morning briefing"
    cron: "0 8 * * 1-5"
    action: "send_daily_summary"
    params:
      channels: ["telegram", "email"]
      
  - name: "Weekly report"
    cron: "0 9 * * 1"
    action: "generate_weekly_report"
    
  - name: "Price monitoring"
    cron: "*/30 * * * *"
    action: "check_prices"
    params:
      targets: ["amazon/product1", "ebay/product2"]
      
  - name: "Data backup"
    cron: "0 2 * * *"
    action: "backup_data"

# Content publishing workflow
async def publish_content(topic: str, platforms: list):
    # 1. Research
    research = await browser.search(topic)
    await memory.save(f"research:{topic}", research)
    
    # 2. Generate content
    article = await llm.generate_article(topic, research)
    
    # 3. Format for each platform
    for platform in platforms:
        formatted = format_for_platform(article, platform)
        
        if platform == "twitter":
            await twitter.post(formatted)
        elif platform == "linkedin":
            await linkedin.post(formatted)
        elif platform == "blog":
            await blog.publish(formatted)
    
    return f"Published to {len(platforms)} platforms"

# Auto-respond to support tickets
async def handle_support_ticket(ticket_id: str):
    # 1. Fetch ticket details
    ticket = await api.get_ticket(ticket_id)
    
    # 2. Analyze and categorize
    category = await llm.classify(ticket.message)
    
    # 3. Check knowledge base
    solutions = await search_knowledge_base(category, ticket.message)
    
    if solutions:
        # 4. Send response
        response = await llm.generate_response(ticket, solutions)
        await api.reply_ticket(ticket_id, response)
    else:
        # Escalate to human
        await api.escalate(ticket_id)
        await notify("#support", f"New escalated ticket: {ticket_id}")
    
    return "Ticket processed"

# ETL pipeline
async def run_etl_pipeline():
    # Extract
    raw_data = []
    for source in ["api", "database", "files"]:
        data = await extract_from(source)
        raw_data.extend(data)
    
    # Transform
    cleaned = []
    for record in raw_data:
        transformed = transform_record(record)
        if validate_record(transformed):
            cleaned.append(transformed)
    
    # Load
    await load_to_database(cleaned)
    await load_to_analytics(cleaned)
    
    # Notify
    await send_notification(f"Pipeline complete: {len(cleaned)} records")
    
    return f"Processed {len(cleaned)} records"

# llm.yaml
providers:
  anthropic:
    enabled: true
    model: "claude-sonnet-4-20250514"
    temperature: 0.7
    max_tokens: 4096
    
  openai:
    enabled: false
    model: "gpt-4o"
    temperature: 0.7
    
  local:
    enabled: false
    provider: "ollama"
    model: "qwen2.5:14b"
    base_url: "http://localhost:11434"

# skills.yaml
skills:
  browser:
    enabled: true
    default_timeout: 30000
    headless: false
    
  file_operations:
    enabled: true
    allowed_paths:
      - "./workspace"
      - "./data"
    max_file_size: 100MB
    
  shell:
    enabled: true
    allowed_commands:
      - "git *"
      - "npm *"
      - "python *"
    blocked_commands:
      - "rm -rf /"
      - "dd *"
      
  http:
    enabled: true
    timeout: 30000
    allowed_domains:
      - "*.example.com"
      - "api.github.com"

# security.yaml
authentication:
  type: "api_key"
  keys:
    - "sk-openclaw-user1-xxxx"
    - "sk-openclaw-user2-xxxx"
    
authorization:
  user_permissions:
    user1:
      skills: ["*"]
      channels: ["telegram", "discord"]
    user2:
      skills: ["browser", "file_operations"]
      channels: ["telegram"]
      
rate_limiting:
  enabled: true
  requests_per_minute: 10
  burst: 20

# Good: Start simple, expand gradually
async def my_first_skill(context: Context):
    # Simple file organization
    files = await list_files("./downloads")
    for file in files:
        category = categorize(file)
        await move_file(file, f"./organized/{category}/")
    return "Files organized"

# Bad: Too complex for initial testing
async def complex_workflow(context: Context):
    # Too many dependencies
    data = await fetch_api_1()  # Might fail
    processed = await process_1(data)  # Depends on above
    await save_to_db(processed)  # Depends on above
    await send_notifications()  # Depends on above
    # If any step fails, entire workflow fails

# Good: Robust error handling
async def robust_workflow(context: Context):
    try:
        # Try primary method
        result = await primary_method()
    except PrimaryError:
        # Fallback to alternative
        try:
            result = await fallback_method()
        except FallbackError:
            # Notify and log
            await notify_admin("Workflow failed")
            raise
        
    # Save checkpoint
    await save_checkpoint(context.workflow_id, result)
    
    return result

# Good: Respect API limits
class RateLimitedSkill:
    def __init__(self):
        self.limiter = TokenBucket(rate=10, capacity=10)
    
    async def execute(self, params):
        # Wait for rate limit
        await self.limiter.acquire()
        
        # Execute with retry
        for attempt in range(3):
            try:
                return await self._do_work(params)
            except RateLimitError:
                await asyncio.sleep(2 ** attempt)  # Exponential backoff
        raise MaxRetriesExceeded()

# Development mode with hot reload
uv run openclaw --dev

# Debug specific channel
uv run openclaw --debug telegram

# Production with systemd
sudo cp openclaw.service /etc/systemd/system/
sudo systemctl enable openclaw
sudo systemctl start openclaw

# openclaw.service
[Unit]
Description=OpenClaw AI Agent
After=network.target

[Service]
Type=simple
User=openclaw
WorkingDirectory=/opt/openclaw
ExecStart=/usr/bin/uv run openclaw
Restart=always
Environment=ANTHROPIC_API_KEY=xxx

[Install]
WantedBy=multi-user.target

# Railway/Render render.yaml
services:
  - name: openclaw
    buildCommand: uv sync
    startCommand: uv run openclaw
    envVars:
      - key: LLM_PROVIDER
        value: anthropic
      - key: ANTHROPIC_API_KEY
        sync: false

# Install browser dependencies
npx playwright install chromium

# Or use system browser
BROWSER_TYPE=firefox

# Implement exponential backoff
async def with_retry(func, max_retries=3):
    for i in range(max_retries):
        try:
            return await func()
        except RateLimitError:
            await asyncio.sleep(2 ** i)
    raise

# Configure memory limits
memory:
  max_conversation_length: 1000
  vector_store_cleanup:
    enabled: true
    older_than_days: 30