使用 Python 和 JavaScript 实现 Telegram Bot 并部署到 AWS

并部署新的 Telegram 机器人到 AWS

目录

以下是我在如何实现并部署到AWS的Telegram机器人方面的笔记和逐步教程。
我添加了快速入门（长轮询）和生产就绪路径（webhook），并提供了Python和Node.js的示例。

机器人在IT领域工作

Telegram机器人是在您的服务器上运行的应用程序，通过Telegram与用户进行交流。
我们将注册一个机器人，编写一个小程序，并将其连接到Telegram的Bot API。

TL;DR

  1. 通过 @BotFather 创建机器人 → 获取令牌。
  2. 使用 python-telegram-bot 或 Telegraf 构建一个小应用程序。
  3. 从本地的 长轮询 开始；在生产环境中切换到 webhook。

你需要的东西

  • 一个Telegram账户（手机或桌面）

  • 基本终端 + 代码编辑器

  • 以下之一：

    • Python 3.10+ 和 pip
    • Node.js 18+ 和 npm/pnpm/yarn

使用BotFather创建你的机器人（获取令牌）

  1. 在Telegram中搜索 @BotFather 并开始聊天。
  2. 发送 /newbot 并按照提示命名你的机器人并选择一个唯一的用户名（必须以 bot 结尾），例如 example_helper_bot。
  3. BotFather会回复一个像 123456:ABC-DEF... 的 API令牌。请将其视为密码；不要将其提交到Git中。如果泄露，你总是可以通过BotFather重新生成它。

提示：Telegram的官方教程会逐步引导你完成这些步骤，并说明如何使用 getMe 验证你的令牌。

选择你的技术栈

你可以自己调用原始的HTTP Bot API，但使用维护良好的库会更快。

  • Python: python-telegram-bot（异步、现代）。([https://docs.python-telegram-bot.org])
  • Node.js: telegraf（成熟、TypeScript友好）。([https://telegraf.js.org])

Bot API本身在这里有文档，并且是 getUpdates 和 setWebhook 等方法的权威来源。

快速入门：使用长轮询在本地运行

长轮询非常适合本地开发：你的机器人通过 getUpdates 反复向Telegram请求新更新。（在生产环境中，你可能会切换到webhook。）

Telegram支持两种互斥的方式来接收更新：getUpdates（轮询）或 webhook；更新最多保存24小时。

选项A：Python（使用 python-telegram-bot）

安装与搭建

python -m venv .venv && source .venv/bin/activate  # Windows: .venv\Scripts\activate
pip install python-telegram-bot

代码：bot.py

import os
from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, ContextTypes, filters

TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]  # export TELEGRAM_BOT_TOKEN=123:ABC

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text("👋 你好！给我发点什么，我会原样返回。")

async def echo(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text(update.message.text)

def main():
    app = ApplicationBuilder().token(TOKEN).build()
    app.add_handler(CommandHandler("start", start))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo))
    app.run_polling()  # 长轮询

if __name__ == "__main__":
    main()

运行它：

export TELEGRAM_BOT_TOKEN=123456:ABC-DEF...
python bot.py

这使用了当前稳定文档中的 ApplicationBuilder。

选项B：Node.js（使用 telegraf）

安装与搭建

npm init -y
npm i telegraf

代码：index.js

const { Telegraf } = require('telegraf');

const bot = new Telegraf(process.env.BOT_TOKEN); // 设置 BOT_TOKEN 环境变量
bot.start((ctx) => ctx.reply('👋 你好！给我发点什么，我会原样返回。'));
bot.on('text', (ctx) => ctx.reply(ctx.message.text));

bot.launch();

// 优雅关闭
process.once('SIGINT', () => bot.stop('SIGINT'));
process.once('SIGTERM', () => bot.stop('SIGTERM'));

运行它：

export BOT_TOKEN=123456:ABC-DEF...
node index.js

这与官方Telegraf的“入门”示例一致。

让机器人更实用（命令与按钮）

添加 /help 命令

Python

from telegram.ext import CommandHandler

async def help_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text("/start – 问候\n/help – 帮助")

app.add_handler(CommandHandler("help", help_cmd))

CommandHandler 是绑定 /命令 的标准方式。

Node（Telegraf）

bot.help((ctx) => ctx.reply('/start – 问候\n/help – 帮助'));

快速回复按钮（自定义键盘）

Python

from telegram import ReplyKeyboardMarkup

async def start(update: Update, context):
    keyboard = [["帮助", "关于"]]
    await update.message.reply_text(
        "选择一个选项：",
        reply_markup=ReplyKeyboardMarkup(keyboard, resize_keyboard=True)
    )

Node（Telegraf）

const { Markup } = require('telegraf');
bot.start((ctx) => {
  ctx.reply("选择一个选项：", Markup.keyboard([["帮助", "关于"]]).resize());
});

生产环境：切换到webhook

在生产环境中，Telegram会推送更新到你的HTTPS端点。你可以通过 setWebhook 设置，可选地提供一个秘密令牌，以便通过 X-Telegram-Bot-Api-Secret-Token 头验证请求。

使用 curl 设置webhook：

TOKEN=123456:ABC-DEF...
WEBHOOK_URL="https://your-domain.com/telegram/${TOKEN}"  # 包含一个秘密路径
SECRET="my-secret-42"

curl -X POST "https://api.telegram.org/bot${TOKEN}/setWebhook" \
  -d url="${WEBHOOK_URL}" \
  -d secret_token="${SECRET}" \
  -d drop_pending_updates=true

规范中的关键点：

  • setWebhook 需要一个 HTTPS URL 并支持 secret_token。
  • 当设置webhook时，你不能使用 getUpdates。
  • 支持的端口包括 443, 80, 88, 8443。

webhook服务器选项

  • Node（Telegraf）内置webhook：
bot.launch({
  webhook: {
    domain: "your-domain.com",
    port: 8080,
    // 可选的秘密头
    secretToken: process.env.WEBHOOK_SECRET
  }
});

Telegraf提供了一流的webhook选项，并与Express/Fastify/Cloudflare Workers等集成。

  • Python（python-telegram-bot） PTB与ASGI/WSGI框架（FastAPI, Starlette, Flask）配合良好。使用 ApplicationBuilder().token(...).build()，暴露一个POST路由，将传入的JSON更新输入到你的应用程序中，并调用 bot.set_webhook(...)。有关构建模式，请参阅 ApplicationBuilder 文档。([docs.python-telegram-bot.org][2])

本地测试？使用隧道（例如 ngrok）将 https://... 暴露给Telegram，然后使用公开URL调用 setWebhook。

配置机器人命令（可选但对用户友好）

你可以通过以下方式定义机器人命令列表（用户在输入 / 时看到的内容）：

  • 在 BotFather 中通过 /mybots → 编辑机器人 → 编辑命令，或
  • 使用 setMyCommands 通过你的库或原始Bot API编程实现。

Telegram的官方“从BotFather到‘Hello World’”指南链接到更高级的命令处理和示例，如果你想要深入。

部署注意事项

任何支持HTTPS的主机都可以：

  • 一个小型虚拟机（Ubuntu + systemd）
  • 无服务器（AWS Lambda/Cloud Functions）与webhook集成
  • 在Fly.io/Render/Heroku等平台上的容器

Telegraf的文档包括Lambda、GCF、Express、Fastify等的示例。

安全与可靠性检查清单

  • 保持令牌秘密（环境变量、密钥管理器）。如果泄露，通过BotFather撤销。
  • 使用 secret_token 与webhook并验证 X-Telegram-Bot-Api-Secret-Token 头。
  • 不要混合 轮询和webhook；一次选择一个。
  • 处理错误与重试：Telegram会重试非2xx的webhook响应。适当记录日志。
  • 注意更新类型（消息、回调、内联查询、支付等）并只订阅你需要的（allowed_updates）。

直接使用HTTP API（可选）

你可以调用如下端点：

https://api.telegram.org/bot<token>/getMe
https://api.telegram.org/bot<token>/getUpdates
https://api.telegram.org/bot<token>/sendMessage

根据规范使用GET或POST，使用JSON或表单数据。

下一步要做什么和阅读什么

  • 官方Bot API参考： 方法、对象、限制、格式、支付、内联模式。
  • 官方“从BotFather到‘Hello World’”指南： 更深入的走查和多语言示例。
  • python-telegram-bot文档（稳定版）： 现代异步模式。
  • Telegraf文档： 快速配方、webhook助手和TS类型。

部署到AWS的Python版本步骤

我们有两个主要选项用于将Telegram机器人部署到AWS基础设施：

  • A) 无服务器（API Gateway + Lambda + Secrets Manager） — 最便宜/最容易运行，适合中等流量。
  • B) 容器化（ECS Fargate + ALB + ACM） — 适合稳定流量和长期运行的库（如 python-telegram-bot 在webhook模式下）。

A) AWS上的无服务器（API Gateway + Lambda）

当你想要零服务器和接近零的空闲成本时使用此方法。下面的代码直接处理Telegram的webhook（没有长时间运行的事件循环）。

  1. 准备一个最小的Lambda处理程序（Python）

创建 handler.py：

import json, os, urllib.request

BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
SECRET_HEADER = os.environ.get("WEBHOOK_SECRET", "")  # 必须与配置webhook时的secret_token匹配

API_BASE = f"https://api.telegram.org/bot{BOT_TOKEN}"

def reply(chat_id: int, text: str):
    data = json.dumps({"chat_id": chat_id, "text": text}).encode()
    req = urllib.request.Request(f"{API_BASE}/sendMessage", data=data, headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req) as resp:
        return resp.read()

def lambda_handler(event, context):
    # 1) 验证Telegram的secret header（在配置webhook时设置）
    if SECRET_HEADER:
        if event.get("headers", {}).get("X-Telegram-Bot-Api-Secret-Token") != SECRET_HEADER:
            return {"statusCode": 401, "body": "无效的secret"}

    # 2) 解析update
    body = event.get("body") or "{}"
    update = json.loads(body)

    message = update.get("message") or update.get("edited_message")
    if not message:
        # 忽略非消息更新（callback_query, inline_query等）目前
        return {"statusCode": 200, "body": "ok"}

    chat_id = message["chat"]["id"]
    text = (message.get("text") or "").strip()

    # 3) 简单路由
    if text.startswith("/start"):
        reply(chat_id, "👋 来自AWS Lambda的问候！给我发任何文本，我会原样返回。")
    elif text.startswith("/help"):
        reply(chat_id, "/start – 问候\n/help – 帮助\n（部署在AWS Lambda上）")
    elif text:
        reply(chat_id, text)  # 原样返回

    return {"statusCode": 200, "body": "ok"}

这使用原始的HTTPS调用Bot API来保持Lambda轻量且冷启动友好。你可以以后扩展路由。

  1. 打包并部署Lambda
# 项目布局
# .
# ├─ handler.py
# └─ requirements.txt   # （此最小示例留空）

zip -r function.zip handler.py
aws lambda create-function \
  --function-name telegram-bot-webhook \
  --runtime python3.11 \
  --role arn:aws:iam::<ACCOUNT_ID>:role/<LambdaExecutionRole> \
  --handler handler.lambda_handler \
  --timeout 10 \
  --memory-size 256 \
  --zip-file fileb://function.zip \
  --environment Variables='{TELEGRAM_BOT_TOKEN=<TOKEN_FROM_BOTFATHER>,WEBHOOK_SECRET=my-secret-42}'

IAM角色 (<LambdaExecutionRole>) 需要 AWSLambdaBasicExecutionRole 用于CloudWatch日志。

更喜欢将你的令牌存储在 AWS Secrets Manager 中并在初始化时加载它—此示例为了简洁使用环境变量。

  1. 创建HTTPS端点（API Gateway）
# HTTP API（非REST）以降低延迟
API_ID=$(aws apigatewayv2 create-api \
  --name telegram-webhook \
  --protocol-type HTTP \
  --target arn:aws:lambda:<REGION>:<ACCOUNT_ID>:function:telegram-bot-webhook \
  --query 'ApiId' --output text)

# 添加默认路由 POST /webhook
aws apigatewayv2 create-integration \
  --api-id $API_ID \
  --integration-type AWS_PROXY \
  --integration-uri arn:aws:lambda:<REGION>:<ACCOUNT_ID>:function:telegram-bot-webhook \
  --payload-format-version 2.0

aws apigatewayv2 create-route \
  --api-id $API_ID \
  --route-key "POST /webhook" \
  --target "integrations/$(
    aws apigatewayv2 get-integrations --api-id $API_ID --query 'Items[0].IntegrationId' --output text
  )"

aws apigatewayv2 create-deployment --api-id $API_ID
GW_URL=$(aws apigatewayv2 get-apis --query "Items[?ApiId=='$API_ID'].ApiEndpoint" --output text)
echo $GW_URL  # 例如，https://abc123.execute-api.ap-somewhere.amazonaws.com

确保 Lambda权限 自动添加；如果没有：

aws lambda add-permission \
  --function-name telegram-bot-webhook \
  --statement-id apigw \
  --action lambda:InvokeFunction \
  --principal apigateway.amazonaws.com \
  --source-arn "arn:aws:execute-api:<REGION>:<ACCOUNT_ID>:$API_ID/*/*/webhook"
  1. 将Telegram指向你的webhook
TOKEN=<TOKEN_FROM_BOTFATHER>
WEBHOOK_URL="$GW_URL/webhook"
SECRET="my-secret-42"

curl -X POST "https://api.telegram.org/bot${TOKEN}/setWebhook" \
  -d url="${WEBHOOK_URL}" \
  -d secret_token="${SECRET}" \
  -d drop_pending_updates=true

向你的机器人发送 /start — 消息将通过API Gateway → Lambda。

  1. 日志、重试和更新
  • 查看日志：aws logs tail /aws/lambda/telegram-bot-webhook --follow
  • Telegram在你的端点不是2xx时会重试；保持响应快速。
  • 要部署新代码：重新打包 + aws lambda update-function-code --function-name telegram-bot-webhook --zip-file fileb://function.zip。

B) 在ECS Fargate + ALB上运行容器化的 python-telegram-bot

当你想要 python-telegram-bot（PTB）的易用性（PTB）与异步应用和持久连接时使用此方法。我们将PTB运行在HTTPS负载均衡器后面。

  1. 应用代码（FastAPI + PTB webhook）

app.py：

import os, asyncio
from fastapi import FastAPI, Request, Header
from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, ContextTypes, filters

TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
SECRET = os.environ.get("WEBHOOK_SECRET", "")

app = FastAPI()
tg_app = None  # PTB 应用

async def start_cmd(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text("🚀 来自ECS Fargate + PTB的问候！")

async def echo(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text(update.message.text)

@app.on_event("startup")
async def on_startup():
    global tg_app
    tg_app = ApplicationBuilder().token(TOKEN).build()
    tg_app.add_handler(CommandHandler("start", start_cmd))
    tg_app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo))
    # 不调用 run_polling(); 我们将手动处理webhook。

@app.post("/telegram/webhook")
async def telegram_webhook(request: Request, x_telegram_bot_api_secret_token: str | None = Header(default=None)):
    if SECRET and (x_telegram_bot_api_secret_token != SECRET):
        return {"ok": True}  # 静默忽略

    data = await request.json()
    update = Update.de_json(data, tg_app.bot)
    await tg_app.process_update(update)
    return {"ok": True}

uvicorn 入口点：uvicorn app:app --host 0.0.0.0 --port 8080

  1. Dockerfile
FROM python:3.11-slim
WORKDIR /app
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1
RUN pip install --no-cache-dir fastapi uvicorn[standard] python-telegram-bot==21.*
COPY app.py .
EXPOSE 8080
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8080"]

构建并推送：

aws ecr create-repository --repository-name telegram-ptb || true
ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
REGION=<REGION>
ECR="$ACCOUNT_ID.dkr.ecr.$REGION.amazonaws.com/telegram-ptb"

aws ecr get-login-password --region $REGION | docker login --username AWS --password-stdin "$ACCOUNT_ID.dkr.ecr.$REGION.amazonaws.com"
docker build -t telegram-ptb .
docker tag telegram-ptb:latest $ECR:latest
docker push $ECR:latest
  1. ECS Fargate服务在ALB后面
  • 创建一个 安全组 允许ALB的 443 端口入站；ECS任务允许 8080 端口从ALB安全组。
  • 创建一个 应用负载均衡器（ALB）带有 HTTPS监听器（443） 和一个 ACM证书 用于你的域名。
  • 目标组：HTTP 8080（IP目标类型），健康检查路径 /（FastAPI返回404；你可以添加 @app.get("/") 健康路由）。

创建一个 任务定义（Fargate）：

  • 容器镜像：$ECR:latest
  • 端口映射：8080
  • 环境变量：TELEGRAM_BOT_TOKEN，WEBHOOK_SECRET
  • 任务角色：基本的CloudWatch日志。

创建一个 ECS服务：

  • 启动类型 Fargate，期望数量 1+。
  • 附加到ALB目标组。

注意ALB的公共 HTTPS域名（或使用Route 53将你的DNS名称指向）。

  1. 告诉Telegram你的webhook URL
TOKEN=<TOKEN_FROM_BOTFATHER>
SECRET="my-secret-42"
WEBHOOK_URL="https://your.domain.com/telegram/webhook"

curl -X POST "https://api.telegram.org/bot${TOKEN}/setWebhook" \
  -d url="${WEBHOOK_URL}" \
  -d secret_token="${SECRET}" \
  -d drop_pending_updates=true \
  -d allowed_updates='["message","edited_message","callback_query"]'
  1. 扩展与操作
  • 扩展ECS任务上下；ALB将分散流量。
  • 在 Secrets Manager 中轮换令牌，使用新任务环境更新服务。
  • 使用 CloudWatch Logs 用于应用日志和 ALB访问日志（S3）。

我们应该选择哪一个？

  • Lambda + API Gateway： 最简单、最便宜（低流量时）；适合每分钟只进行几次调用的机器人。
  • ECS Fargate + ALB： 当你想要完整的 python-telegram-bot 体验、自定义中间件、后台作业和稳定流量时最佳选择。

快速检查清单（两种方法）

  • 使用 HTTPS 端点 + secret_token 并验证 X-Telegram-Bot-Api-Secret-Token 头。
  • 选择 webhook 或 轮询（不要两者都用）。
  • 将配置持久化在 Secrets Manager 中，而不是代码中。
  • 添加 可观测性：CloudWatch Logs + 指标（5xx警报）。
  • 仅处理你需要的 更新类型；以后可以扩展。

有用的链接