Unified Channel Hub

  ](https://www.npmjs.com/package/unified-channel)   ](https://nodejs.org/) 

统一频道

个人人工智能助理的通信层-在一个统一的API中的IM、电子邮件、语音、日历、物联网。Python+TypeScript+Java+Rust。

5个类别的27个适配器。ServiceBridge、对话记忆、丰富的输出和流媒体。不是聊天机器人平台，而是你的图书馆 pip install / npm install /Maven添加/ cargo add 进入 你的 项目和电话 manager.run().

为什么存在

统一渠道为您提供了消息传递管道，因此您可以专注于您的实际逻辑：服务管理、人工智能代理、警报、客户支持、社区机器人等等。

功能亮点

核心价值：口袋里的远程控制面板

任何服务+统一渠道=通过IM进行完全远程管理。

你在某个地方运行着一个服务——一个部署管道、一个监控栈、一个家庭自动化系统、一个GPU集群。你想检查它，重启东西，查看日志，运行命令——但你在手机上，在火车上，在吃饭的时候。

通过统一渠道 ServiceBridge，您可以在4行中将服务功能作为聊天命令公开：

from unified_channel import ChannelManager, ServiceBridge
from unified_channel.adapters.telegram import TelegramAdapter

manager = ChannelManager()
manager.add_channel(TelegramAdapter("BOT_TOKEN"))

bridge = ServiceBridge(manager)
bridge.expose("deploy", deploy_service, description="Deploy service")
bridge.expose("logs", get_logs, description="View logs")
bridge.expose("restart", restart_service, description="Restart service")
bridge.expose("metrics", get_metrics, description="View metrics")

await manager.run()

现在，通过您的手机（Telegram、Discord、任何地方）：

/deploy prod v2.1     → ✅ deployed to prod
/logs api --tail 50   → [last 50 log lines]
/restart worker-3     → ✅ restarted
/metrics              → CPU: 23% | Mem: 4.2GB | QPS: 1.2k

这不是聊天机器人框架。 它是您的服务的远程控制平面，恰好使用IM作为传输方式。你的手机成为任何你可以为其编写Python/TypeScript/Java函数的终端。

由于它支持MCP，AI代理也可以控制你的服务——与MCP工具相同的公开功能。

┌──────────────────────────────────────────────┐
│               unified-channel                │
│                                              │
│  Your Phone (IM)  ──→  ServiceBridge  ──→  Your Service
│  AI Agent (MCP)   ──→  MCP Server    ──→  Functions
│                                              │
│  Telegram │ Discord │ Slack │ WhatsApp       │
│  iMessage │ Matrix  │ Teams │ LINE           │
│  Feishu   │ Mattermost │ Google Chat         │
│  Nextcloud│ Synology│ Zalo  │ Nostr          │
│  BlueBubbles │ Twitch │ IRC                  │
└──────────────────────────────────────────────┘

对话记忆

使用可插拔后端跟踪每个用户、每个通道的对话历史记录：

from unified_channel import ChannelManager, ConversationMemory
from unified_channel.memory import SQLiteBackend

manager = ChannelManager()
memory = ConversationMemory(backend=SQLiteBackend("conversations.db"))
manager.add_middleware(memory)

@manager.on_message
async def handle(msg):
    history = await memory.get_history(msg.sender_id, limit=10)
    # Pass history to your LLM, search engine, or custom logic
    return generate_response(msg.content.text, history)

后端： InMemoryBackend （默认）， SQLiteBackend, RedisBackend每个都存储了带有发送者、通道、时间戳和元数据的消息。

丰富的回复

发送跨平台自动降级的结构化内容：

from unified_channel import RichReply

reply = (
    RichReply("Here are your options:")
    .add_buttons(["Approve", "Reject", "Defer"])
    .add_image("https://example.com/chart.png", alt="CPU usage graph")
    .set_footer("Reply within 24h")
)
return reply

在Telegram上：内联键盘按钮+图像。在Slack上：Block Kit。IRC上：带有编号选项的纯文本。每个适配器都将丰富的元素映射到最佳的可用平台原语——没有手动的每个平台逻辑。

流媒体

长时间的回复会得到打字指示和分块交付，而不是尴尬的沉默：

from unified_channel import StreamingReply

async def handle(msg):
    stream = StreamingReply(msg)
    await stream.start_typing()

    async for chunk in call_llm_streaming(msg.content.text):
        await stream.send_chunk(chunk)

    await stream.finish()

在支持就地编辑的平台上（Telegram、Discord、Slack）：消息实时更新。在其他方面：块被分批处理，并以合理的间隔发送。打字指示器在间隙期间自动发送。

快速开始

python

pip install unified-channel[telegram]

from unified_channel import ChannelManager, CommandMiddleware
from unified_channel.adapters.telegram import TelegramAdapter

manager = ChannelManager()
manager.add_channel(TelegramAdapter("BOT_TOKEN"))

commands = CommandMiddleware()

@commands.command("status")
async def status(msg):
    return "Agent is running!"

manager.add_middleware(commands)
manager.on_message(lambda msg: f"Echo: {msg.content.text}")

await manager.run()

TypeScript

npm install unified-channel grammy

import { ChannelManager, CommandMiddleware } from "unified-channel";
import { TelegramAdapter } from "unified-channel/adapters/telegram";

const manager = new ChannelManager();
manager.addChannel(new TelegramAdapter("BOT_TOKEN"));

const commands = new CommandMiddleware();
commands.command("status", async () => "Agent is running!");

manager.addMiddleware(commands).onMessage(async (msg) => `Echo: ${msg.content.text}`);

await manager.run();

Java

  io.github.gambletan
  unified-channel
  0.1.0

var manager = new ChannelManager();
manager.addChannel(new TelegramAdapter("BOT_TOKEN"));

var commands = new CommandMiddleware();
commands.command("status", msg -> HandlerResult.text("Agent is running!"));

manager.addMiddleware(commands);
manager.onMessage(msg -> HandlerResult.text("Echo: " + msg.content().text()));
manager.run();

建筑

Incoming Message → [Middleware 1] → [Middleware 2] → ... → [Fallback Handler]
                                                                  ↓
                              Adapter.send() ← reply (string or OutboundMessage)

核心概念：

• 通道适配器：每个平台一个。处理连接/断开、消息解析、发送。
• 中间件：在处理程序之前拦截消息。将它们链接起来进行访问控制、命令路由、日志记录、速率限制。
• 渠道经理：协调一切。注册适配器，添加中间件，设置处理程序，调用 run().

功能对比

|功能|统一频道|OpenClaw|Botpress|LangBot| |---------|:-:|:-:|:-:|:-:| |可嵌入库| 是 |否（完全代理）|否（平台）|部分| |频道计数| 27 | 18 | 4 | 10 | |ServiceBridge（远程控制）| 是 |否|否|否| |对话记忆| 是 |是|是|否| |丰富回复（自动降级）| 是 |部分|是|否| |流媒体输出| 是 |是|否|否| |MCP服务器| 是 |否|否|否| |Python+TypeScript+Java+Rust| 是 |仅限TS |仅限TS |仅限Python| |无需付款| 是 |否|否|否|

支持的频道

用例

• 远程服务管理 （主要）：管理部署、查看日志、重启服务、检查指标——所有这些都可以通过Telegram/Discord/Slack从您的手机上完成。 ServiceBridge 将任何一组功能转换为远程控制面板。
• GPU集群/ML操作：监控训练作业、检查GPU利用率、取消运行、从手机聊天中提取结果。
• 物联网/家庭自动化：控制智能家居设备，检查传感器读数，通过iMessage、Telegram或任何渠道触发例程。
• AI聊天代理：从一个代码库跨多个平台部署LLM驱动的代理。
• DevOps警报：同时向Slack、Discord和Mattermost广播CI/CD、监控和事件警报。
• 客户支持机器人：跨Telegram、WhatsApp、LINE和网络聊天的统一收件箱。
• 社区管理：服务于Discord、Telegram和Matrix社区的单一机器人逻辑。

MCP服务器（用于AI代理）

任何与MCP兼容的代理都可以作为工具控制消息通道：

{
  "mcpServers": {
    "unified-channel": {
      "command": "npx",
      "args": ["@unified-channel/mcp-server"],
      "env": {
        "UC_TELEGRAM_TOKEN": "your-bot-token",
        "UC_DISCORD_TOKEN": "your-bot-token"
      }
    }
  }
}

可用工具： send_message, broadcast_message, get_channel_status, list_channels, get_recent_messages

看 mcp服务器/README.md 查看完整文档。

项目结构

unified-channel/
├── python/          # Python implementation (pip install)
├── typescript/      # TypeScript/Node.js implementation (npm install)
├── java/            # Java implementation (Maven)
├── rust/            # Rust implementation (cargo add)
├── mcp-server/      # MCP server for AI agents
├── .github/
│   ├── workflows/   # CI for all languages
│   └── PULL_REQUEST_TEMPLATE.md
└── CONTRIBUTING.md

贡献

看 贡献.md 作为指导方针。我们欢迎以下PR：

• 新的通道适配器（特别是Java存根到完整实现）
• 中间件添加（速率限制、日志记录、i18n）
• Bug修复和文档改进
• 针对特定语言的改进

许可证

麻省理工学院