第17讲:邮件自动化 Skill 开发 掌握邮件自动化技能,实现邮件自动发送、模板管理、批量投递等功能,让邮件处理效率大幅提升。
一、场景分析 1.1 用户痛点 邮件是商务沟通的主要方式,但重复性工作很多:
批量发送繁琐 :需要给多个客户发送相似内容的邮件,逐一发送效率低模板管理混乱 :常用邮件模板散落在各处,查找使用不便附件处理耗时 :每次发送都要手动添加附件,容易遗漏跟进提醒困难 :需要定时跟进客户,但容易忘记发送记录难查 :历史邮件分散,难以统一管理和统计1.2 典型应用场景 场景 需求描述 Skill 价值 营销邮件 向客户群发产品推广邮件 批量个性化发送 会议通知 自动发送会议邀请和提醒 定时自动发送 报表投递 定期发送数据报表给相关人员 自动附件处理 客户跟进 按设定时间自动跟进客户 智能提醒机制 自动回复 根据邮件内容自动回复 智能应答处理
二、核心功能设计 2.1 Skill 功能架构 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 📧 邮件自动化助手 ├── 邮件发送 │ ├── 单封发送 │ ├── 批量发送 │ ├── 定时发送 │ └── 抄送/密送 ├── 模板管理 │ ├── 模板创建 │ ├── 变量替换 │ ├── 模板分类 │ └── 模板预览 ├── 附件处理 │ ├── 自动附加 │ ├── 批量附加 │ ├── 压缩打包 │ └── 云存储链接 ├── 收件管理 │ ├── 邮件读取 │ ├── 自动分类 │ ├── 智能回复 │ └── 跟进提醒 └── 统计分析 ├── 发送记录 ├── 送达统计 ├── 打开追踪 └── 效果分析
2.2 技术选型 邮件处理的核心技术栈:
功能 技术方案 说明 邮件发送 smtplib / yagmail Python 标准库和第三方库 邮件接收 imaplib / poplib 读取收件箱邮件 邮件解析 email / mail-parser 解析邮件内容 定时任务 schedule / APScheduler 定时发送邮件 邮件服务 SendGrid / Mailgun 专业邮件发送服务
三、技术实现 3.1 Coze 平台实现 3.1.1 基础发送代码 使用 smtplib:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 import smtplibfrom email.mime.text import MIMETextfrom email.mime.multipart import MIMEMultipartfrom email.mime.base import MIMEBasefrom email import encodersimport osdef send_email (smtp_config, to_addrs, subject, body, attachments=None , cc_addrs=None ): """ 发送邮件 Args: smtp_config: SMTP 配置 { 'host': 'smtp.example.com', 'port': 587, 'user': 'user@example.com', 'password': 'password', 'use_tls': True } to_addrs: 收件人地址列表 subject: 邮件主题 body: 邮件正文(支持 HTML) attachments: 附件路径列表 cc_addrs: 抄送地址列表 """ msg = MIMEMultipart() msg['From' ] = smtp_config['user' ] msg['To' ] = ', ' .join(to_addrs) msg['Subject' ] = subject if cc_addrs: msg['Cc' ] = ', ' .join(cc_addrs) msg.attach(MIMEText(body, 'html' , 'utf-8' )) if attachments: for file_path in attachments: if os.path.exists(file_path): with open (file_path, 'rb' ) as f: attachment = MIMEBase('application' , 'octet-stream' ) attachment.set_payload(f.read()) encoders.encode_base64(attachment) filename = os.path.basename(file_path) attachment.add_header( 'Content-Disposition' , f'attachment; filename="{filename} "' ) msg.attach(attachment) with smtplib.SMTP(smtp_config['host' ], smtp_config['port' ]) as server: if smtp_config.get('use_tls' , True ): server.starttls() server.login(smtp_config['user' ], smtp_config['password' ]) all_recipients = to_addrs + (cc_addrs or []) server.sendmail(smtp_config['user' ], all_recipients, msg.as_string()) return True
使用 yagmail(更简单):
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 import yagmaildef send_email_simple (user, password, to, subject, body, attachments=None ): """ 使用 yagmail 简化发送 Args: user: 发件人邮箱 password: 邮箱密码或授权码 to: 收件人地址 subject: 邮件主题 body: 邮件正文 attachments: 附件路径列表 """ yag = yagmail.SMTP(user, password) yag.send( to=to, subject=subject, contents=body, attachments=attachments ) return True
3.1.2 模板邮件代码 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 from jinja2 import Templatedef render_email_template (template_str, variables ): """ 渲染邮件模板 Args: template_str: 模板字符串 variables: 变量字典 Returns: 渲染后的邮件内容 """ template = Template(template_str) return template.render(**variables) EMAIL_TEMPLATES = { 'welcome' : """ <html> <body> <h2>欢迎 {{ name }}!</h2> <p>您好,感谢您注册我们的服务。</p> <p>您的账户信息:</p> <ul> <li>用户名:{{ username }}</li> <li>注册时间:{{ register_time }}</li> </ul> <p>如有任何问题,请随时联系我们。</p> <p>祝好!</p> </body> </html> """ , 'meeting_invite' : """ <html> <body> <h2>会议邀请</h2> <p>您好 {{ name }},</p> <p>诚邀您参加以下会议:</p> <table border="1" cellpadding="10"> <tr><td>会议主题</td><td>{{ meeting_title }}</td></tr> <tr><td>时间</td><td>{{ meeting_time }}</td></tr> <tr><td>地点</td><td>{{ meeting_location }}</td></tr> <tr><td>主持人</td><td>{{ host }}</td></tr> </table> <p>请确认是否能参加。</p> </body> </html> """ , 'report_delivery' : """ <html> <body> <h2>{{ report_name }}</h2> <p>您好 {{ name }},</p> <p>{{ report_period }} 的数据报表已生成,请查收附件。</p> <p>报表概要:</p> <ul> {% for item in summary %} <li>{{ item }}</li> {% endfor %} </ul> <p>如有疑问,请联系数据分析团队。</p> </body> </html> """ } def send_template_email (smtp_config, template_name, to, variables, attachments=None ): """ 发送模板邮件 Args: smtp_config: SMTP 配置 template_name: 模板名称 to: 收件人地址 variables: 模板变量 attachments: 附件列表 """ template = EMAIL_TEMPLATES.get(template_name) if not template: raise ValueError(f"模板 {template_name} 不存在" ) body = render_email_template(template, variables) subject_map = { 'welcome' : '欢迎加入!' , 'meeting_invite' : f"会议邀请:{variables.get('meeting_title' , '' )} " , 'report_delivery' : f"报表:{variables.get('report_name' , '' )} " } subject = subject_map.get(template_name, '邮件通知' ) return send_email(smtp_config, [to], subject, body, attachments)
3.1.3 批量发送代码 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 def batch_send_emails (smtp_config, recipient_list, template_name, delay=1 , progress_callback=None ): """ 批量发送邮件 Args: smtp_config: SMTP 配置 recipient_list: 收件人列表 [ { 'email': 'user@example.com', 'name': '用户名', 'variables': {...} } ] template_name: 模板名称 delay: 每封邮件间隔(秒) progress_callback: 进度回调函数 """ import time results = [] total = len (recipient_list) for i, recipient in enumerate (recipient_list): try : variables = recipient.get('variables' , {}) variables['name' ] = recipient.get('name' , '' ) send_template_email( smtp_config, template_name, recipient['email' ], variables ) results.append({ 'email' : recipient['email' ], 'status' : 'success' , 'message' : '发送成功' }) except Exception as e: results.append({ 'email' : recipient['email' ], 'status' : 'failed' , 'message' : str (e) }) if progress_callback: progress_callback(i + 1 , total) if delay > 0 and i < total - 1 : time.sleep(delay) return results
3.1.4 定时发送代码 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 import scheduleimport timefrom datetime import datetimedef schedule_email (smtp_config, send_time, to, subject, body, attachments=None ): """ 定时发送邮件 Args: smtp_config: SMTP 配置 send_time: 发送时间(datetime 对象) to: 收件人地址 subject: 邮件主题 body: 邮件正文 attachments: 附件列表 """ def job (): send_email(smtp_config, [to], subject, body, attachments) print (f"邮件已发送给 {to} 于 {datetime.now()} " ) schedule_time = send_time.strftime("%H:%M" ) schedule.every().day.at(schedule_time).do(job) print (f"已设置定时任务,将在 {send_time} 发送邮件" ) def schedule_recurring_email (smtp_config, schedule_config, to, subject, body ): """ 设置周期性邮件 Args: smtp_config: SMTP 配置 schedule_config: 调度配置 { 'frequency': 'daily' / 'weekly' / 'monthly', 'day': 'monday' / 'tuesday' / ... (weekly 时), 'date': 1-31 (monthly 时), 'time': '09:00' } """ def job (): send_email(smtp_config, [to], subject, body) freq = schedule_config['frequency' ] send_time = schedule_config['time' ] if freq == 'daily' : schedule.every().day.at(send_time).do(job) elif freq == 'weekly' : day = schedule_config.get('day' , 'monday' ) getattr (schedule.every(), day).at(send_time).do(job) elif freq == 'monthly' : def monthly_job (): today = datetime.now() if today.day == schedule_config.get('date' , 1 ): job() schedule.every().day.at(send_time).do(monthly_job)
3.2 OpenClaw 平台实现 OpenClaw 的邮件 Skill 示例:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 from openclaw import Skill, Toolimport yagmailclass EmailSkill (Skill ): name = "邮件自动化助手" description = "自动化发送和管理邮件" def __init__ (self, smtp_config ): self.smtp_config = smtp_config self.yag = yagmail.SMTP(smtp_config['user' ], smtp_config['password' ]) @Tool def send (self, to: str , subject: str , body: str , attachments: list = None ) -> str : """发送单封邮件""" self.yag.send(to=to, subject=subject, contents=body, attachments=attachments) return f"邮件已发送至 {to} " @Tool def send_template (self, to: str , template: str , variables: dict , attachments: list = None ) -> str : """发送模板邮件""" from jinja2 import Template t = Template(template) body = t.render(**variables) return self.send(to, variables.get('subject' , '邮件通知' ), body, attachments) @Tool def batch_send (self, recipients: list , template: str , variables_list: list , delay: int = 1 ) -> list : """批量发送邮件""" results = [] for recipient, vars in zip (recipients, variables_list): try : result = self.send_template(recipient, template, vars ) results.append({'email' : recipient, 'status' : 'success' }) except Exception as e: results.append({'email' : recipient, 'status' : 'failed' , 'error' : str (e)}) time.sleep(delay) return results
四、Prompt 设计 4.1 系统 Prompt 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 你是邮件自动化助手,专门帮助用户高效处理邮件发送任务。 你可以执行以下操作: 1. 单封发送:发送单封邮件,支持 HTML 格式和附件 2. 批量发送:批量发送个性化邮件 3. 模板邮件:使用预设模板发送邮件 4. 定时发送:设置邮件定时发送 5. 邮件管理:查看发送记录和统计 工作流程: 1. 理解用户的邮件发送需求 2. 收集必要信息(收件人、主题、内容、附件等) 3. 确认邮件内容 4. 执行发送 5. 返回发送结果 注意事项: - 批量发送时注意发送频率,避免被判定为垃圾邮件 - 重要邮件建议添加已读回执 - 敏感信息注意加密保护 - 遵守邮件发送规范和法律法规
4.2 意图识别示例 用户输入 识别意图 提取参数 "给张三发封邮件" 单封发送 收件人、主题、内容 "批量发送会议邀请" 批量发送 收件人列表、模板 "明天上午9点发提醒邮件" 定时发送 时间、收件人、内容 "用欢迎模板给新用户发邮件" 模板邮件 模板名称、收件人、变量 "查看邮件发送记录" 记录查询 查询条件
五、实战案例 5.1 案例一:营销邮件群发 场景 :市场部门需要向客户群发新品推广邮件。
解决方案 :
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 def send_marketing_campaign (smtp_config, customer_data, product_info ): """ 发送营销邮件活动 Args: smtp_config: SMTP 配置 customer_data: 客户数据列表 product_info: 产品信息 """ template = """ <html> <body style="font-family: Arial, sans-serif;"> <h2 style="color: #333;">🎉 新品上市:{{ product_name }}</h2> <p>尊敬的 {{ name }},</p> <p>我们很高兴向您推荐我们的最新产品!</p> <div style="border: 1px solid #ddd; padding: 15px; margin: 15px 0;"> <h3>{{ product_name }}</h3> <p>{{ product_description }}</p> <p style="color: #e74c3c; font-size: 18px;"> 特惠价:¥{{ price }} <span style="text-decoration: line-through; color: #999;">¥{{ original_price }}</span> </p> </div> <p>限时优惠,先到先得!</p> <p><a href="{{ link }}" style="background: #3498db; color: white; padding: 10px 20px; text-decoration: none; border-radius: 5px;">立即购买</a></p> <p>如您不感兴趣,<a href="{{ unsubscribe_link }}">点击退订</a></p> </body> </html> """ recipient_list = [] for customer in customer_data: recipient_list.append({ 'email' : customer['email' ], 'name' : customer['name' ], 'variables' : { 'name' : customer['name' ], 'product_name' : product_info['name' ], 'product_description' : product_info['description' ], 'price' : product_info['price' ], 'original_price' : product_info['original_price' ], 'link' : product_info['purchase_link' ], 'unsubscribe_link' : f"https://example.com/unsubscribe?email={customer['email' ]} " } }) results = batch_send_emails( smtp_config, recipient_list, 'marketing' , delay=5 ) success_count = sum (1 for r in results if r['status' ] == 'success' ) print (f"发送完成:成功 {success_count} /{len (results)} " ) return results
5.2 案例二:自动报表投递 场景 :需要定期向管理层发送数据报表。
解决方案 :
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 def schedule_report_delivery (smtp_config, report_config ): """ 设置自动报表投递 Args: smtp_config: SMTP 配置 report_config: 报表配置 { 'recipients': ['manager@example.com'], 'report_generator': generate_weekly_report, # 报表生成函数 'schedule': { 'frequency': 'weekly', 'day': 'monday', 'time': '09:00' } } """ import schedule def job (): report_path = report_config['report_generator' ]() report_name = os.path.basename(report_path) report_date = datetime.now().strftime("%Y年%m月%d日" ) for recipient in report_config['recipients' ]: send_template_email( smtp_config, 'report_delivery' , recipient, { 'name' : '领导' , 'report_name' : report_name, 'report_period' : report_date, 'summary' : [ '本周销售额同比增长 15%' , '新增客户 23 家' , '订单转化率提升 3%' ] }, attachments=[report_path] ) print (f"报表已发送:{report_date} " ) schedule_config = report_config['schedule' ] day = schedule_config['day' ] time = schedule_config['time' ] getattr (schedule.every(), day).at(time).do(job) print (f"已设置定时报表投递:每周{day} {time} " )
六、实战练习 练习 1:欢迎邮件发送 创建一个 Skill,实现以下功能:
使用欢迎邮件模板 接收新用户信息(姓名、邮箱等) 个性化填充模板变量 发送欢迎邮件 练习 2:会议邀请批量发送 创建一个 Skill,实现以下功能:
读取参会人员列表 使用会议邀请模板 批量发送个性化邀请邮件 记录发送状态 练习 3:定时提醒邮件 创建一个 Skill,实现以下功能:
设置提醒任务(时间、收件人、内容) 定时检查并发送提醒邮件 支持周期性提醒(每日/每周/每月) 管理提醒任务列表 七、常见问题 Q1:邮件被判定为垃圾邮件怎么办? 解决方案 :
使用专业的邮件发送服务(SendGrid、Mailgun) 控制发送频率,避免短时间内大量发送 添加退订链接 使用 SPF、DKIM、DMARC 等邮件验证 保持邮件内容质量,避免敏感词汇 Q2:附件太大无法发送? 解决方案 :
压缩附件 使用云存储链接代替附件 分割大文件 使用支持大附件的邮件服务 Q3:如何确保邮件送达? 解决方案 :
使用可靠的 SMTP 服务 实现重试机制 添加送达回执 监控退信和投诉率 八、下节预告 下一讲我们将进入 进阶开发技巧 章节,学习:
多平台 Skill 适配 数据持久化 安全与权限 性能优化 加入学习群 👉 加入AI编程学习交流群
本讲是《AI Skills 从入门到实践》系列课程的第17讲。
🎓 AI 编程实战课程 想系统学习 AI 编程?程序员晚枫的 AI 编程实战课 帮你从零上手!