OpenClaw/Moltbot 调研报告 & MBE-Gateway 设计方案

文档版本: v1.0
创建日期: 2026-02-03
适用范围: MBE生态扩展、多渠道AI助手开发

📋 目录

1. OpenClaw/Moltbot 概述
2. 核心特性与技术架构
3. MBE vs OpenClaw 能力对比
4. MBE-Gateway 设计方案
5. 技术实现方案
6. 快速开始：MVP实现
7. 开发路线图
8. 附录

1. OpenClaw/Moltbot 概述

1.1 项目背景

OpenClaw（曾用名：Clawdbot → Moltbot → OpenClaw）是由奥地利开发者 Peter Steinberger 创建的开源自托管个人AI助手项目。

1.2 爆火程度

1.3 核心理念

"任何OS + WhatsApp/Telegram/Discord/iMessage网关的AI代理"

设计原则：

• 本地优先：数据完全归用户所有
• 数据主权：你的数据，你的硬件，你的控制
• 开源透明：MIT协议，完全开源
• 高度可定制：支持自定义工作流和技能

2. 核心特性与技术架构

2.1 核心特性

🎯 本地优先架构

• 在用户自己的硬件上运行（Mac Mini、VPS、树莓派、旧笔记本等）
• 不依赖云端，数据完全归用户所有
• 确保数据隐私和完全控制权

📱 多平台深度集成

支持 50+ 通讯平台：

🧠 持久化记忆系统

• 跨会话保留对话历史（天、周、月）
• 自动将对话沉淀为Markdown知识库
• 记住用户偏好、任务历史和上下文信息

🔄 自主工作流引擎 (Lobster)

• 执行复杂的多步骤任务
• 主动发送提醒和定时任务
• 晨间简报、监控告警等功能
• 强类型工作流编排

🤖 多模型支持

• AI模型: Claude、GPT系列、Gemini、Ollama（本地）
• 多模型冗余: 热切换与失败回退机制
• 唯一代理: Pi Agent (RPC模式)

🛠️ 强大的技能系统

• 565+ 内置技能
• 50+ 平台集成
• 系统操作: Shell命令、文件管理、浏览器控制
• 邮件日历: 读写邮件与日历

2.2 技术架构

┌─────────────────────────────────────────────────────────────┐
│                    OpenClaw 架构                             │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  消息平台 (WhatsApp/Telegram/Discord/iMessage等)            │
│                           ↓                                 │
│  ┌──────────────────────────────────────────────────────┐  │
│  │         Gateway 服务 (核心枢纽)                       │  │
│  │  - WebSocket控制平面 (ws://127.0.0.1:18789)          │  │
│  │  - HTTP Canvas服务 (port 18793)                      │  │
│  │  - 会话管理                                           │  │
│  │  - 协议转换                                           │  │
│  └───────────────────────┬──────────────────────────────┘  │
│                          │                                  │
│                          ↓                                  │
│  ┌──────────────────────────────────────────────────────┐  │
│  │         Pi Agent (RPC模式)                            │  │
│  │  - 代码执行                                           │  │
│  │  - 工具调用                                           │  │
│  │  - 上下文管理                                         │  │
│  └──────────────────────────────────────────────────────┘  │
│                                                             │
│  其他组件:                                                  │
│  - CLI工具 (openclaw命令)                                  │
│  - macOS/iOS/Android应用                                   │
│  - Web控制面板                                             │
│                                                             │
└─────────────────────────────────────────────────────────────┘

2.3 系统要求

2.4 部署方式

• 命令行安装: npm install -g openclaw@latest
• Docker部署: 完整容器化方案
• 云平台一键部署: Railway、Render、Northflank、Hetzner VPS等

3. MBE vs OpenClaw 能力对比

3.1 能力对比矩阵

评分说明:

• ⭐ = 无需开发
• ⭐⭐ = 简单
• ⭐⭐⭐ = 中等
• ⭐⭐⭐⭐ = 困难
• ⭐⭐⭐⭐⭐ = 很困难

3.2 MBE的核心优势

1️⃣ 更强的AI引擎

OpenClaw架构:
Pi Agent (单一代理) → 处理所有任务

MBE架构:
MOE (43专家)
  ↓
  ├─ 专家市场层 (17个专家)
  ├─ TITANS 记忆系统
  ├─ MIRAS 多尺度检索
  └─ HOPE 在线学习

优势对比：

2️⃣ 已有MCP支持

OpenClaw 不支持 MCP协议，而MBE已实现 完整的MCP Server：

MBE MCP工具：

1. mbe_surprise - 惊讶度检测
2. mbe_route - 智能路由
3. mbe_memory_write - 记忆写入
4. mbe_memory_read - 记忆读取
5. mbe_analyze_learning - 学习分析

集成能力：

• ✅ Cursor IDE
• ✅ Claude Desktop
• ✅ 所有MCP兼容工具

3️⃣ 完整的商业基础设施

# MBE已实现的企业级功能
✅ 用户系统 (6种角色: super_admin/admin/operator/developer/enterprise/user)
✅ 订阅管理 (4种套餐: free/personal/pro/enterprise)
✅ API密钥管理 (api_clients + api_keys表)
✅ Token计费系统 (token_usage_logs + usage_records)
✅ 支付集成 (支付宝/微信支付)
✅ 管理后台 (Admin UI - Next.js)
✅ 多租户支持 (企业子账号)
✅ SLA监控 (sla_incidents表)
✅ 成本分析 (llm_cost_records + 11个价格爬虫)

OpenClaw作为开源项目，没有任何商业功能。

4️⃣ 中国市场优势

3.3 OpenClaw的核心优势

1️⃣ 成熟的多渠道集成

OpenClaw的核心价值就是Gateway，已实现：

• ✅ WhatsApp (Baileys协议)
• ✅ Telegram (Bot API)
• ✅ Discord (Bot API)
• ✅ Slack (Bot API)
• ✅ iMessage (macOS)
• ✅ Mattermost (插件)
• ✅ 50+ 其他平台

这是MBE最大的缺口。

2️⃣ Lobster工作流引擎

OpenClaw的Lobster引擎支持：

• ✅ 强类型工作流编排
• ✅ 定时任务和条件触发
• ✅ 多步任务自动化
• ✅ 主动提醒和简报

MBE需要自行开发类似功能。

3️⃣ 活跃的开源社区

• GitHub Stars: 144,000+
• Fork: 21,600+
• 社区: 活跃的开发者生态
• 文档: 完善的官方文档

4. MBE-Gateway 设计方案

4.1 核心理念

完全隔离架构：

独立的Gateway项目 → 通过SDK/MCP调用MBE → 零侵入MBE核心

4.2 推荐架构

┌─────────────────────────────────────────────────────────────────┐
│                     MBE-Gateway 项目                             │
│                  (完全独立的代码库)                               │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │          多渠道接入层 (Channel Adapters)                  │  │
│  │  ┌──────┐  ┌──────┐  ┌──────┐  ┌──────┐  ┌──────┐      │  │
│  │  │WhatsApp│ │Telegram│ │Discord│ │ Slack │ │ 微信 │      │  │
│  │  └───┬──┘  └───┬──┘  └───┬──┘  └───┬──┘  └───┬──┘      │  │
│  └──────┼─────────┼─────────┼─────────┼─────────┼──────────┘  │
│         └─────────┴─────────┴─────────┴─────────┘              │
│                          │                                      │
│  ┌───────────────────────▼──────────────────────────────────┐  │
│  │         Gateway Core (WebSocket + Session)               │  │
│  │  ┌──────────┐  ┌──────────┐  ┌──────────┐              │  │
│  │  │会话管理   │  │消息队列   │  │协议转换   │              │  │
│  │  └──────────┘  └──────────┘  └──────────┘              │  │
│  └───────────────────────┬──────────────────────────────────┘  │
│                          │                                      │
│  ┌───────────────────────▼──────────────────────────────────┐  │
│  │        MBE Client Layer (SDK/MCP)                        │  │
│  │  ┌──────────────────┐  ┌──────────────────┐             │  │
│  │  │   MBE Python SDK │  │   MBE MCP Client │             │  │
│  │  │   (HTTP/REST)    │  │   (stdio/TCP)    │             │  │
│  │  └────────┬─────────┘  └────────┬─────────┘             │  │
│  └───────────┼──────────────────────┼───────────────────────┘  │
│              │                      │                          │
└──────────────┼──────────────────────┼──────────────────────────┘
               │                      │
               ▼                      ▼
┌──────────────────────────────────────────────────────────────────┐
│                    MBE 核心引擎 (不变)                            │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐            │
│  │  MOE    │  │ TITANS  │  │  MIRAS  │  │  HOPE   │            │
│  │ 43专家  │  │  记忆   │  │ 多尺度  │  │ 在线学习 │            │
│  └─────────┘  └─────────┘  └─────────┘  └─────────┘            │
└──────────────────────────────────────────────────────────────────┘

4.3 两种接入方案对比

方案1: 使用 MBE Python SDK (推荐) ⭐⭐⭐⭐⭐

优点：

• ✅ 成熟的HTTP REST接口
• ✅ 完整的类型提示和错误处理
• ✅ 支持异步和同步调用
• ✅ 可以远程部署（Gateway和MBE分离部署）
• ✅ 更好的负载均衡和扩展性

缺点：

• ⚠️ 需要网络调用开销 (20-50ms延迟)

适用场景：生产环境、分布式部署

方案2: 使用 MBE MCP (轻量级) ⭐⭐⭐⭐

优点：

• ✅ 标准MCP协议
• ✅ 更低的延迟（本地通信 5-10ms）
• ✅ 与AI工具生态兼容（Cursor、Claude Desktop）
• ✅ 进程间通信更高效

缺点：

• ⚠️ 需要Gateway和MBE在同一机器
• ⚠️ 扩展性受限

适用场景：单机部署、低延迟要求

4.4 性能对比

4.5 核心优势

1️⃣ 完全隔离 ✅

MBE核心：零修改，持续独立迭代
Gateway：独立代码库，独立部署

• MBE核心引擎无需任何修改
• Gateway可以独立开发、测试、部署
• 互不影响，降低耦合

2️⃣ 灵活扩展 ✅

# 添加新渠道只需实现适配器
class DiscordAdapter(BaseAdapter):
    async def send_message(self, channel_id, content):
        # 调用MBE SDK
        result = await self.mbe.llm.generate(content)
        # 发送到Discord
        await self.discord_client.send(channel_id, result.content)

3️⃣ 独立部署 ✅

# Gateway可以部署在不同机器
Machine A: MBE Core (GPU服务器) - 部署在高性能服务器
Machine B: MBE Gateway (轻量级服务器) - 部署在边缘节点
Machine C: MBE Gateway (另一个地区) - 全球多地部署

4️⃣ 多租户支持 ✅

# 一个MBE Core可以服务多个Gateway
Gateway A (Telegram) → MBE Core
Gateway B (WhatsApp) → MBE Core  
Gateway C (Discord)  → MBE Core
Gateway D (企业专用) → MBE Core

5. 技术实现方案

5.1 项目结构

mbe-gateway/                    # Gateway项目根目录
├── gateway/                    # Gateway核心
│   ├── __init__.py
│   ├── server.py              # WebSocket服务器
│   ├── session.py             # 会话管理
│   ├── protocol.py            # 消息协议
│   └── config.py              # 配置管理
│
├── channels/                   # 渠道适配器
│   ├── __init__.py
│   ├── base.py                # 基础适配器接口
│   ├── telegram/              # Telegram适配器
│   │   ├── __init__.py
│   │   ├── bot.py             # Bot主程序
│   │   └── handlers.py        # 消息处理器
│   ├── whatsapp/              # WhatsApp适配器
│   │   ├── __init__.py
│   │   └── client.py          # WhatsApp客户端
│   ├── discord/               # Discord适配器
│   │   ├── __init__.py
│   │   └── bot.py             # Discord Bot
│   └── wechat/                # 微信适配器
│       ├── __init__.py
│       └── handler.py         # 微信消息处理
│
├── mbe/                        # MBE客户端层
│   ├── __init__.py
│   ├── client.py              # MBE SDK封装
│   ├── mcp_client.py          # MBE MCP封装
│   └── models.py              # 数据模型
│
├── storage/                    # 本地存储
│   ├── __init__.py
│   ├── redis_store.py         # Redis会话存储
│   └── sqlite_store.py        # SQLite消息历史
│
├── utils/                      # 工具类
│   ├── __init__.py
│   ├── logger.py              # 日志工具
│   └── metrics.py             # 指标监控
│
├── tests/                      # 测试
│   ├── __init__.py
│   ├── test_gateway.py
│   ├── test_telegram.py
│   └── test_mbe_client.py
│
├── docker/                     # Docker相关
│   ├── Dockerfile
│   ├── docker-compose.yml
│   └── .dockerignore
│
├── scripts/                    # 脚本
│   ├── setup.sh               # 安装脚本
│   ├── run_telegram.sh        # 运行Telegram Bot
│   └── deploy.sh              # 部署脚本
│
├── docs/                       # 文档
│   ├── README.md
│   ├── ARCHITECTURE.md
│   └── API.md
│
├── config.yaml                 # 配置文件
├── config.example.yaml         # 配置模板
├── requirements.txt            # Python依赖
├── setup.py                    # 安装配置
├── README.md                   # 项目说明
├── LICENSE                     # 开源协议
└── .env.example                # 环境变量模板

5.2 技术栈选型

后端框架

# Gateway服务
- Python 3.11+ (与MBE一致)
- FastAPI (WebSocket + REST)
- Redis (会话缓存)
- PostgreSQL/SQLite (消息持久化)

渠道SDK

- WhatsApp: baileys (Node.js) 或 whatsapp-web.py
- Telegram: python-telegram-bot
- Discord: discord.py
- Slack: slack-bolt
- 微信: itchat/wxpy (注意风控)

工作流引擎

- Prefect (推荐，与MBE生态兼容)
- Airflow (更重量级，企业级)
- 自研DSL (轻量级，快速实现)

5.3 核心模块设计

5.3.1 基础适配器接口

# channels/base.py
from abc import ABC, abstractmethod
from typing import Any, Dict, Optional
from mbe_client import MBEClient

class BaseAdapter(ABC):
    """渠道适配器基类"""
    
    def __init__(self, mbe_client: MBEClient, config: Dict[str, Any]):
        self.mbe = mbe_client
        self.config = config
    
    @abstractmethod
    async def start(self):
        """启动适配器"""
        pass
    
    @abstractmethod
    async def stop(self):
        """停止适配器"""
        pass
    
    @abstractmethod
    async def send_message(self, channel_id: str, content: str, **kwargs):
        """发送消息"""
        pass
    
    @abstractmethod
    async def handle_message(self, message: Dict[str, Any]):
        """处理接收到的消息"""
        pass
    
    async def process_with_mbe(self, user_id: str, message: str) -> str:
        """使用MBE处理消息 (通用逻辑)"""
        # 1. 惊讶度检测
        surprise = await self.mbe.surprise.calculate(
            input=message,
            context={"user_id": user_id}
        )
        
        # 2. 智能路由
        routing = await self.mbe.route.select(
            input=message,
            task_type="chat",
            user_profile={"user_id": user_id}
        )
        
        # 3. 生成回复
        result = await self.mbe.llm.generate(
            prompt=message,
            system_prompt=f"使用{routing.selected_experts[0].id}专家知识回答"
        )
        
        # 4. 存储记忆
        await self.mbe.memory.write(
            user_id=user_id,
            memory_type="working",
            content={
                "user_message": message,
                "assistant_reply": result.content,
                "surprise_score": surprise.surprise_score
            },
            importance=0.5 + surprise.surprise_score * 0.5
        )
        
        return result.content

5.3.2 会话管理

# gateway/session.py
from typing import Dict, Optional
from dataclasses import dataclass, field
from datetime import datetime
import redis.asyncio as redis

@dataclass
class Session:
    """会话对象"""
    session_id: str
    user_id: str
    channel: str
    created_at: datetime = field(default_factory=datetime.now)
    last_active: datetime = field(default_factory=datetime.now)
    context: Dict = field(default_factory=dict)

class SessionManager:
    """会话管理器"""
    
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis = redis.from_url(redis_url)
        self.ttl = 3600  # 1小时过期
    
    async def get_session(self, session_id: str) -> Optional[Session]:
        """获取会话"""
        data = await self.redis.get(f"session:{session_id}")
        if data:
            import json
            session_dict = json.loads(data)
            return Session(**session_dict)
        return None
    
    async def create_session(
        self, 
        session_id: str, 
        user_id: str, 
        channel: str
    ) -> Session:
        """创建会话"""
        session = Session(
            session_id=session_id,
            user_id=user_id,
            channel=channel
        )
        await self.save_session(session)
        return session
    
    async def save_session(self, session: Session):
        """保存会话"""
        import json
        from dataclasses import asdict
        data = asdict(session)
        # 转换datetime为字符串
        data['created_at'] = data['created_at'].isoformat()
        data['last_active'] = data['last_active'].isoformat()
        await self.redis.setex(
            f"session:{session.session_id}",
            self.ttl,
            json.dumps(data)
        )
    
    async def delete_session(self, session_id: str):
        """删除会话"""
        await self.redis.delete(f"session:{session_id}")

5.3.3 配置管理

# config.yaml
gateway:
  host: "0.0.0.0"
  port: 18789
  
mbe:
  api_url: "http://localhost:8001"
  api_key: "${MBE_API_KEY}"
  timeout: 30.0

telegram:
  bot_token: "${TELEGRAM_BOT_TOKEN}"
  allowed_users: []  # 空则允许所有用户
  
whatsapp:
  enabled: false
  session_path: "./whatsapp_session"

discord:
  enabled: false
  bot_token: "${DISCORD_BOT_TOKEN}"

redis:
  host: "localhost"
  port: 6379
  db: 0

logging:
  level: "INFO"
  file: "logs/gateway.log"
  
monitoring:
  enabled: true
  prometheus_port: 9090

6. 快速开始：MVP实现

6.1 MBE-Telegram-Gateway (最小可行版本)

完整代码实现

# telegram_gateway.py
"""
MBE Telegram Gateway - 最小可行版本
通过MBE SDK实现Telegram Bot，完全隔离MBE核心
"""

import asyncio
import os
from typing import Optional
from telegram import Update
from telegram.ext import (
    ApplicationBuilder, 
    CommandHandler, 
    MessageHandler, 
    filters,
    ContextTypes
)
from mbe_client import MBEClient  # MBE Python SDK
from loguru import logger

class MBETelegramGateway:
    """MBE Telegram Gateway"""
    
    def __init__(
        self, 
        telegram_token: str, 
        mbe_api_url: str, 
        mbe_api_key: str
    ):
        """
        初始化Gateway
        
        Args:
            telegram_token: Telegram Bot Token
            mbe_api_url: MBE API地址
            mbe_api_key: MBE API密钥
        """
        # 初始化MBE客户端
        self.mbe = MBEClient(
            api_url=mbe_api_url,
            api_key=mbe_api_key
        )
        
        # 初始化Telegram Bot
        self.app = ApplicationBuilder().token(telegram_token).build()
        
        # 注册命令处理器
        self.app.add_handler(CommandHandler("start", self.cmd_start))
        self.app.add_handler(CommandHandler("help", self.cmd_help))
        self.app.add_handler(CommandHandler("stats", self.cmd_stats))
        
        # 注册消息处理器
        self.app.add_handler(
            MessageHandler(filters.TEXT & ~filters.COMMAND, self.handle_message)
        )
        
        logger.info("MBE Telegram Gateway 初始化完成")
    
    async def cmd_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """处理 /start 命令"""
        user = update.effective_user
        logger.info(f"用户 {user.id} ({user.username}) 发送了 /start")
        
        welcome_message = (
            f"👋 你好 {user.first_name}！\n\n"
            "我是由 **MBE (米塞斯行为引擎)** 驱动的智能助手\n\n"
            "🧠 **核心能力**：\n"
            "• 43个专家领域知识\n"
            "• 智能学习与记忆\n"
            "• 个性化建议\n\n"
            "💬 直接发送消息，我会为你提供帮助！\n"
            "📖 发送 /help 查看更多命令"
        )
        
        await update.message.reply_text(
            welcome_message, 
            parse_mode='Markdown'
        )
    
    async def cmd_help(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """处理 /help 命令"""
        help_message = (
            "📖 **可用命令**：\n\n"
            "/start - 开始使用\n"
            "/help - 查看帮助\n"
            "/stats - 查看统计信息\n\n"
            "💡 **使用提示**：\n"
            "直接发送消息即可与我对话，我会：\n"
            "• 分析消息的新颖度\n"
            "• 智能路由到最合适的专家\n"
            "• 记住我们的对话历史\n"
            "• 提供个性化的建议"
        )
        
        await update.message.reply_text(help_message, parse_mode='Markdown')
    
    async def cmd_stats(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """处理 /stats 命令"""
        user_id = str(update.effective_user.id)
        
        try:
            # 从MBE读取用户记忆统计
            memories = await self.mbe.memory.read(
                user_id=user_id,
                query="conversation",
                top_k=100
            )
            
            stats_message = (
                f"📊 **你的统计信息**：\n\n"
                f"💬 对话记录数：{memories.total_count}\n"
                f"🧠 记忆项目数：{len(memories.memories)}\n"
            )
            
            await update.message.reply_text(stats_message, parse_mode='Markdown')
            
        except Exception as e:
            logger.error(f"获取统计信息失败: {e}")
            await update.message.reply_text("暂时无法获取统计信息")
    
    async def handle_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """处理普通消息"""
        user = update.effective_user
        user_id = str(user.id)
        user_message = update.message.text
        
        logger.info(f"收到用户 {user_id} 的消息: {user_message}")
        
        # 显示"正在输入"状态
        await update.message.chat.send_action("typing")
        
        try:
            # 1️⃣ 调用MBE惊讶度检测
            logger.debug("调用 MBE Surprise API")
            surprise = await self.mbe.surprise.calculate(
                input=user_message,
                context={
                    "user_id": user_id,
                    "username": user.username,
                    "domain": "chat"
                }
            )
            
            logger.info(
                f"惊讶度分析: score={surprise.surprise_score:.2f}, "
                f"novel={surprise.is_novel}, mode={surprise.learning_mode.value}"
            )
            
            # 2️⃣ 调用MBE智能路由选择专家
            logger.debug("调用 MBE Route API")
            routing = await self.mbe.route.select(
                input=user_message,
                task_type="chat",
                user_profile={
                    "user_id": user_id,
                    "surprise_score": surprise.surprise_score
                }
            )
            
            selected_expert = routing.selected_experts[0] if routing.selected_experts else None
            logger.info(
                f"路由选择: expert={selected_expert.id if selected_expert else 'none'}, "
                f"confidence={routing.routing_confidence:.2f}"
            )
            
            # 3️⃣ 调用MBE专家生成回复
            logger.debug("调用 MBE LLM API")
            
            system_prompt = "你是一个友好、专业的AI助手，由MBE（米塞斯行为引擎）驱动。"
            if selected_expert:
                system_prompt += f" 当前使用 {selected_expert.id} 专家的知识回答。"
            
            result = await self.mbe.llm.generate(
                prompt=user_message,
                system_prompt=system_prompt,
                temperature=0.7
            )
            
            # 4️⃣ 存储对话到记忆
            logger.debug("存储对话到 MBE Memory")
            await self.mbe.memory.write(
                user_id=user_id,
                memory_type="working",
                content={
                    "user_message": user_message,
                    "assistant_reply": result.content,
                    "surprise_score": surprise.surprise_score,
                    "expert_id": selected_expert.id if selected_expert else None,
                    "timestamp": update.message.date.isoformat()
                },
                importance=0.5 + surprise.surprise_score * 0.5
            )
            
            # 5️⃣ 构建回复
            reply_text = result.content
            
            # 如果是新知识，添加提示
            if surprise.is_novel:
                reply_text += f"\n\n💡 _这对你来说是新知识！(惊讶度: {surprise.surprise_score:.2f})_"
            
            # 添加专家信息（可选）
            if selected_expert and routing.routing_confidence > 0.8:
                reply_text += f"\n\n🤖 _由 {selected_expert.id} 专家提供_"
            
            # 6️⃣ 发送回复
            await update.message.reply_text(reply_text, parse_mode='Markdown')
            
            logger.info(f"成功处理消息，回复已发送")
            
        except Exception as e:
            logger.exception(f"处理消息时出错: {e}")
            error_message = (
                "❌ 抱歉，处理消息时出现了错误。\n\n"
                f"错误信息：`{str(e)}`\n\n"
                "请稍后重试或联系管理员。"
            )
            await update.message.reply_text(error_message, parse_mode='Markdown')
    
    async def start(self):
        """启动Gateway"""
        logger.info("🚀 MBE Telegram Gateway 启动中...")
        
        # 测试MBE连接
        try:
            health = await self.mbe.health()
            logger.info(f"✅ MBE连接成功: {health}")
        except Exception as e:
            logger.error(f"❌ MBE连接失败: {e}")
            raise
        
        # 启动Telegram Bot
        await self.app.initialize()
        await self.app.start()
        await self.app.updater.start_polling()
        
        logger.info("✅ Gateway 已启动，等待消息...")
    
    async def stop(self):
        """停止Gateway"""
        logger.info("停止 Gateway...")
        await self.app.stop()
        await self.mbe.close()
        logger.info("✅ Gateway 已停止")


# ==================== 主程序 ====================

async def main():
    """主程序"""
    # 从环境变量读取配置
    telegram_token = os.getenv("TELEGRAM_BOT_TOKEN")
    mbe_api_url = os.getenv("MBE_API_URL", "http://localhost:8001")
    mbe_api_key = os.getenv("MBE_API_KEY")
    
    # 验证配置
    if not telegram_token:
        logger.error("未设置 TELEGRAM_BOT_TOKEN 环境变量")
        return
    
    if not mbe_api_key:
        logger.warning("未设置 MBE_API_KEY，将使用匿名访问")
    
    # 创建Gateway
    gateway = MBETelegramGateway(
        telegram_token=telegram_token,
        mbe_api_url=mbe_api_url,
        mbe_api_key=mbe_api_key
    )
    
    try:
        await gateway.start()
        
        # 保持运行
        await asyncio.Event().wait()
        
    except KeyboardInterrupt:
        logger.info("收到停止信号")
    finally:
        await gateway.stop()


if __name__ == "__main__":
    # 配置日志
    logger.add(
        "logs/telegram_gateway.log",
        rotation="500 MB",
        retention="10 days",
        level="INFO"
    )
    
    # 运行
    asyncio.run(main())

6.2 配置文件

.env 文件

# .env
# MBE配置
MBE_API_URL=http://localhost:8001
MBE_API_KEY=your_mbe_api_key_here

# Telegram配置
TELEGRAM_BOT_TOKEN=your_telegram_bot_token_here

# Redis配置（可选）
REDIS_URL=redis://localhost:6379/0

# 日志级别
LOG_LEVEL=INFO

requirements.txt

# requirements.txt
# MBE客户端
mbe-sdk>=1.0.0

# Telegram Bot
python-telegram-bot>=20.0

# 异步HTTP
httpx>=0.24.0

# 日志
loguru>=0.7.0

# 配置管理
python-dotenv>=1.0.0

# Redis（可选）
redis>=5.0.0

# 数据验证
pydantic>=2.0.0

6.3 Docker部署

Dockerfile

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y \
    gcc \
    && rm -rf /var/lib/apt/lists/*

# 复制依赖文件
COPY requirements.txt .

# 安装Python依赖
RUN pip install --no-cache-dir -r requirements.txt

# 复制代码
COPY telegram_gateway.py .
COPY config.yaml .

# 创建日志目录
RUN mkdir -p logs

# 暴露端口（可选）
EXPOSE 18789

# 启动命令
CMD ["python", "telegram_gateway.py"]

docker-compose.yml

# docker-compose.yml
version: '3.8'

services:
  # MBE核心引擎 (已存在)
  mbe-core:
    image: mbe:latest
    container_name: mbe-core
    ports:
      - "8001:8001"
    environment:
      - MBE_ENV=production
      - DATABASE_URL=postgresql://mbe:password@postgres:5432/mbe
    volumes:
      - mbe-data:/app/data
    networks:
      - mbe-network
    restart: unless-stopped

  # MBE Telegram Gateway (新增)
  mbe-telegram-gateway:
    build: .
    container_name: mbe-telegram-gateway
    environment:
      - MBE_API_URL=http://mbe-core:8001
      - MBE_API_KEY=${MBE_API_KEY}
      - TELEGRAM_BOT_TOKEN=${TELEGRAM_BOT_TOKEN}
      - REDIS_URL=redis://redis:6379/0
      - LOG_LEVEL=INFO
    depends_on:
      - mbe-core
      - redis
    volumes:
      - ./config.yaml:/app/config.yaml
      - gateway-logs:/app/logs
    networks:
      - mbe-network
    restart: unless-stopped

  # Redis (会话存储)
  redis:
    image: redis:7-alpine
    container_name: mbe-redis
    ports:
      - "6379:6379"
    volumes:
      - redis-data:/data
    networks:
      - mbe-network
    restart: unless-stopped

volumes:
  mbe-data:
  redis-data:
  gateway-logs:

networks:
  mbe-network:
    driver: bridge

6.4 部署脚本

setup.sh

#!/bin/bash
# setup.sh - 一键部署脚本

set -e

echo "🚀 MBE Telegram Gateway 部署脚本"
echo "================================="

# 检查环境变量
if [ -z "$TELEGRAM_BOT_TOKEN" ]; then
    echo "❌ 错误: 未设置 TELEGRAM_BOT_TOKEN"
    echo "请运行: export TELEGRAM_BOT_TOKEN='your_token'"
    exit 1
fi

if [ -z "$MBE_API_KEY" ]; then
    echo "⚠️  警告: 未设置 MBE_API_KEY，将使用匿名访问"
fi

# 创建必要目录
echo "📁 创建目录..."
mkdir -p logs

# 复制配置文件
if [ ! -f .env ]; then
    echo "📝 创建 .env 文件..."
    cp .env.example .env
    echo "请编辑 .env 文件填入配置"
fi

# 安装依赖
echo "📦 安装依赖..."
pip install -r requirements.txt

# 测试MBE连接
echo "🔍 测试MBE连接..."
python -c "
import asyncio
from mbe_client import MBEClient
async def test():
    client = MBEClient(api_url='${MBE_API_URL:-http://localhost:8001}')
    try:
        health = await client.health()
        print(f'✅ MBE连接成功: {health}')
    except Exception as e:
        print(f'❌ MBE连接失败: {e}')
    finally:
        await client.close()
asyncio.run(test())
"

echo ""
echo "✅ 部署完成！"
echo ""
echo "启动方式："
echo "  开发模式: python telegram_gateway.py"
echo "  Docker模式: docker-compose up -d"
echo ""

6.5 使用说明

快速启动

# 1. 克隆或创建项目
mkdir mbe-telegram-gateway
cd mbe-telegram-gateway

# 2. 创建必要文件
touch telegram_gateway.py
touch requirements.txt
touch Dockerfile
touch docker-compose.yml
touch .env

# 3. 配置环境变量
export TELEGRAM_BOT_TOKEN="your_bot_token"
export MBE_API_KEY="your_mbe_api_key"
export MBE_API_URL="http://localhost:8001"

# 4. 安装依赖
pip install -r requirements.txt

# 5. 启动Gateway
python telegram_gateway.py

Docker部署

# 1. 构建镜像
docker-compose build

# 2. 启动服务
docker-compose up -d

# 3. 查看日志
docker-compose logs -f mbe-telegram-gateway

# 4. 停止服务
docker-compose down

7. 开发路线图

7.1 MVP阶段 (1周)

目标: 完成基础Telegram Gateway，验证方案可行性

交付物：

• ✅ 可运行的Telegram Bot
• ✅ Docker部署方案
• ✅ 基础文档

7.2 扩展阶段 (4-6周)

目标: 支持多个渠道，增强功能

Phase 1: 多渠道支持 (3周)

Phase 2: 高级功能 (2-3周)

7.3 生产优化阶段 (2-3周)

目标: 性能优化、监控、文档完善

7.4 里程碑

v0.1.0 (MVP) - Week 1
  ✅ Telegram Bot基础功能
  ✅ MBE SDK集成
  ✅ Docker部署

v0.2.0 (多渠道) - Week 4
  ✅ Discord支持
  ✅ WhatsApp支持
  ✅ 统一的适配器接口

v0.3.0 (高级功能) - Week 7
  ✅ WebSocket控制面板
  ✅ 工作流引擎
  ✅ 定时任务

v1.0.0 (生产就绪) - Week 10
  ✅ 性能优化
  ✅ 监控系统
  ✅ 完整文档
  ✅ 测试覆盖>80%

附录

A. OpenClaw官方资源

B. MBE相关资源

C. 技术栈对比

D. 安全隐患警告

OpenClaw 2026年2月安全警告

多个安全研究人员指出，OpenClaw的不当部署会导致：

1. 未授权访问风险: 通过反向代理暴露的控制界面可能被未授权访问
2. 数据泄露: 敏感数据、API密钥、OAuth令牌可能暴露
3. 凭证盗窃: 因OpenClaw自动批准"本地"连接，导致互联网流量被错误识别为可信

安全建议

• ❌ 不要将控制面板直接暴露到公网
• ✅ 使用Tailscale等VPN进行远程访问
• ✅ 启用Gateway令牌认证
• ✅ 遵循官方安全配置指南

E. 开发者注意事项

使用SDK方案的优势

1. 完全隔离: MBE核心零修改
2. 灵活部署: Gateway和MBE可分离部署
3. 负载均衡: 多Gateway实例共享一个MBE
4. 独立迭代: Gateway和MBE独立开发
5. 商业友好: 可以闭源或商业化Gateway

潜在挑战

1. 多渠道开发复杂度: 每个渠道需要独立适配器
2. 工作流引擎: 需要自行实现类似Lobster的引擎
3. WebSocket架构: 与MBE的REST API需要转换
4. 维护成本: 需要跟踪各渠道API的变化

F. 参考文献

1. OpenClaw官方文档: https://docs.molt.bot/
2. MBE系统架构: docs/architecture/MBE_COMPLETE_SYSTEM_ARCHITECTURE.md
3. MBE用户手册: docs/guides/MBE_USER_OPERATION_MANUAL.md
4. Python Telegram Bot: https://python-telegram-bot.org/
5. MCP协议规范: https://modelcontextprotocol.io

🎯 总结

核心结论

1. OpenClaw是优秀的多渠道AI助手框架

   • 成熟的Gateway架构
   • 50+渠道集成
   • 活跃的开源社区

2. MBE具备更强的AI能力

   • 43个专家 vs 单一Pi代理
   • HOPE在线学习
   • TITANS记忆系统
   • 完整的商业基础设施

3. 通过SDK/MCP开发Gateway是最优方案

   • ✅ 完全隔离MBE核心
   • ✅ 独立开发和部署
   • ✅ 灵活扩展渠道
   • ✅ 复用MBE全部能力

下一步行动

推荐: 从Telegram Gateway MVP开始，1周完成验证，然后逐步扩展到其他渠道。

文档维护者: MBE团队
最后更新: 2026-02-03
版本: v1.0