server.py•5.17 kB
import os
import requests
from typing import Optional, Dict, Any
from fastmcp import FastMCP
# Initialize the FastMCP server
mcp = FastMCP(
name="AtData SafeToSend Email Verification Server",
instructions="""
This server provides email verification services using AtData's SafeToSend API.
The SafeToSend API verifies email addresses to filter out invalid and high-risk ones,
which results in higher open rates, clicks, and conversions.
""",
)
@mcp.tool()
def verify_email(email: str, api_key: Optional[str] = None) -> Dict[str, Any]:
"""
Verify an email address using AtData's SafeToSend API.
This tool verifies email addresses to filter out invalid and high-risk ones,
which results in higher open rates, clicks, and conversions.
Args:
email: The email address to verify
api_key: AtData API key (if not provided, will try to get from ATDATA_API_KEY env var)
Returns:
Dictionary containing the verification results including:
- email: The email address that was verified
- status: The verification status
- deliverable: Whether the email is deliverable
- risk_level: Risk assessment of the email
- additional verification details
"""
# Get API key from parameter or environment variable
if api_key is None:
api_key = os.getenv("ATDATA_API_KEY")
if not api_key:
return {
"error": "API key is required. Provide it as a parameter or set ATDATA_API_KEY environment variable.",
"email": email,
}
# AtData SafeToSend API endpoint
url = "https://api.atdata.com/v5/ev"
# Set up headers
headers = {"Accept": "application/json", "User-Agent": "AtData-MCP-Server/1.0"}
# Set up parameters
params = {"email": email, "api_key": api_key}
try:
# Make the API request
response = requests.get(url, headers=headers, params=params, timeout=30)
# Check if request was successful
if response.status_code == 200:
result = response.json()
return {"success": True, "email": email, "verification_result": result}
elif response.status_code == 401:
return {
"error": "Authentication failed. Please check your API key.",
"email": email,
"status_code": response.status_code,
}
elif response.status_code == 400:
return {
"error": "Bad request. Please check the email format.",
"email": email,
"status_code": response.status_code,
"details": response.text,
}
elif response.status_code == 429:
return {
"error": "Rate limit exceeded. Please try again later.",
"email": email,
"status_code": response.status_code,
}
else:
return {
"error": f"API request failed with status code {response.status_code}",
"email": email,
"status_code": response.status_code,
"details": response.text,
}
except requests.exceptions.Timeout:
return {
"error": "Request timeout. The API did not respond within 30 seconds.",
"email": email,
}
except requests.exceptions.ConnectionError:
return {
"error": "Connection error. Unable to reach the AtData API.",
"email": email,
}
except requests.exceptions.RequestException as e:
return {"error": f"Request failed: {str(e)}", "email": email}
except Exception as e:
return {"error": f"Unexpected error: {str(e)}", "email": email}
@mcp.tool()
def batch_verify_emails(
emails: list[str], api_key: Optional[str] = None
) -> Dict[str, Any]:
"""
Verify multiple email addresses using AtData's SafeToSend API.
This tool allows you to verify multiple email addresses in batch,
processing each one individually through the SafeToSend API.
Args:
emails: List of email addresses to verify
api_key: AtData API key (if not provided, will try to get from ATDATA_API_KEY env var)
Returns:
Dictionary containing:
- results: List of verification results for each email
- summary: Summary statistics of the batch verification
"""
if not emails:
return {
"error": "No emails provided for verification",
"results": [],
"summary": {"total": 0, "successful": 0, "failed": 0},
}
results = []
successful = 0
failed = 0
for email in emails:
result = verify_email(email, api_key)
results.append(result)
if result.get("success"):
successful += 1
else:
failed += 1
summary = {
"total": len(emails),
"successful": successful,
"failed": failed,
"success_rate": (successful / len(emails)) * 100 if emails else 0,
}
return {"results": results, "summary": summary}
if __name__ == "__main__":
mcp.run()