Webhook Bot间消息广播系统

概述

这个Webhook系统解决了Telegram Bot无法直接看到其他Bot消息的问题。当一个Bot在群组中发送消息时，它会将消息广播到Webhook服务器，服务器再将消息分发给所有订阅了该群组的其他Bot，使得Bot之间可以感知彼此的消息。

系统架构

Bot A 发送消息 → Telegram Group
     ↓
Bot A 广播到 Webhook Server
     ↓
Webhook Server 分发给 Bot B, C, D
     ↓
Bot B, C, D 接收消息并处理

核心组件

1. Webhook Server (FastAPI)

接收Bot消息广播
管理Bot注册和订阅
消息分发和路由
Token认证

2. Webhook Client

Bot连接到Webhook服务器
发送消息广播
接收其他Bot消息

3. Telegram Bot Integration

自动在Bot发送消息时广播
接收其他Bot消息并集成到上下文

配置说明

Webhook配置 (configs/default.toml)

# Webhook配置（Bot间消息广播）
[webhook]
# 是否启用Webhook功能
enabled = true
# Webhook服务器地址
server_url = "http://localhost:8080"
# 认证Token
auth_token = "your-secure-webhook-token"

# 服务器配置（如果作为Webhook服务器运行）
[webhook.server]
# 服务器绑定地址
host = "0.0.0.0"
# 服务器端口
port = 8080

# 客户端配置（Bot连接到Webhook服务器）
[webhook.client]
# 订阅的群组ID列表（Bot会接收这些群组中其他Bot的消息）
subscribed_groups = [-1001234567890, -1001987654321]
# 回调服务器端口（用于接收来自Webhook服务器的消息）
callback_port = 8081
# 连接超时时间(秒)
connection_timeout = 10.0
# 请求超时时间(秒)
request_timeout = 5.0
# 重试次数
max_retries = 3
# 重试延迟(秒)
retry_delay = 1.0

使用方法

1. 启动Webhook服务器

# 使用默认配置启动
python scripts/webhook_server.py

# 使用指定配置启动
python scripts/webhook_server.py production

2. 配置Bot订阅

在每个Bot的配置文件中：

[webhook]
enabled = true
server_url = "http://your-webhook-server:8080"
auth_token = "your-secure-webhook-token"

[webhook.client]
subscribed_groups = [-1001234567890]  # 要监听的群组ID
callback_port = 8081  # 每个Bot使用不同的端口

3. 启动Bot

Bot会自动连接到Webhook服务器并开始接收其他Bot的消息。

API接口

注册Bot

POST /register
Authorization: Bearer <token>
Content-Type: application/json

{
  "bot_username": "mybot",
  "auth_token": "<token>",
  "subscribed_groups": [-1001234567890],
  "webhook_endpoint": "http://localhost:8081/webhook"
}

广播消息

POST /broadcast
Authorization: Bearer <token>
Content-Type: application/json

{
  "bot_message": {
    "bot_username": "mybot",
    "group_id": -1001234567890,
    "message_content": "Hello from Bot!",
    "sender_info": {
      "user_id": 123456789,
      "username": "mybot",
      "is_bot": true
    },
    "timestamp": "2024-01-01T12:00:00Z"
  }
}

获取注册的Bot

GET /bots
Authorization: Bearer <token>

健康检查

GET /health

消息格式

BotMessage结构

{
  "message_id": "uuid",
  "bot_username": "mybot",
  "group_id": -1001234567890,
  "timestamp": "2024-01-01T12:00:00Z",
  "message_content": "Hello world",
  "message_type": "text",
  "sender_info": {
    "user_id": 123456789,
    "username": "mybot",
    "first_name": "My",
    "last_name": "Bot",
    "is_bot": true
  },
  "reply_info": {
    "user_info": {
      "user_id": 987654321,
      "username": "user",
      "is_bot": false
    },
    "timestamp": "2024-01-01T11:59:00Z",
    "content": "Original message",
    "message_id": 123
  },
  "telegram_message_id": 456,
  "chat_type": "group"
}

安全考虑

Token认证: 所有API请求都需要有效的Bearer Token
HTTPS: 生产环境建议使用HTTPS
防火墙: 限制Webhook服务器的访问
Token轮换: 定期更换认证Token

故障处理

常见问题

连接超时
- 检查Webhook服务器是否运行
- 验证网络连接
- 检查防火墙设置
认证失败
- 验证Token配置
- 检查Token是否匹配
消息丢失
- 检查群组ID配置
- 验证Bot订阅设置
- 查看服务器日志

日志和监控

Webhook服务器日志: webhook_server.log
Bot日志中包含Webhook相关信息
使用 /health 端点进行健康检查

扩展和自定义

自定义消息处理

async def custom_message_handler(message: BotMessage):
    # 自定义处理逻辑
    print(f"收到来自 {message.bot_username} 的消息: {message.message_content}")

    # 可以进行额外的处理，如保存到数据库
    # await save_to_database(message)

# 在创建WebhookClient时传入
client = WebhookClient(
    config=webhook_config,
    bot_username="mybot",
    subscribed_groups=[-123],
    message_handler=custom_message_handler
)

消息过滤

可以在处理函数中添加过滤逻辑：

async def filtered_message_handler(message: BotMessage):
    # 只处理特定Bot的消息
    if message.bot_username in ["important_bot", "admin_bot"]:
        await process_important_message(message)

    # 只处理包含特定关键词的消息
    if "urgent" in message.message_content.lower():
        await handle_urgent_message(message)

测试

运行单元测试

python -m pytest tests/unit/test_webhook.py -v

运行集成测试

python -m pytest tests/integration/test_webhook_integration.py -v

手动测试

启动Webhook服务器
启动多个Bot实例
在群组中发送消息
观察其他Bot是否能接收到消息