Python实现AWS Lambda通过Websocket API向指定connectionId发消息

你预期的直接在Lambda返回值中携带connectionId列表的方式无法实现需求，AWS API Gateway WebSocket 不会自动根据返回值中的字段向指定连接推送消息，你需要主动调用API Gateway提供的Management API的post_to_connection方法向目标连接推送内容。

1. 前置IAM权限配置 #

首先需要为你的Lambda函数的执行角色添加如下权限，允许其调用WebSocket连接管理接口：

{
    "Effect": "Allow",
    "Action": "execute-api:ManageConnections",
    "Resource": "arn:aws:execute-api:<区域ID>:<AWS账号ID>:<WebSocket API ID>/<部署阶段>/POST/@connections/*"
}

上述占位符需要替换为你实际的资源信息。

2. Python Lambda 实现代码 #

Lambda运行环境默认自带boto3依赖，无需额外打包：

import json
import boto3
from botocore.exceptions import ClientError

# 初始化API Gateway管理客户端，替换为你自己的WebSocket API端点
apigw_mgmt_client = boto3.client(
    'apigatewaymanagementapi',
    endpoint_url='https://<API ID>.execute-api.<区域ID>.amazonaws.com/<部署阶段>'
)

def create_message(message, conversation):
    # 持久化消息到数据库
    save_message_in_db(message)
    # 获取当前会话所有参与者的connectionId
    target_conn_ids = conversation["participantsID"]
    
    # 遍历所有目标连接推送消息
    for conn_id in target_conn_ids:
        try:
            apigw_mgmt_client.post_to_connection(
                ConnectionId=conn_id,
                Data=json.dumps({"message": message})
            )
        except ClientError as e:
            if e.response['Error']['Code'] == 'GoneException':
                # 连接已失效，清理数据库中对应的无效connectionId
                delete_invalid_conn_from_db(conn_id)
            else:
                # 其他异常按需处理，比如记录错误日志
                print(f"向连接{conn_id}推送失败：{str(e)}")
    
    # 该返回值仅响应触发当前Lambda的客户端请求，和批量推送逻辑无关
    return {
        "statusCode": 200,
        "body": json.dumps({"status": "success"})
    }

注意事项 #

• endpoint_url可在API Gateway控制台的WebSocket API详情页中查询到
• 捕获GoneException的逻辑是必要的，避免因为单个用户断连未及时清理数据库，导致整个批量推送流程失败
• 如果单次需要推送的连接数较多，可以引入并发逻辑（比如concurrent.futures.ThreadPoolExecutor）提升推送效率

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