从零构建发送邮件Agent Skill：AWS Lambda实战指南

2次阅读

共计 2402 个字符，预计需要花费 7 分钟才能阅读完成。

在构建现代应用时，邮件通知是刚需功能（如注册验证、订单确认）。传统方案需要维护邮件服务器，而 Serverless 架构让开发者只需关注业务逻辑：

无服务器管理 ：Lambda 自动扩容，按执行计费
事件驱动 ：可轻松对接 API Gateway、S3 等 AWS 服务
成本优势 ：低频场景下成本远低于常驻服务器

但挑战也很明显：

权限管控复杂（尤其跨服务访问）
冷启动可能导致延迟敏感场景超时
邮件服务有严格的发送限制和反垃圾策略

服务	优点	缺点
SES	原生 AWS 集成 / 高送达率 / 免费额度	新账户有发送限额
SendGrid	丰富的 API/ 模板功能	需管理第三方 API 密钥
SMTP 自建	完全可控	维护成本高 / 易进垃圾邮箱

推荐场景 ：
– 已使用 AWS 生态 → SES
– 需要高级邮件分析 → SendGrid
– 企业级合规要求 → 自建 +DKIM

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "ses:SendEmail",
        "ses:SendRawEmail"
      ],
      "Resource": "*"
    }
  ]
}

关键点 ：
– 生产环境应限制 Resource 为特定域名
– 通过策略变量动态限制发件人：arn:aws:ses:${AWS::Region}:${AWS::AccountId}:identity/yourdomain.com

import boto3
from email.utils import formataddr

def lambda_handler(event, context):
    # 从事件中提取参数
    recipient = event.get('email')
    subject = event.get('subject', '默认主题')
    body_text = event.get('body', '')

    # 初始化 SES 客户端
    ses = boto3.client('ses', region_name='us-east-1')

    try:
        response = ses.send_email(Source=formataddr(('发件人名称', 'no-reply@yourdomain.com')),
            Destination={'ToAddresses': [recipient]},
            Message={'Subject': {'Data': subject},
                'Body': {'Text': {'Data': body_text},
                    # 'Html': {'Data': '<h1>HTML 内容 </h1>'}  # 可选 HTML 版本
                }
            },
            # 配置集可用于跟踪打开率等
            # ConfigurationSetName='ConfigSet'
        )
        return {'messageId': response['MessageId']}
    except Exception as e:
        print(f"发送失败: {str(e)}")
        raise

优化技巧 ：
– 复用 boto3 客户端（放在函数外部）减少冷启动时间
– 使用 formataddr 规范发件人格式提升送达率

SES 模板方案 ：

在 AWS 控制台创建模板（含变量占位符）

{
  "TemplateName": "WelcomeEmail",
  "SubjectPart": "欢迎, {{name}}!",
  "HtmlPart": "<p> 您的验证码是: {{code}}</p>"
}

Lambda 调用时注入数据：

ses.send_templated_email(
    Source='...',
    Destination={'ToAddresses': [recipient]},
    Template='WelcomeEmail',
    TemplateData='{"name":" 李雷 ","code":"123456"}'
)

预置并发 ：为关键函数配置预置并发实例
精简依赖 ：
使用 AWS SDK 分层（减少部署包大小）
避免大型库（如 Pandas，改用内置 JSON 处理）
定时预热 ：通过 CloudWatch Events 每 5 分钟触发空调用

重试策略 ：
1. SES 限流错误（ThrottlingException）→ 指数退避重试
2. 临时故障 → 配置 DLQ（死信队列）异步处理
3. 硬错误（如无效邮箱）→ 记录到 DynamoDB 人工核查

示例代码片段 ：

from botocore.config import Config

retry_config = Config(
    retries={
        'max_attempts': 3,
        'mode': 'adaptive'  # 自动调整重试间隔
    }
)
ses = boto3.client('ses', config=retry_config)