打造迷你版 OpenClaw：一步步教你实现

# 建项目文件夹
mkdir mini-openclaw && cd mini-openclaw

# 建技能和前端目录
mkdir -p skills/datetime skills/memory_work skills/browser_use

# 核心组件文件
touch main.py agent_runtime.py context_builder.py session_manager.py telegram_channel.py memory_store.py skill_loader.py SOUL.md .env

# 日期时间技能
touch skills/datetime/SKILL.md skills/datetime/handler.py

# 记忆笔记技能
touch skills/memory_work/SKILL.md skills/memory_work/handler.py

# 浏览器技能
touch skills/browser_use/SKILL.md skills/browser_use/handler.py

# 创建并激活虚拟环境
uv venv
source .venv/bin/activate

# 装包
uv pip install httpx python-dotenv python-telegram-bot playwright

# 给 Playwright 装浏览器内核
playwright install chromium

# ./memory_store.py

import json
import os

class Memory:
    def __init__(self, path="MEMORY.json"):
        self.path = path
        if os.path.exists(path):
            with open(path) as f:
                self._data = json.load(f)
        else:
            self._data = {}

    def set(self, key, value):
        self._data[key] = value
        self._save()

    def get(self, key):
        return self._data.get(key)

    def keys(self):
        return list(self._data.keys())

    def _save(self):
        with open(self.path, "w") as f:
            json.dump(self._data, f, indent=2, default=str)

# ./session_manager.py

import json
import os
import time

class SessionManager:
    def __init__(self, path="SESSIONS.json"):
        self.path = path
        if os.path.exists(path):
            with open(path) as f:
                self.sessions = json.load(f)
            print("✅ 从磁盘恢复了历史会话！")
        else:
            self.sessions = {}

    def get_or_create_session(self, client_id, channel):
        session_id = f"{channel}:{client_id}"
        if session_id notin self.sessions:
            self.sessions[session_id] = {
                "client_id": client_id,
                "channel": channel,
                "created_at": time.time(),
                "history": [],
            }
        return session_id

    def add_message(self, session_id, message):
        session = self.sessions.get(session_id)
        if session:
            session["history"].append(message)
            self._save()

    def get_history(self, session_id):
        session = self.sessions.get(session_id)
        return session["history"] if session else []

    def _save(self):
        with open(self.path, "w") as f:
            json.dump(self.sessions, f, indent=2, default=str)

{
  "telegram:1191237804": {
    "client_id": "1191237804",
    "channel": "telegram",
    "created_at": 1774270494.7631621,
    "history": [
      {"role": "user", "content": "Hi!", "timestamp": 1774270494.7631638},
      {"role": "assistant", "content": "Hello! I'm your personal AI assistant...", "timestamp": 1774270497.646961},
      {"role": "user", "content": "Remember my name is ashish", "timestamp": 1774270609.259111},
      {"role": "assistant", "content": "Perfect! I've saved your name...", "timestamp": 1774270611.7315981}
    ]
  }
}

# skills/datetime/SKILL.md
---
name: datetime
description: 获取当前日期和时间。
---

# skills/datetime/handler.py
from datetime import datetime, timezone

tools = [
    {
        "name": "get_current_datetime",
        "description": "获取当前日期和时间。",
        "parameters": {
            "type": "object",
            "properties": {},
            "required": [],
        },
    }
]

asyncdef execute(tool_name, tool_input, context):
    if tool_name == "get_current_datetime":
        now = datetime.now(timezone.utc)
        return {
            "readable": now.strftime("%A, %B %d, %Y %I:%M:%S %p UTC"),
        }
    return {"error": f"未知工具：{tool_name}"}

# skills/memory_work/SKILL.md
---
name: memory_work
description: 把用户的个人信息存到记忆里。
---

# skills/memory_work/handler.py
tools = [
    {
        "name": "save_note",
        "description": "把关于用户的笔记存到记忆里。",
        "parameters": {
            "type": "object",
            "properties": {
                "key": {"type": "string", "description": "简短的键名"},
                "content": {"type": "string", "description": "笔记内容"},
            },
            "required": ["key", "content"],
        },
    }
]

asyncdef execute(tool_name, tool_input, context):
    memory = context["memory"]
    if tool_name == "save_note":
        memory.set(f"note:{tool_input['key']}", {
            "content": tool_input["content"],
        })
        return {"success": True, "key": tool_input["key"]}
    return {"error": f"未知工具：{tool_name}"}

# skills/browser_use/SKILL.md
---
name: browser_use
description: 网页浏览、提取文本、点击元素、填写并提交表单。用户让你访问网站、读页面、操作网页时使用。
---

# skills/browser_use/handler.py
from playwright.async_api import async_playwright

_browser = None
_page = None

asyncdef _get_page():
    global _browser, _page
    if _browser and _page:
        return _page
    pw = await async_playwright().start()
    _browser = await pw.chromium.launch(headless=True)
    _page = await _browser.new_page()
    return _page

tools = [
    {
        "name": "browse_url",
        "description": "访问网址并返回标题和文本内容",
        "parameters": {
            "type": "object",
            "properties": {"url": {"type": "string", "description": "要访问的URL"}},
            "required": ["url"],
        },
    },
    {
        "name": "click_element",
        "description": "通过选择器或文本点击页面元素",
        "parameters": {
            "type": "object",
            "properties": {"selector": {"type": "string", "description": "CSS 选择器或文本"}},
            "required": ["selector"],
        },
    },
    {
        "name": "fill_input",
        "description": "在输入框输入内容",
        "parameters": {
            "type": "object",
            "properties": {
                "selector": {"type": "string"},
                "text": {"type": "string"}
            },
            "required": ["selector", "text"],
        },
    },
    {
        "name": "get_page_content",
        "description": "获取当前页面文本",
        "parameters": {
            "type": "object",
            "properties": {"selector": {"type": "string", "description": "可选选择器"}},
            "required": [],
        },
    }
]

asyncdef execute(tool_name, tool_input, context):
    try:
        page = await _get_page()
        if tool_name == "browse_url":
            url = tool_input["url"]
            ifnot url.startswith("http"):
                url = "https://" + url
            await page.goto(url, wait_until="domcontentloaded", timeout=10000)
            title = await page.title()
            text = await page.inner_text("body")
            return {
                "title": title,
                "url": page.url,
                "content_preview": text.strip()[:3000],
            }
        elif tool_name == "click_element":
            await page.click(tool_input["selector"], timeout=3000)
            await page.wait_for_load_state("domcontentloaded")
            return {
                "clicked": tool_input["selector"],
                "new_url": page.url,
                "new_title": await page.title(),
            }
        elif tool_name == "fill_input":
            await page.fill(tool_input["selector"], tool_input["text"])
            return {
                "filled": tool_input["selector"],
                "text": tool_input["text"],
            }
        elif tool_name == "get_page_content":
            selector = tool_input.get("selector") or"body"
            text = await page.inner_text(selector)
            return {
                "url": page.url,
                "content": text.strip()[:5000],
            }
        return {"error": f"未知工具：{tool_name}"}
    except Exception as e:
        return {"error": str(e)}

# skill_loader.py
import os
import importlib.util

class SkillLoader:
    def __init__(self):
        self.skills = {}

    def load_from_directory(self, skills_dir):
        ifnot os.path.isdir(skills_dir):
            print("未找到技能目录")
            return
        for entry in os.listdir(skills_dir):
            skill_dir = os.path.join(skills_dir, entry)
            skill_md = os.path.join(skill_dir, "SKILL.md")
            handler_py = os.path.join(skill_dir, "handler.py")
            ifnot os.path.isdir(skill_dir):
                continue
            ifnot os.path.exists(skill_md) ornot os.path.exists(handler_py):
                continue
            try:
                with open(skill_md) as f:
                    name, desc = self._parse_skill_md(f.read())
                spec = importlib.util.spec_from_file_location(f"skill_{entry}", handler_py)
                module = importlib.util.module_from_spec(spec)
                spec.loader.exec_module(module)
                self.skills[name] = {
                    "name": name,
                    "description": desc,
                    "tools": getattr(module, "tools", []),
                    "execute": getattr(module, "execute", None),
                }
                print(f"✅ 已加载技能：{name}")
            except Exception as e:
                print(f"❌ 加载失败 {entry}：{e}")

    def get_active_skills(self):
        return [{"name": s["name"], "description": s["description"]} for s in self.skills.values()]

    def get_tools(self):
        tools = []
        for s in self.skills.values():
            tools.extend(s["tools"])
        return tools

    asyncdef execute_tool(self, tool_name, tool_input, context):
        for s in self.skills.values():
            if any(t["name"] == tool_name for t in s["tools"]):
                if s["execute"]:
                    returnawait s["execute"](tool_name, tool_input, context)
        return {"error": f"未知工具：{tool_name}"}

    def _parse_skill_md(self, content):
        name = "unknown"
        desc = ""
        for line in content.split("\n"):
            if line.startswith("name:"):
                name = line.split(":", 1)[1].strip()
            elif line.startswith("description:"):
                desc = line.split(":", 1)[1].strip()
        return name, desc

# Soul
你是 Mini-OpenClaw，一个跑在用户本地机器上的私人 AI 助手。你可以调用已安装技能提供的工具。

## 性格
- 友好、简洁，偶尔皮一下
- 语气轻松，像跟聪明朋友聊天
- 不懂就直说，别瞎编

## 规则
- 存笔记时用简短统一的键，比如 name、location、job
- 上网搜索只用 DuckDuckGo，不用 Google
- 该用工具就用工具
- 执行危险操作前必须先问用户
- 没要求详细说明时，回复控制在 300 字以内

# context_builder.py
import os
from datetime import datetime, timezone

BASE_PROMPT = """你是一个由 Mini OpenClaw 驱动的贴心私人 AI 助手。
简洁、友好、乐于助人，该用工具就用工具。"""

def load_soul():
    path = os.path.join(os.path.dirname(__file__), "SOUL.md")
    try:
        with open(path, "r") as f:
            return f.read()
    except FileNotFoundError:
        return BASE_PROMPT

def build_system_prompt(active_skills, memory=None):
    prompt = load_soul()
    if active_skills:
        prompt += "\n\n## 可用技能\n"
        for s in active_skills:
            prompt += f"### {s['name']}\n{s['description']}\n\n"
    if memory:
        prefix = "note:"
        notes = {k[len(prefix):]: memory.get(k) for k in memory.keys() if k.startswith(prefix)}
        if notes:
            prompt += "\n\n## 你知道的用户信息\n"
            for k, v in notes.items():
                content = v.get("content", v) if isinstance(v, dict) else v
                prompt += f"- {k}：{content}\n"
    prompt += f"\n当前时间：{datetime.now(timezone.utc).isoformat()}"
    return prompt

# agent_runtime.py
import json
import httpx
from context_builder import build_system_prompt

MAX_TOOL_ROUNDS = 5

class AgentRuntime:
    def __init__(self, provider, model, api_key, skills, memory):
        self.provider = provider
        self.model = model
        self.api_key = api_key
        self.skills = skills
        self.memory = memory

    asyncdef run(self, history, session_id, callbacks):
        on_token = callbacks.get("on_token")
        on_tool_use = callbacks.get("on_tool_use")

        system_prompt = build_system_prompt(self.skills.get_active_skills(), self.memory)
        messages = [{"role": m["role"], "content": m["content"]} for m in history]
        tools = self.skills.get_tools()

        response = ""
        rounds = 0

        while rounds < MAX_TOOL_ROUNDS:
            rounds += 1
            result = await self._call_anthropic(system_prompt, messages, tools)

            if result["tool_calls"]:
                messages.append({"role": "assistant", "content": result["raw_content"]})
                for tc in result["tool_calls"]:
                    if on_tool_use:
                        await on_tool_use(tc["name"], tc["input"])
                    tool_res = await self.skills.execute_tool(
                        tc["name"], tc["input"],
                        {"session_id": session_id, "memory": self.memory}
                    )
                    messages.append({
                        "role": "user",
                        "content": [{
                            "type": "tool_result",
                            "tool_use_id": tc["id"],
                            "content": json.dumps(tool_res),
                        }]
                    })
                continue

            if result["text"]:
                if on_token:
                    await on_token(result["text"])
                response = result["text"]
            break
        return response

    asyncdef _call_anthropic(self, system_prompt, messages, tools):
        body = {
            "model": self.model,
            "max_tokens": 4096,
            "system": system_prompt,
            "messages": messages,
        }
        if tools:
            body["tools"] = [{
                "name": t["name"],
                "description": t["description"],
                "input_schema": t["parameters"],
            } for t in tools]

        asyncwith httpx.AsyncClient(timeout=120) as client:
            res = await client.post(
                "https://api.anthropic.com/v1/messages",
                headers={
                    "Content-Type": "application/json",
                    "x-api-key": self.api_key,
                    "anthropic-version": "2023-06-01",
                },
                json=body
            )

        if res.status_code != 200:
            raise Exception(f"Anthropic API 报错 {res.status_code}：{res.text}")
        data = res.json()
        text_parts = []
        tool_calls = []
        for b in data["content"]:
            if b["type"] == "text":
                text_parts.append(b["text"])
            elif b["type"] == "tool_use":
                tool_calls.append({
                    "id": b["id"],
                    "name": b["name"],
                    "input": b["input"],
                })
        return {
            "text": "".join(text_parts),
            "tool_calls": tool_calls orNone,
            "raw_content": data["content"],
        }

# telegram_channel.py
import time
import asyncio
from telegram import Update
from telegram.ext import Application, MessageHandler, filters

class TelegramChannel:
    def __init__(self, token, agent, sessions):
        self.token = token
        self.agent = agent
        self.sessions = sessions

    asyncdef start(self):
        app = Application.builder().token(self.token).build()
        app.add_handler(MessageHandler(filters.TEXT, self._on_message))
        await app.initialize()
        await app.start()
        await app.updater.start_polling()
        await asyncio.Future()

    asyncdef _on_message(self, update: Update, context):
        chat_id = str(update.effective_chat.id)
        user_text = update.message.text
        ifnot user_text:
            return

        session_id = self.sessions.get_or_create_session(chat_id, "telegram")
        self.sessions.add_message(session_id, {
            "role": "user",
            "content": user_text,
            "timestamp": time.time(),
        })

        await update.effective_chat.send_action("typing")

        try:
            history = self.sessions.get_history(session_id)
            full_response = ""

            asyncdef on_token(token):
                nonlocal full_response
                full_response += token

            asyncdef on_tool_use(name, input):
                await update.effective_chat.send_action("typing")

            await self.agent.run(history, session_id, {
                "on_token": on_token,
                "on_tool_use": on_tool_use,
            })

            if full_response:
                for i in range(0, len(full_response), 4096):
                    await update.message.reply_text(full_response[i:i+4096])

            self.sessions.add_message(session_id, {
                "role": "assistant",
                "content": full_response,
                "timestamp": time.time(),
            })
        except Exception as e:
            await update.message.reply_text(f"报错啦：{e}")

# main.py
import asyncio
import os
from dotenv import load_dotenv

from memory_store import Memory
from session_manager import SessionManager
from skill_loader import SkillLoader
from agent_runtime import AgentRuntime
from telegram_channel import TelegramChannel

load_dotenv()

asyncdef main():
    print("Mini OpenClaw 启动中……")

    memory = Memory()
    sessions = SessionManager()
    skills = SkillLoader()
    skills.load_from_directory(os.path.join(os.path.dirname(__file__), "skills"))

    agent = AgentRuntime(
        provider=os.getenv("MODEL_PROVIDER"),
        model=os.getenv("MODEL_NAME"),
        api_key=os.getenv("ANTHROPIC_API_KEY"),
        skills=skills,
        memory=memory
    )

    telegram = TelegramChannel(
        token=os.getenv("TELEGRAM_BOT_TOKEN"),
        agent=agent,
        sessions=sessions
    )

    print("\nMini OpenClaw 已经在 Telegram 上跑起来啦！")
    print("\nGO CLAW!!! 🦞")

    await telegram.start()

if __name__ == "__main__":
    asyncio.run(main())

ANTHROPIC_API_KEY=<你的密钥>
MODEL_PROVIDER=anthropic
MODEL_NAME=claude-opus-4-6
TELEGRAM_BOT_TOKEN=<你的 Telegram 机器人 token>