Telegram结合Flask搭建自定义Webhook接收器：从零实现机器人主动推送

引言：为什么需要自定义Webhook接收器？

Telegram Bot API 提供了两种获取更新（Update）的方式：长轮询（Long Polling）和 Webhook。长轮询需要客户端持续请求服务器，不仅浪费资源，而且存在延迟。而 Webhook 允许 Telegram 服务器主动将新消息推送到你的服务器，响应更快、资源占用更低。官方默认的 Webhook 配置虽然简单，但灵活性不足。通过 Flask 构建自定义接收器，你可以完全控制请求处理逻辑，轻松实现消息过滤、多机器人管理、与数据库联动等高级功能。

本文将从零开始，指导你使用 Python 的 Flask 框架搭建一个安全、可扩展的 Webhook 接收器，并与 Telegram Bot 无缝集成。无论你是想开发客服机器人、自动化工具，还是学习 Webhook 原理，这篇文章都能为你打下坚实基础。

准备工作：前置依赖与开发环境

在开始编码之前，请确保你具备以下条件：

• Python 3.7+ 环境，建议使用虚拟环境隔离项目依赖。
• Telegram Bot Token：通过 @BotFather 创建机器人并获取 Token。
• 可公网访问的 HTTPS 地址：Telegram 要求 Webhook URL 必须为 HTTPS。开发阶段可使用 ngrok 或 localtunnel 将本地端口映射到公网。
• 基础 Flask 知识：了解路由和请求对象即可。

安装依赖：

pip install flask requests

创建项目目录并编写入口文件 app.py。

第一步：创建最小 Flask 应用

我们先构建一个最简单的 Flask 应用，用于接收 Telegram 的 POST 请求。Telegram 会以 JSON 格式发送更新到一个指定的端点（例如 /webhook）。

from flask import Flask, request, jsonify
import json

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def webhook():
    update = request.get_json()
    # 打印更新内容，便于调试
    print(json.dumps(update, indent=2, ensure_ascii=False))
    return jsonify({'ok': True})

if __name__ == '__main__':
    app.run(port=5000, debug=True)

运行此应用后，访问 http://localhost:5000/webhook 会返回 404，因为只接受 POST 请求。接下来我们需要将 Telegram 的更新转发到此地址。

第二步：设置 Webhook 地址

使用 Telegram Bot API 的 setWebhook 方法，将你的机器人更新发送到 Flask 应用的公网 URL。命令如下：

curl -F "url=https://your-ngrok-url.ngrok.io/webhook" https://api.telegram.org/bot<YOUR_BOT_TOKEN>/setWebhook

若返回 true，则设置成功。你可以通过 getWebhookInfo 方法检查当前 Webhook 状态：

curl https://api.telegram.org/bot<YOUR_BOT_TOKEN>/getWebhookInfo

如果出现 SSL 证书错误，请确认你的公网地址确实为有效的 HTTPS。本地开发时，用 ngrok http 5000 获取临时域名即可。

第三步：安全验证：防止伪造请求

任何人都可以向你公开的 Webhook URL 发送 POST 请求，伪造更新。为了安全，必须验证请求确实来自 Telegram。Bot API 支持在 setWebhook 时指定一个 secret_token，Telegram 会在每次请求的 X-Telegram-Bot-Api-Secret-Token 头中携带该 token。我们只需校验这个头即可。

修改 setWebhook 命令：

curl -F "url=https://your-ngrok-url.ngrok.io/webhook" -F "secret_token=your-custom-secret" https://api.telegram.org/bot<YOUR_BOT_TOKEN>/setWebhook

然后在 Flask 中校验：

import os
from flask import abort

SECRET_TOKEN = os.getenv('WEBHOOK_SECRET', 'your-custom-secret')

@app.route('/webhook', methods=['POST'])
def webhook():
    received_token = request.headers.get('X-Telegram-Bot-Api-Secret-Token')
    if received_token != SECRET_TOKEN:
        abort(401)
    # 处理更新...
    return jsonify({'ok': True})

这样，只有携带正确 token 的请求才会被处理，大大增强了安全性。

第四步：解析并响应不同类型的更新

Telegram 的更新对象涵盖消息、回调查询、内联查询等。我们在 Flask 中可以根据 update 的字段进行路由。

处理文字消息

@app.route('/webhook', methods=['POST'])
def webhook():
    update = request.get_json()
    if 'message' in update:
        chat_id = update['message']['chat']['id']
        text = update['message'].get('text')
        print(f'收到来自  的消息: ')
        # 可在此处调用 sendMessage API 回复
    elif 'callback_query' in update:
        # 处理按钮回调
        callback = update['callback_query']
        user_id = callback['from']['id']
        data = callback['data']
        print(f'回调:  - ')
    return jsonify({'ok': True})

注意：为了不阻塞 Webhook 响应，建议将处理逻辑放到后台线程或任务队列中。Telegram 要求 Webhook 必须在 60 秒内响应，否则会重试。

一键回复：调用 sendMessage

在 Flask 中集成 requests 库发送回复消息：

import requests

TELEGRAM_API = f"https://api.telegram.org/bot"

def send_message(chat_id, text):
    requests.post(f"/sendMessage", json={
        'chat_id': chat_id,
        'text': text
    })

# 在 webhook 函数中调用 send_message(chat_id, '已收到！')

第五步：高级路由与多机器人支持

如果你的服务器需要同时处理多个机器人，可以在 URL 中加入机器人的标识符。例如，将 Webhook 设置为 https://your-domain.com/webhook/bot1 和 /webhook/bot2，然后在 Flask 中动态获取 token 和 secret。

BOTS = {
    'bot1': {'token': '111:token1', 'secret': 'secret1'},
    'bot2': {'token': '222:token2', 'secret': 'secret2'}
}

@app.route('/webhook/<bot_name>', methods=['POST'])
def webhook(bot_name):
    if bot_name not in BOTS:
        return jsonify({'ok': False, 'error': 'Bot not found'}), 404
    bot = BOTS[bot_name]
    received_token = request.headers.get('X-Telegram-Bot-Api-Secret-Token')
    if received_token != bot['secret']:
        abort(401)
    update = request.get_json()
    # 使用 bot['token'] 进行 API 调用
    return jsonify({'ok': True})

同时设置每个机器人的 Webhook 时，将 URL 指向对应的子路径即可。

第六步：实用技巧与调试建议

• 使用环境变量：防止 Token 和 Secret 硬编码，利用 os.getenv 管理。
• 日志记录：使用 Flask 的 app.logger 记录收到的更新和错误，便于排查问题。
• 快速重试机制：如果处理逻辑可能失败，建议先返回成功避免 Telegram 重复推送，再在后台处理。
• 压测与性能：Webhook 接收器应快速响应，避免在请求线程中执行耗时操作。可使用 Celery 或 Redis 队列解耦。
• 更新丢失问题：如果收到 409 错误，说明有另一个 Webhook 或轮询实例在运行，确保只有一个消费者。

第七步：完整示例：回显机器人

让我们将所有内容整合，构建一个简单的回显机器人。用户发送任何消息，机器人都原样返回。

import os
from flask import Flask, request, jsonify, abort
import requests

app = Flask(__name__)

TOKEN = os.getenv('BOT_TOKEN')
SECRET = os.getenv('WEBHOOK_SECRET')
API_URL = f"https://api.telegram.org/bot"

def send_message(chat_id, text):
    try:
        requests.post(f"/sendMessage", json={
            'chat_id': chat_id,
            'text': text
        }, timeout=10)
    except Exception as e:
        app.logger.error(f"Failed to send message: ")

@app.route('/webhook', methods=['POST'])
def webhook():
    if request.headers.get('X-Telegram-Bot-Api-Secret-Token') != SECRET:
        abort(401)
    update = request.get_json()
    if 'message' in update:
        msg = update['message']
        chat_id = msg['chat']['id']
        text = msg.get('text')
        if text:
            send_message(chat_id, f"你说了：")
    return jsonify({'ok': True})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

部署到公网后，设置 Webhook 和 secret_token，即可让你的机器人自动回显消息。

总结

本文详细介绍了如何利用 Flask 搭建 Telegram 自定义 Webhook 接收器，从环境准备、安全验证到高级路由，每一步都提供了可落地的代码示例。通过这种方式，你可以将 Telegram 机器人无缝集成到任何 Python 工作流中，实现自动化推送、智能客服、IoT 控制等无限可能。Webhook 不仅仅是接收更新，更是打通 Telegram 与外部世界的桥梁。掌握它，你的机器人开发能力将迈上一个新台阶。