agentic_implementation/email_mcp_server.py

import gradio as gr
import json
import os
from typing import Dict, List
from datetime import datetime, timedelta
from dotenv import load_dotenv

# Import your existing modules
from tools import extract_query_info, analyze_emails
from email_scraper import scrape_emails_by_text_search_with_credentials, _load_email_db
from logger import logger

load_dotenv()

def search_emails(email_address: str, app_password: str, query: str) -> str:
    """
    Search for emails based on a natural language query and return a summary.
    
    Args:
        email_address (str): The Gmail address to connect to
        app_password (str): The Gmail app password for authentication  
        query (str): Natural language query (e.g., "show me mails from swiggy last week")
        
    Returns:
        str: JSON string containing email search results and analysis
    """
    try:
        logger.info("Email MCP tool called with query: %s", query)
        
        # Extract sender keyword and date range from query
        query_info = extract_query_info(query)
        sender_keyword = query_info.get("sender_keyword", "")
        start_date = query_info.get("start_date")
        end_date = query_info.get("end_date")
        
        print(f"Searching for emails with keyword '{sender_keyword}' between {start_date} and {end_date}")
        
        # Use the modified scraper function that accepts credentials
        full_emails = scrape_emails_by_text_search_with_credentials(
            email_address, app_password, sender_keyword, start_date, end_date
        )
        
        if not full_emails:
            result = {
                "query_info": query_info,
                "email_summary": [],
                "analysis": {"summary": f"No emails found for '{sender_keyword}' in the specified date range.", "insights": []},
                "email_count": 0
            }
            return json.dumps(result, indent=2)
        
        # Create summary version without full content
        email_summary = []
        for email in full_emails:
            summary_email = {
                "date": email.get("date"),
                "time": email.get("time"),
                "subject": email.get("subject"),
                "from": email.get("from", "Unknown Sender"),
                "message_id": email.get("message_id")
            }
            email_summary.append(summary_email)
        
        # Auto-analyze the emails for insights
        analysis = analyze_emails(full_emails)
        
        # Return summary info with analysis
        result = {
            "query_info": query_info,
            "email_summary": email_summary,
            "analysis": analysis,
            "email_count": len(full_emails)
        }
        
        return json.dumps(result, indent=2)
        
    except Exception as e:
        logger.error("Error in search_emails: %s", e)
        error_result = {
            "error": str(e),
            "query": query,
            "message": "Failed to search emails. Please check your credentials and try again."
        }
        return json.dumps(error_result, indent=2)


def get_email_details(email_address: str, app_password: str, message_id: str) -> str:
    """
    Get full details of a specific email by its message ID.
    
    Args:
        email_address (str): The Gmail address to connect to
        app_password (str): The Gmail app password for authentication
        message_id (str): The message ID of the email to retrieve
        
    Returns:
        str: JSON string containing the full email details
    """
    try:
        logger.info("Getting email details for message_id: %s", message_id)
        
        # Load from local cache first
        db = _load_email_db()
        
        # Search each sender's email list
        for sender_data in db.values():
            for email in sender_data.get("emails", []):
                if email.get("message_id") == message_id:
                    return json.dumps(email, indent=2)
        
        # If not found in cache
        error_result = {
            "error": f"No email found with message_id '{message_id}'",
            "message": "Email may not be in local cache. Try searching for emails first."
        }
        return json.dumps(error_result, indent=2)
        
    except Exception as e:
        logger.error("Error in get_email_details: %s", e)
        error_result = {
            "error": str(e),
            "message_id": message_id,
            "message": "Failed to retrieve email details."
        }
        return json.dumps(error_result, indent=2)


def analyze_email_patterns(email_address: str, app_password: str, sender_keyword: str, days_back: str = "30") -> str:
    """
    Analyze email patterns from a specific sender over a given time period.
    
    Args:
        email_address (str): The Gmail address to connect to
        app_password (str): The Gmail app password for authentication
        sender_keyword (str): The sender/company keyword to analyze (e.g., "amazon", "google")
        days_back (str): Number of days to look back (default: "30")
        
    Returns:
        str: JSON string containing email pattern analysis
    """
    try:
        logger.info("Analyzing email patterns for sender: %s, days_back: %s", sender_keyword, days_back)
        
        # Calculate date range
        days_int = int(days_back)
        end_date = datetime.today()
        start_date = end_date - timedelta(days=days_int)
        
        start_date_str = start_date.strftime("%d-%b-%Y")
        end_date_str = end_date.strftime("%d-%b-%Y")
        
        # Search for emails
        full_emails = scrape_emails_by_text_search_with_credentials(
            email_address, app_password, sender_keyword, start_date_str, end_date_str
        )
        
        if not full_emails:
            result = {
                "sender_keyword": sender_keyword,
                "date_range": f"{start_date_str} to {end_date_str}",
                "analysis": {"summary": f"No emails found from '{sender_keyword}' in the last {days_back} days.", "insights": []},
                "email_count": 0
            }
            return json.dumps(result, indent=2)
        
        # Analyze the emails
        analysis = analyze_emails(full_emails)
        
        result = {
            "sender_keyword": sender_keyword,
            "date_range": f"{start_date_str} to {end_date_str}",
            "analysis": analysis,
            "email_count": len(full_emails)
        }
        
        return json.dumps(result, indent=2)
        
    except Exception as e:
        logger.error("Error in analyze_email_patterns: %s", e)
        error_result = {
            "error": str(e),
            "sender_keyword": sender_keyword,
            "message": "Failed to analyze email patterns."
        }
        return json.dumps(error_result, indent=2)


# Create the Gradio interface for email search
search_interface = gr.Interface(
    fn=search_emails,
    inputs=[
        gr.Textbox(label="Email Address", placeholder="[email protected]"),
        gr.Textbox(label="App Password", type="password", placeholder="Your Gmail app password"),
        gr.Textbox(label="Query", placeholder="Show me emails from amazon last week")
    ],
    outputs=gr.Textbox(label="Search Results", lines=20),
    title="Email Search",
    description="Search your emails using natural language queries"
)

# Create the Gradio interface for email details
details_interface = gr.Interface(
    fn=get_email_details,
    inputs=[
        gr.Textbox(label="Email Address", placeholder="[email protected]"),
        gr.Textbox(label="App Password", type="password", placeholder="Your Gmail app password"),
        gr.Textbox(label="Message ID", placeholder="Email message ID from search results")
    ],
    outputs=gr.Textbox(label="Email Details", lines=20),
    title="Email Details",
    description="Get full details of a specific email by message ID"
)

# Create the Gradio interface for email pattern analysis
analysis_interface = gr.Interface(
    fn=analyze_email_patterns,
    inputs=[
        gr.Textbox(label="Email Address", placeholder="[email protected]"),
        gr.Textbox(label="App Password", type="password", placeholder="Your Gmail app password"),
        gr.Textbox(label="Sender Keyword", placeholder="amazon, google, linkedin, etc."),
        gr.Textbox(label="Days Back", value="30", placeholder="Number of days to analyze")
    ],
    outputs=gr.Textbox(label="Analysis Results", lines=20),
    title="Email Pattern Analysis",
    description="Analyze email patterns from a specific sender over time"
)

# Combine interfaces into a tabbed interface
demo = gr.TabbedInterface(
    [search_interface, details_interface, analysis_interface],
    ["Email Search", "Email Details", "Pattern Analysis"],
    title="📧 Email Assistant MCP Server"
)

if __name__ == "__main__":
    # Set environment variable to enable MCP server
    import os
    os.environ["GRADIO_MCP_SERVER"] = "True"
    
    # Launch the server
    demo.launch(share=False)
    
    print("\n🚀 MCP Server is running!")
    print("📍 MCP Endpoint: http://localhost:7860/gradio_api/mcp/sse")
    print("📖 Copy this URL to your Claude Desktop MCP configuration")