mac-messages-mcp

#!/usr/bin/env python3 """ Mac Messages MCP - Entry point fixed for proper MCP protocol implementation """ import sys import logging import asyncio from mcp.server.fastmcp import FastMCP, Context from mac_messages_mcp.messages import ( get_recent_messages, send_message, find_contact_by_name, check_messages_db_access, get_cached_contacts, check_addressbook_access, query_messages_db ) # Configure logging to stderr for debugging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', stream=sys.stderr ) logger = logging.getLogger("mac_messages_mcp") # Initialize the MCP server with proper description mcp = FastMCP( "MessageBridge", description="A bridge for interacting with macOS Messages app" ) @mcp.tool() def tool_get_recent_messages(ctx: Context, hours: int = 24, contact: str = None) -> str: """ Get recent messages from the Messages app. Args: hours: Number of hours to look back (default: 24) contact: Filter by contact name, phone number, or email (optional) Use "contact:N" to select a specific contact from previous matches """ logger.info(f"Getting recent messages: hours={hours}, contact={contact}") try: # Handle contacts that are passed as numbers if contact is not None: contact = str(contact) result = get_recent_messages(hours=hours, contact=contact) return result except Exception as e: logger.error(f"Error in get_recent_messages: {str(e)}") return f"Error getting messages: {str(e)}" @mcp.tool() def tool_send_message(ctx: Context, recipient: str, message: str, group_chat: bool = False) -> str: """ Send a message using the Messages app. Args: recipient: Phone number, email, contact name, or "contact:N" to select from matches For example, "contact:1" selects the first contact from a previous search message: Message text to send group_chat: Whether to send to a group chat (uses chat ID instead of buddy) """ logger.info(f"Sending message to: {recipient}, group_chat: {group_chat}") try: # Ensure recipient is a string (handles numbers properly) recipient = str(recipient) result = send_message(recipient=recipient, message=message, group_chat=group_chat) return result except Exception as e: logger.error(f"Error in send_message: {str(e)}") return f"Error sending message: {str(e)}" @mcp.tool() def tool_find_contact(ctx: Context, name: str) -> str: """ Find a contact by name using fuzzy matching. Args: name: The name to search for """ logger.info(f"Finding contact: {name}") try: matches = find_contact_by_name(name) if not matches: return f"No contacts found matching '{name}'." if len(matches) == 1: contact = matches[0] return f"Found contact: {contact['name']} ({contact['phone']}) with confidence {contact['score']:.2f}" else: # Format multiple matches result = [f"Found {len(matches)} contacts matching '{name}':"] for i, contact in enumerate(matches[:10]): # Limit to top 10 result.append(f"{i+1}. {contact['name']} ({contact['phone']}) - confidence {contact['score']:.2f}") if len(matches) > 10: result.append(f"...and {len(matches) - 10} more.") return "\n".join(result) except Exception as e: logger.error(f"Error in find_contact: {str(e)}") return f"Error finding contact: {str(e)}" @mcp.tool() def tool_check_db_access(ctx: Context) -> str: """ Diagnose database access issues. """ logger.info("Checking database access") try: return check_messages_db_access() except Exception as e: logger.error(f"Error checking database access: {str(e)}") return f"Error checking database access: {str(e)}" @mcp.tool() def tool_check_contacts(ctx: Context) -> str: """ List available contacts in the address book. """ logger.info("Checking available contacts") try: contacts = get_cached_contacts() if not contacts: return "No contacts found in AddressBook." contact_count = len(contacts) sample_entries = list(contacts.items())[:10] # Show first 10 contacts formatted_samples = [f"{number} -> {name}" for number, name in sample_entries] result = [ f"Found {contact_count} contacts in AddressBook.", "Sample entries (first 10):", *formatted_samples ] return "\n".join(result) except Exception as e: logger.error(f"Error checking contacts: {str(e)}") return f"Error checking contacts: {str(e)}" @mcp.tool() def tool_check_addressbook(ctx: Context) -> str: """ Diagnose AddressBook access issues. """ logger.info("Checking AddressBook access") try: return check_addressbook_access() except Exception as e: logger.error(f"Error checking AddressBook: {str(e)}") return f"Error checking AddressBook: {str(e)}" @mcp.tool() def tool_get_chats(ctx: Context) -> str: """ List available group chats from the Messages app. """ logger.info("Getting available chats") try: query = "SELECT chat_identifier, display_name FROM chat WHERE display_name IS NOT NULL" results = query_messages_db(query) if not results: return "No group chats found." if "error" in results[0]: return f"Error accessing chats: {results[0]['error']}" # Filter out chats without display names and format the results chats = [r for r in results if r.get('display_name')] if not chats: return "No named group chats found." formatted_chats = [] for i, chat in enumerate(chats, 1): formatted_chats.append(f"{i}. {chat['display_name']} (ID: {chat['chat_identifier']})") return "Available group chats:\n" + "\n".join(formatted_chats) except Exception as e: logger.error(f"Error getting chats: {str(e)}") return f"Error getting chats: {str(e)}" @mcp.resource("messages://recent/{hours}") def get_recent_messages_resource(hours: int = 24) -> str: """Resource that provides recent messages.""" return get_recent_messages(hours=hours) @mcp.resource("messages://contact/{contact}/{hours}") def get_contact_messages_resource(contact: str, hours: int = 24) -> str: """Resource that provides messages from a specific contact.""" return get_recent_messages(hours=hours, contact=contact) def run_server(): """Run the MCP server with proper error handling""" try: logger.info("Starting Mac Messages MCP server...") mcp.run() except Exception as e: logger.error(f"Failed to start server: {str(e)}") sys.exit(1) if __name__ == "__main__": run_server()