如何在Lambda中调用Fargate镜像的handler并传递event与context？

核心原理 #

Lambda以容器模式运行时，会将event数据通过标准输入(stdin)传递给容器，而context的核心属性则通过环境变量提供（比如函数名、内存限制等），并非通过命令行参数传递。需要调整镜像的启动逻辑，让程序能读取stdin的event并从环境变量构造context，进而调用handler函数。

解决方案（两种可选） #

方案一：修改main.py，兼容Fargate和Lambda两种执行模式 #

保留原有的click命令逻辑，同时在入口处判断执行模式：如果没有命令行参数（Lambda调用场景），则读取stdin的event、从环境变量构造context，直接调用handler；如果有run参数（Fargate场景），则执行原有业务逻辑。

修改后的main.py示例：

import click
import orjson
import sys
import os

def handler(event, context):
    print("Handler called!")
    print(orjson.dumps(event))
    print(orjson.dumps(context))

@click.group()
def cli():
    pass

@cli.command()
def run():
    # 原有Fargate运行的业务逻辑
    print("Running in Fargate mode...")

if __name__ == "__main__":
    # 判断是否为Lambda调用：无命令行参数时进入handler模式
    if len(sys.argv) == 1:
        # 从标准输入读取event
        event_raw = sys.stdin.read()
        event = orjson.loads(event_raw)
        
        # 从环境变量构造context（按需添加所需属性）
        context = {
            "function_name": os.getenv("AWS_LAMBDA_FUNCTION_NAME"),
            "function_version": os.getenv("AWS_LAMBDA_FUNCTION_VERSION"),
            "memory_limit_in_mb": os.getenv("AWS_LAMBDA_FUNCTION_MEMORY_SIZE"),
            "request_id": os.getenv("AWS_LAMBDA_REQUEST_ID")
        }
        
        # 调用handler函数
        handler(event, context)
    else:
        # 原有Fargate模式，执行click命令
        cli()

调整Docker的ENTRYPOINT：

ENTRYPOINT [ "python3", "/app/main.py" ]

• Fargate侧仍使用原调用方式：python3 /app/main.py run，完全兼容原有逻辑
• Lambda侧启动容器时无额外参数，自动进入handler模式

方案二：新增启动脚本，隔离两种执行逻辑 #

创建独立的启动脚本start.py，作为容器入口，根据命令行参数判断执行模式：

start.py示例：

import sys
import os
import orjson
from main import handler, cli

def lambda_mode():
    # 读取stdin中的event数据
    event_raw = sys.stdin.read()
    event = orjson.loads(event_raw)
    
    # 从环境变量构造context
    context = {
        "function_name": os.getenv("AWS_LAMBDA_FUNCTION_NAME"),
        "function_version": os.getenv("AWS_LAMBDA_FUNCTION_VERSION"),
        "memory_limit_in_mb": os.getenv("AWS_LAMBDA_FUNCTION_MEMORY_SIZE"),
        "request_id": os.getenv("AWS_LAMBDA_REQUEST_ID")
    }
    
    handler(event, context)

if __name__ == "__main__":
    # 存在"run"参数时，执行Fargate原有逻辑
    if len(sys.argv) > 1 and sys.argv[1] == "run":
        cli()
    # 无参数时进入Lambda handler模式
    else:
        lambda_mode()

调整Docker的ENTRYPOINT：

ENTRYPOINT [ "python3", "/app/start.py" ]

• Fargate调用：python3 /app/start.py run，与原有流程一致
• Lambda调用：直接执行启动脚本，自动触发handler

注意事项 #

• 如果你的handler依赖Lambda context对象的方法（比如get_remaining_time_in_millis()），需要在构造的context字典中模拟这些方法，或者调整handler逻辑适配字典形式的context
• 测试Lambda调用时，SQS的消息会被封装到event的Records字段中，handler可以通过event["Records"]获取具体消息内容

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