发送邮件agent skill的实现原理与生产环境最佳实践

3次阅读

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

邮件发送看似基础的功能，在实际生产环境中却面临诸多挑战。以下是开发过程中最常见的三类问题：

连接管理难题 ：高频创建 SMTP 连接会导致服务器主动断开，而未正确关闭的连接会逐渐耗尽系统资源
性能瓶颈 ：同步发送模式在批量处理时会产生严重延迟，特别是带有大附件的邮件可能阻塞整个发送队列
安全风险 ：缺乏 DKIM 等认证的邮件容易被标记为垃圾邮件，某些云服务商默认屏蔽未加密的 25 端口通信

维度	原生 SMTP 方案	第三方邮件 API
开发成本	需要自主实现所有底层协议逻辑	简单 HTTP 调用即可完成
发送速度	依赖自建服务器性能	通常具有全球加速节点
可达率	需自行处理各类拦截策略	内置白名单和信誉度管理
费用模型	仅服务器成本	按发送量阶梯计费
功能扩展性	可完全自定义	受限于提供商功能

对于日发送量低于 1 万封的中等规模应用，推荐采用混合架构：

核心业务通知使用 SendGrid 等专业 API 确保到达率
非关键邮件（如营销通知）通过自建 SMTP 服务发送
两种通道统一通过抽象层对接，便于后期切换

import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.utils import formatdate

class SMTPAgent:
    def __init__(self, host, port, username, password, use_tls=True):
        self.pool = []  # 连接池
        self.host = host
        self.port = port
        self.auth = (username, password)
        self.use_tls = use_tls

    def _get_connection(self):
        """智能获取连接：优先使用池中空闲连接"""
        try:
            return self.pool.pop()
        except IndexError:
            conn = smtplib.SMTP(self.host, self.port, timeout=30)
            if self.use_tls:
                conn.starttls()
            conn.login(*self.auth)
            return conn

    def _release_connection(self, conn):
        """归还连接到池"""
        try:
            # 测试连接是否仍然有效
            conn.noop()
            self.pool.append(conn)
        except:
            conn.close()

    def send(self, to, subject, body, attachments=None):
        msg = MIMEMultipart()
        msg['From'] = self.auth[0]
        msg['To'] = ','.join(to) if isinstance(to, list) else to
        msg['Date'] = formatdate(localtime=True)
        msg['Subject'] = subject

        # 正文处理
        msg.attach(MIMEText(body, 'html' if '<html>' in body else 'plain'))

        # 附件处理（省略具体实现）conn = self._get_connection()
        try:
            conn.sendmail(self.auth[0], to, msg.as_string())
        finally:
            self._release_connection(conn)

结合 Celery 实现异步发送的关键配置：

from celery import Celery
from celery.utils.log import get_task_logger

app = Celery('email_tasks', broker='redis://localhost:6379/0')
logger = get_task_logger(__name__)

@app.task(bind=True, max_retries=3)
def async_send(self, recipient, subject, template_name, context):
    try:
        # 渲染邮件模板
        html_content = render_template(template_name, **context)

        # 获取 SMTP 连接代理实例
        agent = get_smtp_agent()  # 需实现单例管理

        # 实际发送
        agent.send(recipient, subject, html_content)
    except smtplib.SMTPException as e:
        logger.error(f'邮件发送失败: {str(e)}')
        # 指数退避重试
        raise self.retry(exc=e, countdown=2 ** self.request.retries)

def add_dkim_signature(msg, domain, selector, private_key):
    """
    :param msg: 构建好的 MIMEMultipart 对象
    :param domain: 发件域名如 example.com
    :param selector: DKIM 选择器通常是 default
    :param private_key: RSA 私钥内容
    """
    import dkim

    headers = ['From', 'To', 'Subject']
    sig = dkim.sign(message=msg.as_bytes(),
        selector=str(selector).encode(),
        domain=str(domain).encode(),
        privkey=private_key.encode(),
        include_headers=headers
    )
    msg['DKIM-Signature'] = sig.decode().split(':', 1)[1].strip()

DNS 需要添加的 TXT 记录：

_dmarc.example.com. 300 IN TXT "v=DMARC1; p=quarantine; rua=mailto:postmaster@example.com"

指标名称	采集方式	告警阈值
发送成功率	成功回调 / 日志分析	15 分钟内 <95%
平均延迟	任务队列处理时间统计	P99 > 5 秒
连接池利用率	活跃连接数 / 最大连接数	>80% 持续 10 分钟
退信率	接收 SMTP 错误码统计	每小时 >3%