如何创建支持响应流式输出的Python自定义Runtime Lambda函数？

方案选择建议 #

• 优先选自定义Runtime（基于Bootstrap）：流式输出的核心是对接Lambda的响应流式API，这需要Bootstrap文件处理与Lambda服务的通信（包括获取请求、发送流式响应）。自定义Runtime模式直接围绕Bootstrap构建，逻辑更清晰，官方的流式输出指引也基于该模式。
• 容器部署作为补充：如果函数需要复杂依赖或自定义运行环境，容器方式可以打包整个环境，但本质上还是需要在容器中配置自定义Bootstrap作为入口，核心逻辑和自定义Runtime一致，只是部署方式不同。

自定义Runtime（适配Python）的具体操作步骤 #

1. 编写Python版Bootstrap脚本 #

Lambda的Bootstrap需要循环完成「获取请求→调用业务函数→返回流式响应」的流程，直接用Python编写即可（注意文件名必须是小写bootstrap，且添加执行权限），示例框架如下：

#!/usr/bin/env python3
import os
import sys
import json
import urllib3

# 初始化HTTP客户端，对接Lambda Runtime API
http = urllib3.PoolManager()
LAMBDA_RUNTIME_API = os.environ.get("AWS_LAMBDA_RUNTIME_API")

def invoke_handler(event):
    # 导入你的业务处理函数
    from your_handler import stream_handler
    # 返回生成器，实现流式输出逻辑
    return stream_handler(event)

def main():
    while True:
        # 获取下一个Lambda请求
        response = http.request("GET", f"http://{LAMBDA_RUNTIME_API}/2018-06-01/runtime/invocation/next")
        invocation_id = response.headers.get("Lambda-Runtime-Aws-Request-Id")
        event = json.loads(response.data.decode("utf-8"))

        try:
            # 调用流式处理函数，获取数据生成器
            stream_generator = invoke_handler(event)
            # 初始化流式响应连接
            resp_conn = http.request(
                "POST",
                f"http://{LAMBDA_RUNTIME_API}/2018-06-01/runtime/invocation/{invocation_id}/response",
                headers={"Transfer-Encoding": "chunked"},
                preload_content=False
            )
            # 逐块发送流式数据
            for chunk in stream_generator:
                resp_conn.body.write(chunk)
                resp_conn.body.flush()
            # 结束响应
            resp_conn.body.close()
        except Exception as e:
            # 异常处理，返回错误信息
            error_msg = json.dumps({"errorMessage": str(e)})
            http.request(
                "POST",
                f"http://{LAMBDA_RUNTIME_API}/2018-06-01/runtime/invocation/{invocation_id}/error",
                body=error_msg,
                headers={"Content-Type": "application/json"}
            )

if __name__ == "__main__":
    main()

2. 编写流式业务处理函数（your_handler.py） #

实现返回生成器的业务逻辑，模拟流式输出示例：

def stream_handler(event):
    # 模拟分块返回数据，可替换为实际业务逻辑
    content_blocks = ["Hello ", "this ", "is ", "a ", "streaming ", "response!"]
    for block in content_blocks:
        yield block.encode("utf-8")
        # 可选：添加延迟，测试流式效果
        import time
        time.sleep(0.5)

3. 打包部署 #

• 给bootstrap添加执行权限：chmod +x bootstrap
• 将bootstrap、your_handler.py以及依赖包（如果有）打包成zip文件
• 在Lambda控制台创建函数，选择「自定义Runtime」，上传zip包，配置Runtime为「提供自己的Runtime」

容器部署方式的Bootstrap配置 #

如果用容器部署，核心是将自定义Bootstrap设置为镜像入口点：

1. 基于Lambda Python基础镜像构建自定义镜像，Dockerfile示例：

FROM public.ecr.aws/lambda/python:3.11

# 复制业务代码和Bootstrap脚本
COPY your_handler.py ${LAMBDA_TASK_ROOT}
COPY bootstrap /var/runtime/bootstrap

# 给Bootstrap添加执行权限
RUN chmod +x /var/runtime/bootstrap

# 指定镜像入口为Bootstrap
ENTRYPOINT ["/var/runtime/bootstrap"]

2. 构建镜像并推送到ECR，然后在Lambda控制台创建容器镜像函数即可。

关键注意事项 #

• Bootstrap必须有执行权限，否则Lambda无法启动服务
• 流式响应必须通过Transfer-Encoding: chunked头分块发送，不能一次性返回全部内容
• 自定义Runtime打包时，需将依赖包（如urllib3）一同放入zip；容器部署则直接在Dockerfile中安装依赖

内容的提问来源于stack exchange，提问作者tobixoxo