Serverless Framework 实战：在 AWS Lambda 上部署 Python Telegram 回声机器人（Echo Bot）

本指南基于仓库中的 aws-python-telegram-bot 示例，完整讲解如何用 Serverless Framework 将基于 python-telegram-bot v21+ 的异步 Telegram 回声机器人部署到 AWS Lambda + API Gateway。读完本文，你将掌握 Telegram Bot 的 Webhook 接入原理、异步 Bot API 与同步 Lambda 处理器的桥接方式，并能从零开始完成「申请 Token → 部署 → 配置 Webhook → 对话验证」的完整闭环。

示例概览：一个基于 Webhook 的异步回声机器人

该示例实现的是一个经典 Telegram「回声机器人」（echo bot）：用户向机器人发送任意文本，机器人把同样的文本原样回复给用户；当收到 /start 命令时，则返回一段定制的欢迎语。整个服务按 Serverless 方式构建，运行在 AWS Lambda 上，由 API Gateway 通过 Webhook 模式把 Telegram 的更新事件推送给 Lambda 处理，无需常驻进程、无需固定公网 IP。

示例有两个关键设计点：

• 完全异步的 SDK：依赖 requirements.txt 中声明的 python-telegram-bot>=21。该版本 SDK 全面拥抱 async，Bot 实例需作为异步上下文管理器使用，所有 Bot API 调用都需要 await。
• 异步与同步的桥接：Lambda 的处理器入口是普通同步函数，示例通过在处理器内部调用 asyncio.run(...) 来驱动异步的 Bot API 调用，从而把两套编程模型优雅地衔接起来。

项目结构一览

整个示例文件极少、结构清晰，适合作为学习 Telegram Serverless 集成的起点：

文件	作用
handler.py	Lambda 处理器，包含 Webhook 接收、消息回显、Webhook 配置三个核心逻辑
serverless.yml	Serverless Framework 服务定义：函数、事件、运行时、环境变量
requirements.txt	Python 依赖声明：python-telegram-bot>=21
package.json	npm 元数据（当前未声明运行时依赖）
README.md	官方使用说明（本文的依据文档）
LICENSE	MIT 许可证

该示例在仓库中的定位是 Python 语言族的一个「即时可用（ready-to-deploy）」模板，其 README 头部的 frontmatter 注释（framework: v4、platform: AWS、language: Python）会被仓库的文档生成脚本自动读取，用于在根目录 README.md 的示例清单中登记。

核心代码解析：handler.py

整个服务的业务逻辑都集中在 handler.py 中，可以拆解为五个职责明确的函数。

1. 日志初始化与统一响应结构

import asyncio
import json
import logging
import os

import telegram

logger = logging.getLogger()
if logger.handlers:
    for handler in logger.handlers:
        logger.removeHandler(handler)
logging.basicConfig(level=logging.INFO)

OK_RESPONSE = {
    'statusCode': 200,
    'headers': {'Content-Type': 'application/json'},
    'body': json.dumps('ok')
}
ERROR_RESPONSE = {
    'statusCode': 400,
    'body': json.dumps('Oops, something went wrong!')
}

代码先重置 Logger 的已有 handler 并设置 INFO 级别，保证 Lambda 的 CloudWatch 日志清晰可读；随后定义了两个统一的 API Gateway 响应结构：OK_RESPONSE（200，Content-Type: application/json，body 为 ok）和 ERROR_RESPONSE（400，body 为错误提示）。后续所有分支都复用这两个常量，让返回路径保持简洁一致。

2. 读取 Token：get_telegram_token

def get_telegram_token():
    telegram_token = os.environ.get('TELEGRAM_TOKEN')
    if not telegram_token:
        logger.error('The TELEGRAM_TOKEN must be set')
        raise NotImplementedError
    return telegram_token

Token 通过环境变量 TELEGRAM_TOKEN 注入（具体注入方式见下文 serverless.yml 的 environment 配置）。函数在 Token 缺失时记录错误日志并抛出 NotImplementedError，属于快速失败（fail-fast）策略，避免带着空 Token 继续请求 Telegram Bot API。

3. 异步发送消息与设置 Webhook

async def send_message(chat_id, text):
    async with telegram.Bot(get_telegram_token()) as bot:
        await bot.send_message(chat_id=chat_id, text=text)


async def configure_webhook(url):
    async with telegram.Bot(get_telegram_token()) as bot:
        return await bot.set_webhook(url)

这是示例中最能体现 python-telegram-bot v21+ 异步特性的部分：

• telegram.Bot(token) 被用作异步上下文管理器（async with），其职责是自动完成底层 HTTP 会话的打开与关闭；
• 每一次 Bot API 调用（send_message、set_webhook）都必须显式 await；
• 两个函数都是短生命周期设计：每次调用创建一个临时 Bot 实例，用后即弃，不需要在 Lambda 全局维护长连接（从源码结构看，这也意味着每次调用都会重建 HTTP 会话，对冷启动延迟有一定影响——这是示例为了简洁所做的取舍，生产环境可考虑在全局复用 Bot 实例）。

4. 入口一：webhook —— 接收并回显消息

def webhook(event, context):
    logger.info('Event: {}'.format(event))

    if event.get('httpMethod') == 'POST' and event.get('body'):
        logger.info('Message received')
        update = telegram.Update.de_json(json.loads(event.get('body')), None)
        chat_id = update.message.chat.id
        text = update.message.text

        if text == '/start':
            text = """Hello, human! ..."""

        asyncio.run(send_message(chat_id, text))
        logger.info('Message sent')

        return OK_RESPONSE

    return ERROR_RESPONSE

webhook 对应 serverless.yml 中绑定的 POST / 事件，处理流程如下：

1. 校验请求必须为 POST 且携带 body，否则直接返回 ERROR_RESPONSE；
2. 用 telegram.Update.de_json 把 API Gateway 传入的 JSON 请求体解析为 Update 对象；
3. 从 update.message 中取出 chat.id（会话标识）和 text（用户消息文本）；
4. 若文本是 /start，替换为一段欢迎语（这也是机器人对外展示的「自我介绍」入口）；
5. 通过 asyncio.run(send_message(chat_id, text)) 把普通同步函数桥接到异步调用，向同一个 chat 回显文本；
6. 成功则返回 OK_RESPONSE。

5. 入口二：set_webhook —— 动态配置 Webhook

def set_webhook(event, context):
    logger.info('Event: {}'.format(event))
    url = 'https://{}/{}/'.format(
        event.get('headers').get('Host'),
        event.get('requestContext').get('stage'),
    )
    webhook_set = asyncio.run(configure_webhook(url))

    if webhook_set:
        return OK_RESPONSE

    return ERROR_RESPONSE

set_webhook 对应 serverless.yml 中绑定的 POST /set_webhook 事件。它的巧妙之处在于不需要手动粘贴部署后生成的 URL：函数从 API Gateway 事件中读取 headers.Host（即网关的公共域名）和 requestContext.stage（部署阶段，默认 dev），自动拼出形如 https://<your_url>.amazonaws.com/dev/ 的 Webhook 地址，再调用 Telegram Bot API 的 set_webhook 接口完成注册。

6. 完整调用链

Telegram 用户发消息
   │ (Telegram Bot API 回调)
   ▼
API Gateway POST / (公开 HTTPS 端点)
   ▼
Lambda webhook(event, context)
   ├─ 校验 POST 与 body
   ├─ Update.de_json 解析消息
   ├─ asyncio.run(send_message(chat_id, text))
   └─ 返回 OK_RESPONSE (200)

部署时手动触发：
curl -X POST https://<your_url>.amazonaws.com/dev/set_webhook
   ▼
Lambda set_webhook(event, context)
   ├─ 由 Host + stage 拼出 Webhook URL
   ├─ asyncio.run(configure_webhook(url))
   └─ 返回 OK_RESPONSE (200)

serverless.yml 服务配置逐项解读

serverless.yml 是 Serverless Framework 的服务定义文件，全文如下：

service: serverless-telegram-bot

frameworkVersion: '4'

package:
  patterns:
    - '!node_modules/**'

provider:
  name: aws
  runtime: python3.14
  architecture: arm64
  environment:
    TELEGRAM_TOKEN: ${file(./serverless.env.yml):TELEGRAM_TOKEN, ''}

functions:
  webhook:
    handler: handler.webhook
    events:
      - http: POST /

  set_webhook:
    handler: handler.set_webhook
    events:
      - http: POST /set_webhook

custom:
  # Built-in python requirements packaging activates on this key.
  pythonRequirements: {}

逐项说明：

• service / frameworkVersion：服务名 serverless-telegram-bot，且显式声明使用 Serverless Framework '4'。
• package.patterns：排除 node_modules/**，避免无关文件被打进 Lambda 部署包。
• provider.runtime：python3.14，即 Lambda 的 Python 运行时（部署时需确认目标区域已提供该运行时版本）。
• provider.architecture：arm64，Lambda 采用 Arm 架构（Graviton），通常成本更低。
• provider.environment.TELEGRAM_TOKEN：这里用到了 Serverless 的变量解析语法 ${file(./serverless.env.yml):TELEGRAM_TOKEN, ''}——从同目录的 serverless.env.yml 读取 TELEGRAM_TOKEN 键的值注入环境变量，'' 是兜底默认值（文件或键缺失时不报错）。这就是 handler.py 中 os.environ.get('TELEGRAM_TOKEN') 的数据来源。注意 serverless.env.yml 应纳入 .gitignore，避免 Token 泄漏到版本库。
• functions.webhook：处理器 handler.webhook，事件为 POST /，即 API Gateway 的根路径，用于接收 Telegram 的消息回调。
• functions.set_webhook：处理器 handler.set_webhook，事件为 POST /set_webhook，用于部署后注册 Webhook。
• custom.pythonRequirements：按文件内注释，Serverless Framework v4 的内置 Python 依赖打包能力在检测到该键时自动激活，会依据 requirements.txt 把 python-telegram-bot 等依赖打入部署包。这一点与 package.json 未声明任何 dependencies 的现状一致——README 中的 npm install 步骤在当前版本下实际不会安装额外插件，保留它主要是兼容旧版 serverless-python-requirements 插件的工作流。

部署前置条件

按照 README.md 的要求，开始前需要准备：

• 本地配置好的 AWS 凭证：确保本地环境已配置 AWS Key（可参考 Serverless Framework 官方的 AWS 凭证指南），否则 serverless deploy 无法创建 CloudFormation 栈。
• Node.js：用于安装 Serverless Framework CLI。
• 一个 Telegram 账号：用于通过 @BotFather 创建机器人并获取 Token。

从零到一的部署实战

以下步骤完整继承自示例的 使用说明，并补充了必要的背景解释。

步骤 1：安装 Serverless Framework

npm install serverless -g

全局安装 Serverless Framework CLI。

步骤 2：安装依赖

npm install

进入示例目录执行（如 aws-python-telegram-bot），如上一节所述，当前 package.json 未声明 npm 依赖，该步骤主要是兼容旧版插件流程，执行无副作用。

步骤 3：向 @BotFather 申请机器人

在 Telegram 中向 @BotFather 发送：

/newbot

按提示为机器人命名，随后会收到一个形如 123456789:ABCdef... 的 Bot Token，请妥善保存。

步骤 4：把 Token 写入 serverless.env.yml

在 aws-python-telegram-bot 目录下创建 serverless.env.yml：

cat serverless.env.yml
TELEGRAM_TOKEN: <your_token>

该文件会被 serverless.yml 中的 ${file(...)} 语法读取，把 Token 注入 Lambda 环境变量。

步骤 5：部署服务

serverless deploy

部署完成后，Serverless CLI 会在输出中返回 API Gateway 的端点 URL，形如 https://<your_url>.amazonaws.com/dev。以该 URL 作为 <your_url> 参与下一步。

步骤 6：配置 Webhook

curl -X POST https://<your_url>.amazonaws.com/dev/set_webhook

这一步调用我们部署的第二个 Lambda 函数。它内部会自动读取 API Gateway 的 Host 与 stage，把 Webhook 注册到 Telegram 指向部署好的端点。

步骤 7：开始对话验证

在 Telegram 中给机器人发任意文本，机器人会原样回显；发送 /start，则收到定制欢迎语。至此整个闭环打通。

Webhook 自动配置原理

在常见的 Telegram Serverless 教程里，开发者往往需要手动把 API Gateway 的完整 URL 粘贴到 Telegram Bot API 的 setWebhook 接口。本示例则把这一步自动化了，值得单独理解：

• API Gateway 为每次 HTTP 请求注入包含 headers.Host（网关公共域名）和 requestContext.stage（部署阶段名）的 event；
• set_webhook 函数 据此拼出 https://{Host}/{stage}/，例如 https://xxxx.execute-api.us-east-1.amazonaws.com/dev/（末尾带斜杠，对应 POST / 根路径端点）；
• 再通过 asyncio.run(configure_webhook(url)) 调用 bot.set_webhook(url) 完成注册，并根据返回值决定返回 OK_RESPONSE 还是 ERROR_RESPONSE。

由于 API Gateway 天然以 HTTPS 暴露端点，无需额外配置即可满足 Telegram Bot API 对 Webhook 使用 HTTPS 的要求。这一设计让「部署 + 注册」两步操作都可以脚本化，适合 CI/CD 流程。

扩展与生产化建议

从源码结构看，本示例是一个最小可用（minimal viable）演示，若将其用于生产环境，可考虑以下增强方向：

• 请求来源校验：webhook 端点对外完全公开，且未校验 Telegram 官方签名。Telegram Bot API 支持在设置 Webhook 时指定 secret_token，回调请求会附带 X-Telegram-Bot-Api-Secret-Token 请求头，可在 handler.py 的入口处校验该头以拒绝伪造请求。
• 消息类型覆盖：当前逻辑只读取 update.message.text，对图片、语音、回调按钮等消息类型未做分支处理；若业务需要，可按 update.message 下的不同字段扩展。
• 健壮性加固：set_webhook 直接链式调用 event.get('headers').get('Host')，在缺头场景下会抛异常，可增加取值兜底；同时建议为函数配置 CloudWatch 告警，及时发现异常返回。
• 本地调试：可结合 serverless-offline 插件在本地模拟 API Gateway 环境，快速验证 Webhook 解析与回显逻辑，无需每次改动都完整部署。

如果你更熟悉 Node.js，本仓库还提供了同主题的兄弟示例 aws-node-telegram-echo-bot，两者架构与部署流程完全对应，可对照学习。

总结

aws-python-telegram-bot 用极少的代码和配置，展示了 Serverless 与聊天机器人集成的完整范式：API Gateway 暴露 HTTPS 端点接收 Webhook → Lambda 同步处理器通过 asyncio.run 桥接异步 Bot API → set_webhook 函数自动完成端点注册。其中环境变量注入（${file(...)}）、内置 Python 依赖打包（custom.pythonRequirements）、以及异步 SDK 在 Lambda 中的桥接写法，都是可以直接复用到其他 Telegram/Serverless 项目的关键技能点。