- itsskofficial + itsskofficial (Sarthak)
This is a test email.
", + to_recipients=["recipient@example.com"] +) + +# Search for emails +search_results = await search_emails(query="meeting", top=5) +``` + +## Privacy and Security + +- All credentials are encrypted using AES encryption +- User privacy filters are applied to email content +- Access tokens are stored securely in MongoDB +- The integration respects Microsoft's data handling policies + +## Troubleshooting + +### Common Issues + +1. **OAuth Error**: Ensure your redirect URI matches exactly in Azure app registration +2. **Permission Denied**: Verify all required API permissions are granted +3. **Token Expired**: The integration handles token refresh automatically +4. **Connection Issues**: Check that the MCP server is running on the correct port + +### Debug Mode + +Enable debug logging by setting the environment variable: +```bash +ENVIRONMENT=dev-local +``` + +## API Reference + +The integration uses Microsoft Graph API v1.0. For detailed API documentation, visit: +https://docs.microsoft.com/en-us/graph/api/overview + +## Contributing + +When contributing to this integration: + +1. Follow the existing code patterns +2. Add appropriate error handling +3. Include privacy filter considerations +4. Update this README with any new features +5. Test thoroughly with different email scenarios diff --git a/src/server/mcp_hub/outlook/auth.py b/src/server/mcp_hub/outlook/auth.py new file mode 100644 index 00000000..bca1bfd3 --- /dev/null +++ b/src/server/mcp_hub/outlook/auth.py @@ -0,0 +1,98 @@ +import os +import json +import logging +from typing import Optional, Dict, Any +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes +from cryptography.hazmat.primitives import padding +from cryptography.hazmat.backends import default_backend +from motor.motor_asyncio import AsyncIOMotorClient +from dotenv import load_dotenv + +# Load environment variables +load_dotenv() + +logger = logging.getLogger(__name__) + +# MongoDB connection +MONGO_URI = os.getenv("MONGO_URI", "mongodb://localhost:27017") +MONGO_DB_NAME = os.getenv("MONGO_DB_NAME", "sentient") +mongo_client = AsyncIOMotorClient(MONGO_URI) +db = mongo_client[MONGO_DB_NAME] + +# Encryption key for credentials +ENCRYPTION_KEY = os.getenv("ENCRYPTION_KEY", "your-32-byte-encryption-key-here").encode() + +def aes_decrypt(encrypted_data: str) -> str: + """Decrypt AES encrypted data.""" + try: + # Decode from base64 + encrypted_bytes = bytes.fromhex(encrypted_data) + + # Extract IV and ciphertext + iv = encrypted_bytes[:16] + ciphertext = encrypted_bytes[16:] + + # Create cipher + cipher = Cipher(algorithms.AES(ENCRYPTION_KEY), modes.CBC(iv), backend=default_backend()) + decryptor = cipher.decryptor() + + # Decrypt + padded_data = decryptor.update(ciphertext) + decryptor.finalize() + + # Remove padding + unpadder = padding.PKCS7(128).unpadder() + data = unpadder.update(padded_data) + unpadder.finalize() + + return data.decode('utf-8') + except Exception as e: + logger.error(f"Error decrypting data: {e}") + raise + +def get_user_id_from_context(ctx) -> str: + """Extract user_id from MCP context.""" + try: + # Extract user_id from context metadata + user_id = ctx.metadata.get("user_id") + if not user_id: + raise ValueError("user_id not found in context metadata") + return user_id + except Exception as e: + logger.error(f"Error extracting user_id from context: {e}") + raise + +async def get_outlook_credentials(user_id: str) -> Dict[str, Any]: + """Get Outlook credentials for a user from MongoDB.""" + try: + user_profile = await db.user_profiles.find_one({"user_id": user_id}) + if not user_profile: + raise ValueError(f"User profile not found for user_id: {user_id}") + + integrations = user_profile.get("userData", {}).get("integrations", {}) + outlook_integration = integrations.get("outlook", {}) + + if not outlook_integration.get("connected", False): + raise ValueError("Outlook not connected for this user") + + encrypted_creds = outlook_integration.get("credentials") + if not encrypted_creds: + raise ValueError("No credentials found for Outlook integration") + + # Decrypt credentials + decrypted_creds = aes_decrypt(encrypted_creds) + return json.loads(decrypted_creds) + + except Exception as e: + logger.error(f"Error getting Outlook credentials for user {user_id}: {e}") + raise + +async def get_user_info(user_id: str) -> Dict[str, Any]: + """Get user information including privacy filters.""" + try: + user_profile = await db.user_profiles.find_one({"user_id": user_id}) + if not user_profile: + return {} + + return user_profile.get("userData", {}) + except Exception as e: + logger.error(f"Error getting user info for {user_id}: {e}") + return {} diff --git a/src/server/mcp_hub/outlook/main.py b/src/server/mcp_hub/outlook/main.py new file mode 100644 index 00000000..c046c712 --- /dev/null +++ b/src/server/mcp_hub/outlook/main.py @@ -0,0 +1,269 @@ +import os +import asyncio +import logging +from typing import Dict, Any, List, Optional + +from dotenv import load_dotenv +from fastmcp import FastMCP, Context +from fastmcp.prompts.prompt import Message +from fastmcp.utilities.logging import configure_logging, get_logger +from fastmcp.exceptions import ToolError + +# Local imports +from . import auth +from . import prompts +from . import utils as helpers + +# --- Standardized Logging Setup --- +configure_logging(level="INFO") +logger = get_logger(__name__) + +# Conditionally load .env for local development +ENVIRONMENT = os.getenv('ENVIRONMENT', 'dev-local') +if ENVIRONMENT == 'dev-local': + dotenv_path = os.path.join(os.path.dirname(__file__), '..', '..', '.env') + if os.path.exists(dotenv_path): + load_dotenv(dotenv_path=dotenv_path) + +# --- Server Initialization --- +mcp = FastMCP( + name="OutlookServer", + instructions="Provides a comprehensive suite of tools to read, search, send, and manage emails in Outlook using Microsoft Graph API.", +) + +# --- Prompt Registration --- +@mcp.resource("prompt://outlook-agent-system") +def get_outlook_system_prompt() -> str: + """Provides the system prompt for the Outlook agent.""" + return prompts.outlook_agent_system_prompt + +@mcp.prompt(name="outlook_user_prompt_builder") +def build_outlook_user_prompt(query: str, username: str, previous_tool_response: str = "{}") -> Message: + """Builds a formatted user prompt for the Outlook agent.""" + content = prompts.outlook_agent_user_prompt.format( + query=query, + username=username, + previous_tool_response=previous_tool_response + ) + return Message(role="user", content=content) + +# --- Tool Helper --- +async def _execute_outlook_action(ctx: Context, action_name: str, **kwargs) -> Dict[str, Any]: + """Helper to handle auth and execution for all Outlook tools.""" + try: + user_id = auth.get_user_id_from_context(ctx) + credentials = await auth.get_outlook_credentials(user_id) + + if not credentials or "access_token" not in credentials: + raise ToolError("Outlook not connected or access token missing") + + # Create Outlook API client + outlook_api = helpers.OutlookAPI(credentials["access_token"]) + + # Execute the requested action + if action_name == "get_emails": + return await outlook_api.get_emails(**kwargs) + elif action_name == "get_email": + return await outlook_api.get_email(**kwargs) + elif action_name == "send_email": + return await outlook_api.send_email(**kwargs) + elif action_name == "get_folders": + return await outlook_api.get_folders(**kwargs) + elif action_name == "search_emails": + return await outlook_api.search_emails(**kwargs) + else: + raise ToolError(f"Unknown action: {action_name}") + + except Exception as e: + logger.error(f"Error executing Outlook action {action_name}: {e}") + raise ToolError(f"Outlook action failed: {str(e)}") + +# --- Tool Definitions --- +@mcp.tool() +async def get_emails(ctx: Context, folder: str = "inbox", top: int = 10, skip: int = 0, + search: Optional[str] = None) -> Dict[str, Any]: + """Get emails from a specific folder in Outlook.""" + try: + result = await _execute_outlook_action(ctx, "get_emails", folder=folder, top=top, skip=skip, search=search) + + # Get user info for privacy filters + user_id = auth.get_user_id_from_context(ctx) + user_info = await auth.get_user_info(user_id) + privacy_filters = user_info.get("privacy_filters", {}) + + # Apply privacy filters and format emails + emails = result.get("value", []) + filtered_emails = helpers.apply_privacy_filters(emails, privacy_filters) + formatted_emails = [helpers.format_email_summary(email) for email in filtered_emails] + + return { + "emails": formatted_emails, + "total_count": len(formatted_emails), + "folder": folder + } + + except Exception as e: + logger.error(f"Error getting emails: {e}") + raise ToolError(f"Failed to get emails: {str(e)}") + +@mcp.tool() +async def get_email(ctx: Context, message_id: str) -> Dict[str, Any]: + """Get a specific email by ID.""" + try: + result = await _execute_outlook_action(ctx, "get_email", message_id=message_id) + + # Format the email for better readability + email = result + from_info = email.get("from", {}) + to_info = email.get("toRecipients", []) + cc_info = email.get("ccRecipients", []) + bcc_info = email.get("bccRecipients", []) + + formatted_email = { + "id": email.get("id"), + "subject": email.get("subject", "No Subject"), + "from": from_info.get("emailAddress", {}).get("address", "Unknown"), + "from_name": from_info.get("emailAddress", {}).get("name", "Unknown"), + "to": [recipient.get("emailAddress", {}).get("address", "") for recipient in to_info], + "cc": [recipient.get("emailAddress", {}).get("address", "") for recipient in cc_info], + "bcc": [recipient.get("emailAddress", {}).get("address", "") for recipient in bcc_info], + "received_date": email.get("receivedDateTime"), + "sent_date": email.get("sentDateTime"), + "is_read": email.get("isRead", False), + "has_attachments": email.get("hasAttachments", False), + "body": email.get("body", {}).get("content", ""), + "body_preview": email.get("bodyPreview", "") + } + + return formatted_email + + except Exception as e: + logger.error(f"Error getting email {message_id}: {e}") + raise ToolError(f"Failed to get email: {str(e)}") + +@mcp.tool() +async def send_email(ctx: Context, subject: str, body: str, to_recipients: List[str], + cc_recipients: Optional[List[str]] = None, + bcc_recipients: Optional[List[str]] = None, + reply_to_message_id: Optional[str] = None) -> Dict[str, Any]: + """Send an email through Outlook.""" + try: + result = await _execute_outlook_action( + ctx, "send_email", + subject=subject, + body=body, + to_recipients=to_recipients, + cc_recipients=cc_recipients, + bcc_recipients=bcc_recipients, + reply_to_message_id=reply_to_message_id + ) + + return { + "success": True, + "message": "Email sent successfully", + "to": to_recipients, + "subject": subject + } + + except Exception as e: + logger.error(f"Error sending email: {e}") + raise ToolError(f"Failed to send email: {str(e)}") + +@mcp.tool() +async def get_folders(ctx: Context) -> Dict[str, Any]: + """Get email folders in Outlook.""" + try: + result = await _execute_outlook_action(ctx, "get_folders") + + folders = result.get("value", []) + formatted_folders = [] + + for folder in folders: + formatted_folders.append({ + "id": folder.get("id"), + "name": folder.get("displayName"), + "message_count": folder.get("messageCount", 0), + "unread_count": folder.get("unreadItemCount", 0) + }) + + return { + "folders": formatted_folders, + "total_folders": len(formatted_folders) + } + + except Exception as e: + logger.error(f"Error getting folders: {e}") + raise ToolError(f"Failed to get folders: {str(e)}") + +@mcp.tool() +async def search_emails(ctx: Context, query: str, top: int = 10) -> Dict[str, Any]: + """Search emails in Outlook.""" + try: + result = await _execute_outlook_action(ctx, "search_emails", query=query, top=top) + + # Get user info for privacy filters + user_id = auth.get_user_id_from_context(ctx) + user_info = await auth.get_user_info(user_id) + privacy_filters = user_info.get("privacy_filters", {}) + + # Apply privacy filters and format emails + emails = result.get("value", []) + filtered_emails = helpers.apply_privacy_filters(emails, privacy_filters) + formatted_emails = [helpers.format_email_summary(email) for email in filtered_emails] + + return { + "emails": formatted_emails, + "total_count": len(formatted_emails), + "search_query": query + } + + except Exception as e: + logger.error(f"Error searching emails: {e}") + raise ToolError(f"Failed to search emails: {str(e)}") + +@mcp.tool() +async def reply_to_email(ctx: Context, message_id: str, body: str, + cc_recipients: Optional[List[str]] = None, + bcc_recipients: Optional[List[str]] = None) -> Dict[str, Any]: + """Reply to an existing email.""" + try: + # First get the original email to extract recipients + original_email = await _execute_outlook_action(ctx, "get_email", message_id=message_id) + + # Extract the original sender as recipient for reply + from_info = original_email.get("from", {}) + reply_to_email = from_info.get("emailAddress", {}).get("address") + + if not reply_to_email: + raise ToolError("Could not determine recipient for reply") + + # Create reply subject + original_subject = original_email.get("subject", "") + reply_subject = f"Re: {original_subject}" if not original_subject.startswith("Re:") else original_subject + + # Send the reply + result = await _execute_outlook_action( + ctx, "send_email", + subject=reply_subject, + body=body, + to_recipients=[reply_to_email], + cc_recipients=cc_recipients, + bcc_recipients=bcc_recipients, + reply_to_message_id=message_id + ) + + return { + "success": True, + "message": "Reply sent successfully", + "to": [reply_to_email], + "subject": reply_subject, + "original_message_id": message_id + } + + except Exception as e: + logger.error(f"Error replying to email: {e}") + raise ToolError(f"Failed to reply to email: {str(e)}") + +if __name__ == "__main__": + import uvicorn + uvicorn.run(mcp.app, host="0.0.0.0", port=9027) diff --git a/src/server/mcp_hub/outlook/prompts.py b/src/server/mcp_hub/outlook/prompts.py new file mode 100644 index 00000000..d893111b --- /dev/null +++ b/src/server/mcp_hub/outlook/prompts.py @@ -0,0 +1,16 @@ +outlook_agent_system_prompt = """You are an Outlook email assistant that helps users manage their emails through Microsoft Graph API. You can: + +1. List emails from different folders (Inbox, Sent Items, etc.) +2. Read email content and details +3. Send new emails +4. Reply to existing emails +5. Search for specific emails +6. Manage email folders + +Always be helpful, concise, and respect user privacy. When reading emails, focus on the most relevant information and summarize when appropriate.""" + +outlook_agent_user_prompt = """User Query: {query} +Username: {username} +Previous Tool Response: {previous_tool_response} + +Please help the user with their Outlook email management request. Use the available tools to perform the requested action and provide a clear, helpful response.""" diff --git a/src/server/mcp_hub/outlook/requirements.txt b/src/server/mcp_hub/outlook/requirements.txt new file mode 100644 index 00000000..f591ea1d --- /dev/null +++ b/src/server/mcp_hub/outlook/requirements.txt @@ -0,0 +1,6 @@ +fastmcp +httpx +python-dotenv +cryptography +motor +requests diff --git a/src/server/mcp_hub/outlook/test_client.py b/src/server/mcp_hub/outlook/test_client.py new file mode 100644 index 00000000..6f1a3900 --- /dev/null +++ b/src/server/mcp_hub/outlook/test_client.py @@ -0,0 +1,82 @@ +#!/usr/bin/env python3 +""" +Test client for Outlook MCP Server +""" + +import asyncio +import json +import logging +from typing import Dict, Any + +from fastmcp import FastMCPClient +from fastmcp.utilities.logging import configure_logging, get_logger + +# Configure logging +configure_logging(level="INFO") +logger = get_logger(__name__) + +async def test_outlook_server(): + """Test the Outlook MCP server functionality.""" + + # Create client + client = FastMCPClient( + name="OutlookTestClient", + server_url="http://localhost:9027/sse" + ) + + try: + # Test context with mock user_id + test_context = { + "metadata": { + "user_id": "test_user_123" + } + } + + logger.info("Testing Outlook MCP Server...") + + # Test 1: Get folders + logger.info("Test 1: Getting email folders...") + try: + folders_result = await client.call_tool( + "get_folders", + context=test_context + ) + logger.info(f"Folders result: {json.dumps(folders_result, indent=2)}") + except Exception as e: + logger.error(f"Error getting folders: {e}") + + # Test 2: Get emails from inbox + logger.info("Test 2: Getting emails from inbox...") + try: + emails_result = await client.call_tool( + "get_emails", + context=test_context, + folder="inbox", + top=5 + ) + logger.info(f"Emails result: {json.dumps(emails_result, indent=2)}") + except Exception as e: + logger.error(f"Error getting emails: {e}") + + # Test 3: Search emails + logger.info("Test 3: Searching emails...") + try: + search_result = await client.call_tool( + "search_emails", + context=test_context, + query="test", + top=3 + ) + logger.info(f"Search result: {json.dumps(search_result, indent=2)}") + except Exception as e: + logger.error(f"Error searching emails: {e}") + + logger.info("Outlook MCP Server tests completed!") + + except Exception as e: + logger.error(f"Error testing Outlook MCP server: {e}") + finally: + await client.close() + +if __name__ == "__main__": + asyncio.run(test_outlook_server()) diff --git a/src/server/mcp_hub/outlook/utils.py b/src/server/mcp_hub/outlook/utils.py new file mode 100644 index 00000000..957d499a --- /dev/null +++ b/src/server/mcp_hub/outlook/utils.py @@ -0,0 +1,188 @@ +import httpx +import logging +from typing import Dict, Any, List, Optional +from datetime import datetime, timezone + +logger = logging.getLogger(__name__) + +class OutlookAPI: + """Helper class for Microsoft Graph API calls.""" + + def __init__(self, access_token: str): + self.access_token = access_token + self.base_url = "https://graph.microsoft.com/v1.0" + self.headers = { + "Authorization": f"Bearer {access_token}", + "Content-Type": "application/json" + } + + async def get_emails(self, folder: str = "inbox", top: int = 10, skip: int = 0, + search: Optional[str] = None) -> Dict[str, Any]: + """Get emails from a specific folder.""" + try: + url = f"{self.base_url}/me/mailFolders/{folder}/messages" + params = { + "$top": top, + "$skip": skip, + "$orderby": "receivedDateTime desc", + "$select": "id,subject,from,toRecipients,receivedDateTime,isRead,hasAttachments,bodyPreview" + } + + if search: + params["$filter"] = f"contains(subject,'{search}') or contains(body/content,'{search}')" + + async with httpx.AsyncClient() as client: + response = await client.get(url, headers=self.headers, params=params) + response.raise_for_status() + return response.json() + + except Exception as e: + logger.error(f"Error getting emails: {e}") + raise + + async def get_email(self, message_id: str) -> Dict[str, Any]: + """Get a specific email by ID.""" + try: + url = f"{self.base_url}/me/messages/{message_id}" + params = { + "$select": "id,subject,from,toRecipients,ccRecipients,bccRecipients,receivedDateTime,sentDateTime,isRead,hasAttachments,body,bodyPreview" + } + + async with httpx.AsyncClient() as client: + response = await client.get(url, headers=self.headers, params=params) + response.raise_for_status() + return response.json() + + except Exception as e: + logger.error(f"Error getting email {message_id}: {e}") + raise + + async def send_email(self, subject: str, body: str, to_recipients: List[str], + cc_recipients: Optional[List[str]] = None, + bcc_recipients: Optional[List[str]] = None, + reply_to_message_id: Optional[str] = None) -> Dict[str, Any]: + """Send an email.""" + try: + url = f"{self.base_url}/me/sendMail" + + # Prepare recipients + to_emails = [{"emailAddress": {"address": email}} for email in to_recipients] + cc_emails = [{"emailAddress": {"address": email}} for email in (cc_recipients or [])] + bcc_emails = [{"emailAddress": {"address": email}} for email in (bcc_recipients or [])] + + # Prepare message + message = { + "subject": subject, + "body": { + "contentType": "HTML", + "content": body + }, + "toRecipients": to_emails + } + + if cc_emails: + message["ccRecipients"] = cc_emails + if bcc_emails: + message["bccRecipients"] = bcc_emails + if reply_to_message_id: + message["replyTo"] = [{"id": reply_to_message_id}] + + payload = {"message": message, "saveToSentItems": True} + + async with httpx.AsyncClient() as client: + response = await client.post(url, headers=self.headers, json=payload) + response.raise_for_status() + return {"success": True, "message": "Email sent successfully"} + + except Exception as e: + logger.error(f"Error sending email: {e}") + raise + + async def get_folders(self) -> Dict[str, Any]: + """Get email folders.""" + try: + url = f"{self.base_url}/me/mailFolders" + params = { + "$select": "id,displayName,messageCount,unreadItemCount" + } + + async with httpx.AsyncClient() as client: + response = await client.get(url, headers=self.headers, params=params) + response.raise_for_status() + return response.json() + + except Exception as e: + logger.error(f"Error getting folders: {e}") + raise + + async def search_emails(self, query: str, top: int = 10) -> Dict[str, Any]: + """Search emails using Microsoft Graph search.""" + try: + url = f"{self.base_url}/me/messages" + params = { + "$search": f'"{query}"', + "$top": top, + "$orderby": "receivedDateTime desc", + "$select": "id,subject,from,toRecipients,receivedDateTime,isRead,hasAttachments,bodyPreview" + } + + async with httpx.AsyncClient() as client: + response = await client.get(url, headers=self.headers, params=params) + response.raise_for_status() + return response.json() + + except Exception as e: + logger.error(f"Error searching emails: {e}") + raise + +def format_email_summary(email: Dict[str, Any]) -> Dict[str, Any]: + """Format email data for better readability.""" + try: + from_info = email.get("from", {}) + to_info = email.get("toRecipients", []) + + return { + "id": email.get("id"), + "subject": email.get("subject", "No Subject"), + "from": from_info.get("emailAddress", {}).get("address", "Unknown"), + "from_name": from_info.get("emailAddress", {}).get("name", "Unknown"), + "to": [recipient.get("emailAddress", {}).get("address", "") for recipient in to_info], + "received_date": email.get("receivedDateTime"), + "is_read": email.get("isRead", False), + "has_attachments": email.get("hasAttachments", False), + "body_preview": email.get("bodyPreview", "") + } + except Exception as e: + logger.error(f"Error formatting email: {e}") + return email + +def apply_privacy_filters(emails: List[Dict[str, Any]], privacy_filters: Dict[str, Any]) -> List[Dict[str, Any]]: + """Apply privacy filters to email list.""" + try: + if not privacy_filters: + return emails + + keyword_filters = privacy_filters.get("keywords", []) + email_filters = [email.lower() for email in privacy_filters.get("emails", [])] + + filtered_emails = [] + for email in emails: + # Skip if email contains filtered keywords + subject = email.get("subject", "").lower() + body = email.get("bodyPreview", "").lower() + + if any(keyword.lower() in subject or keyword.lower() in body for keyword in keyword_filters): + continue + + # Skip if from filtered email addresses + from_email = email.get("from", "").lower() + if from_email in email_filters: + continue + + filtered_emails.append(email) + + return filtered_emails + + except Exception as e: + logger.error(f"Error applying privacy filters: {e}") + return emails diff --git a/src/server/workers/executor/config.py b/src/server/workers/executor/config.py index 874193c1..c2fb2774 100644 --- a/src/server/workers/executor/config.py +++ b/src/server/workers/executor/config.py @@ -212,5 +212,15 @@ "name": "tasks_server", "url": os.getenv("TASKS_MCP_SERVER_URL", "http://localhost:9018/sse/") } + }, + "outlook": { + "display_name": "Outlook", + "description": "Connect to read, send, and manage emails in Outlook. The agent can list emails, read message content, send new emails, reply to messages, and manage folders.", + "auth_type": "oauth", + "icon": "IconMail", + "mcp_server_config": { + "name": "outlook_server", + "url": os.getenv("OUTLOOK_MCP_SERVER_URL", "http://localhost:9027/sse") + } } } \ No newline at end of file diff --git a/src/server/workers/planner/config.py b/src/server/workers/planner/config.py index 26835b39..a3a79053 100644 --- a/src/server/workers/planner/config.py +++ b/src/server/workers/planner/config.py @@ -220,6 +220,16 @@ "name": "whatsapp_server", "url": os.getenv("WHATSAPP_MCP_SERVER_URL", "http://localhost:9024/sse") } + }, + "outlook": { + "display_name": "Outlook", + "description": "Connect to read, send, and manage emails in Outlook. The agent can list emails, read message content, send new emails, reply to messages, and manage folders.", + "auth_type": "oauth", + "icon": "IconMail", + "mcp_server_config": { + "name": "outlook_server", + "url": os.getenv("OUTLOOK_MCP_SERVER_URL", "http://localhost:9027/sse") + } } }