OpenClaw WebSocket Channel 开发实战：从零打造自定义 AI 通信通道

OpenClaw WebSocket Channel 开发实战：从零打造自定义 AI 通信通道 | 极客日志

前端层：Vue 3 + WebSocket
服务端：Python + aiohttp + uv
通道层：Node.js + ws + OpenClaw Plugin SDK
AI 层：OpenClaw Gateway + LLM Provider

openclaw-websocket-channel/
├── websocket-service/ # Python WebSocket 服务端
│   ├── app.py # aiohttp 主程序
│   └── requirements.txt # Python 依赖
├── websocket-web/ # Vue 3 前端
│   ├── src/
│   │   └── App.vue # 主界面
│   └── package.json
└── websocket-channel/ # OpenClaw 通道插件
    ├── index.ts # 插件主逻辑
    └── openclaw.plugin.json

# 进入服务端目录
cd websocket-service
# 使用 uv 安装依赖
uv sync
# 启动服务端
python app.py # 默认监听：ws://localhost:8765

# 进入前端目录
cd websocket-web
# 安装依赖
npm install
# 开发模式运行
npm run dev # 访问：http://localhost:3000

# 进入通道插件目录
cd websocket-channel
# 安装到 OpenClaw
openclaw plugins install .
# 验证安装
openclaw plugins list # 应该看到：websocket-channel

{
  "channels": {
    "websocket-channel": {
      "enabled": true,
      "config": {
        "enabled": true,
        "wsUrl": "ws://localhost:8765/openclaw"
      }
    }
  }
}

# 如果使用 macOS 应用
# 点击菜单栏 OpenClaw → Restart Gateway
# 或命令行重启
pkill -f openclaw-gateway
openclaw gateway run

你：你好，请介绍一下自己
AI：你好！我是你的个人 AI 助手，基于 OpenClaw 框架运行。我可以帮助你回答问题、编写代码、分析数据等。有什么我可以帮你的吗？

┌─────────────────┐
│ 用户浏览器      │
│ (Vue 前端)      │
└────────┬────────┘
         │ WebSocket
         │ ws://localhost:8765
         ▼
┌─────────────────┐
│ Python 服务端   │
│ (aiohttp)       │
└────────┬────────┘
         │ WebSocket
         │ 长连接
         ▼
┌─────────────────┐
│ Node.js 通道    │
│ (ws 库)         │
└────────┬────────┘
         │ OpenClaw Plugin API
         ▼
┌─────────────────┐
│ OpenClaw        │
│ Gateway         │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│ AI Provider     │
│ (Qwen/Bailian)  │
└─────────────────┘

# 创建插件目录
mkdir -p openclaw-websocket-channel/websocket-channel
cd openclaw-websocket-channel/websocket-channel
# 创建基础文件
touch index.ts openclaw.plugin.json package.json

import type { ReplyPayload } from "openclaw/auto-reply/types";
import type { ChannelPlugin, OpenClawConfig } from "openclaw/plugin-sdk";
import { createDefaultChannelRuntimeState } from "openclaw/plugin-sdk";

interface WebSocketChannelConnection {
  ws: any;
  accountId: string;
}

interface WebSocketChannelAccount {
  accountId: string;
  wsUrl: string;
  enabled?: boolean;
  configured?: boolean;
  dmPolicy?: "pairing" | "allowlist" | "open" | "disabled";
}

const connections = new Map<string, WebSocketChannelConnection>();
let pluginRuntime: any = null;

const WebSocketChannel: ChannelPlugin<WebSocketChannelAccount> = {
  id: "websocket-channel",
  meta: {
    id: "websocket-channel",
    label: "Websocket Channel",
    selectionLabel: "Websocket Channel (Custom)",
    docsPath: "/channels/websocket-channel",
    blurb: "WebSocket based messaging channel.",
    aliases: ["ws"],
  },
  // ... 其他配置
};

config: {
  /**
   * 列出所有配置的账户 ID
   * @returns 固定返回 ["default"]
   */
  listAccountIds: (cfg: OpenClawConfig) => {
    return ["default"];
  },
  /**
   * 解析账户配置
   */
  resolveAccount: (cfg: OpenClawConfig, accountId: string) => {
    const channelCfg = cfg.channels?.["websocket-channel"];
    if (!channelCfg || !channelCfg.config) {
      return undefined;
    }
    const config = channelCfg.config as any;
    return {
      accountId: "default",
      wsUrl: config.wsUrl || "ws://localhost:8765/openclaw",
      enabled: config.enabled !== false,
    };
  },
  /**
   * 检查账户是否已配置
   */
  isConfigured: async (account, cfg) => {
    return Boolean(account.wsUrl && account.wsUrl.trim() !== "");
  },
}

status: {
  /**
   * 默认运行时状态模板
   * ⚠️ 必须实现这个方法，否则 UI 会显示 "0/1 connected"
   */
  defaultRuntime: createDefaultChannelRuntimeState("default", {
    wsUrl: null,
    connected: false,
    groupPolicy: null,
  }),
  /**
   * 构建通道摘要（用于 UI 显示）
   */
  buildChannelSummary: ({ snapshot }) => ({
    wsUrl: snapshot.wsUrl ?? null,
    connected: snapshot.connected ?? null,
    groupPolicy: snapshot.groupPolicy ?? null,
  }),
  /**
   * 构建账户完整快照
   */
  buildAccountSnapshot: ({ account, runtime }) => ({
    accountId: account.accountId,
    enabled: account.enabled,
    configured: account.configured,
    wsUrl: account.wsUrl,
    running: runtime?.running ?? false,
    connected: runtime?.connected ?? false,
    groupPolicy: runtime?.groupPolicy ?? null,
    lastStartAt: runtime?.lastStartAt ?? null,
    lastStopAt: runtime?.lastStopAt ?? null,
    lastError: runtime?.lastError ?? null,
  }),
}

gateway: {
  /**
   * 启动 WebSocket 账户连接
   */
  startAccount: async (ctx) => {
    const { log, account, abortSignal, cfg } = ctx;
    log?.info(`[websocket-channel] Starting WebSocket Channel for ${account.accountId}`);

    // 获取 runtime API
    const runtime = pluginRuntime;

    // ⭐ 关键：设置初始状态为已连接
    ctx.setStatus({
      accountId: account.accountId,
      wsUrl: account.wsUrl,
      running: true,
      connected: true,
    });
    log?.info(`[websocket-channel] Status set: connected=true, running=true`);

    // 创建 WebSocket 连接
    const WebSocketLib = await import("ws");
    const ws = new (WebSocketLib.default as any)(account.wsUrl);

    // 存储连接
    connections.set(account.accountId, { ws, accountId: account.accountId });

    // 监听消息事件
    ws.on("message", async (data: Buffer) => {
      try {
        // 1. 解析原始消息
        const rawData = data.toString();
        const eventData = JSON.parse(rawData);
        const innerData = eventData.data || {};

        // 2. 标准化消息
        const normalizedMessage = {
          id: `${eventData.source || "websocket"}-${Date.now()}`,
          channel: "websocket-channel",
          accountId: account.accountId,
          senderId: innerData.source || eventData.source || "unknown",
          senderName: innerData.source || eventData.source || "Unknown",
          text: innerData.content || innerData.text || "",
          timestamp: innerData.timestamp || Date.now().toISOString(),
          isGroup: false,
          groupId: undefined,
          attachments: [],
          metadata: {},
        };
        log?.info(`[websocket-channel] 📨 Received: "${normalizedMessage.text}" from ${normalizedMessage.senderId}`);

        // 3. 解析路由
        const route = runtime.channel.routing.resolveAgentRoute({
          cfg,
          channel: "websocket-channel",
          accountId: account.accountId,
          peer: {
            kind: "direct",
            id: normalizedMessage.senderId,
          },
        });

        // 4. 构建消息上下文
        const ctxPayload = runtime.channel.reply.finalizeInboundContext({
          Body: normalizedMessage.text,
          BodyForAgent: normalizedMessage.text,
          From: normalizedMessage.senderId,
          To: undefined,
          SessionKey: route.sessionKey,
          AccountId: route.accountId,
          ChatType: "direct",
          SenderName: normalizedMessage.senderName,
          SenderId: normalizedMessage.senderId,
          Provider: "websocket-channel",
          Surface: "websocket-channel",
          MessageSid: normalizedMessage.id,
          Timestamp: Date.now(),
        });

        // 5. 调用框架调度器
        await runtime.channel.reply.dispatchReplyWithBufferedBlockDispatcher({
          ctx: ctxPayload,
          cfg: cfg,
          dispatcherOptions: {
            deliver: async (payload: ReplyPayload, { kind }) => {
              log?.info(`[websocket-channel] Delivering ${kind} reply via WebSocket...`);
              const currentConn = connections.get(account.accountId);
              if (!currentConn || !currentConn.ws || currentConn.ws.readyState !== 1) {
                throw new Error("No WebSocket connection available");
              }
              // 发送 AI 回复
              currentConn.ws.send(JSON.stringify({ type: "reply", content: payload.text || "", kind }));
            },
            onError: (err, { kind }) => {
              log?.error(`[websocket-channel] Delivery error for ${kind}: ${err.message}`);
            },
          },
        });
        log?.info(`[websocket-channel] Message dispatched successfully`);
      } catch (err) {
        log?.error(`[websocket-channel] Failed to process message: ${err.message}`);
      }
    });

    // 监听错误和关闭
    const connectionPromise = new Promise<void>((resolve, reject) => {
      ws.on("error", (err: Error) => {
        log?.error(`[websocket-channel] ❌ WebSocket error: ${err.message}`);
        connections.delete(account.accountId);
        reject(err);
      });

      ws.on("close", () => {
        log?.info(`[websocket-channel] 🔴 Connection closed`);
        connections.delete(account.accountId);
        resolve();
      });

      // 监听中止信号
      abortSignal.addEventListener("abort", () => {
        log?.info(`[websocket-channel] ⏹️ Abort requested`);
        ws.close();
        resolve();
      });

      // 保持连接运行
      Promise.race([
        connectionPromise,
        new Promise<void>((res) => {
          abortSignal.addEventListener("abort", () => res());
        }),
      ]).then(() => {
        connections.delete(account.accountId);
      });
    });
  },
}

/**
 * 注册插件入口
 * @param api - 插件 API
 */
export default function register(api: any) {
  console.log("[websocket-channel] Registering WebSocket Channel plugin");
  pluginRuntime = api.runtime;
  api.registerChannel({ plugin: WebSocketChannel });
}

OpenClaw WebSocket Channel 开发实战：从零打造自定义 AI 通信通道

项目背景

为什么做这个项目？

技术栈

快速开始

项目结构

1. 启动 Python WebSocket 服务端

2. 启动 Vue 前端

3. 安装 WebSocket Channel

4. 配置 OpenClaw

5. 重启 OpenClaw Gateway

6. 测试

程序架构

整体架构图

数据流详解

入站消息（前端 → AI）

出站消息（AI → 前端）

Channel 开发详解

1. 项目初始化

2. 定义插件元数据

3. 实现配置适配器

4. 实现状态管理适配器 ⭐关键

5. 实现网关适配器（核心）

6. 注册插件入口

参考链接

官方文档

示例项目

更多推荐文章

相关免费在线工具

OpenClaw WebSocket Channel 开发实战：从零打造自定义 AI 通信通道

项目背景

为什么做这个项目？

技术栈

快速开始

项目结构

1. 启动 Python WebSocket 服务端

2. 启动 Vue 前端

3. 安装 WebSocket Channel

4. 配置 OpenClaw

5. 重启 OpenClaw Gateway

6. 测试

程序架构

整体架构图

数据流详解

入站消息（前端 → AI）

出站消息（AI → 前端）

Channel 开发详解

1. 项目初始化

2. 定义插件元数据

3. 实现配置适配器

4. 实现状态管理适配器 ⭐关键

5. 实现网关适配器（核心）

6. 注册插件入口

参考链接

官方文档

示例项目

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具