"""
邮件发送 MCP 服务器
基于 email_sender.py 工具类提供邮件发送功能
"""
from mcp.server.fastmcp import FastMCP
from email_sender import EmailSender, SMTP_CONFIGS
from typing import List, Optional, Dict, Any
import os
import json
# 尝试加载 .env 文件(如果存在)
try:
from dotenv import load_dotenv
load_dotenv()
except ImportError:
# 如果没有安装 python-dotenv,跳过加载
pass
# 创建 MCP 服务器
mcp = FastMCP("EmailSender", json_response=True)
def get_email_sender(custom_config: Optional[Dict[str, Any]] = None) -> EmailSender:
"""
获取邮件发送器实例
Args:
custom_config: 自定义配置字典,包含 smtp_server, smtp_port, sender_email,
sender_password, use_tls, use_ssl
Returns:
EmailSender 实例
"""
if custom_config:
return EmailSender(
smtp_server=custom_config.get('smtp_server'),
smtp_port=custom_config.get('smtp_port'),
sender_email=custom_config.get('sender_email'),
sender_password=custom_config.get('sender_password'),
use_tls=custom_config.get('use_tls'),
use_ssl=custom_config.get('use_ssl'),
use_default_config=False
)
else:
return EmailSender(use_default_config=True)
@mcp.tool()
def send_simple_email(
receiver_email: str,
subject: str,
content: str,
content_type: str = "plain"
) -> str:
"""
发送简单邮件(单个收件人,无附件)
Args:
receiver_email: 收件人邮箱地址
subject: 邮件主题
content: 邮件内容(纯文本或HTML)
content_type: 内容类型,'plain' 表示纯文本,'html' 表示HTML格式
Returns:
发送结果信息
"""
try:
sender = get_email_sender()
result = sender.send_simple_email(
receiver_email=receiver_email,
subject=subject,
content=content,
content_type=content_type
)
if result['success']:
return f"✅ 邮件发送成功!收件人: {receiver_email}"
else:
return f"❌ 邮件发送失败: {result['message']}"
except Exception as e:
return f"❌ 发生错误: {str(e)}"
@mcp.tool()
def send_email(
receiver_emails: List[str],
subject: str,
content: str,
content_type: str = "plain",
attachments: Optional[List[str]] = None,
cc_emails: Optional[List[str]] = None,
bcc_emails: Optional[List[str]] = None
) -> str:
"""
发送完整邮件(支持多个收件人、附件、CC、BCC)
Args:
receiver_emails: 收件人邮箱地址列表
subject: 邮件主题
content: 邮件内容(纯文本或HTML)
content_type: 内容类型,'plain' 表示纯文本,'html' 表示HTML格式
attachments: 附件文件路径列表(可选)
cc_emails: 抄送邮箱地址列表(可选)
bcc_emails: 密送邮箱地址列表(可选)
Returns:
发送结果信息
"""
try:
sender = get_email_sender()
result = sender.send_email(
receiver_emails=receiver_emails,
subject=subject,
content=content,
content_type=content_type,
attachments=attachments,
cc_emails=cc_emails,
bcc_emails=bcc_emails
)
if result['success']:
recipients_info = f"收件人: {', '.join(result['recipients'])}"
return f"✅ 邮件发送成功!{recipients_info}"
else:
return f"❌ 邮件发送失败: {result['message']}"
except Exception as e:
return f"❌ 发生错误: {str(e)}"
@mcp.tool()
def send_email_with_custom_config(
receiver_emails: List[str],
subject: str,
content: str,
smtp_server: str,
smtp_port: int,
sender_email: str,
sender_password: str,
content_type: str = "plain",
use_tls: bool = False,
use_ssl: bool = False,
attachments: Optional[List[str]] = None,
cc_emails: Optional[List[str]] = None,
bcc_emails: Optional[List[str]] = None
) -> str:
"""
使用自定义SMTP配置发送邮件
Args:
receiver_emails: 收件人邮箱地址列表
subject: 邮件主题
content: 邮件内容
smtp_server: SMTP服务器地址(如:smtp.qq.com, smtp.gmail.com)
smtp_port: SMTP端口(25/465/587)
sender_email: 发件人邮箱
sender_password: 发件人密码或授权码
content_type: 内容类型,'plain' 或 'html'
use_tls: 是否使用TLS(端口587通常需要)
use_ssl: 是否使用SSL(端口465通常需要)
attachments: 附件文件路径列表(可选)
cc_emails: 抄送邮箱地址列表(可选)
bcc_emails: 密送邮箱地址列表(可选)
Returns:
发送结果信息
"""
try:
custom_config = {
'smtp_server': smtp_server,
'smtp_port': smtp_port,
'sender_email': sender_email,
'sender_password': sender_password,
'use_tls': use_tls,
'use_ssl': use_ssl
}
sender = get_email_sender(custom_config)
result = sender.send_email(
receiver_emails=receiver_emails,
subject=subject,
content=content,
content_type=content_type,
attachments=attachments,
cc_emails=cc_emails,
bcc_emails=bcc_emails
)
if result['success']:
recipients_info = f"收件人: {', '.join(result['recipients'])}"
return f"✅ 邮件发送成功!{recipients_info}"
else:
return f"❌ 邮件发送失败: {result['message']}"
except Exception as e:
return f"❌ 发生错误: {str(e)}"
@mcp.tool()
def get_smtp_configs() -> str:
"""
获取常用邮箱的SMTP配置信息
Returns:
常用邮箱SMTP配置信息(JSON格式)
"""
try:
configs = {}
for name, config in SMTP_CONFIGS.items():
configs[name] = {
'server': config['server'],
'port': config['port'],
'use_tls': config['use_tls'],
'use_ssl': config['use_ssl']
}
return json.dumps(configs, indent=2, ensure_ascii=False)
except Exception as e:
return f"❌ 获取配置失败: {str(e)}"
@mcp.tool()
def get_current_config() -> str:
"""
获取当前使用的SMTP配置(从环境变量读取)
Returns:
当前配置信息
"""
try:
config = EmailSender.get_default_config()
# 隐藏密码
safe_config = config.copy()
if 'sender_password' in safe_config:
safe_config['sender_password'] = '***' if safe_config['sender_password'] else ''
return json.dumps(safe_config, indent=2, ensure_ascii=False)
except Exception as e:
return f"❌ 获取配置失败: {str(e)}"
# 运行服务器
if __name__ == "__main__":
print("📧 邮件发送 MCP 服务器启动中...")
print("可用工具:")
print(" - send_simple_email: 发送简单邮件(单个收件人)")
print(" - send_email: 发送完整邮件(支持附件、CC、BCC)")
print(" - send_email_with_custom_config: 使用自定义配置发送邮件")
print(" - get_smtp_configs: 获取常用邮箱SMTP配置")
print(" - get_current_config: 获取当前配置")
mcp.run(transport="stdio")