MCP connection (#9)

- mcp server refactor done (08d701a79043690507e7125c35ceb5c1e9fa4063)

agentic_implementation/README_MCP.md

@@ -0,0 +1,115 @@
+# Email Assistant MCP Server
+
+This is a Gradio-based MCP (Model Context Protocol) server that allows Claude Desktop to interact with your Gmail emails.
+
+## Features
+
+- **Email Search**: Search your emails using natural language queries
+- **Email Details**: Get full details of specific emails by message ID
+- **Pattern Analysis**: Analyze email patterns from specific senders over time
+
+## Setup
+
+1. **Install Dependencies**:
+   ```bash
+   pip install -r requirements_mcp.txt
+   ```
+
+2. **Set up Gmail App Password**:
+   - Enable 2-Factor Authentication on your Gmail account
+   - Generate an App Password: https://support.google.com/accounts/answer/185833
+   - Keep your Gmail address and app password ready
+
+3. **Run the MCP Server**:
+   ```bash
+   python email_mcp_server.py
+   ```
+
+   The server will start and show you the MCP endpoint URL, typically:
+   ```
+   http://localhost:7860/gradio_api/mcp/sse
+   ```
+
+## Claude Desktop Configuration
+
+Add this configuration to your Claude Desktop MCP settings:
+
+**For SSE-supported clients:**
+```json
+{
+  "mcpServers": {
+    "email-assistant": {
+      "url": "http://localhost:7860/gradio_api/mcp/sse"
+    }
+  }
+}
+```
+
+**For Claude Desktop (requires mcp-remote):**
+```json
+{
+  "mcpServers": {
+    "email-assistant": {
+      "command": "npx",
+      "args": [
+        "mcp-remote",
+        "http://localhost:7860/gradio_api/mcp/sse"
+      ]
+    }
+  }
+}
+```
+
+## Available Tools
+
+### 1. search_emails
+Search your emails using natural language queries.
+
+**Parameters:**
+- `email_address`: Your Gmail address
+- `app_password`: Your Gmail app password
+- `query`: Natural language query (e.g., "show me emails from amazon last week")
+
+**Example Usage in Claude:**
+> "Can you search my emails for messages from Swiggy in the last week? My email is [email protected] and my app password is xxxx-xxxx-xxxx-xxxx"
+
+### 2. get_email_details
+Get full details of a specific email by message ID.
+
+**Parameters:**
+- `email_address`: Your Gmail address
+- `app_password`: Your Gmail app password
+- `message_id`: Message ID from search results
+
+### 3. analyze_email_patterns
+Analyze email patterns from a specific sender over time.
+
+**Parameters:**
+- `email_address`: Your Gmail address
+- `app_password`: Your Gmail app password
+- `sender_keyword`: Sender to analyze (e.g., "amazon", "google")
+- `days_back`: Number of days to analyze (default: "30")
+
+## Security Notes
+
+- Your email credentials are only used for the duration of each tool call
+- Credentials are not stored or logged by the server
+- All communication happens locally on your machine
+- The server only exposes the MCP interface, not a public web interface
+
+## Troubleshooting
+
+1. **Connection Issues**: Make sure your Gmail app password is correct and 2FA is enabled
+2. **MCP Client Issues**: Try restarting Claude Desktop after configuration changes
+3. **Search Issues**: The tool searches in FROM, SUBJECT, and BODY fields for keywords
+
+## Example Queries
+
+Once configured with Claude Desktop, you can ask:
+
+- "Search my emails for messages from Amazon in the last month"
+- "Show me emails from my bank from last week" 
+- "Analyze my LinkedIn email patterns over the last 60 days"
+- "Find emails from Swiggy today"
+
+Claude will automatically call the appropriate tools with your provided credentials. 

agentic_implementation/email_mcp_server.py

@@ -0,0 +1,245 @@
+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") 

agentic_implementation/email_scraper.py

@@ -287,6 +287,144 @@ def scrape_emails_by_text_search(keyword: str, start_date: str, end_date: str) -
         print(f"Email text search failed: {e}")
         raise
 
+def scrape_emails_by_text_search_with_credentials(email_id: str, app_password: str, keyword: str, start_date: str, end_date: str) -> List[Dict]:
+    """
+    Scrape emails containing a specific keyword (like company name) within date range.
+    Uses provided credentials instead of environment variables.
+    
+    Args:
+        email_id: Gmail address
+        app_password: Gmail app password
+        keyword: Keyword to search for
+        start_date: Start date in DD-MMM-YYYY format
+        end_date: End date in DD-MMM-YYYY format
+    """
+    print(f"Searching emails containing '{keyword}' between {start_date} and {end_date}") 
+    
+    if not email_id or not app_password:
+        raise Exception("Email ID and App Password are required")
+    
+    try:
+        # Connect using provided credentials
+        print("=== IMAP Connection Debug ===")
+        print(f"Email ID: {email_id[:5]}...@{email_id.split('@')[1] if '@' in email_id else 'INVALID'}")
+        print("App password: [PROVIDED]")
+        
+        print("🔄 Attempting IMAP SSL connection to imap.gmail.com:993...")
+        mail = imaplib.IMAP4_SSL("imap.gmail.com")
+        print("✅ SSL connection established")
+        
+        print("🔄 Attempting login...")
+        result = mail.login(email_id, app_password)
+        print(f"✅ Login successful: {result}")
+        
+        print("🔄 Selecting mailbox: [Gmail]/All Mail...")
+        result = mail.select('"[Gmail]/All Mail"')
+        print(f"✅ Mailbox selected: {result}")
+        
+        # Prepare IMAP search criteria with text search
+        start_imap = _date_to_imap_format(start_date)
+        # Add one day to end_date for BEFORE criteria (IMAP BEFORE is exclusive)
+        end_dt = datetime.strptime(end_date, "%d-%b-%Y") + timedelta(days=1)
+        end_imap = end_dt.strftime("%d-%b-%Y")
+        
+        # Search for emails containing the keyword in FROM field or SUBJECT or BODY
+        # We'll search multiple criteria and combine results
+        search_criteria_list = [
+            f'FROM "{keyword}" SINCE "{start_imap}" BEFORE "{end_imap}"',
+            f'SUBJECT "{keyword}" SINCE "{start_imap}" BEFORE "{end_imap}"',
+            f'BODY "{keyword}" SINCE "{start_imap}" BEFORE "{end_imap}"'
+        ]
+        
+        all_email_ids = set()
+        
+        # Search with multiple criteria to catch emails containing the keyword
+        for search_criteria in search_criteria_list:
+            try:
+                print(f"IMAP search: {search_criteria}")
+                status, data = mail.search(None, search_criteria)
+                if status == 'OK' and data[0]:
+                    email_ids = data[0].split()
+                    all_email_ids.update(email_ids)
+                    print(f"Found {len(email_ids)} emails with this criteria")
+            except Exception as e:
+                print(f"Search criteria failed: {search_criteria}, error: {e}")
+                continue
+        
+        print(f"Total unique emails found: {len(all_email_ids)}")
+        scraped_emails = []
+        
+        # Process each email
+        for i, email_id in enumerate(all_email_ids):
+            try:
+                print(f"Processing email {i+1}/{len(all_email_ids)}")
+                
+                # Fetch email
+                status, msg_data = mail.fetch(email_id, "(RFC822)")
+                if status != 'OK':
+                    continue
+                
+                # Parse email
+                msg = message_from_bytes(msg_data[0][1])
+                
+                # Extract information
+                subject = msg.get("Subject", "No Subject")
+                from_header = msg.get("From", "Unknown Sender")
+                content = _email_to_clean_text(msg)
+                
+                # Check if the keyword is actually present (case-insensitive)
+                keyword_lower = keyword.lower()
+                if not any(keyword_lower in text.lower() for text in [subject, from_header, content]):
+                    continue
+                
+                # Parse date
+                date_header = msg.get("Date", "")
+                if date_header:
+                    try:
+                        dt_obj = parsedate_to_datetime(date_header)
+                        # Convert to IST
+                        ist_dt = dt_obj.astimezone(ZoneInfo("Asia/Kolkata"))
+                        email_date = ist_dt.strftime("%d-%b-%Y")
+                        email_time = ist_dt.strftime("%H:%M:%S")
+                    except:
+                        email_date = datetime.today().strftime("%d-%b-%Y")
+                        email_time = "00:00:00"
+                else:
+                    email_date = datetime.today().strftime("%d-%b-%Y")
+                    email_time = "00:00:00"
+                
+                # Double-check date range
+                if not _is_date_in_range(email_date, start_date, end_date):
+                    continue
+                
+                # Get message ID for deduplication
+                message_id = msg.get("Message-ID", f"missing-{email_id.decode()}")
+                
+                scraped_emails.append({
+                    "date": email_date,
+                    "time": email_time,
+                    "subject": subject,
+                    "from": from_header,
+                    "content": content[:2000],  # Limit content length
+                    "message_id": message_id
+                })
+                
+            except Exception as e:
+                print(f"Error processing email {email_id}: {e}")
+                continue
+        
+        mail.logout()
+        
+        # Sort by date (newest first)
+        scraped_emails.sort(key=lambda x: datetime.strptime(f"{x['date']} {x['time']}", "%d-%b-%Y %H:%M:%S"), reverse=True)
+        
+        print(f"Successfully processed {len(scraped_emails)} emails containing '{keyword}'")
+        return scraped_emails
+        
+    except Exception as e:
+        print(f"Email text search failed: {e}")
+        raise
+
 # Test the scraper
 if __name__ == "__main__":
     # Test scraping