使用说明

统一频道

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

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

为什么存在

项目	它是什么	差距
OpenClaw/Nanobot	完整的AI代理框架	如果你只需要消息传递，那就太重了。你会得到一个完整的运行时、配置系统和固执己见的代理循环。
Botpress/Hexabot	聊天机器人平台	不可嵌入。你在里面建造他们的平台，不是你的。
语言机器人	AI到IM桥	没有ServiceBridge（远程功能控制），没有对话内存，没有丰富的输出，没有流媒体。
统一频道	您在项目中嵌入的轻量级库	这就是重点——没有差距。

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

功能亮点

功能	它的作用
27个适配器，5个类别	即时通讯：Telegram、Discord、Slack、WhatsApp、iMessage、Matrix、Teams、LINE、Feishu、Mattermost、谷歌聊天、Nextcloud、Synology、Zalo、Nost、BlueBubbles、Twitch、IRC· 电子邮件Gmail/IMAP 语音：提里奥之声· 短信：Twilio短信· 日历：谷歌日历· 物联网：家庭助理
服务桥	将任何功能显示为聊天命令。您的手机将成为您服务的遥控器。
会话记忆	每个用户/每个频道的对话历史记录。在内存中，SQLite或Redis后端。
RichReply	发送按钮、旋转木马、图像、文件——在不支持它们的平台上自动降级。
流媒体	为长时间运行的响应键入指示符+分块消息传递。
MCP 服务器	AI代理（Claude、GPT、本地LLM）通过标准MCP工具调用控制您的服务。
4种语言	Python，TypeScript，Java，Rust-相同的架构，相同的API形状。

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

任何服务+统一渠道=通过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

<dependency>
  <groupId>io.github.gambletan</groupId>
  <artifactId>unified-channel</artifactId>
  <version>0.1.0</version>
</dependency>

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
无需付款	是	否	否	否

支持的频道

通道	协议	Python	TypeScript	Java
Telegram	Bot API（轮询）	是	是	否
Discord	网关WebSocket	Yes	Yes	是
松弛	套接字模式	是	是	有
WhatsApp	云API/WhatsApp-web.js	是	是	存根
iMessage	macOS SQLite+AppleScript	是	是	存根
矩阵	客户端-服务器API	是	是	存根
MS Teams	机器人框架	是	是	存根
LINE	消息API	是	是	存根
飞树/云雀	活动订阅	是	是	存根
Mattermost	WebSocket+REST	是	是	有
谷歌聊天	服务帐户	是	是	存根
Nextcloud Talk	REST轮询	是	是	存根
Synology聊天	Webhook	是	是	存根
Zalo	OA API	是	是	存根
Nostr	NIP-04 DM	是	是	存根
BlueBubbles	REST轮询	是	是	存根
Twitch	IRC/TMI	是	是	存根
IRC	原始IRC	是	是	有
电子邮件（Gmail/IMAP）	IMAP+SMTP	是	是	存根
Twilio短信	REST+Webhook	是	是	存根
Twilio Voice	REST+Webhook	是	是	存根
Google日历	REST API v3	是	是	存根
主助手	WebSocket API	是	是	存根

用例

远程服务管理 （主要）：管理部署、查看日志、重启服务、检查指标——所有这些都可以通过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修复和文档改进
针对特定语言的改进

许可证

麻省理工学院

使用说明

统一频道

为什么存在

功能亮点

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

对话记忆

丰富的回复

流媒体

快速开始

python

TypeScript

Java

建筑

功能对比

支持的频道

用例

MCP服务器（用于AI代理）

项目结构

贡献

许可证

相关 MCP

远程MacOS控制服务器

终端命令执行服务器

AI代理编排系统

视频播放控制服务器

SFTP-SSH MCP服务器

调试器HTTP接口插件