"""
MCP Email Server - SendGrid Integration
This MCP server provides email sending capabilities through SendGrid API.
It exposes tools for sending emails with various features:
- Basic text/HTML emails
- Template-based emails
- Attachments support
- CC/BCC recipients
"""
import os
import base64
import logging
from typing import Optional, List
from pathlib import Path
from fastmcp import FastMCP
from dotenv import load_dotenv
from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import (
Mail,
Email,
To,
Cc,
Bcc,
Content,
Attachment,
FileContent,
FileName,
FileType,
Disposition,
)
from pydantic import BaseModel, EmailStr, Field
# Load environment variables
load_dotenv()
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# Initialize FastMCP server
mcp = FastMCP(name="email-server", version="0.1.0")
# Get SendGrid configuration from environment
SENDGRID_API_KEY = os.getenv("SENDGRID_API_KEY")
DEFAULT_FROM_EMAIL = os.getenv("DEFAULT_FROM_EMAIL")
DEFAULT_FROM_NAME = os.getenv("DEFAULT_FROM_NAME", "")
# Validate configuration
if not SENDGRID_API_KEY:
logger.warning("SENDGRID_API_KEY not set. Email sending will fail.")
if not DEFAULT_FROM_EMAIL:
logger.warning("DEFAULT_FROM_EMAIL not set. You must provide from_email in each request.")
# Initialize SendGrid client
sg = SendGridAPIClient(api_key=SENDGRID_API_KEY) if SENDGRID_API_KEY else None
@mcp.resource("file://airports_in_poland")
def airports_in_poland():
"""
Returns a list of airports in Poland with their IATA codes.
"""
return [
{"name": "Warsaw Chopin Airport", "code": "WAW"},
{"name": "Kraków John Paul II International Airport", "code": "KRK"},
{"name": "Gdańsk Lech Wałęsa Airport", "code": "GDN"},
{"name": "Katowice International Airport", "code": "KTW"},
{"name": "Wrocław Copernicus Airport", "code": "WRO"},
{"name": "Poznań–Ławica Airport", "code": "POZ"},
{"name": "Rzeszów–Jasionka Airport", "code": "RZE"},
{"name": "Szczecin–Goleniów Airport", "code": "SZZ"},
{"name": "Bydgoszcz Ignacy Jan Paderewski Airport", "code": "BZG"},
{"name": "Lublin Airport", "code": "LUZ"},
{"name": "Łódź Władysław Reymont Airport", "code": "LCJ"}
]
@mcp.tool()
def send_email(
to_email: str,
subject: str,
body: str,
from_email: Optional[str] = None,
from_name: Optional[str] = None,
cc_emails: Optional[List[str]] = None,
bcc_emails: Optional[List[str]] = None,
is_html: bool = False,
) -> str:
"""
Send an email using SendGrid.
Args:
to_email: Recipient email address
subject: Email subject line
body: Email body content (text or HTML)
from_email: Sender email address (uses DEFAULT_FROM_EMAIL if not provided)
from_name: Sender name (uses DEFAULT_FROM_NAME if not provided)
cc_emails: List of CC email addresses (optional)
bcc_emails: List of BCC email addresses (optional)
is_html: Whether the body is HTML (default: False for plain text)
Returns:
Success message with status code or error message
"""
try:
if not sg:
return "Error: SendGrid client not initialized. Check SENDGRID_API_KEY."
# Use defaults if not provided
sender_email = from_email or DEFAULT_FROM_EMAIL
sender_name = from_name or DEFAULT_FROM_NAME
if not sender_email:
return "Error: No from_email provided and DEFAULT_FROM_EMAIL not set."
# Create the email message
from_email_obj = Email(sender_email, sender_name)
to_email_obj = To(to_email)
# Set content type based on is_html flag
content_type = "text/html" if is_html else "text/plain"
content = Content(content_type, body)
# Create mail object
mail = Mail(from_email_obj, to_email_obj, subject, content)
# Add CC recipients if provided
if cc_emails:
for cc_email in cc_emails:
mail.add_cc(Cc(cc_email))
# Add BCC recipients if provided
if bcc_emails:
for bcc_email in bcc_emails:
mail.add_bcc(Bcc(bcc_email))
# Send the email
response = sg.send(mail)
logger.info(f"Email sent successfully to {to_email}. Status: {response.status_code}")
return f"Email sent successfully! Status code: {response.status_code}"
except Exception as e:
error_msg = f"Failed to send email: {str(e)}"
logger.error(error_msg)
return error_msg
@mcp.tool()
def send_email_with_template(
to_email: str,
template_id: str,
dynamic_data: dict,
from_email: Optional[str] = None,
from_name: Optional[str] = None,
subject: Optional[str] = None,
) -> str:
"""
Send an email using a SendGrid dynamic template.
Args:
to_email: Recipient email address
template_id: SendGrid template ID
dynamic_data: Dictionary of dynamic template data (key-value pairs)
from_email: Sender email address (uses DEFAULT_FROM_EMAIL if not provided)
from_name: Sender name (uses DEFAULT_FROM_NAME if not provided)
subject: Email subject (optional, can be defined in template)
Returns:
Success message with status code or error message
"""
try:
if not sg:
return "Error: SendGrid client not initialized. Check SENDGRID_API_KEY."
# Use defaults if not provided
sender_email = from_email or DEFAULT_FROM_EMAIL
sender_name = from_name or DEFAULT_FROM_NAME
if not sender_email:
return "Error: No from_email provided and DEFAULT_FROM_EMAIL not set."
# Create the email message
mail = Mail(
from_email=Email(sender_email, sender_name),
to_emails=To(to_email)
)
# Set template ID
mail.template_id = template_id
# Add dynamic template data
mail.dynamic_template_data = dynamic_data
# Add subject if provided (overrides template subject)
if subject:
mail.subject = subject
# Send the email
response = sg.send(mail)
logger.info(f"Template email sent to {to_email}. Template: {template_id}, Status: {response.status_code}")
return f"Template email sent successfully! Status code: {response.status_code}"
except Exception as e:
error_msg = f"Failed to send template email: {str(e)}"
logger.error(error_msg)
return error_msg
@mcp.tool()
def send_email_with_attachments(
to_email: str,
subject: str,
body: str,
attachment_paths: List[str],
from_email: Optional[str] = None,
from_name: Optional[str] = None,
is_html: bool = False,
) -> str:
"""
Send an email with file attachments.
Args:
to_email: Recipient email address
subject: Email subject line
body: Email body content (text or HTML)
attachment_paths: List of file paths to attach
from_email: Sender email address (uses DEFAULT_FROM_EMAIL if not provided)
from_name: Sender name (uses DEFAULT_FROM_NAME if not provided)
is_html: Whether the body is HTML (default: False for plain text)
Returns:
Success message with status code or error message
"""
try:
if not sg:
return "Error: SendGrid client not initialized. Check SENDGRID_API_KEY."
# Use defaults if not provided
sender_email = from_email or DEFAULT_FROM_EMAIL
sender_name = from_name or DEFAULT_FROM_NAME
if not sender_email:
return "Error: No from_email provided and DEFAULT_FROM_EMAIL not set."
# Create the email message
from_email_obj = Email(sender_email, sender_name)
to_email_obj = To(to_email)
content_type = "text/html" if is_html else "text/plain"
content = Content(content_type, body)
mail = Mail(from_email_obj, to_email_obj, subject, content)
# Process and attach files
for file_path in attachment_paths:
try:
path = Path(file_path)
if not path.exists():
logger.warning(f"Attachment file not found: {file_path}")
continue
# Read and encode file
with open(path, 'rb') as f:
file_data = f.read()
encoded_file = base64.b64encode(file_data).decode()
# Create attachment
attached_file = Attachment(
FileContent(encoded_file),
FileName(path.name),
FileType('application/octet-stream'),
Disposition('attachment')
)
mail.add_attachment(attached_file)
logger.info(f"Added attachment: {path.name}")
except Exception as e:
logger.error(f"Failed to attach file {file_path}: {str(e)}")
return f"Error processing attachment {file_path}: {str(e)}"
# Send the email
response = sg.send(mail)
logger.info(f"Email with attachments sent to {to_email}. Status: {response.status_code}")
return f"Email with attachments sent successfully! Status code: {response.status_code}"
except Exception as e:
error_msg = f"Failed to send email with attachments: {str(e)}"
logger.error(error_msg)
return error_msg
if __name__ == "__main__":
logger.info("Starting MCP Email Server...")
logger.info(f"SendGrid configured: {bool(SENDGRID_API_KEY)}")
logger.info(f"Default from email: {DEFAULT_FROM_EMAIL}")
# Check transport mode from environment
transport = os.getenv("MCP_TRANSPORT", "stdio").lower()
logger.info(f"Running in {transport.upper()} mode")
# Run with the specified transport
mcp.run(transport=transport, host="0.0.0.0", port=8000)
